windows: fix live streaming for Update-all and companion staging

install.bat

@@ -166,7 +166,16 @@ for /l %%i in (1,1,5) do (
     if not errorlevel 1 (
       set "move_ok=1"
     ) else (
-      timeout /t 1 /nobreak >nul
+      rem timeout.exe refuses to run at all without a real console --
+      rem "ERROR: Input redirection is not supported, exiting the
+      rem process immediately." -- confirmed live when this runs
+      rem non-interactively (a Windows Service, e.g. the companion
+      rem re-invoking this script for a self-update has no console
+      rem at all). That failure was silent here (redirected to nul)
+      rem but meant every retry fired back-to-back with zero actual
+      rem delay, defeating this whole workaround. ping needs no
+      rem console at all; -n 2 against localhost gives ~1s.
+      ping -n 2 127.0.0.1 >nul
     )
   )
 )
@@ -227,7 +236,12 @@ if not errorlevel 1 goto :eof
 sc stop "%~1" >nul 2>&1
 set "tries=0"
 :stop_wait_loop
-timeout /t 1 /nobreak >nul
+rem timeout.exe needs a real console (see download_binary's own
+rem comment) -- without one this whole 30-iteration wait would blow
+rem through in milliseconds instead of ~30s, giving up on a service
+rem that's genuinely still stopping and force-killing it far too
+rem early. ping needs no console at all.
+ping -n 2 127.0.0.1 >nul
 sc query "%~1" | find "STOPPED" >nul
 if not errorlevel 1 goto :eof
 set /a tries+=1
@@ -702,7 +716,9 @@ if errorlevel 1 (
 ) else (
   echo install.bat: removing update-detector ^(agent^)...
   sc stop update-detector >nul 2>&1
-  timeout /t 2 /nobreak >nul
+  rem ping, not timeout -- see download_binary's own comment on why
+  rem timeout.exe can't be used here (needs a real console).
+  ping -n 3 127.0.0.1 >nul
   sc delete update-detector >nul 2>&1
   del /f /q "%BIN_DIR%\update-detector.exe" >nul 2>&1
   echo install.bat: removing %ProgramData%\update-detector ^(includes this agent's aggregator identity^)
@@ -722,7 +738,7 @@ if errorlevel 1 (
 ) else (
   echo install.bat: removing update-aggregator...
   sc stop update-aggregator >nul 2>&1
-  timeout /t 2 /nobreak >nul
+  ping -n 3 127.0.0.1 >nul
   sc delete update-aggregator >nul 2>&1
   del /f /q "%BIN_DIR%\update-aggregator.exe" >nul 2>&1
   echo install.bat: removing %ProgramData%\update-aggregator ^(includes the fleet registry -- all enrolled/approved hosts^)
@@ -741,7 +757,7 @@ rem Companion is always native, needs real administrator rights to run
 rem winget -- no Docker case to check here, same as install.sh's own
 rem uninstall_companion.
 sc stop update-detector-companion >nul 2>&1
-timeout /t 2 /nobreak >nul
+ping -n 3 127.0.0.1 >nul
 sc delete update-detector-companion >nul 2>&1
 del /f /q "%BIN_DIR%\update-detector-companion.exe" >nul 2>&1
 del /f /q "%CACHED_INSTALL_BAT%" >nul 2>&1

internal/aggregator/companion.go

@@ -214,14 +214,30 @@ type ConnectResult struct {
 }
 
 // Connect registers agentID as having a live stream, with the following
-// arbitration (companion always outranks agent -- see ActionType.requiresCompanion):
+// arbitration (companion always outranks agent for the main slot -- see
+// ActionType.requiresCompanion):
 //   - no existing entry: accept unconditionally, either kind.
 //   - existing kind == new kind (a reconnect, e.g. after a restart):
 //     replace as before, no signal fired -- this is not a priority change.
-//   - existing=agent, new=companion: companion preempts. The old entry's
-//     superseded channel is closed so its holder notices and tears down.
-//   - existing=companion, new=agent: rejected outright; the existing
-//     companion stream is left untouched.
+//   - existing=agent, new=companion: companion preempts the main slot.
+//     The old entry's superseded channel is closed so its holder notices
+//     and tears down.
+//   - existing=companion, new=agent: the companion keeps the main slot,
+//     but the agent is still accepted into agentStreams (see below), so
+//     agent-only actions (recheck, ActionCompleteCompanionSwap) can still
+//     reach it even while a companion is connected.
+//
+// Whenever this replaces an existing entry (main slot or agentStreams),
+// it also clears that agentID's corresponding pending/agentPending marker
+// -- confirmed live as a real stuck-forever bug otherwise: if the old
+// connection died uncleanly (crash, network drop, anything that never
+// delivers a clean TCP close), its own deferred Disconnect either never
+// runs at all, or runs too late and no-ops against the channel it's
+// comparing against (already replaced by the time it gets there) -- so
+// without clearing it here too, an action that will now never resolve
+// (its executor is gone) blocks every future Push for that agent with
+// ErrActionInFlight/"agent stream busy", forever, until the whole
+// aggregator process restarts.
 //
 // companionVersion is only recorded when kind is KindCompanion -- it has
 // no meaning for an agent-only connection, and must not clobber the last
@@ -243,6 +259,7 @@ func (h *CompanionHub) Connect(agentID string, kind ClientKind, companionVersion
 		// Close any previous agent stream for this ID
 		if prev, ok := h.agentStreams[agentID]; ok {
 			close(prev.superseded)
+			delete(h.agentPending, agentID)
 		}
 		h.agentStreams[agentID] = agentEntry
 		return ConnectResult{Accepted: true, Ch: agentEntry.ch, Superseded: agentEntry.superseded}
@@ -253,14 +270,18 @@ func (h *CompanionHub) Connect(agentID string, kind ClientKind, companionVersion
 		kind:       kind,
 		superseded: make(chan struct{}),
 	}
-	if hasExisting && existing.kind != kind {
-		close(existing.superseded)
+	if hasExisting {
+		if existing.kind != kind {
+			close(existing.superseded)
+		}
+		delete(h.pending, agentID)
 	}
 	h.streams[agentID] = entry
 	if kind == KindAgent {
 		// Also track in agentStreams for agent-only action routing
 		if prev, ok := h.agentStreams[agentID]; ok {
 			close(prev.superseded)
+			delete(h.agentPending, agentID)
 		}
 		h.agentStreams[agentID] = entry
 	}

internal/aggregator/companion_test.go

@@ -106,6 +106,57 @@ func TestCompanionHubDisconnectClearsInFlight(t *testing.T) {
 	}
 }
 
+// TestCompanionHubReconnectClearsStalePendingWithoutDisconnect is the
+// regression test for a real stuck-forever bug: if the old connection
+// died uncleanly (crash, network drop -- anything that never delivers a
+// clean TCP close), its own deferred Disconnect either never runs, or
+// runs after Connect already replaced the entry and so no-ops (its
+// cur.ch == ch guard fails). Without Connect itself clearing the stale
+// marker, a fresh reconnect from that same agent would still get
+// ErrActionInFlight forever for an action whose executor no longer
+// exists -- exactly "self-update failed (409): agent already has an
+// action in flight, but nothing happens" observed live.
+func TestCompanionHubReconnectClearsStalePendingWithoutDisconnect(t *testing.T) {
+	h := NewCompanionHub()
+	h.Connect("a1", KindCompanion, "")
+
+	if err := h.Push("a1", Action{ID: "act1", Type: ActionUpgrade}); err != nil {
+		t.Fatalf("push failed: %v", err)
+	}
+
+	// Reconnect *without* ever calling Disconnect for the old connection --
+	// simulates the old one having died uncleanly.
+	h.Connect("a1", KindCompanion, "")
+
+	if err := h.Push("a1", Action{ID: "act2", Type: ActionUpgrade}); err != nil {
+		t.Fatalf("expected push to succeed after an unclean reconnect, got: %v", err)
+	}
+}
+
+// TestCompanionHubAgentReconnectClearsStaleAgentPending is the same
+// regression, for the agentStreams slot (e.g. a stuck recheck or
+// companion-swap action) instead of the main companion slot.
+func TestCompanionHubAgentReconnectClearsStaleAgentPending(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 act1 to be tracked as pending")
+	}
+
+	h.Connect("a1", KindAgent, "")
+
+	if h.IsPending("a1", "act1") {
+		t.Fatal("expected the stale agentPending marker to be cleared by the reconnect")
+	}
+	if err := h.Push("a1", Action{ID: "act2", Type: ActionRecheck}); err != nil {
+		t.Fatalf("expected push to succeed after an unclean reconnect, got: %v", err)
+	}
+}
+
 func TestCompanionHubTracksCompanionVersion(t *testing.T) {
 	h := NewCompanionHub()
 

internal/aggregator/server.go

@@ -232,7 +232,17 @@ func (s *Server) handleCompanionStream(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "superseded: a companion is already connected for this agent", http.StatusConflict)
 		return
 	}
-	defer s.hub.Disconnect(rec.ID, result.Ch)
+	// Connect/Disconnect change what the admin page's "connected"/"offline"
+	// badges show for this agent, independently of any registry mutation
+	// (enroll/report/approve) -- without notifying here too, a reconnect
+	// (e.g. every agent/companion in the fleet, right after an aggregator
+	// restart) was invisible until something else happened to trigger a
+	// reload, or the operator refreshed manually.
+	s.adminHub.Notify()
+	defer func() {
+		s.hub.Disconnect(rec.ID, result.Ch)
+		s.adminHub.Notify()
+	}()
 	// Only meaningful for a real companion (see SetAggregatorPresent) --
 	// an agent-only connection has never run the aggregator-colocation
 	// check at all, so a missing/unparseable header here (including one

internal/aggregator/server_test.go

@@ -344,6 +344,55 @@ func TestHandleCompanionStreamRequiresAuth(t *testing.T) {
 	}
 }
 
+// TestHandleCompanionStreamNotifiesAdminHubOnConnectAndDisconnect is the
+// regression test for the admin page never showing a reconnect live:
+// Connect/Disconnect change the "connected"/"offline" badges independently
+// of any registry mutation (enroll/report/approve), so without this the
+// only existing notify triggers (registry changes) had no reason to fire
+// on their own -- an operator watching the page during, say, a fleet-wide
+// reconnect after an aggregator restart would see nothing update until
+// something else happened to trigger a reload.
+func TestHandleCompanionStreamNotifiesAdminHubOnConnectAndDisconnect(t *testing.T) {
+	s, reg := newTestServer(t)
+	approvedAgent(t, s, reg, "a1", "web01", "tok")
+
+	adminCh, cancel := s.adminHub.Subscribe()
+	defer cancel()
+
+	httpSrv := httptest.NewServer(s.Handler())
+	defer httpSrv.Close()
+
+	req, err := http.NewRequest(http.MethodGet, httpSrv.URL+"/companion/stream", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	req.Header.Set("X-Agent-ID", "a1")
+	req.Header.Set("Authorization", "Bearer tok")
+
+	client := &http.Client{Timeout: 5 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp.StatusCode != http.StatusOK {
+		t.Fatalf("got status %d, want 200", resp.StatusCode)
+	}
+
+	select {
+	case <-adminCh:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for a notify on connect")
+	}
+
+	resp.Body.Close()
+
+	select {
+	case <-adminCh:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for a notify on disconnect")
+	}
+}
+
 func TestHandleCompanionStreamPushesAction(t *testing.T) {
 	s, reg := newTestServer(t)
 	approvedAgent(t, s, reg, "a1", "web01", "tok")

internal/checker/windows/reboot.go

@@ -34,10 +34,12 @@ func keyExists(path string) bool {
 }
 
 // pendingFileRenameOperationsSet reports whether Session Manager's
-// PendingFileRenameOperations value is set and non-empty -- a non-empty
-// REG_MULTI_SZ here means the OS has files staged to be renamed/deleted
-// on next boot, the same signal Windows Update and many installers use
-// to indicate a pending reboot.
+// PendingFileRenameOperations value has any entry that isn't routine
+// noise (see isRoutinePendingRename/anyRealPendingRename in
+// reboot_parse.go) -- a real, non-empty REG_MULTI_SZ entry here means
+// the OS has files staged to be renamed/deleted on next boot, the same
+// signal Windows Update and many installers use to indicate a pending
+// reboot.
 func pendingFileRenameOperationsSet() bool {
 	k, err := registry.OpenKey(registry.LOCAL_MACHINE, `SYSTEM\CurrentControlSet\Control\Session Manager`, registry.QUERY_VALUE)
 	if err != nil {
@@ -48,5 +50,5 @@ func pendingFileRenameOperationsSet() bool {
 	if err != nil {
 		return false
 	}
-	return len(values) > 0
+	return anyRealPendingRename(values)
 }

internal/checker/windows/reboot_parse.go

@@ -0,0 +1,57 @@
+// Deliberately untagged, same reasoning as packages_parse.go: pure
+// string matching with no OS-specific imports, testable on any platform,
+// not just Windows.
+package windows
+
+import "strings"
+
+// pendingRenameNoisePatterns lists (lowercased) path substrings for
+// entries that reappear in PendingFileRenameOperations after nearly
+// every single boot, forever, regardless of whether anything meaningful
+// is actually waiting -- confirmed live on a real host: Windows Gaming
+// Services' own proxy DLL and Microsoft Edge's background auto-updater
+// both re-queue an entry here almost immediately after every reboot. A
+// naive "list is non-empty" check is permanently true on any host with
+// either installed (i.e. nearly all of them) and useless as a signal.
+// Entries matching one of these are ignored; anything else still counts
+// as a real pending change -- when in doubt, this errs toward reporting
+// reboot-required, never toward hiding one.
+var pendingRenameNoisePatterns = []string{
+	`\gamingservicesproxy`,
+	// No trailing backslash -- confirmed live, Edge's updater queues the
+	// bare Temp *folder* itself as one of the pending entries (not just
+	// files nested inside it), which a `...\temp\` pattern would miss
+	// entirely and report as a real pending change.
+	`\microsoft\edge\temp`,
+}
+
+// isRoutinePendingRename reports whether entry (one raw string from
+// PendingFileRenameOperations -- either half of a rename pair, or a
+// delete pair's always-empty second half) matches a known-routine
+// pattern. An empty string is routine by definition: it's never a real
+// path on its own, only ever the "delete" half of a pair whose other
+// half is what actually identifies what's pending.
+func isRoutinePendingRename(entry string) bool {
+	if entry == "" {
+		return true
+	}
+	lower := strings.ToLower(entry)
+	for _, pattern := range pendingRenameNoisePatterns {
+		if strings.Contains(lower, pattern) {
+			return true
+		}
+	}
+	return false
+}
+
+// anyRealPendingRename reports whether entries (the raw string list from
+// PendingFileRenameOperations) contains anything other than routine
+// noise -- see isRoutinePendingRename.
+func anyRealPendingRename(entries []string) bool {
+	for _, v := range entries {
+		if !isRoutinePendingRename(v) {
+			return true
+		}
+	}
+	return false
+}