recheck: stream live output (verbose real command output or narration)

README.md

@@ -1,10 +1,16 @@
 # update-detector
 
-A small agent that detects (never applies) available OS updates on a host:
-package updates, security updates, pending-reboot state, and OS release
-upgrades. It exposes the result over HTTP for [Gatus](https://gatus.io) to
-poll, and can notify a channel (Telegram today) when something meaningful
-changes. Ships as a single Docker image, one container per host.
+A small agent that detects available OS updates on a host: package
+updates, security updates, pending-reboot state, and OS release upgrades.
+It exposes the result over HTTP for [Gatus](https://gatus.io) to poll, and
+can notify a channel (Telegram today) when something meaningful changes.
+Ships as its own Docker image (`update-detector`), one container per host.
+The agent itself never writes to the host — an optional aggregator (a
+separate Docker image, one instance for your whole fleet) and companion
+(always native, never containerized) add a central dashboard and
+push-button apply on top — see
+[Fleet dashboard and push-button updates](#fleet-dashboard-and-push-button-updates)
+below.
 
 ## Supported platforms
 
@@ -14,7 +20,7 @@ changes. Ships as a single Docker image, one container per host.
 | Plain Debian / Raspberry Pi OS (bare metal or VM) | ✅ supported now — see [OS flavors](docs/reference.md#os-flavors) |
 | Raspberry Pi 4B (arm64, either flavor above) | ✅ supported now — see [Releases](docs/reference.md#releases) |
 | WSL2 Ubuntu/Debian distro on Windows | ✅ supported now — see [WSL2](docs/wsl2.md) (Docker Desktop's WSL2 integration is usually a CLI shim, not a real engine — `install.sh` offers a native, no-Docker install for this reason) |
-| Actual Windows OS (Windows Update, winget) | 🧪 experimental — detection, `install.bat`, and companion apply/self-update all exist, see [Limitations](docs/reference.md#platform-limitations); none of it verified against a real Windows host yet |
+| Actual Windows OS (Windows Update) | 🧪 experimental — detection, `install.bat`, and companion apply/self-update confirmed against a real Windows host, see [Limitations](docs/reference.md#platform-limitations); **winget is not supported** |
 | Actual macOS host (`softwareupdate`, `brew`) | 🚧 planned — same reason |
 
 ## Installation

cmd/update-detector-companion/main.go

@@ -15,7 +15,6 @@ import (
 	"log"
 	"os"
 	"os/signal"
-	"runtime"
 	"syscall"
 	"time"
 
@@ -103,43 +102,44 @@ func run(ctx context.Context) error {
 		}()
 
 		// Companion self-update has fundamentally different behavior
-		// on Linux vs Windows:
+		// on Linux vs Windows, but both now run Apply the same way as
+		// every other action -- synchronously, in the foreground, with
+		// its real output tee'd to actionCtx's sink as it happens --
+		// rather than reporting a guessed outcome *before* running it,
+		// which used to end this action's output stream (EventDone,
+		// via report below) before Apply/install.sh had produced
+		// anything at all. Confirmed live: that was why "Update
+		// companion" never showed any real streamed output.
 		//
-		// Linux: install.sh restarts this process (systemctl restart).
-		// The companion process dies, but install.sh survives (Linux
-		// inode semantics let the running process keep its open fd even
-		// after the binary is renamed). So we must report optimistically
-		// *before* calling Apply, since code after it may never run.
+		// Windows: the companion stages the new binary to .exe.new and
+		// returns a real Staged result (no restart of this process at
+		// all) -- Apply always returns normally, nothing more to
+		// special-case here.
 		//
-		// Windows: the companion stages the new binary to .exe.new
-		// and returns a Staged result (no restart). The agent (a
-		// separate Windows Service on the same host) will later be
-		// told to stop the companion, swap the binary, and restart it.
-		// The result is reported normally.
-		//
-		// If Apply returns having failed on Linux, that's only a
-		// *real* failure to correct the record with if ctx is still
-		// alive -- once systemd's restart reaches this process
-		// (SIGTERM, via the same ctx), the in-flight install.sh child
-		// gets killed too, and Apply surfaces that as an ordinary-
-		// looking failure ("signal: terminated") even though the
-		// swap+restart actually succeeded. Confirmed live: without
-		// this check, that spurious failure overwrote the correct
-		// optimistic success report every time.
+		// Linux: install.sh restarts this process (systemctl restart)
+		// as its own last step. By the time that reaches this process
+		// (SIGTERM, canceling ctx), install.sh's own child process gets
+		// killed too (exec.CommandContext's own doing), which makes
+		// Apply return a spurious-looking failure ("signal: terminated")
+		// even though the swap+restart had, by that point, already
+		// actually succeeded. Only in that specific situation --
+		// !result.Success with ctx already canceled -- is the failure
+		// replaced with the optimistic message instead of reported as a
+		// real failure. Confirmed live: without this check, that
+		// spurious failure overwrote what was actually a successful
+		// update every time.
 		if action.Type == aggregator.ActionSelfUpdate && action.Component == "companion" {
-			if runtime.GOOS == "windows" {
-				result := companion.Apply(actionCtx, cfg.AgentStatusURL, cfg.AggregatorURL, identity, action)
-				report(result)
-			} else {
-				report(aggregator.ActionResult{
+			result := companion.Apply(actionCtx, cfg.AgentStatusURL, cfg.AggregatorURL, identity, action)
+			switch {
+			case !result.Success && ctx.Err() != nil:
+				result = aggregator.ActionResult{
 					ActionID: action.ID, Success: true,
 					Message: "update installing, restarting shortly", CompletedAt: time.Now(),
-				})
-				if result := companion.Apply(actionCtx, cfg.AgentStatusURL, cfg.AggregatorURL, identity, action); !result.Success && ctx.Err() == nil {
-					log.Printf("companion: self-update of companion failed before restarting: %s", result.Message)
-					report(result)
 				}
+			case !result.Success:
+				log.Printf("companion: self-update of companion failed: %s", result.Message)
 			}
+			report(result)
 			return
 		}
 

cmd/update-detector/main.go

@@ -6,10 +6,12 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"net/http"
 	"os"
 	"os/signal"
+	"sync"
 	"syscall"
 	"time"
 
@@ -131,58 +133,34 @@ func run(ctx context.Context) error {
 		}
 	}()
 
-	if aggClient != nil {
-		// Holds the aggregator's stream connection whenever no companion
-		// is running (or hasn't connected yet) -- the aggregator's
-		// CompanionHub always lets a companion preempt this, since only
-		// it can carry out apply-type actions; this only ever receives
-		// (and can only ever receive, per that same server-side gate)
-		// ActionRecheck. Handled in-process, unlike the companion's own
-		// loopback HTTP call, since the agent already is that process.
-		onAction := func(action aggregator.Action) {
-			switch action.Type {
-			case aggregator.ActionRecheck:
-				srv.TriggerRecheck()
-				resultCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
-				if err := aggClient.ReportActionResult(resultCtx, action.ID, true, "recheck triggered"); err != nil {
-					log.Printf("aggregator: reporting recheck result for %s: %v", action.ID, err)
-				}
-				cancel()
-			case aggregator.ActionCompleteCompanionSwap:
-				result := companion.CompleteCompanionSwap(ctx, action)
-				resultCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
-				if err := aggClient.ReportActionResult(resultCtx, action.ID, result.Success, result.Message); err != nil {
-					log.Printf("aggregator: reporting companion swap result for %s: %v", action.ID, err)
-				}
-				cancel()
-			default:
-				log.Printf("aggregator: ignoring unexpected action type %q on agent stream", action.Type)
-			}
-		}
-		// aggregatorPresent is meaningless for a plain agent connection
-		// (only a companion ever runs the aggregator-colocation check --
-		// see CompanionHub.SetAggregatorPresent), so always false here.
-		go agentstream.Run(ctx, cfg.AggregatorURL, identity, aggregator.KindAgent, false, false, onAction)
-	}
+	// checkMu serializes every actual detection cycle -- the ticker-driven
+	// background loop below and a synchronous, admin-triggered recheck
+	// (see onAction's ActionRecheck case) must never run concurrently:
+	// they'd otherwise race on `previous` and risk two overlapping
+	// apt-get invocations against the same host state.
+	var checkMu sync.Mutex
 
-	if aggClient != nil {
-		enrollCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
-		aggStatus, err := aggClient.Enroll(enrollCtx, cfg.Hostname)
-		cancel()
-		if err != nil {
-			log.Printf("aggregator: enroll failed (will retry on next report): %v", err)
-		} else {
-			log.Printf("aggregator: enrollment status: %s", aggStatus)
-		}
-	}
+	// runCheck runs one detection cycle. lineSink, if non-nil, is attached
+	// to the check's own context for the duration of this call only (see
+	// checker.WithLineSink) -- a verbose recheck's real-command-output
+	// tap; the periodic ticker-driven cycle always passes nil, so its
+	// behavior is completely unchanged by this parameter's existence.
+	// Returns the resulting Status and chk.Check's own error, so a caller
+	// invoking this synchronously (onAction's ActionRecheck case) can
+	// build a real ActionResult instead of always claiming success.
+	runCheck := func(first bool, lineSink func(string)) (checker.Status, error) {
+		checkMu.Lock()
+		defer checkMu.Unlock()
 
-	runCheck := func(first bool) {
 		checkCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
+		if lineSink != nil {
+			checkCtx = checker.WithLineSink(checkCtx, lineSink)
+		}
 		status, err := chk.Check(checkCtx, previous)
 		cancel()
 		if err != nil {
 			log.Printf("check failed: %v", err)
-			return
+			return status, err
 		}
 		if len(status.Errors) > 0 {
 			log.Printf("check completed with errors: %v", status.Errors)
@@ -217,13 +195,117 @@ func run(ctx context.Context) error {
 
 		srv.SetStatus(status)
 		previous = &status
+		return status, nil
 	}
 
-	runCheck(true)
-
 	ticker := time.NewTicker(cfg.CheckInterval)
 	defer ticker.Stop()
 
+	if aggClient != nil {
+		// Holds the aggregator's stream connection whenever no companion
+		// is running (or hasn't connected yet) -- the aggregator's
+		// CompanionHub always lets a companion preempt this, since only
+		// it can carry out apply-type actions; this only ever receives
+		// (and can only ever receive, per that same server-side gate)
+		// ActionRecheck. Handled in-process, unlike the companion's own
+		// loopback HTTP call, since the agent already is that process.
+		onAction := func(action aggregator.Action) {
+			switch action.Type {
+			case aggregator.ActionRecheck:
+				// Streams this recheck's output back to the aggregator
+				// exactly like the companion binary streams an apply's --
+				// same sink/StreamOutput/report-before-close pattern (see
+				// cmd/update-detector-companion/main.go), so "Force
+				// recheck" gets a live console whether or not a companion
+				// is even installed on this host.
+				sink := companion.NewOutputSink(1000)
+				streamCtx, cancelStream := context.WithCancel(ctx)
+				go func() {
+					if err := companion.StreamOutput(streamCtx, cfg.AggregatorURL, identity, action.ID, sink); err != nil {
+						log.Printf("aggregator: streaming recheck output for %s: %v", action.ID, err)
+					}
+				}()
+
+				var lineSink func(string)
+				if action.Verbose {
+					lineSink = sink.Push
+				} else {
+					sink.Push("Running detection cycle...")
+				}
+
+				status, checkErr := runCheck(false, lineSink)
+				ticker.Reset(cfg.CheckInterval) // same as the srv.Recheck() case below
+
+				success := checkErr == nil
+				message := "recheck complete"
+				switch {
+				case checkErr != nil:
+					message = fmt.Sprintf("recheck failed: %v", checkErr)
+				case !action.Verbose:
+					message = fmt.Sprintf("Recheck complete: %d upgradable (%d security)",
+						status.Packages.UpgradableTotal, status.Packages.UpgradableSecurity)
+					sink.Push(message)
+				}
+
+				// Reported *before* closing the sink/stream, deliberately
+				// -- OutputHub.End's "first call wins" race means this
+				// must land as EventDone before the /companion/output
+				// body closing would otherwise mark it EventDisconnected.
+				resultCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+				if err := aggClient.ReportActionResult(resultCtx, action.ID, success, message); err != nil {
+					log.Printf("aggregator: reporting recheck result for %s: %v", action.ID, err)
+				}
+				cancel()
+
+				sink.Close()
+				cancelStream()
+			case aggregator.ActionCompleteCompanionSwap:
+				// CompleteCompanionSwap already calls emitFromContext(ctx)
+				// and runCapped throughout (stop/swap/start the service) --
+				// it was always ready to stream, it just never had a sink
+				// attached to actually stream to. Same sink/StreamOutput/
+				// report-before-close pattern as ActionRecheck above.
+				sink := companion.NewOutputSink(1000)
+				streamCtx, cancelStream := context.WithCancel(ctx)
+				go func() {
+					if err := companion.StreamOutput(streamCtx, cfg.AggregatorURL, identity, action.ID, sink); err != nil {
+						log.Printf("aggregator: streaming companion-swap output for %s: %v", action.ID, err)
+					}
+				}()
+
+				result := companion.CompleteCompanionSwap(companion.WithOutputSink(ctx, sink), action)
+
+				resultCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+				if err := aggClient.ReportActionResult(resultCtx, action.ID, result.Success, result.Message); err != nil {
+					log.Printf("aggregator: reporting companion swap result for %s: %v", action.ID, err)
+				}
+				cancel()
+
+				sink.Close()
+				cancelStream()
+			default:
+				log.Printf("aggregator: ignoring unexpected action type %q on agent stream", action.Type)
+			}
+		}
+		// aggregatorPresent is meaningless for a plain agent connection
+		// (only a companion ever runs the aggregator-colocation check --
+		// see CompanionHub.SetAggregatorPresent), so always false here.
+		go agentstream.Run(ctx, cfg.AggregatorURL, identity, aggregator.KindAgent, false, false, onAction)
+	}
+
+	if aggClient != nil {
+		enrollCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
+		aggStatus, err := aggClient.Enroll(enrollCtx, cfg.Hostname)
+		cancel()
+		if err != nil {
+			log.Printf("aggregator: enroll failed (will retry on next report): %v", err)
+		} else {
+			log.Printf("aggregator: enrollment status: %s", aggStatus)
+		}
+	}
+
+	runCheck(true, nil)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -235,10 +317,10 @@ func run(ctx context.Context) error {
 			defer cancel()
 			return httpSrv.Shutdown(shutdownCtx)
 		case <-ticker.C:
-			runCheck(false)
+			runCheck(false, nil)
 		case <-srv.Recheck():
 			log.Println("out-of-band recheck requested")
-			runCheck(false)
+			runCheck(false, nil)
 			ticker.Reset(cfg.CheckInterval)
 		}
 	}