From 5272debf8d629d69ef7f89a4ee6417f0fd7cedd7 Mon Sep 17 00:00:00 2001 From: Eve Date: Mon, 13 Jul 2026 17:50:06 +0800 Subject: [PATCH] fix(daemon): reconcile profiles after WS reconnect Co-authored-by: multica-agent --- .../internal/daemon/coauthor_enabled_test.go | 2 +- server/internal/daemon/daemon.go | 36 +++++++++------- .../daemon/runtime_profile_drift_test.go | 42 ++++++++++++++++++- 3 files changed, 64 insertions(+), 16 deletions(-) diff --git a/server/internal/daemon/coauthor_enabled_test.go b/server/internal/daemon/coauthor_enabled_test.go index 4d0c920090..bac0bf3f91 100644 --- a/server/internal/daemon/coauthor_enabled_test.go +++ b/server/internal/daemon/coauthor_enabled_test.go @@ -113,7 +113,7 @@ func TestSyncWorkspacesSkipsReposRefreshOnExistingWorkspace(t *testing.T) { t.Fatalf("precondition: expected co-author hook to start enabled") } - if err := d.syncWorkspacesFromAPI(context.Background()); err != nil { + if err := d.syncWorkspacesFromAPI(context.Background(), false); err != nil { t.Fatalf("syncWorkspacesFromAPI: %v", err) } diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index b1ae845599..2f9cf077cb 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -1514,8 +1514,8 @@ func (d *Daemon) refreshWorkspaceRuntimeProfiles(ctx context.Context, workspaceI // The workspaceState pointer is preserved: the workspace itself is still a // valid workspace the user belongs to, just one with no agents on this // daemon for the moment. If the user re-enables a profile or installs a -// built-in agent, the next sync tick's profile-drift detection (or a daemon -// restart) will register it again. +// built-in agent, the profile-change notification or the next daemon WS +// reconnect will register it again. func (d *Daemon) convergeWorkspaceRuntimesToZero(ctx context.Context, workspaceID, profileSig string) error { d.mu.Lock() ws, ok := d.workspaces[workspaceID] @@ -1640,7 +1640,7 @@ const DefaultTokenRenewalInterval = 3 * 24 * time.Hour // register runtimes once one appears. func (d *Daemon) preflightAuth(ctx context.Context) error { d.tryRenewToken(ctx) - return d.syncWorkspacesFromAPI(ctx) + return d.syncWorkspacesFromAPI(ctx, false) } // tokenRenewalLoop keeps the daemon's PAT alive by periodically asking the @@ -1699,10 +1699,10 @@ func (d *Daemon) tryRenewToken(ctx context.Context) { // workspaceSyncLoop periodically fetches the user's workspaces from the API // and registers runtimes for any new ones. A WS connect/reconnect broadcast -// triggers an immediate sync so workspace and runtime changes the server -// applied during the WS gap are picked up sub-second instead of after the -// next 30s tick. Repository bindings and workspace settings refresh on demand -// when a checkout needs them. +// triggers an immediate sync and one runtime-profile reconciliation so changes +// made during the WS gap are picked up sub-second without putting profile +// fetches back on the 30s ticker. Repository bindings and workspace settings +// refresh on demand when a checkout needs them. func (d *Daemon) workspaceSyncLoop(ctx context.Context) { ticker := time.NewTicker(DefaultWorkspaceSyncInterval) defer ticker.Stop() @@ -1712,8 +1712,8 @@ func (d *Daemon) workspaceSyncLoop(ctx context.Context) { reconcileCh = d.reconcile.notify() } - sync := func() { - if err := d.syncWorkspacesFromAPI(ctx); err != nil { + sync := func(reconcileProfiles bool) { + if err := d.syncWorkspacesFromAPI(ctx, reconcileProfiles); err != nil { d.logger.Debug("workspace sync failed", "error", err) } } @@ -1726,17 +1726,20 @@ func (d *Daemon) workspaceSyncLoop(ctx context.Context) { if d.reconcile != nil { reconcileCh = d.reconcile.notify() } - sync() + sync(true) case <-ticker.C: - sync() + sync(false) } } } // syncWorkspacesFromAPI fetches all workspaces the user belongs to and -// registers runtimes for any that aren't already tracked. Workspaces the user -// has left are cleaned up. -func (d *Daemon) syncWorkspacesFromAPI(ctx context.Context) error { +// registers runtimes for any that aren't already tracked. When +// reconcileProfiles is true (after a daemon WS connect/reconnect), tracked +// workspaces reconcile custom runtime profiles once so a change made while the +// WS was unavailable is not lost. The normal 30s ticker passes false and makes +// no runtime-profile requests. Workspaces the user has left are cleaned up. +func (d *Daemon) syncWorkspacesFromAPI(ctx context.Context, reconcileProfiles bool) error { d.reloading.Lock() defer d.reloading.Unlock() @@ -1765,6 +1768,11 @@ func (d *Daemon) syncWorkspacesFromAPI(ctx context.Context) error { var removed int for id, name := range apiIDs { if currentIDs[id] { + if reconcileProfiles { + if err := d.refreshWorkspaceRuntimeProfiles(ctx, id); err != nil { + d.logger.Debug("workspace reconcile: profile refresh failed", "workspace_id", id, "error", err) + } + } // Only intervene further if the workspace lost all of its // runtimes (most commonly because handleRuntimeGone pruned them // and its inline re-register failed). The pointer is not replaced diff --git a/server/internal/daemon/runtime_profile_drift_test.go b/server/internal/daemon/runtime_profile_drift_test.go index 90378a9e72..0339826d80 100644 --- a/server/internal/daemon/runtime_profile_drift_test.go +++ b/server/internal/daemon/runtime_profile_drift_test.go @@ -236,7 +236,7 @@ func TestSyncWorkspacesSkipsRuntimeProfileRefreshOnExistingWorkspace(t *testing. d.workspaces[workspaceID].profileSetSig = profileSetSignature(nil) d.runtimeIndex["rt-1"] = Runtime{ID: "rt-1", Provider: "codex"} - if err := d.syncWorkspacesFromAPI(context.Background()); err != nil { + if err := d.syncWorkspacesFromAPI(context.Background(), false); err != nil { t.Fatalf("syncWorkspacesFromAPI: %v", err) } @@ -245,6 +245,46 @@ func TestSyncWorkspacesSkipsRuntimeProfileRefreshOnExistingWorkspace(t *testing. } } +// TestSyncWorkspacesRefreshesRuntimeProfilesOnReconcile preserves the +// reliability backstop for a profile mutation that happened while the daemon +// WebSocket was disconnected. The reconnect broadcast must make one on-demand +// request for each tracked workspace; otherwise removing the ticker polling +// would leave the daemon permanently stale until it restarted. +func TestSyncWorkspacesRefreshesRuntimeProfilesOnReconcile(t *testing.T) { + t.Parallel() + + const workspaceID = "ws-1" + var profileCalls atomic.Int32 + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/api/workspaces": + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode([]WorkspaceInfo{{ID: workspaceID, Name: "ws"}}) + case "/api/daemon/workspaces/" + workspaceID + "/runtime-profiles": + profileCalls.Add(1) + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(RuntimeProfilesResponse{WorkspaceID: workspaceID}) + default: + http.NotFound(w, r) + } + })) + t.Cleanup(srv.Close) + + d := freshDaemon(srv.URL) + d.workspaces[workspaceID] = newWorkspaceState(workspaceID, []string{"rt-1"}, "", nil, nil) + d.workspaces[workspaceID].profileSetSig = profileSetSignature(nil) + d.runtimeIndex["rt-1"] = Runtime{ID: "rt-1", Provider: "codex"} + + if err := d.syncWorkspacesFromAPI(context.Background(), true); err != nil { + t.Fatalf("syncWorkspacesFromAPI: %v", err) + } + + if got := profileCalls.Load(); got != 1 { + t.Fatalf("reconcile fetched runtime-profiles %d times, want 1", got) + } +} + // TestRefreshWorkspaceRuntimeProfiles_NoDrift_DoesNotReregister verifies the // hot-path: when the server's profile set has not changed since the daemon // last registered the workspace, a duplicate notification must NOT fire a