From a1d5fdecf3fabec9626956e8293425ff7f103e2c Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 12:14:02 +0800 Subject: [PATCH 1/7] recheck: stream live output (verbose real command output or narration) Force recheck previously did nothing visible: the agent's ActionRecheck handling was fire-and-forget (queue a signal, report success unconditionally) and never opened the same output stream Apply/Upgrade already use. It now runs synchronously and streams to the same live console, whether or not a companion is installed on that host -- the agent always holds its own dedicated stream for non-companion actions like recheck, so one fix covers both cases. A per-row "verbose" checkbox controls what streams: checked taps the real apt-get/apt-check/winget/powershell output (via a new optional checker.WithLineSink context hook, wired through aptutil and each platform backend with zero behavior change when absent); unchecked (default) emits short synthetic progress narration instead. OutputHub now buffers each in-flight action's lines so a browser reconnecting mid-recheck (e.g. a page refresh) replays what it missed before continuing live -- scoped to in-flight only, nothing persisted once an action ends. Also: promoted internal/companion's private lineTee into a shared internal/linetee package (checker's backends need the same line-splitting tee, and companion already imports checker, so checker can't import companion back); exported OutputSink.Push so the agent binary can push into a sink it owns directly. Deleted internal/aggregator/adminbus.go, an unused, superseded-by-OutputHub pub/sub left over from an earlier attempt at this same problem. --- cmd/update-detector/main.go | 163 ++++++++++++------ internal/aggregator/companion.go | 7 + internal/aggregator/output.go | 97 +++++++---- internal/aggregator/output_test.go | 75 +++++++- internal/aggregator/server.go | 61 +++++-- internal/aggregator/server_test.go | 4 +- internal/aggregator/templates.go | 11 +- internal/aptutil/exec.go | 28 ++- internal/aptutil/exec_test.go | 64 +++++++ internal/checker/debian/exec_test.go | 59 +++++++ internal/checker/debian/packages.go | 15 +- internal/checker/linesink.go | 25 +++ internal/checker/linesink_test.go | 27 +++ internal/checker/ubuntu/exec_test.go | 62 +++++++ internal/checker/ubuntu/packages.go | 30 +++- internal/checker/windows/packages.go | 16 +- internal/checker/windows/windowsupdate.go | 16 +- internal/companion/execute.go | 9 +- internal/companion/linetee.go | 58 ------- internal/companion/outputsink.go | 17 +- internal/companion/outputstream_test.go | 6 +- internal/linetee/linetee.go | 63 +++++++ .../{companion => linetee}/linetee_test.go | 26 +-- 23 files changed, 741 insertions(+), 198 deletions(-) create mode 100644 internal/aptutil/exec_test.go create mode 100644 internal/checker/debian/exec_test.go create mode 100644 internal/checker/linesink.go create mode 100644 internal/checker/linesink_test.go create mode 100644 internal/checker/ubuntu/exec_test.go delete mode 100644 internal/companion/linetee.go create mode 100644 internal/linetee/linetee.go rename internal/{companion => linetee}/linetee_test.go (58%) diff --git a/cmd/update-detector/main.go b/cmd/update-detector/main.go index aa392b5..2913e1d 100644 --- a/cmd/update-detector/main.go +++ b/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,100 @@ 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: + 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) + } + + 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 +300,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) } } diff --git a/internal/aggregator/companion.go b/internal/aggregator/companion.go index 3a08b4e..272d43a 100644 --- a/internal/aggregator/companion.go +++ b/internal/aggregator/companion.go @@ -100,6 +100,13 @@ type Action struct { // release tag to update it to. Component string `json:"component,omitempty"` TargetVersion string `json:"target_version,omitempty"` + + // Verbose is only meaningful for ActionRecheck: true streams the real + // shell command output the check runs (apt-get/apt-check/winget/ + // powershell), matching apply/upgrade's own fidelity; false (default) + // streams synthetic progress narration lines instead. Ignored for + // every other action type, which always stream real output already. + Verbose bool `json:"verbose,omitempty"` } // ActionResult is what a companion reports back after attempting an Action. diff --git a/internal/aggregator/output.go b/internal/aggregator/output.go index 910c43c..ff81cf4 100644 --- a/internal/aggregator/output.go +++ b/internal/aggregator/output.go @@ -28,56 +28,66 @@ type outputEvent struct { Line string // only set when Kind == EventLine } -// OutputHub fans a companion's live action output out to however many -// browsers are watching that agent's admin-page row, for however long its -// action is in flight. Deliberately parallel to CompanionHub: in-memory -// only, nothing persisted -- a subscriber just reconnects and picks up -// whatever's live from that point on (see the "no history replay" scoping -// decision), and losing everything on an aggregator restart is an -// acceptable trade-off for the same reason CompanionHub already accepts -// it for its own result log. +// OutputHub fans a companion's (or agent's) live action output out to +// however many browsers are watching that agent's admin-page row, for +// however long its action is in flight. Buffers the in-flight action's +// lines in memory (backlog) so a reconnecting subscriber -- e.g. a page +// refresh mid-action -- can replay everything published so far before +// continuing live; see Subscribe. Deliberately scoped to in-flight +// replay only: End clears the backlog the moment an action ends, so +// nothing is retained once it's over, and an aggregator restart still +// loses everything, same trade-off CompanionHub's own result log already +// accepts for the same reason. A completed action's output is not +// retrievable after the fact -- only its final summary lives on in the +// existing recent-actions log. type OutputHub struct { - mu sync.Mutex - active map[string]string // agentID -> the action ID currently streaming - subs map[string]map[chan outputEvent]struct{} + mu sync.Mutex + active map[string]string // agentID -> the action ID currently streaming + backlog map[string][]string // agentID -> lines published so far for the active action + subs map[string]map[chan outputEvent]struct{} } func NewOutputHub() *OutputHub { return &OutputHub{ - active: map[string]string{}, - subs: map[string]map[chan outputEvent]struct{}{}, + active: map[string]string{}, + backlog: map[string][]string{}, + subs: map[string]map[chan outputEvent]struct{}{}, } } -// Begin marks actionID as agentID's currently-streaming action. Called -// once an action is successfully pushed (handleAdminApply and friends), -// not when a companion's output stream actually shows up -- some actions -// never have one at all (a bare recheck served by an agent-only -// connection never opens POST /companion/output; an old companion that -// predates output streaming entirely never will either), and those must -// still resolve to a correct "done" via End once their real result -// arrives, rather than a live pane that waits forever for an End call -// that never comes. Unconditional: CompanionHub's own in-flight guard -// (Push's ErrActionInFlight) already ensures at most one action is ever -// in flight per agent, so there's never a second concurrent Begin to -// race against. +// Begin marks actionID as agentID's currently-streaming action, and +// resets its backlog. Called once an action is successfully pushed +// (handleAdminApply and friends), not when a companion's output stream +// actually shows up -- some actions never have one at all (a bare +// recheck served by an agent-only connection never opens POST +// /companion/output; an old companion that predates output streaming +// entirely never will either), and those must still resolve to a correct +// "done" via End once their real result arrives, rather than a live pane +// that waits forever for an End call that never comes. Unconditional: +// CompanionHub's own in-flight guard (Push's ErrActionInFlight) already +// ensures at most one action is ever in flight per agent, so there's +// never a second concurrent Begin to race against. func (h *OutputHub) Begin(agentID, actionID string) { h.mu.Lock() defer h.mu.Unlock() h.active[agentID] = actionID + h.backlog[agentID] = nil } // Publish fans a line out to every current subscriber for agentID, // non-blocking per subscriber -- the same drop-on-full tolerance the // existing heartbeat-based SSE precedent already accepts for a slow -// consumer. A no-op if actionID isn't (or is no longer) the active one for -// agentID, e.g. a late line from a superseded/ended stream. +// consumer -- and appends it to agentID's backlog so a subscriber that +// (re)connects later still sees it (see Subscribe). A no-op if actionID +// isn't (or is no longer) the active one for agentID, e.g. a late line +// from a superseded/ended stream. func (h *OutputHub) Publish(agentID, actionID, line string) { h.mu.Lock() defer h.mu.Unlock() if h.active[agentID] != actionID { return } + h.backlog[agentID] = append(h.backlog[agentID], line) h.broadcast(agentID, outputEvent{Kind: EventLine, ActionID: actionID, Line: line}) } @@ -89,7 +99,8 @@ func (h *OutputHub) Publish(agentID, actionID, line string) { // fires first wins, and the second must not overwrite it (a normal // success's clean stream-close must never show as "disconnected" after // handleCompanionResult already recorded "done", regardless of which of -// the two HTTP requests happens to finish first). +// the two HTTP requests happens to finish first). Clears the backlog too +// -- once an action is over there's nothing left to replay for it. func (h *OutputHub) End(agentID, actionID string, kind outputEventKind) { h.mu.Lock() defer h.mu.Unlock() @@ -97,6 +108,7 @@ func (h *OutputHub) End(agentID, actionID string, kind outputEventKind) { return } delete(h.active, agentID) + delete(h.backlog, agentID) h.broadcast(agentID, outputEvent{Kind: kind, ActionID: actionID}) } @@ -110,27 +122,40 @@ func (h *OutputHub) broadcast(agentID string, event outputEvent) { } } -// Subscribe registers a new subscriber for agentID's live output. The -// returned cancel func must be called exactly once when the caller +// Subscribe registers a new subscriber for agentID's live output and +// atomically returns a snapshot of whatever's already been published for +// the currently active action (nil if none, or if there's no action in +// flight at all) -- the caller should replay this backlog before +// forwarding the live channel, so a browser reconnecting mid-action (e.g. +// a page refresh) resumes from where it left off instead of missing +// everything published before it (re)connected. Snapshotting the backlog +// and registering the channel happen under one lock acquisition +// specifically so no line published in between could be lost (missed by +// the snapshot, then never broadcast because the channel wasn't +// registered yet) or double-delivered (in the snapshot and then broadcast +// again after registration). +// +// The returned cancel func must be called exactly once when the caller // (typically an SSE handler, on the request context ending) is done, to // unregister the channel and avoid leaking it. -func (h *OutputHub) Subscribe(agentID string) (<-chan outputEvent, func()) { - ch := make(chan outputEvent, 32) +func (h *OutputHub) Subscribe(agentID string) (backlog []string, ch <-chan outputEvent, cancel func()) { + c := make(chan outputEvent, 32) h.mu.Lock() + backlog = append([]string(nil), h.backlog[agentID]...) if h.subs[agentID] == nil { h.subs[agentID] = map[chan outputEvent]struct{}{} } - h.subs[agentID][ch] = struct{}{} + h.subs[agentID][c] = struct{}{} h.mu.Unlock() - cancel := func() { + cancel = func() { h.mu.Lock() - delete(h.subs[agentID], ch) + delete(h.subs[agentID], c) if len(h.subs[agentID]) == 0 { delete(h.subs, agentID) } h.mu.Unlock() } - return ch, cancel + return backlog, c, cancel } diff --git a/internal/aggregator/output_test.go b/internal/aggregator/output_test.go index 39ac03a..1594f32 100644 --- a/internal/aggregator/output_test.go +++ b/internal/aggregator/output_test.go @@ -8,7 +8,7 @@ import ( func TestOutputHubPublishReachesSubscriberForActiveAction(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") - ch, cancel := h.Subscribe("a1") + _, ch, cancel := h.Subscribe("a1") defer cancel() h.Publish("a1", "act1", "hello") @@ -26,7 +26,7 @@ func TestOutputHubPublishReachesSubscriberForActiveAction(t *testing.T) { func TestOutputHubPublishIgnoredForStaleActionID(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") - ch, cancel := h.Subscribe("a1") + _, ch, cancel := h.Subscribe("a1") defer cancel() h.Publish("a1", "act2", "stale") // not the active action for a1 @@ -46,7 +46,7 @@ func TestOutputHubPublishIgnoredForStaleActionID(t *testing.T) { func TestOutputHubEndDoneThenLateDisconnectedIsNoop(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") - ch, cancel := h.Subscribe("a1") + _, ch, cancel := h.Subscribe("a1") defer cancel() h.End("a1", "act1", EventDone) @@ -70,7 +70,7 @@ func TestOutputHubEndDoneThenLateDisconnectedIsNoop(t *testing.T) { func TestOutputHubEndDisconnectedForRestartCase(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") - ch, cancel := h.Subscribe("a1") + _, ch, cancel := h.Subscribe("a1") defer cancel() h.End("a1", "act1", EventDisconnected) @@ -87,9 +87,9 @@ func TestOutputHubEndDisconnectedForRestartCase(t *testing.T) { func TestOutputHubMultipleSubscribersAllReceive(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") - ch1, cancel1 := h.Subscribe("a1") + _, ch1, cancel1 := h.Subscribe("a1") defer cancel1() - ch2, cancel2 := h.Subscribe("a1") + _, ch2, cancel2 := h.Subscribe("a1") defer cancel2() h.Publish("a1", "act1", "hi") @@ -106,10 +106,71 @@ func TestOutputHubMultipleSubscribersAllReceive(t *testing.T) { } } +// TestOutputHubSubscribeReplaysBacklogForActiveAction is the regression +// test for browser-refresh resume: a subscriber that connects *after* +// some lines were already published must still see them, then keep +// receiving new ones live. +func TestOutputHubSubscribeReplaysBacklogForActiveAction(t *testing.T) { + h := NewOutputHub() + h.Begin("a1", "act1") + h.Publish("a1", "act1", "one") + h.Publish("a1", "act1", "two") + + backlog, ch, cancel := h.Subscribe("a1") + defer cancel() + + if len(backlog) != 2 || backlog[0] != "one" || backlog[1] != "two" { + t.Fatalf("got backlog %#v, want [one two]", backlog) + } + + h.Publish("a1", "act1", "three") + select { + case ev := <-ch: + if ev.Kind != EventLine || ev.Line != "three" { + t.Fatalf("got %#v, want a live line event saying three", ev) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for the live line published after Subscribe") + } +} + +// TestOutputHubEndClearsBacklog: once an action ends, there's nothing +// left to replay for it -- only in-flight output is meant to survive a +// reconnect, not a completed action's history. +func TestOutputHubEndClearsBacklog(t *testing.T) { + h := NewOutputHub() + h.Begin("a1", "act1") + h.Publish("a1", "act1", "one") + h.End("a1", "act1", EventDone) + + backlog, _, cancel := h.Subscribe("a1") + defer cancel() + if len(backlog) != 0 { + t.Fatalf("got backlog %#v after End, want empty", backlog) + } +} + +// TestOutputHubBeginResetsBacklogForNewAction guards against a new +// action's backlog leaking lines left over from a prior one on the same +// agent. +func TestOutputHubBeginResetsBacklogForNewAction(t *testing.T) { + h := NewOutputHub() + h.Begin("a1", "act1") + h.Publish("a1", "act1", "from act1") + h.End("a1", "act1", EventDone) + + h.Begin("a1", "act2") + backlog, _, cancel := h.Subscribe("a1") + defer cancel() + if len(backlog) != 0 { + t.Fatalf("got backlog %#v for a freshly begun action, want empty", backlog) + } +} + func TestOutputHubCancelStopsFurtherDelivery(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") - ch, cancel := h.Subscribe("a1") + _, ch, cancel := h.Subscribe("a1") cancel() // Must not panic or block just because every subscriber already left. diff --git a/internal/aggregator/server.go b/internal/aggregator/server.go index 2fae7c3..9fb3554 100644 --- a/internal/aggregator/server.go +++ b/internal/aggregator/server.go @@ -458,7 +458,7 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, return } - ch, cancel := s.outputHub.Subscribe(id) + backlog, ch, cancel := s.outputHub.Subscribe(id) defer cancel() w.Header().Set("Content-Type", "text/event-stream") @@ -467,6 +467,18 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, w.WriteHeader(http.StatusOK) flusher.Flush() + // Replay whatever was already published for the currently in-flight + // action before forwarding anything live -- see OutputHub.Subscribe. + // A reconnecting browser (e.g. a page refresh mid-recheck/apply) gets + // caught up first instead of just missing everything before it + // (re)connected. + for _, line := range backlog { + if err := writeOutputEvent(w, string(EventLine), id, line); err != nil { + return + } + } + flusher.Flush() + heartbeat := time.NewTicker(30 * time.Second) defer heartbeat.Stop() @@ -483,14 +495,7 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, } flusher.Flush() case event := <-ch: - payload, err := json.Marshal(struct { - ActionID string `json:"action_id"` - Line string `json:"line,omitempty"` - }{ActionID: event.ActionID, Line: event.Line}) - if err != nil { - continue - } - if _, err := fmt.Fprintf(w, "event: %s\ndata: %s\n\n", event.Kind, payload); err != nil { + if err := writeOutputEvent(w, string(event.Kind), event.ActionID, event.Line); err != nil { return } flusher.Flush() @@ -498,6 +503,23 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, } } +// writeOutputEvent writes one SSE frame in the exact shape a browser's +// EventSource listener expects (see openLiveOutput in templates.go). +// Shared by handleAdminOutputStream's backlog replay and its live-forward +// loop so a replayed line is byte-identical to a live one -- the browser +// must not be able to tell the difference. +func writeOutputEvent(w io.Writer, kind, actionID, line string) error { + payload, err := json.Marshal(struct { + ActionID string `json:"action_id"` + Line string `json:"line,omitempty"` + }{ActionID: actionID, Line: line}) + if err != nil { + return nil // malformed payload is skipped, not fatal to the stream + } + _, err = fmt.Fprintf(w, "event: %s\ndata: %s\n\n", kind, payload) + return err +} + func (s *Server) handleAdmin(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) @@ -677,6 +699,13 @@ func (s *Server) handleAdminApply(w http.ResponseWriter, r *http.Request, id str writeJSON(w, http.StatusAccepted, map[string]string{"action_id": action.ID}) } +type recheckRequest struct { + // Verbose, if true, asks the agent to stream the recheck's real shell + // command output instead of synthetic progress narration -- see + // Action.Verbose. + Verbose bool `json:"verbose,omitempty"` +} + // handleAdminRecheck pushes a recheck Action to a connected companion, so // the agent runs an out-of-band detection cycle instead of waiting for the // next CHECK_INTERVAL -- e.g. after the admin page's data still looks @@ -684,14 +713,24 @@ func (s *Server) handleAdminApply(w http.ResponseWriter, r *http.Request, id str // no shared secret: it can't change anything on the host, only make it // report what's already true sooner, so it gets the same trust model as // the rest of /admin (approve/reject/view). -func (s *Server) handleAdminRecheck(w http.ResponseWriter, _ *http.Request, id string) { +func (s *Server) handleAdminRecheck(w http.ResponseWriter, r *http.Request, id string) { rec, ok := s.registry.Get(id) if !ok || rec.Status != StatusApproved { http.Error(w, "agent not found or not approved", http.StatusNotFound) return } - action := Action{ID: newActionID(), Type: ActionRecheck, CreatedAt: time.Now()} + // Best-effort decode: an empty/missing body (e.g. a bare POST with no + // content) is not an error here, just means verbose=false -- unlike + // handleAdminApply's request body, which is mandatory (it selects + // which packages to touch), this one's body is purely an optional + // preference. + var req recheckRequest + if r.Body != nil { + _ = json.NewDecoder(r.Body).Decode(&req) + } + + action := Action{ID: newActionID(), Type: ActionRecheck, Verbose: req.Verbose, CreatedAt: time.Now()} if err := s.hub.Push(id, action); err != nil { http.Error(w, err.Error(), http.StatusConflict) return diff --git a/internal/aggregator/server_test.go b/internal/aggregator/server_test.go index e1d6f59..faf166c 100644 --- a/internal/aggregator/server_test.go +++ b/internal/aggregator/server_test.go @@ -610,7 +610,7 @@ func TestHandleCompanionResultEndsOutputStreamAsDone(t *testing.T) { // before handleCompanionResult ever runs. s.outputHub.Begin("a1", "act1") - ch, cancel := s.outputHub.Subscribe("a1") + _, ch, cancel := s.outputHub.Subscribe("a1") defer cancel() rec := doJSON(t, s, http.MethodPost, "/companion/result", companionResultRequest{ActionID: "act1", Success: true}, map[string]string{"X-Agent-ID": "a1", "Authorization": "Bearer tok"}) @@ -1198,7 +1198,7 @@ func TestHandleAdminRecheckWithNoCompanionStillResolvesLiveViewAsDone(t *testing t.Fatal(err) } - ch, cancel := s.outputHub.Subscribe("a1") + _, ch, cancel := s.outputHub.Subscribe("a1") defer cancel() resultRec := doJSON(t, s, http.MethodPost, "/companion/result", diff --git a/internal/aggregator/templates.go b/internal/aggregator/templates.go index f8dd79f..ea81c95 100644 --- a/internal/aggregator/templates.go +++ b/internal/aggregator/templates.go @@ -559,6 +559,9 @@ const adminTemplateSrc = ` {{end}}
+ {{if .CompanionConnected}} @@ -766,9 +769,15 @@ const adminTemplateSrc = ` return false; } async function forceRecheck(id) { + const verboseBox = document.getElementById('verbose-' + id); + const verbose = verboseBox ? verboseBox.checked : false; const es = openLiveOutput(id); try { - const resp = await fetch('/admin/agents/' + id + '/recheck', {method: 'POST'}); + const resp = await fetch('/admin/agents/' + id + '/recheck', { + method: 'POST', + headers: {'Content-Type': 'application/json'}, + body: JSON.stringify({verbose: verbose}), + }); if (!resp.ok) { closeLiveOutput(id, es); alert('recheck failed (' + resp.status + '): ' + await resp.text()); diff --git a/internal/aptutil/exec.go b/internal/aptutil/exec.go index 88082c2..66c1b71 100644 --- a/internal/aptutil/exec.go +++ b/internal/aptutil/exec.go @@ -4,19 +4,43 @@ import ( "bytes" "context" "fmt" + "io" "os" "os/exec" "strings" + + "update-detector/internal/checker" + "update-detector/internal/linetee" ) // Update runs `apt-get update` against the given apt.conf (see Write), // refreshing the container-local package index cache. Shared by every -// checker flavor. +// checker flavor. If ctx carries a line sink (see checker.WithLineSink), +// this command's real stdout/stderr are also tapped live, one line at a +// time -- purely additive, e.g. for a UI-triggered verbose recheck; the +// normal periodic detection cycle never attaches one, so this behaves +// exactly as before in that case. func Update(ctx context.Context, aptConfigPath string) error { cmd := exec.CommandContext(ctx, "apt-get", "update", "-q", "-o", "Acquire::Retries=2") cmd.Env = Env(aptConfigPath) var stderr bytes.Buffer - cmd.Stderr = &stderr + var out, errOut io.Writer = io.Discard, &stderr + if sink := checker.LineSinkFromContext(ctx); sink != nil { + // Both point at the same *linetee.Writer value (not two separate + // tees, one per stream) -- os/exec only guarantees single- + // goroutine-at-a-time writes when Stdout and Stderr are the same + // comparable writer, and sink itself relies on that guarantee + // (see OutputSink.Push). Side effect only while verbose streaming + // is active: stdout's routine progress output also lands in the + // stderr buffer below, so a failure's error message may include + // some of it too -- an acceptable trade-off for a live/best-effort + // view, not the normal (no sink) path this function's callers see. + tee := linetee.New(&stderr, sink) + defer tee.Flush() + out, errOut = tee, tee + } + cmd.Stdout = out + cmd.Stderr = errOut if err := cmd.Run(); err != nil { return fmt.Errorf("apt-get update: %w: %s", err, strings.TrimSpace(stderr.String())) } diff --git a/internal/aptutil/exec_test.go b/internal/aptutil/exec_test.go new file mode 100644 index 0000000..b3083e9 --- /dev/null +++ b/internal/aptutil/exec_test.go @@ -0,0 +1,64 @@ +package aptutil + +import ( + "context" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + + "update-detector/internal/checker" +) + +// writeFakeAptGet puts a fake "apt-get" script at the front of PATH for +// the duration of the test, so Update's exec.Command call hits it instead +// of a real apt-get -- same technique as +// internal/companion/execute_test.go's writeFakeAptGet. +func writeFakeAptGet(t *testing.T, script string) { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "apt-get") + if err := os.WriteFile(path, []byte("#!/bin/sh\n"+script+"\n"), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) +} + +func TestUpdateTapsRealOutputToLineSinkWhenPresent(t *testing.T) { + writeFakeAptGet(t, "echo 'Get:1 http://archive.ubuntu.com noble InRelease'\nexit 0") + + var lines []string + ctx := checker.WithLineSink(context.Background(), func(line string) { lines = append(lines, line) }) + + if err := Update(ctx, "/dev/null"); err != nil { + t.Fatal(err) + } + + want := []string{"Get:1 http://archive.ubuntu.com noble InRelease"} + if !reflect.DeepEqual(lines, want) { + t.Fatalf("got sink lines %v, want %v", lines, want) + } +} + +func TestUpdateUnchangedWithoutLineSink(t *testing.T) { + writeFakeAptGet(t, "echo 'Get:1 http://archive.ubuntu.com noble InRelease'\nexit 0") + + if err := Update(context.Background(), "/dev/null"); err != nil { + t.Fatal(err) + } +} + +func TestUpdateReturnsStderrOnFailureWithLineSinkAttached(t *testing.T) { + writeFakeAptGet(t, "echo 'some progress noise'\necho 'E: real failure reason' >&2\nexit 1") + + ctx := checker.WithLineSink(context.Background(), func(string) {}) + + err := Update(ctx, "/dev/null") + if err == nil { + t.Fatal("expected an error") + } + if !strings.Contains(err.Error(), "real failure reason") { + t.Fatalf("got error %q, want it to contain the real stderr failure reason", err.Error()) + } +} diff --git a/internal/checker/debian/exec_test.go b/internal/checker/debian/exec_test.go new file mode 100644 index 0000000..09d1dcf --- /dev/null +++ b/internal/checker/debian/exec_test.go @@ -0,0 +1,59 @@ +//go:build !windows + +package debian + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "update-detector/internal/checker" +) + +// writeFakeAptGet puts a fake "apt-get" script at the front of PATH for +// the duration of the test, so checkUpgradable's exec.Command call hits +// it instead of a real apt-get -- same technique as +// internal/companion/execute_test.go's writeFakeAptGet. +func writeFakeAptGet(t *testing.T, script string) { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "apt-get") + if err := os.WriteFile(path, []byte("#!/bin/sh\n"+script+"\n"), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) +} + +func TestCheckUpgradableTapsRealOutputToLineSinkWhenPresent(t *testing.T) { + writeFakeAptGet(t, "cat <<'EOF'\n"+sampleDistUpgradeOutput+"EOF") + + var lines []string + ctx := checker.WithLineSink(context.Background(), func(line string) { lines = append(lines, line) }) + + result, err := checkUpgradable(ctx, "/dev/null") + if err != nil { + t.Fatal(err) + } + if result.Total != 5 { + t.Fatalf("got Total=%d, want 5 -- attaching a sink must not change parsed output: %#v", result.Total, result) + } + + want := strings.Split(strings.TrimSuffix(sampleDistUpgradeOutput, "\n"), "\n") + if strings.Join(lines, "\n") != strings.Join(want, "\n") { + t.Fatalf("got sink lines %v, want %v", lines, want) + } +} + +func TestCheckUpgradableUnchangedWithoutLineSink(t *testing.T) { + writeFakeAptGet(t, "cat <<'EOF'\n"+sampleDistUpgradeOutput+"EOF") + + result, err := checkUpgradable(context.Background(), "/dev/null") + if err != nil { + t.Fatal(err) + } + if result.Total != 5 { + t.Fatalf("got Total=%d, want 5: %#v", result.Total, result) + } +} diff --git a/internal/checker/debian/packages.go b/internal/checker/debian/packages.go index 0fa3a2d..5e306a2 100644 --- a/internal/checker/debian/packages.go +++ b/internal/checker/debian/packages.go @@ -6,12 +6,14 @@ import ( "bytes" "context" "fmt" + "io" "os/exec" "regexp" "strings" "update-detector/internal/aptutil" "update-detector/internal/checker" + "update-detector/internal/linetee" ) type packageResult struct { @@ -30,11 +32,22 @@ type packageResult struct { // version, group 4: origin/archive/arch info. var instLineRE = regexp.MustCompile(`^Inst\s+(\S+)(?:\s+\[([^\]]+)\])?\s+\(([^\s]+)\s+([^)]*)\)`) +// Only stdout (the simulated dist-upgrade output actually parsed) is +// tapped when a sink is present -- stderr keeps its own independent, +// un-tapped buffer, same reasoning as ubuntu's aptListUpgradable: merging +// the two would risk stderr diagnostics landing inside the very text +// parseDistUpgrade parses, during a verbose recheck specifically. func checkUpgradable(ctx context.Context, aptConfigPath string) (packageResult, error) { var stdout, stderr bytes.Buffer cmd := exec.CommandContext(ctx, "apt-get", "-s", "dist-upgrade") cmd.Env = aptutil.Env(aptConfigPath) - cmd.Stdout = &stdout + var out io.Writer = &stdout + if sink := checker.LineSinkFromContext(ctx); sink != nil { + tee := linetee.New(&stdout, sink) + defer tee.Flush() + out = tee + } + cmd.Stdout = out cmd.Stderr = &stderr if err := cmd.Run(); err != nil { return packageResult{}, fmt.Errorf("apt-get -s dist-upgrade: %w: %s", err, strings.TrimSpace(stderr.String())) diff --git a/internal/checker/linesink.go b/internal/checker/linesink.go new file mode 100644 index 0000000..8ede2c7 --- /dev/null +++ b/internal/checker/linesink.go @@ -0,0 +1,25 @@ +package checker + +import "context" + +type lineSinkKey struct{} + +// WithLineSink attaches sink to ctx so a Checker's command-running code +// can tap real command stdout/stderr as it's produced (e.g. apt-get, +// apt-check, winget, powershell), without threading an extra parameter +// through every Check/checkPackages/checkUpgradable call in every +// platform package. Optional: its absence must never change behavior -- +// see LineSinkFromContext. Intended for a caller that wants a live, +// real-command-output view of one specific Check call (e.g. a UI- +// triggered "verbose" recheck), not for the normal periodic detection +// cycle, which never attaches one. +func WithLineSink(ctx context.Context, sink func(string)) context.Context { + return context.WithValue(ctx, lineSinkKey{}, sink) +} + +// LineSinkFromContext returns the sink attached via WithLineSink, or nil +// if none -- callers must treat nil as "no live tap," not an error. +func LineSinkFromContext(ctx context.Context) func(string) { + sink, _ := ctx.Value(lineSinkKey{}).(func(string)) + return sink +} diff --git a/internal/checker/linesink_test.go b/internal/checker/linesink_test.go new file mode 100644 index 0000000..1e225f6 --- /dev/null +++ b/internal/checker/linesink_test.go @@ -0,0 +1,27 @@ +package checker + +import ( + "context" + "testing" +) + +func TestLineSinkRoundTrip(t *testing.T) { + var got []string + ctx := WithLineSink(context.Background(), func(line string) { got = append(got, line) }) + + sink := LineSinkFromContext(ctx) + if sink == nil { + t.Fatal("expected a non-nil sink after WithLineSink") + } + sink("hello") + + if len(got) != 1 || got[0] != "hello" { + t.Fatalf("got %v, want [hello]", got) + } +} + +func TestLineSinkFromContextNilWhenAbsent(t *testing.T) { + if sink := LineSinkFromContext(context.Background()); sink != nil { + t.Fatal("expected nil sink when none attached") + } +} diff --git a/internal/checker/ubuntu/exec_test.go b/internal/checker/ubuntu/exec_test.go new file mode 100644 index 0000000..6bbc22c --- /dev/null +++ b/internal/checker/ubuntu/exec_test.go @@ -0,0 +1,62 @@ +//go:build !windows + +package ubuntu + +import ( + "context" + "os" + "path/filepath" + "reflect" + "testing" + + "update-detector/internal/checker" +) + +// writeFakeApt puts a fake "apt" script at the front of PATH for the +// duration of the test, so aptListUpgradable's exec.Command call hits it +// instead of a real apt -- same technique as +// internal/companion/execute_test.go's writeFakeAptGet. +func writeFakeApt(t *testing.T, script string) { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "apt") + if err := os.WriteFile(path, []byte("#!/bin/sh\n"+script+"\n"), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) +} + +const fakeAptListOutput = "Listing...\n" + + "docker-compose-plugin/noble 5.3.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.2.0-1~ubuntu.24.04~noble]\n" + +func TestAptListUpgradableTapsRealOutputToLineSinkWhenPresent(t *testing.T) { + writeFakeApt(t, "printf '"+fakeAptListOutput+"'") + + var lines []string + ctx := checker.WithLineSink(context.Background(), func(line string) { lines = append(lines, line) }) + + upgrades, err := aptListUpgradable(ctx, "/dev/null") + if err != nil { + t.Fatal(err) + } + if len(upgrades) != 1 || upgrades[0].Name != "docker-compose-plugin" { + t.Fatalf("got %#v, want one docker-compose-plugin upgrade -- attaching a sink must not change parsed output", upgrades) + } + + want := []string{"Listing...", "docker-compose-plugin/noble 5.3.0-1~ubuntu.24.04~noble amd64 [upgradable from: 5.2.0-1~ubuntu.24.04~noble]"} + if !reflect.DeepEqual(lines, want) { + t.Fatalf("got sink lines %v, want %v", lines, want) + } +} + +func TestAptListUpgradableUnchangedWithoutLineSink(t *testing.T) { + writeFakeApt(t, "printf '"+fakeAptListOutput+"'") + + upgrades, err := aptListUpgradable(context.Background(), "/dev/null") + if err != nil { + t.Fatal(err) + } + if len(upgrades) != 1 || upgrades[0].Name != "docker-compose-plugin" { + t.Fatalf("got %#v, want one docker-compose-plugin upgrade", upgrades) + } +} diff --git a/internal/checker/ubuntu/packages.go b/internal/checker/ubuntu/packages.go index 495d2f9..da7565f 100644 --- a/internal/checker/ubuntu/packages.go +++ b/internal/checker/ubuntu/packages.go @@ -6,12 +6,14 @@ import ( "bytes" "context" "fmt" + "io" "os/exec" "strconv" "strings" "update-detector/internal/aptutil" "update-detector/internal/checker" + "update-detector/internal/linetee" ) // aptCheckPath is Ubuntu's own upgradable-package counter, shipped by the @@ -45,22 +47,46 @@ func checkUpgradable(ctx context.Context, aptConfigPath string) (packageResult, return packageResult{Total: total, Security: security, Upgrades: upgrades}, nil } +// aptCheckCounts's real content is on stderr (see parseAptCheckCounts) -- +// stdout is unused, so only stderr needs a sink tap; a single tapped +// writer value is enough (no cross-stream merge, so no risk of a +// diagnostic line landing where parsing doesn't expect it). func aptCheckCounts(ctx context.Context, aptConfigPath string) (total int, security int, err error) { var stderr bytes.Buffer cmd := exec.CommandContext(ctx, aptCheckPath) cmd.Env = aptutil.Env(aptConfigPath) - cmd.Stderr = &stderr + var errOut io.Writer = &stderr + if sink := checker.LineSinkFromContext(ctx); sink != nil { + tee := linetee.New(&stderr, sink) + defer tee.Flush() + errOut = tee + } + cmd.Stderr = errOut if err := cmd.Run(); err != nil { return 0, 0, fmt.Errorf("apt-check: %w: %s", err, strings.TrimSpace(stderr.String())) } return parseAptCheckCounts(stderr.String()) } +// Only stdout (the parsed list itself) is tapped when a sink is present -- +// stderr keeps its own independent, un-tapped buffer exactly as before. +// Tapping both would mean sharing one writer value across both streams +// (see internal/aptutil.Update's own comment on why), which here would +// also merge stderr's diagnostic text into the very stdout buffer +// parseUpgradableList parses -- a real risk of breaking parsing during a +// verbose recheck specifically, not just a cosmetic trade-off, so it's +// deliberately avoided. func aptListUpgradable(ctx context.Context, aptConfigPath string) ([]checker.PackageUpgrade, error) { var stdout, stderr bytes.Buffer cmd := exec.CommandContext(ctx, "apt", "list", "--upgradable") cmd.Env = aptutil.Env(aptConfigPath) - cmd.Stdout = &stdout + var out io.Writer = &stdout + if sink := checker.LineSinkFromContext(ctx); sink != nil { + tee := linetee.New(&stdout, sink) + defer tee.Flush() + out = tee + } + cmd.Stdout = out cmd.Stderr = &stderr if err := cmd.Run(); err != nil { return nil, fmt.Errorf("apt list --upgradable: %w: %s", err, strings.TrimSpace(stderr.String())) diff --git a/internal/checker/windows/packages.go b/internal/checker/windows/packages.go index d10047f..358bce6 100644 --- a/internal/checker/windows/packages.go +++ b/internal/checker/windows/packages.go @@ -7,8 +7,12 @@ import ( "context" "errors" "fmt" + "io" "os/exec" "strings" + + "update-detector/internal/checker" + "update-detector/internal/linetee" ) // ErrWingetNotFound means winget itself isn't runnable from this @@ -24,12 +28,20 @@ var ErrWingetNotFound = errors.New("winget not found on PATH for this account") // checkUpgradable shells out to winget and parses its table output (see // packages_parse.go, deliberately untagged so that parsing logic is -// testable on any platform). +// testable on any platform). Only stdout (the table itself) is tapped +// when a sink is present -- stderr keeps its own independent, un-tapped +// buffer, same reasoning as the ubuntu/debian checkers' own equivalents. func checkUpgradable(ctx context.Context) (packageResult, error) { var stdout, stderr bytes.Buffer cmd := exec.CommandContext(ctx, "winget", "upgrade", "--include-unknown", "--accept-source-agreements", "--disable-interactivity") - cmd.Stdout = &stdout + var out io.Writer = &stdout + if sink := checker.LineSinkFromContext(ctx); sink != nil { + tee := linetee.New(&stdout, sink) + defer tee.Flush() + out = tee + } + cmd.Stdout = out cmd.Stderr = &stderr if err := cmd.Run(); err != nil { if errors.Is(err, exec.ErrNotFound) { diff --git a/internal/checker/windows/windowsupdate.go b/internal/checker/windows/windowsupdate.go index 7037720..e2c2c3d 100644 --- a/internal/checker/windows/windowsupdate.go +++ b/internal/checker/windows/windowsupdate.go @@ -6,8 +6,12 @@ import ( "bytes" "context" "fmt" + "io" "os/exec" "strings" + + "update-detector/internal/checker" + "update-detector/internal/linetee" ) // windowsUpdateScript queries the Windows Update Agent API (the same COM @@ -58,7 +62,17 @@ ConvertTo-Json -InputObject $updates -Compress func checkWindowsUpdates(ctx context.Context) (packageResult, error) { var stdout, stderr bytes.Buffer cmd := exec.CommandContext(ctx, "powershell", "-NoProfile", "-NonInteractive", "-Command", windowsUpdateScript) - cmd.Stdout = &stdout + var out io.Writer = &stdout + if sink := checker.LineSinkFromContext(ctx); sink != nil { + // -Compress means this is one long single-line JSON blob -- a + // verbose stream shows it as one raw line, same "no special- + // casing for readability" posture Upgrade's own real-output + // streaming already has. + tee := linetee.New(&stdout, sink) + defer tee.Flush() + out = tee + } + cmd.Stdout = out cmd.Stderr = &stderr if err := cmd.Run(); err != nil { return packageResult{}, fmt.Errorf("querying Windows Update: %w: %s", err, strings.TrimSpace(stderr.String())) diff --git a/internal/companion/execute.go b/internal/companion/execute.go index 46f1e87..a3299b4 100644 --- a/internal/companion/execute.go +++ b/internal/companion/execute.go @@ -15,6 +15,7 @@ import ( "update-detector/internal/aggregator" "update-detector/internal/aggregatorclient" "update-detector/internal/checker" + "update-detector/internal/linetee" ) const outputTruncateLimit = 4000 @@ -145,23 +146,23 @@ func triggerRecheck(ctx context.Context, agentStatusURL string) { // live as it's written -- purely additive, the returned string is // identical either way. cmd.Stdout and cmd.Stderr are always set to the // exact same writer value (whichever one that is), preserving os/exec's -// single-writer-at-a-time guarantee that lineTee itself relies on. +// single-writer-at-a-time guarantee that linetee.Writer itself relies on. // runCappedImpl is the real implementation used by default. Tests may // replace the package-level runCapped variable with a mock to capture and // validate executed commands without actually running them. func runCappedImpl(ctx context.Context, cmd *exec.Cmd) (string, error) { var buf bytes.Buffer var w io.Writer = &buf - var tee *lineTee + var tee *linetee.Writer if sink := sinkFromContext(ctx); sink != nil { - tee = newLineTee(&buf, sink.push) + tee = linetee.New(&buf, sink.Push) w = tee } cmd.Stdout = w cmd.Stderr = w err := cmd.Run() if tee != nil { - tee.flush() + tee.Flush() } out := buf.String() if len(out) > outputTruncateLimit { diff --git a/internal/companion/linetee.go b/internal/companion/linetee.go deleted file mode 100644 index 26a2343..0000000 --- a/internal/companion/linetee.go +++ /dev/null @@ -1,58 +0,0 @@ -package companion - -import ( - "bytes" - "io" -) - -// lineTee wraps dst, forwarding every byte written to it (so callers that -// only care about the final accumulated buffer see no behavior change) while -// also invoking onLine once per complete '\n'-terminated line as it's -// written, so a caller can tap the stream incrementally without changing -// what's ultimately captured. -// -// Safe to use as both exec.Cmd.Stdout and exec.Cmd.Stderr simultaneously via -// the same *lineTee value -- os/exec guarantees at most one goroutine writes -// to a shared writer value at a time when Stdout and Stderr are the same -// comparable value, and this type relies on that guarantee rather than -// duplicating its own locking. -type lineTee struct { - dst io.Writer - onLine func(string) - partial []byte -} - -func newLineTee(dst io.Writer, onLine func(string)) *lineTee { - return &lineTee{dst: dst, onLine: onLine} -} - -func (t *lineTee) Write(p []byte) (int, error) { - n, err := t.dst.Write(p) - if err == nil && t.onLine != nil { - t.emit(p) - } - return n, err -} - -func (t *lineTee) emit(p []byte) { - t.partial = append(t.partial, p...) - for { - i := bytes.IndexByte(t.partial, '\n') - if i < 0 { - break - } - line := string(t.partial[:i]) - t.partial = t.partial[i+1:] - t.onLine(line) - } -} - -// flush emits any trailing partial line that never got a terminating '\n' -// -- e.g. a script's last line, or output still mid-line when the process -// exited. Call once after the command has finished producing output. -func (t *lineTee) flush() { - if t.onLine != nil && len(t.partial) > 0 { - t.onLine(string(t.partial)) - t.partial = nil - } -} diff --git a/internal/companion/outputsink.go b/internal/companion/outputsink.go index 78d605a..ae5ba2f 100644 --- a/internal/companion/outputsink.go +++ b/internal/companion/outputsink.go @@ -11,8 +11,9 @@ import ( // absence must never change behavior: runCapped works exactly as it did // before this existed when no sink is in context. // -// push is only ever called from the single goroutine os/exec guarantees -// when Stdout and Stderr are the same writer value (see lineTee) -- so +// Push is only ever called from the single goroutine os/exec guarantees +// when Stdout and Stderr are the same writer value (see internal/linetee), +// or from a single caller-owned goroutine outside this package -- so // OutputSink needs no locking of its own. type OutputSink struct { ch chan string @@ -28,12 +29,16 @@ func NewOutputSink(bufSize int) *OutputSink { return &OutputSink{ch: make(chan string, bufSize)} } -// push is non-blocking. While the buffer is full it just counts drops; +// Push is non-blocking. While the buffer is full it just counts drops; // once a slot frees up, one synthetic marker line reports how many were // lost before resuming normal lines -- mirrors runCapped's own // "...(truncated)..." convention for the same reason (an incomplete live -// view should say so, not silently skip ahead). -func (s *OutputSink) push(line string) { +// view should say so, not silently skip ahead). Exported (not just used +// via emitFromContext/WithOutputSink from within this package) so a +// caller outside package companion -- e.g. cmd/update-detector's own +// recheck handling -- can construct a sink and push directly into it +// without needing a context indirection of its own. +func (s *OutputSink) Push(line string) { if s.dropped > 0 { select { case s.ch <- fmt.Sprintf("...(%d line(s) dropped)...", s.dropped): @@ -90,6 +95,6 @@ func emitFromContext(ctx context.Context) func(string, ...any) { return func(string, ...any) {} } return func(format string, args ...any) { - sink.push(fmt.Sprintf(format, args...)) + sink.Push(fmt.Sprintf(format, args...)) } } diff --git a/internal/companion/outputstream_test.go b/internal/companion/outputstream_test.go index ea510b7..8d0a452 100644 --- a/internal/companion/outputstream_test.go +++ b/internal/companion/outputstream_test.go @@ -41,8 +41,8 @@ func TestStreamOutputDeliversLinesInOrder(t *testing.T) { defer srv.Close() sink := NewOutputSink(10) - sink.push("line one") - sink.push("line two") + sink.Push("line one") + sink.Push("line two") sink.Close() identity := aggregatorclient.Identity{AgentID: "agent1", Token: "tok"} @@ -76,7 +76,7 @@ func TestStreamOutputReturnsPromptlyWhenAggregatorUnreachable(t *testing.T) { sink := NewOutputSink(10) go func() { - sink.push("line one") + sink.Push("line one") sink.Close() }() diff --git a/internal/linetee/linetee.go b/internal/linetee/linetee.go new file mode 100644 index 0000000..4b1f1ef --- /dev/null +++ b/internal/linetee/linetee.go @@ -0,0 +1,63 @@ +// Package linetee provides a writer that tees written bytes through +// unchanged while also invoking a callback once per complete line, so a +// caller can tap a command's output incrementally (e.g. for live +// streaming) without changing what's ultimately captured. +package linetee + +import ( + "bytes" + "io" +) + +// Writer wraps Dst, forwarding every byte written to it (so callers that +// only care about the final accumulated buffer see no behavior change) +// while also invoking OnLine once per complete '\n'-terminated line as +// it's written, so a caller can tap the stream incrementally without +// changing what's ultimately captured. +// +// Safe to use as both exec.Cmd.Stdout and exec.Cmd.Stderr simultaneously +// via the same *Writer value -- os/exec guarantees at most one goroutine +// writes to a shared writer value at a time when Stdout and Stderr are +// the same comparable value, and this type relies on that guarantee +// rather than duplicating its own locking. +type Writer struct { + dst io.Writer + onLine func(string) + partial []byte +} + +// New returns a Writer teeing to dst and invoking onLine per complete line. +func New(dst io.Writer, onLine func(string)) *Writer { + return &Writer{dst: dst, onLine: onLine} +} + +func (w *Writer) Write(p []byte) (int, error) { + n, err := w.dst.Write(p) + if err == nil && w.onLine != nil { + w.emit(p) + } + return n, err +} + +func (w *Writer) emit(p []byte) { + w.partial = append(w.partial, p...) + for { + i := bytes.IndexByte(w.partial, '\n') + if i < 0 { + break + } + line := string(w.partial[:i]) + w.partial = w.partial[i+1:] + w.onLine(line) + } +} + +// Flush emits any trailing partial line that never got a terminating '\n' +// -- e.g. a script's last line, or output still mid-line when the process +// exited. Call once after the command has finished producing output. +func (w *Writer) Flush() { + if w.onLine != nil && len(w.partial) > 0 { + w.onLine(string(w.partial)) + w.partial = nil + } +} diff --git a/internal/companion/linetee_test.go b/internal/linetee/linetee_test.go similarity index 58% rename from internal/companion/linetee_test.go rename to internal/linetee/linetee_test.go index 3a87ff3..107be45 100644 --- a/internal/companion/linetee_test.go +++ b/internal/linetee/linetee_test.go @@ -1,4 +1,4 @@ -package companion +package linetee import ( "bytes" @@ -6,20 +6,20 @@ import ( "testing" ) -func TestLineTeeEmitsCompleteLinesAcrossArbitraryChunking(t *testing.T) { +func TestWriterEmitsCompleteLinesAcrossArbitraryChunking(t *testing.T) { var buf bytes.Buffer var lines []string - tee := newLineTee(&buf, func(line string) { lines = append(lines, line) }) + w := New(&buf, func(line string) { lines = append(lines, line) }) // Deliberately split mid-line across writes, at a boundary that // doesn't align with any line at all. chunks := []string{"first li", "ne\nseco", "nd line\nthir"} for _, c := range chunks { - if _, err := tee.Write([]byte(c)); err != nil { + if _, err := w.Write([]byte(c)); err != nil { t.Fatal(err) } } - tee.flush() // "thir" has no trailing newline yet + w.Flush() // "thir" has no trailing newline yet want := []string{"first line", "second line", "thir"} if !reflect.DeepEqual(lines, want) { @@ -30,27 +30,27 @@ func TestLineTeeEmitsCompleteLinesAcrossArbitraryChunking(t *testing.T) { } } -func TestLineTeeFlushNoopWhenNoTrailingPartial(t *testing.T) { +func TestWriterFlushNoopWhenNoTrailingPartial(t *testing.T) { var buf bytes.Buffer var lines []string - tee := newLineTee(&buf, func(line string) { lines = append(lines, line) }) - if _, err := tee.Write([]byte("one\ntwo\n")); err != nil { + w := New(&buf, func(line string) { lines = append(lines, line) }) + if _, err := w.Write([]byte("one\ntwo\n")); err != nil { t.Fatal(err) } - tee.flush() + w.Flush() want := []string{"one", "two"} if !reflect.DeepEqual(lines, want) { t.Fatalf("got %v, want %v", lines, want) } } -func TestLineTeeNilOnLineIsSafe(t *testing.T) { +func TestWriterNilOnLineIsSafe(t *testing.T) { var buf bytes.Buffer - tee := newLineTee(&buf, nil) - if _, err := tee.Write([]byte("hello\n")); err != nil { + w := New(&buf, nil) + if _, err := w.Write([]byte("hello\n")); err != nil { t.Fatal(err) } - tee.flush() + w.Flush() if buf.String() != "hello\n" { t.Fatalf("got %q", buf.String()) } From c07bb6f30f71bef73b8acc8028496f0be28be1bf Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 12:16:56 +0800 Subject: [PATCH 2/7] aptutil: skip the new fake-apt-get sink tests on Windows They fake apt-get as a #!/bin/sh script on PATH, which windows-latest's exec.LookPath can't run (no shebang support, PATHEXT-based lookup expects a known executable extension) -- confirmed by CI failing on exactly this. apt-get itself is Linux-only in practice anyway, matching the same //go:build !windows already on the equivalent ubuntu/debian checker tests added alongside these. --- internal/aptutil/exec_test.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/internal/aptutil/exec_test.go b/internal/aptutil/exec_test.go index b3083e9..24f13bc 100644 --- a/internal/aptutil/exec_test.go +++ b/internal/aptutil/exec_test.go @@ -1,3 +1,11 @@ +//go:build !windows + +// These tests fake "apt-get" as a #!/bin/sh script on PATH -- apt-get +// (and this package generally) is Linux-only in practice anyway, and a +// POSIX shell script isn't runnable via Windows' exec.LookPath the same +// way (no shebang support, and PATHEXT-based lookup expects a known +// executable extension), so there's no meaningful way to run this same +// fake-script test on windows-latest CI. package aptutil import ( From 76f1ce5c60244ad8cfaa01dfcd90cdc3690cf6d4 Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 12:56:07 +0800 Subject: [PATCH 3/7] aggregator: fix recheck never actually streaming (agentPending was never set) Confirmed on real Ubuntu, Windows, and WSL hosts: Force recheck opened the live-output pane but never showed anything, verbose or not, on any platform. Root cause: CompanionHub.Push's agentStreams branch (the one that routes recheck, since it doesn't require a companion) pushed the action but never recorded it in any in-flight map. handleCompanionOutput authorizes every output-stream POST against that same tracking (previously just `pending`), so the agent's own StreamOutput call for a recheck was rejected with 409 on its very first request -- silently, since the failure only ever logged on the agent side, never surfaced to the browser. Added a separate agentPending map (recheck must never be blocked by, or clobber the tracking of, an already in-flight companion action on the same host -- Push deliberately skips the ErrActionInFlight check for it) and a new IsPending(agentID, actionID) that checks both maps, used by handleCompanionOutput instead of comparing against the single-value Pending() getter (which stays as-is for its own separate "what to resume watching after a page reload" purpose). --- internal/aggregator/companion.go | 63 +++++++++++++++++++++----- internal/aggregator/companion_test.go | 64 +++++++++++++++++++++++++++ internal/aggregator/server.go | 14 +++--- internal/aggregator/server_test.go | 63 ++++++++++++++++++++++++++ 4 files changed, 188 insertions(+), 16 deletions(-) diff --git a/internal/aggregator/companion.go b/internal/aggregator/companion.go index 272d43a..48e8d0f 100644 --- a/internal/aggregator/companion.go +++ b/internal/aggregator/companion.go @@ -170,8 +170,17 @@ type CompanionHub struct { // last reported detecting agent running there (natively or as a // Docker container) -- see SetAgentPresent. agentPresent map[string]bool - pending map[string]string // agentID -> in-flight action ID - results map[string][]ActionResult + pending map[string]string // agentID -> in-flight action ID pushed via the main (usually companion) stream + // agentPending is agentID -> in-flight action ID pushed via + // agentStreams instead (e.g. a recheck) -- tracked separately from + // pending, deliberately: a non-companion-required action is never + // blocked by (and must never clobber the tracking of) an already + // in-flight companion action on the same agent, since Push's own + // agentStreams branch skips the ErrActionInFlight check entirely. See + // IsPending, which is what actually authorizes output streaming + // (handleCompanionOutput) against either map. + agentPending map[string]string + results map[string][]ActionResult } func NewCompanionHub() *CompanionHub { @@ -182,6 +191,7 @@ func NewCompanionHub() *CompanionHub { aggregatorPresent: map[string]bool{}, agentPresent: map[string]bool{}, pending: map[string]string{}, + agentPending: map[string]string{}, results: map[string][]ActionResult{}, } } @@ -342,6 +352,7 @@ func (h *CompanionHub) Disconnect(agentID string, ch chan Action) { } if cur, ok := h.agentStreams[agentID]; ok && cur.ch == ch { delete(h.agentStreams, agentID) + delete(h.agentPending, agentID) } } @@ -364,15 +375,21 @@ func (h *CompanionHub) Push(agentID string, action Action) error { h.mu.Lock() defer h.mu.Unlock() - // Agent-only actions (like ActionCompleteCompanionSwap) are routed - // to the agent stream, even if a companion holds the main slot. - // If no agent stream exists, fall through to the main stream - // (companion can handle it too, e.g. ActionRecheck on a host - // where only the companion is connected). + // Agent-only actions (like ActionCompleteCompanionSwap and + // ActionRecheck) are routed to the agent stream, even if a companion + // holds the main slot. If no agent stream exists, fall through to the + // main stream (companion can handle it too, e.g. ActionRecheck on a + // host where only the companion is connected). Tracked in + // agentPending, not pending -- deliberately no ErrActionInFlight + // check here: these actions are never blocked by (and must never be + // mistaken, via a shared marker, for) an already in-flight companion + // action on the same agent. See IsPending, which authorizes output + // streaming (handleCompanionOutput) against both maps. if !action.Type.requiresCompanion() { if agentEntry, ok := h.agentStreams[agentID]; ok { select { case agentEntry.ch <- action: + h.agentPending[agentID] = action.ID return nil default: return errors.New("agent stream busy, action dropped") @@ -404,13 +421,18 @@ func (h *CompanionHub) Push(agentID string, action Action) error { // RecordResult appends result to agentID's capped action log, and clears // the in-flight marker Push set for this action -- but only if it's still // the current one, so a stale/duplicate result for an already-superseded -// action can't clobber a newer in-flight marker. +// action can't clobber a newer in-flight marker. Checks both pending and +// agentPending since either could hold it, depending on which of Push's +// two branches originally routed this action. func (h *CompanionHub) RecordResult(agentID string, result ActionResult) { h.mu.Lock() defer h.mu.Unlock() if h.pending[agentID] == result.ActionID { delete(h.pending, agentID) } + if h.agentPending[agentID] == result.ActionID { + delete(h.agentPending, agentID) + } log := append(h.results[agentID], result) if len(log) > actionLogLimit { log = log[len(log)-actionLogLimit:] @@ -429,14 +451,35 @@ func (h *CompanionHub) Results(agentID string) []ActionResult { // Pending returns the action ID currently in flight for agentID, if any -- // lets a page load/reload notice an already-running action (e.g. triggered // from another tab, or before a reload) and resume watching its live -// output instead of only the tab that started it. +// output instead of only the tab that started it. Checks pending first, +// falling back to agentPending -- a companion action (if any) is +// preferred for this single-slot "what to resume watching" purpose in +// the rare case both happen to be in flight at once (see IsPending for +// the authorization check that doesn't have to pick just one). func (h *CompanionHub) Pending(agentID string) (string, bool) { h.mu.Lock() defer h.mu.Unlock() - id, ok := h.pending[agentID] + if id, ok := h.pending[agentID]; ok { + return id, true + } + id, ok := h.agentPending[agentID] return id, ok } +// IsPending reports whether actionID is agentID's currently in-flight +// action, checking both pending and agentPending -- unlike Pending +// (which returns a single value for the "what should I resume watching" +// UI case), this is used to authorize output streaming +// (handleCompanionOutput), where a recheck and a companion action can +// legitimately be in flight for the same agent at once, and each needs +// its own actionID to check out correctly regardless of which one +// Pending itself would have preferred to report. +func (h *CompanionHub) IsPending(agentID, actionID string) bool { + h.mu.Lock() + defer h.mu.Unlock() + return h.pending[agentID] == actionID || h.agentPending[agentID] == actionID +} + func newActionID() string { buf := make([]byte, 8) if _, err := rand.Read(buf); err != nil { diff --git a/internal/aggregator/companion_test.go b/internal/aggregator/companion_test.go index 214d5cf..e057076 100644 --- a/internal/aggregator/companion_test.go +++ b/internal/aggregator/companion_test.go @@ -274,6 +274,70 @@ func TestCompanionHubPushRequiresCompanionForApplyActions(t *testing.T) { } } +// TestCompanionHubPushSetsAgentPendingForNonCompanionActions is the +// regression test for a real bug: Push's agentStreams branch pushed the +// action but never recorded it as in-flight anywhere, so +// handleCompanionOutput's authorization check (IsPending) always +// rejected a recheck's own output-stream POST with 409 -- "Force +// recheck" silently never streamed anything, on every platform, +// regardless of the verbose flag. +func TestCompanionHubPushSetsAgentPendingForNonCompanionActions(t *testing.T) { + h := NewCompanionHub() + h.Connect("a1", KindAgent, "") + + if err := h.Push("a1", Action{ID: "act1", Type: ActionRecheck}); err != nil { + t.Fatalf("push failed: %v", err) + } + + if !h.IsPending("a1", "act1") { + t.Fatal("expected IsPending to report the recheck's action ID as in flight") + } + if id, ok := h.Pending("a1"); !ok || id != "act1" { + t.Fatalf("got Pending()=%q,%v, want act1,true", id, ok) + } +} + +// TestCompanionHubRecheckDoesNotClobberInFlightCompanionAction covers the +// other half of the same fix: a recheck routed via agentStreams must +// track its own in-flight marker independently of an already in-flight +// companion action for the same agent, so IsPending can authorize output +// streaming for *both* without either clobbering the other's tracking +// (and so RecordResult for either one clears only its own marker). +func TestCompanionHubRecheckDoesNotClobberInFlightCompanionAction(t *testing.T) { + h := NewCompanionHub() + h.Connect("a1", KindCompanion, "") + // Matches real topology: the agent always opens its own KindAgent + // stream too, landing in agentStreams alongside the companion's main + // slot (see Connect) -- without this, Push's agentStreams branch has + // nothing to route the recheck to and it falls through to share the + // companion's own pending slot instead, which is Push's documented, + // legitimate fallback for a host with no separate agent connection at + // all (e.g. an old agent binary), not what this test means to cover. + h.Connect("a1", KindAgent, "") + + if err := h.Push("a1", Action{ID: "upgrade1", Type: ActionUpgrade}); err != nil { + t.Fatalf("push upgrade failed: %v", err) + } + if err := h.Push("a1", Action{ID: "recheck1", Type: ActionRecheck}); err != nil { + t.Fatalf("push recheck failed: %v", err) + } + + if !h.IsPending("a1", "upgrade1") { + t.Fatal("expected the in-flight upgrade to still be authorized") + } + if !h.IsPending("a1", "recheck1") { + t.Fatal("expected the in-flight recheck to also be authorized") + } + + h.RecordResult("a1", ActionResult{ActionID: "recheck1", Success: true, CompletedAt: time.Now()}) + if h.IsPending("a1", "recheck1") { + t.Fatal("expected the recheck's marker to be cleared after its result") + } + if !h.IsPending("a1", "upgrade1") { + t.Fatal("expected the still-in-flight upgrade to be unaffected by the recheck's own result") + } +} + func TestCompanionHubAgentConnectDoesNotTouchCompanionVersion(t *testing.T) { h := NewCompanionHub() res := h.Connect("a1", KindCompanion, "v1.0.0") diff --git a/internal/aggregator/server.go b/internal/aggregator/server.go index 9fb3554..49353db 100644 --- a/internal/aggregator/server.go +++ b/internal/aggregator/server.go @@ -397,18 +397,20 @@ func (s *Server) handleCompanionOutput(w http.ResponseWriter, r *http.Request) { } // Defense in depth, not just a UI nicety: a stream for an action this // agent doesn't currently have in flight (stale retry, mismatched ID) - // must not be allowed to publish anything. - if pending, ok := s.hub.Pending(rec.ID); !ok || pending != actionID { + // must not be allowed to publish anything. IsPending, not Pending -- + // this agent's own recheck (agentPending) and a companion's own apply + // (pending) can be in flight at the same time, each with its own + // actionID, and both need to check out correctly here. + if !s.hub.IsPending(rec.ID, actionID) { http.Error(w, "action_id does not match this agent's in-flight action", http.StatusConflict) return } // Begin already happened when the action was pushed (handleAdminApply // et al.) -- uniformly, whether or not this stream ever actually - // arrives, so an agent-only recheck (never opens this endpoint at - // all) and an old companion that predates output streaming entirely - // both still resolve to a correct "done" via handleCompanionResult - // instead of a live pane that never closes. + // arrives, so an old companion that predates output streaming + // entirely still resolves to a correct "done" via + // handleCompanionResult instead of a live pane that never closes. defer s.outputHub.End(rec.ID, actionID, EventDisconnected) scanner := bufio.NewScanner(r.Body) diff --git a/internal/aggregator/server_test.go b/internal/aggregator/server_test.go index faf166c..2facb50 100644 --- a/internal/aggregator/server_test.go +++ b/internal/aggregator/server_test.go @@ -535,6 +535,69 @@ func TestHandleCompanionOutputFansOutToAdminStream(t *testing.T) { } } +// TestHandleCompanionOutputAcceptsAgentRoutedRecheckAction is the +// regression test for a real bug: a recheck (agentStreams-routed, no +// companion involved) never set CompanionHub's in-flight marker, so this +// exact endpoint always rejected its output-stream POST with 409 -- +// "Force recheck" silently never streamed anything, on every platform. +func TestHandleCompanionOutputAcceptsAgentRoutedRecheckAction(t *testing.T) { + s, reg := newTestServer(t) + approvedAgent(t, s, reg, "a1", "web01", "tok") + res := s.hub.Connect("a1", KindAgent, "") + defer s.hub.Disconnect("a1", res.Ch) + if err := s.hub.Push("a1", Action{ID: "act1", Type: ActionRecheck, CreatedAt: time.Now()}); err != nil { + t.Fatalf("push failed: %v", err) + } + s.outputHub.Begin("a1", "act1") + + httpSrv := httptest.NewServer(s.Handler()) + defer httpSrv.Close() + + streamReq, err := http.NewRequest(http.MethodGet, httpSrv.URL+"/admin/agents/a1/output/stream", nil) + if err != nil { + t.Fatal(err) + } + streamResp, err := (&http.Client{Timeout: 5 * time.Second}).Do(streamReq) + if err != nil { + t.Fatal(err) + } + defer streamResp.Body.Close() + if streamResp.StatusCode != http.StatusOK { + t.Fatalf("got status %d, want 200", streamResp.StatusCode) + } + + pr, pw := io.Pipe() + go func() { + _ = json.NewEncoder(pw).Encode(companionOutputFrame{ActionID: "act1", Line: "Running detection cycle..."}) + pw.Close() + }() + + outputReq, err := http.NewRequest(http.MethodPost, httpSrv.URL+"/companion/output?action_id=act1", pr) + if err != nil { + t.Fatal(err) + } + outputReq.Header.Set("X-Agent-ID", "a1") + outputReq.Header.Set("Authorization", "Bearer tok") + outputResp, err := (&http.Client{Timeout: 5 * time.Second}).Do(outputReq) + if err != nil { + t.Fatal(err) + } + defer outputResp.Body.Close() + if outputResp.StatusCode != http.StatusOK { + t.Fatalf("got status %d, want 200 -- this is exactly the 409 regression", outputResp.StatusCode) + } + + buf := make([]byte, 4096) + n, err := streamResp.Body.Read(buf) + if err != nil && err != io.EOF { + t.Fatal(err) + } + got := string(buf[:n]) + if !strings.Contains(got, "event: line") || !strings.Contains(got, "Running detection cycle") { + t.Fatalf("expected a line event with the recheck's narration in SSE body, got: %q", got) + } +} + // TestHandleCompanionOutputEndsAsDisconnectedWithoutPriorResult is the // regression test for the companion-self-update-restart case: the output // stream's body ending with no prior handleCompanionResult call must From a38dec1524343ce17ac8a7166bab027add89e8a1 Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 12:56:07 +0800 Subject: [PATCH 4/7] companion: add --disable-interactivity to winget apply commands Confirmed on a real Windows host: applying a winget-sourced package via the companion (always running as a Windows Service, so stdin is never a real console) failed with "ERROR: Input redirection is not supported, exiting the process immediately." -- a winget error that --accept-package-agreements/--accept-source-agreements alone don't prevent. The detection side's own winget invocation (internal/checker/windows/packages.go) already passes --disable-interactivity and has no such problem; the apply side's three winget commands (Packages, Upgrade, and FullUpgrade via Upgrade) were missing it. --- internal/companion/applier_winget.go | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/internal/companion/applier_winget.go b/internal/companion/applier_winget.go index df17efb..05d442b 100644 --- a/internal/companion/applier_winget.go +++ b/internal/companion/applier_winget.go @@ -15,11 +15,20 @@ import ( // Update KB (see kbPattern), the same optional/supplementary role // winget already has on the detection side (see // internal/checker/windows/windows.go's own Check). +// +// Every invocation below includes --disable-interactivity, matching the +// detection side's own already-working winget command (see +// internal/checker/windows/packages.go) -- confirmed live, without it +// the companion (always running as a Windows Service, so stdin is never +// a real console) hit `ERROR: Input redirection is not supported, +// exiting the process immediately.` from winget itself. +// --accept-package-agreements/--accept-source-agreements alone aren't +// enough to guarantee winget never falls back to trying to prompt. type wingetApplier struct{} // Packages upgrades each named package individually via: // -// winget upgrade --id --silent --accept-package-agreements --accept-source-agreements +// winget upgrade --id --silent --accept-package-agreements --accept-source-agreements --disable-interactivity // // Winget has no batch form, so packages are applied one at a time. // All output is collected; the first failure aborts the loop. @@ -32,6 +41,7 @@ func (w *wingetApplier) Packages(ctx context.Context, names []string) (string, e "--silent", "--accept-package-agreements", "--accept-source-agreements", + "--disable-interactivity", )) combined.WriteString(out) if err != nil { @@ -43,13 +53,14 @@ func (w *wingetApplier) Packages(ctx context.Context, names []string) (string, e // Upgrade runs: // -// winget upgrade --all --silent --accept-package-agreements --accept-source-agreements +// winget upgrade --all --silent --accept-package-agreements --accept-source-agreements --disable-interactivity func (w *wingetApplier) Upgrade(ctx context.Context) (string, error) { return runCapped(ctx, wingetCommand(ctx, "upgrade", "--all", "--silent", "--accept-package-agreements", "--accept-source-agreements", + "--disable-interactivity", )) } From 3fe21a0f2970473c8a9d53209fa07f5cc321e64d Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 14:44:47 +0800 Subject: [PATCH 5/7] companion: fix "Update companion" self-update never streaming The companion binary reported the self-update-of-itself action's result *before* running Apply at all (Linux), specifically to guarantee an outcome got recorded even though install.sh's own systemctl restart kills this very process partway through. That report ends the action's output stream (OutputHub.End fires on the first /companion/result, and Publish becomes a no-op afterward for that action ID) -- so the live pane always closed itself off before install.sh had produced a single line, on every self-update-of-companion, regardless of how long the process actually survived first. Apply now always runs first, synchronously, exactly like every other action -- its real output streams normally for however long the process survives. The existing (and separately confirmed-live) rule that a failure coinciding with ctx already being canceled is spurious (the process is mid-restart, not actually broken) now substitutes the optimistic message only in that specific case, after the fact, instead of assuming it unconditionally upfront. This also let the Windows-only branch collapse away entirely: Apply already returns a real result there (no restart of this process happens on that platform at all), so both platforms now share one code path. --- cmd/update-detector-companion/main.go | 62 +++++++++++++-------------- 1 file changed, 31 insertions(+), 31 deletions(-) diff --git a/cmd/update-detector-companion/main.go b/cmd/update-detector-companion/main.go index 3b62d88..b62df20 100644 --- a/cmd/update-detector-companion/main.go +++ b/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 } From 0cfea9dbbf618be2c655e6181ec20d81af4763cc Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 17:10:32 +0800 Subject: [PATCH 6/7] docs: correct agent/apply framing, clarify image layout, drop winget README's intro said the agent "never applies" updates -- true of the agent process itself, but no longer true of the project as a whole now that the companion can apply on trigger. Rewrote the intro to say that plainly, and to state clearly that the agent (update-detector) and the aggregator (update-aggregator) are two separate Docker images, not one -- the old "ships as a single Docker image" line was ambiguous now that the paragraph covers all three components. Also: winget is not supported by this project (confirmed) -- it never runs under install.bat's default LocalSystem service account in the first place. Reframed every winget mention in docs/reference.md's Platform limitations from "optional/supplementary signal" to "not supported, here's why," and updated both docs to reflect that detection, install.bat, Force recheck streaming, and companion self-update have now been confirmed against a real Windows host this session. --- README.md | 18 ++++--- docs/reference.md | 132 +++++++++++++++++++++------------------------- 2 files changed, 73 insertions(+), 77 deletions(-) diff --git a/README.md b/README.md index e17d412..bbbac63 100644 --- a/README.md +++ b/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 diff --git a/docs/reference.md b/docs/reference.md index 46be60d..d659207 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -98,55 +98,45 @@ Monitoring the actual Windows or macOS OS needs a native, non-container agent — the checker is designed as an interface specifically so that's a new implementation, not a rewrite (see `docs/plugin-architecture-plan.md`). macOS is still planned; Windows has -an experimental detection-only implementation (`internal/checker/windows`): - -- **Windows Update (primary signal)**: queries the Windows Update Agent - API — the same COM interface (`Microsoft.Update.Session` / +an experimental implementation (`internal/checker/windows`), with +detection, `install.bat`, and companion apply/self-update now confirmed +against a real Windows host. **winget is not supported** — see the +dedicated bullet below for why — Windows Update is the only signal this +project relies on for Windows. + +- **Windows Update (primary, and only supported, signal)**: queries the + Windows Update Agent API — the same COM interface + (`Microsoft.Update.Session` / `CreateUpdateSearcher().Search("IsInstalled=0 and IsHidden=0")`) the - Settings app's own "Check for updates" ultimately goes through — - via a `powershell -Command` one-liner emitting JSON, parsed in Go + Settings app's own "Check for updates" ultimately goes through — via a + `powershell -Command` one-liner emitting JSON, parsed in Go (`internal/checker/windows/windowsupdate.go`/`windowsupdate_parse.go`). This carries a real **MSRC severity rating** per update (`Critical`/ `Important`/`Moderate`/`Low`, or empty for a non-security update) — - counted as security when non-empty, a genuine signal winget has no - equivalent of at all. Expected (not yet confirmed live) to work under - `LocalSystem` without the account workaround winget needs below, since - the Windows Update service is system-level, not tied to a specific - user's own package registration. -- **Apply, via the companion (⚠️ unverified, actually modifies the - system)**: `internal/companion/applier_windows.go` extends the same - Windows Update Agent API to *install* updates, not just detect them — - a checked "Apply selected" item whose name carries a `(KBnnnnnnn)` - marker (every Windows Update title does) is downloaded and installed - via `IUpdateDownloader.Download()`/`IUpdateInstaller.Install()`; - anything else falls back to winget (`winget upgrade --id `), the - same optional/supplementary role it already has for detection. "Upgrade + counted as security when non-empty. Expected (not yet separately + confirmed live) to work under `LocalSystem`, since the Windows Update + service is system-level, not tied to a specific user's own package + registration the way winget is (see below). +- **Apply, via the companion**: `internal/companion/applier_windows.go` + extends the same Windows Update Agent API to *install* updates, not + just detect them — a checked "Apply selected" item whose name carries + a `(KBnnnnnnn)` marker (every Windows Update title does) is downloaded + and installed via + `IUpdateDownloader.Download()`/`IUpdateInstaller.Install()`. "Upgrade all"/"Full upgrade all" install every currently pending Windows Update - (no dist-upgrade/upgrade distinction, same real semantic gap winget's - own `FullUpgrade` already has). **This is the least-tested code path - in this entire Windows implementation** — every other piece here is - read-only detection; this one actually installs updates and can - require a reboot to take effect. Tested only via fixture tests of the - KB-vs-winget name-splitting logic, never against a real Windows Update - install. If you try this, start with a single low-stakes update via - "Apply selected," not "Upgrade all," and watch the live output pane. -- **Packages via winget (optional, supplementary)**: shells out to - `winget upgrade`, parsing its table output the same best-effort way the - Debian checker parses `apt-get -s dist-upgrade`'s text output. Covers - separately-managed packages winget itself tracks — the same - relationship apt has to Ubuntu/Debian's own release upgrades — merged - into the same package list Windows Update above populates. **No - security/severity signal exists in winget at all** (unlike apt's - `-security` pocket or Windows Update's own MSRC ratings above) — every - winget-sourced upgrade reports `security: false`. Winget's own table - format has changed across App Installer versions, and winget itself - may be entirely absent on locked-down or Server Windows machines, or - (see below) simply not runnable from the account the agent/companion - happens to run as; none of that is treated as an error - since winget is optional here — Windows Update above is what actually - matters, and a winget failure only ever means missing out on its - supplementary package list, not degraded detection overall. Any other - winget failure (bad output, a real winget error) is still surfaced. + (no dist-upgrade/upgrade distinction). This actually installs updates + and can require a reboot to take effect — start with a single + low-stakes update via "Apply selected," not "Upgrade all," and watch + the live output pane. +- **Packages via winget: not supported.** The checker/applier code for + it still exists (`winget upgrade`, parsed the same best-effort way the + Debian checker parses `apt-get -s dist-upgrade`'s text output) but + don't rely on it — see the next bullet for why it doesn't work under + this project's own default install, and note that even where it does + run, winget has **no security/severity signal at all** (unlike apt's + `-security` pocket or Windows Update's own MSRC ratings above); every + winget-sourced upgrade would report `security: false`, and its table + output format has changed across App Installer versions. - **Reboot-required**: reads three well-known `HKLM` registry keys — no admin privilege or `winget`/other exec needed, the most reliable part of this checker. No OS-upgrade detection at all in v1 (same @@ -161,39 +151,39 @@ an experimental detection-only implementation (`internal/checker/windows`): around to grant an exception. Config is stored in each service's own registry `Environment` value (`REG_MULTI_SZ`), the native equivalent of systemd's `EnvironmentFile=`. -- **winget effectively never runs under `install.bat`, by design**: every - service `install.bat` creates defaults to running as `LocalSystem`, - under which `winget` simply doesn't exist — `winget.exe` is an App - Execution Alias registered per-*user* (it lives under that user's own +- **Why winget isn't supported**: every service `install.bat` creates + defaults to running as `LocalSystem`, under which `winget` simply + doesn't exist — `winget.exe` is an App Execution Alias registered + per-*user* (it lives under that user's own `AppData\Local\Microsoft\WindowsApps`, on *that user's* `PATH` only), and `SYSTEM` has no such registration at all, confirmed live as `exec.LookPath("winget")` failing outright with "executable file not found in %PATH%" even though `winget` works fine interactively. `install.bat` used to offer to reconfigure a service's logon account to work around this; that prompt/offer has been removed (the code - behind it is still there, just unreferenced) — since winget is only - a supplementary signal (Windows Update above is the real one, and - doesn't have this problem), the account-switching complexity wasn't - worth it, and the installer no longer prompts for or offers it at - all. The winget detection code itself still works if you manually - reconfigure a service to run as a real account - (`sc config obj= ".\" password= ""`, plus - granting it "Log on as a service" via `secpol.msc` if needed) — it's - just not something the installer does for you anymore. -- Also untested against a real Windows machine end-to-end for most of - this: only fixture-based parsing tests and a hosted CI runner (no - `winget`/real registry/real Windows Service state to exercise) have - exercised most of this so far. The Windows Service Control Protocol - fix and the winget account/PATH issue above were both found and fixed - from one real, live install — everything else here carries the same - unverified caveat until it gets the same live exercise. -- **Roadmap, not started**: Windows Update above covers the OS itself, - but this checker's package-manager story is meant to generalize past - winget too — Scoop and Chocolatey as alternative/additional Windows - package sources, a Homebrew-based macOS checker (see the macOS row - above), and Docker image update detection on Linux (tag/digest drift, - a different kind of "update" than an OS package manager reports) are - all intended future checker plugins, none of them started. + behind it is still there, just unreferenced) — since Windows Update + above is the real, actually-supported signal and doesn't have this + problem, the account-switching complexity wasn't worth carrying, and + the installer no longer prompts for or offers it at all. The winget + code itself still runs if you manually reconfigure a service to use a + real account (`sc config obj= ".\" password= + ""`, plus granting it "Log on as a service" via + `secpol.msc`), but that's not something this project sets up for you + or recommends — treat any winget-sourced result as unsupported even + then. +- Confirmed against a real Windows host this way: `install.bat` + install/uninstall, the Windows Service Control Protocol fix, Force + recheck's live-streaming output (verbose and narration), and companion + self-update. Still only fixture-tested, not yet separately confirmed + live: a real Windows Update KB install via "Apply selected"/"Upgrade + all". +- **Roadmap, not started**: Windows Update above covers the OS itself; + a genuinely supported Windows package-manager signal (Scoop and + Chocolatey are candidates, since winget isn't a viable one — see + above), a Homebrew-based macOS checker (see the macOS row above), and + Docker image update detection on Linux (tag/digest drift, a different + kind of "update" than an OS package manager reports) are all intended + future checker plugins, none of them started. On WSL2 specifically, `docker` being on `PATH` doesn't necessarily mean there's a real engine running inside the distro at all — see From aadbfd97cb9c273f01c6ee2f810155b6b70720fb Mon Sep 17 00:00:00 2001 From: Winarto Date: Fri, 21 Aug 2026 20:38:51 +0800 Subject: [PATCH 7/7] fix: stream the swap phase of a Windows companion self-update Windows companion self-update is two actions, not one: the companion stages a new binary (ActionSelfUpdate), then the agent stops the service/swaps the binary/restarts it (ActionCompleteCompanionSwap, auto-pushed on a staged success). CompleteCompanionSwap already called emitFromContext throughout (stopping/swapping/starting narration) -- ready to stream -- but its caller in cmd/update-detector/main.go never attached a sink to its context at all, so there was nothing to stream to regardless. Wired it up with the same sink/StreamOutput pattern already used for recheck and apply. That alone wasn't enough, though: the browser's live-output pane closes itself the moment it sees "done" for the first (staged) action, before the swap action -- now streaming correctly -- even starts, so the second phase would still never be seen without a manual page reload. OutputHub's "done" event now carries whether the underlying result was staged, and the admin page's JS re-subscribes to keep watching instead of switching to version-polling when it sees that. --- cmd/update-detector/main.go | 19 ++++++++++- internal/aggregator/output.go | 16 ++++++++-- internal/aggregator/output_test.go | 32 ++++++++++++++++--- internal/aggregator/server.go | 16 ++++++---- internal/aggregator/server_test.go | 51 ++++++++++++++++++++++++++++++ internal/aggregator/templates.go | 24 ++++++++++++-- 6 files changed, 139 insertions(+), 19 deletions(-) diff --git a/cmd/update-detector/main.go b/cmd/update-detector/main.go index 2913e1d..1b1551e 100644 --- a/cmd/update-detector/main.go +++ b/cmd/update-detector/main.go @@ -260,12 +260,29 @@ func run(ctx context.Context) error { sink.Close() cancelStream() case aggregator.ActionCompleteCompanionSwap: - result := companion.CompleteCompanionSwap(ctx, action) + // 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) } diff --git a/internal/aggregator/output.go b/internal/aggregator/output.go index ff81cf4..bfd074e 100644 --- a/internal/aggregator/output.go +++ b/internal/aggregator/output.go @@ -26,6 +26,14 @@ type outputEvent struct { Kind outputEventKind ActionID string Line string // only set when Kind == EventLine + // Staged is only meaningful when Kind == EventDone: true means this + // "done" is a companion self-update that staged a new binary but + // couldn't restart itself (Windows) -- a follow-up + // ActionCompleteCompanionSwap is about to run on the agent for the + // same host (see handleCompanionResult), so a subscriber shouldn't + // treat this as the action's true end. See openLiveOutput's own + // handling in templates.go. + Staged bool } // OutputHub fans a companion's (or agent's) live action output out to @@ -100,8 +108,10 @@ func (h *OutputHub) Publish(agentID, actionID, line string) { // success's clean stream-close must never show as "disconnected" after // handleCompanionResult already recorded "done", regardless of which of // the two HTTP requests happens to finish first). Clears the backlog too -// -- once an action is over there's nothing left to replay for it. -func (h *OutputHub) End(agentID, actionID string, kind outputEventKind) { +// -- once an action is over there's nothing left to replay for it. staged +// is only meaningful for kind == EventDone (see outputEvent.Staged); +// pass false for EventDisconnected. +func (h *OutputHub) End(agentID, actionID string, kind outputEventKind, staged bool) { h.mu.Lock() defer h.mu.Unlock() if h.active[agentID] != actionID { @@ -109,7 +119,7 @@ func (h *OutputHub) End(agentID, actionID string, kind outputEventKind) { } delete(h.active, agentID) delete(h.backlog, agentID) - h.broadcast(agentID, outputEvent{Kind: kind, ActionID: actionID}) + h.broadcast(agentID, outputEvent{Kind: kind, ActionID: actionID, Staged: staged}) } // broadcast must be called with h.mu held. diff --git a/internal/aggregator/output_test.go b/internal/aggregator/output_test.go index 1594f32..a4c2121 100644 --- a/internal/aggregator/output_test.go +++ b/internal/aggregator/output_test.go @@ -49,7 +49,7 @@ func TestOutputHubEndDoneThenLateDisconnectedIsNoop(t *testing.T) { _, ch, cancel := h.Subscribe("a1") defer cancel() - h.End("a1", "act1", EventDone) + h.End("a1", "act1", EventDone, false) select { case ev := <-ch: if ev.Kind != EventDone { @@ -59,7 +59,7 @@ func TestOutputHubEndDoneThenLateDisconnectedIsNoop(t *testing.T) { t.Fatal("timed out waiting for done event") } - h.End("a1", "act1", EventDisconnected) + h.End("a1", "act1", EventDisconnected, false) select { case ev := <-ch: t.Fatalf("expected no further event after done already fired, got %#v", ev) @@ -67,13 +67,35 @@ func TestOutputHubEndDoneThenLateDisconnectedIsNoop(t *testing.T) { } } +// TestOutputHubEndPropagatesStaged is the regression test for a Windows +// companion self-update never showing its second phase (stop/swap/start +// the service) live: the browser's "done" handler needs to know this +// first "done" was a staged intermediate result, not the real end, so it +// can keep watching instead of switching to version-polling. +func TestOutputHubEndPropagatesStaged(t *testing.T) { + h := NewOutputHub() + h.Begin("a1", "act1") + _, ch, cancel := h.Subscribe("a1") + defer cancel() + + h.End("a1", "act1", EventDone, true) + select { + case ev := <-ch: + if ev.Kind != EventDone || !ev.Staged { + t.Fatalf("got %#v, want a Staged EventDone", ev) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for done event") + } +} + func TestOutputHubEndDisconnectedForRestartCase(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") _, ch, cancel := h.Subscribe("a1") defer cancel() - h.End("a1", "act1", EventDisconnected) + h.End("a1", "act1", EventDisconnected, false) select { case ev := <-ch: if ev.Kind != EventDisconnected { @@ -141,7 +163,7 @@ func TestOutputHubEndClearsBacklog(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") h.Publish("a1", "act1", "one") - h.End("a1", "act1", EventDone) + h.End("a1", "act1", EventDone, false) backlog, _, cancel := h.Subscribe("a1") defer cancel() @@ -157,7 +179,7 @@ func TestOutputHubBeginResetsBacklogForNewAction(t *testing.T) { h := NewOutputHub() h.Begin("a1", "act1") h.Publish("a1", "act1", "from act1") - h.End("a1", "act1", EventDone) + h.End("a1", "act1", EventDone, false) h.Begin("a1", "act2") backlog, _, cancel := h.Subscribe("a1") diff --git a/internal/aggregator/server.go b/internal/aggregator/server.go index 49353db..5504824 100644 --- a/internal/aggregator/server.go +++ b/internal/aggregator/server.go @@ -322,7 +322,7 @@ func (s *Server) handleCompanionResult(w http.ResponseWriter, r *http.Request) { CompletedAt: time.Now(), Staged: req.Staged, }) - s.outputHub.End(rec.ID, req.ActionID, EventDone) + s.outputHub.End(rec.ID, req.ActionID, EventDone, req.Staged) // When the companion reports a staged self-update (downloaded .exe.new // but can't swap because stopping itself would kill the process), @@ -411,7 +411,7 @@ func (s *Server) handleCompanionOutput(w http.ResponseWriter, r *http.Request) { // arrives, so an old companion that predates output streaming // entirely still resolves to a correct "done" via // handleCompanionResult instead of a live pane that never closes. - defer s.outputHub.End(rec.ID, actionID, EventDisconnected) + defer s.outputHub.End(rec.ID, actionID, EventDisconnected, false) scanner := bufio.NewScanner(r.Body) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) @@ -475,7 +475,7 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, // caught up first instead of just missing everything before it // (re)connected. for _, line := range backlog { - if err := writeOutputEvent(w, string(EventLine), id, line); err != nil { + if err := writeOutputEvent(w, string(EventLine), id, line, false); err != nil { return } } @@ -497,7 +497,7 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, } flusher.Flush() case event := <-ch: - if err := writeOutputEvent(w, string(event.Kind), event.ActionID, event.Line); err != nil { + if err := writeOutputEvent(w, string(event.Kind), event.ActionID, event.Line, event.Staged); err != nil { return } flusher.Flush() @@ -509,12 +509,14 @@ func (s *Server) handleAdminOutputStream(w http.ResponseWriter, r *http.Request, // EventSource listener expects (see openLiveOutput in templates.go). // Shared by handleAdminOutputStream's backlog replay and its live-forward // loop so a replayed line is byte-identical to a live one -- the browser -// must not be able to tell the difference. -func writeOutputEvent(w io.Writer, kind, actionID, line string) error { +// must not be able to tell the difference. staged is only meaningful on +// a "done" event -- see outputEvent.Staged's own doc comment. +func writeOutputEvent(w io.Writer, kind, actionID, line string, staged bool) error { payload, err := json.Marshal(struct { ActionID string `json:"action_id"` Line string `json:"line,omitempty"` - }{ActionID: actionID, Line: line}) + Staged bool `json:"staged,omitempty"` + }{ActionID: actionID, Line: line, Staged: staged}) if err != nil { return nil // malformed payload is skipped, not fatal to the stream } diff --git a/internal/aggregator/server_test.go b/internal/aggregator/server_test.go index 2facb50..e2b046e 100644 --- a/internal/aggregator/server_test.go +++ b/internal/aggregator/server_test.go @@ -1661,6 +1661,57 @@ func TestHandleCompanionResultStagedAutoPushesSwapAction(t *testing.T) { } } +// TestHandleCompanionResultStagedPropagatesToOutputStream is the +// regression test for a Windows companion self-update's second phase +// (the actual stop/swap/start-the-service work, done by +// ActionCompleteCompanionSwap on the agent) never appearing in the live +// output pane: the browser needs to know a "done" event was for a staged +// intermediate result, not the real end, so it can keep watching instead +// of switching to version-polling immediately. +func TestHandleCompanionResultStagedPropagatesToOutputStream(t *testing.T) { + s, reg := newTestServer(t) + approvedAgent(t, s, reg, "a1", "web01", "tok") + res := s.hub.Connect("a1", KindCompanion, "v0.0.0-test") + defer s.hub.Disconnect("a1", res.Ch) + if err := s.hub.Push("a1", Action{ID: "act1", Type: ActionSelfUpdate, Component: "companion", CreatedAt: time.Now()}); err != nil { + t.Fatalf("push failed: %v", err) + } + s.outputHub.Begin("a1", "act1") + + httpSrv := httptest.NewServer(s.Handler()) + defer httpSrv.Close() + + streamReq, err := http.NewRequest(http.MethodGet, httpSrv.URL+"/admin/agents/a1/output/stream", nil) + if err != nil { + t.Fatal(err) + } + streamResp, err := (&http.Client{Timeout: 5 * time.Second}).Do(streamReq) + if err != nil { + t.Fatal(err) + } + defer streamResp.Body.Close() + + rec := doJSON(t, s, http.MethodPost, "/companion/result", companionResultRequest{ + ActionID: "act1", + Success: true, + Message: "companion update staged to .exe.new", + Staged: true, + }, map[string]string{"X-Agent-ID": "a1", "Authorization": "Bearer tok"}) + if rec.Code != http.StatusOK { + t.Fatalf("got status %d, body %s", rec.Code, rec.Body.String()) + } + + buf := make([]byte, 4096) + n, err := streamResp.Body.Read(buf) + if err != nil && err != io.EOF { + t.Fatal(err) + } + got := string(buf[:n]) + if !strings.Contains(got, "event: done") || !strings.Contains(got, `"staged":true`) { + t.Fatalf("expected a done event with staged:true in SSE body, got: %q", got) + } +} + // TestHandleCompanionResultNonStagedDoesNotPushSwapAction verifies that // a normal (non-staged) result does NOT trigger an auto-push of // ActionCompleteCompanionSwap. diff --git a/internal/aggregator/templates.go b/internal/aggregator/templates.go index ea81c95..336b021 100644 --- a/internal/aggregator/templates.go +++ b/internal/aggregator/templates.go @@ -649,11 +649,15 @@ const adminTemplateSrc = ` return secret; } - function openLiveOutput(id, selfUpdateExpect) { + // resume=true keeps the pane's existing content instead of clearing it -- + // used when re-subscribing after a staged companion self-update's first + // phase reports "done" but a follow-up ActionCompleteCompanionSwap is + // about to stream on the same agent (see the 'done' handler below). + function openLiveOutput(id, selfUpdateExpect, resume) { const pane = document.getElementById('output-' + id); if (!pane) return null; pane.style.display = 'block'; - pane.textContent = ''; + if (!resume) pane.textContent = ''; const baselinePromise = selfUpdateExpect ? null : fetchAgentVersionInfo(id).then(d => d ? d.last_seen : ''); const es = new EventSource('/admin/agents/' + id + '/output/stream'); es.addEventListener('line', (e) => { @@ -661,8 +665,22 @@ const adminTemplateSrc = ` pane.textContent += data.line + '\n'; pane.scrollTop = pane.scrollHeight; }); - es.addEventListener('done', async () => { + es.addEventListener('done', async (e) => { es.close(); + const data = JSON.parse(e.data); + if (selfUpdateExpect && data.staged) { + // Windows-only case: the companion staged a new binary but + // can't restart itself, so the aggregator already pushed + // ActionCompleteCompanionSwap to the agent for the same host + // (by the time this SSE event reached the browser, that push + // -- and its own outputHub.Begin -- already happened + // server-side, synchronously, in the same request that ended + // this stream). Re-subscribe to keep watching that instead of + // treating this as the real end. + pane.textContent += '--- staged -- continuing to the restart step ---\n'; + openLiveOutput(id, selfUpdateExpect, true); + return; + } if (selfUpdateExpect) { pane.textContent += '--- done -- waiting for ' + selfUpdateExpect.component + ' to report version ' + selfUpdateExpect.targetVersion + ' ---\n';