diff --git a/apps/backend/cmd/kandev/gateway.go b/apps/backend/cmd/kandev/gateway.go index 76ea3492df..1f3256c0b9 100644 --- a/apps/backend/cmd/kandev/gateway.go +++ b/apps/backend/cmd/kandev/gateway.go @@ -25,6 +25,7 @@ import ( sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" userservice "github.com/kandev/kandev/internal/user/service" + ws "github.com/kandev/kandev/pkg/websocket" ) // scriptServiceAdapter adapts the task service to scripts.ScriptService. @@ -44,6 +45,16 @@ func (a *scriptServiceAdapter) GetRepositoryScript(ctx context.Context, id strin }, nil } +// hubModeQuerierAdapter converts the hub's SessionMode enum to lifecycle's +// WorkspacePollMode (same string values) so lifecycle doesn't import websocket. +type hubModeQuerierAdapter struct { + hub *gateways.Hub +} + +func (a *hubModeQuerierAdapter) GetSessionMode(sessionID string) lifecycle.WorkspacePollMode { + return lifecycle.WorkspacePollMode(a.hub.GetSessionMode(sessionID)) +} + // sessionReaderAdapter implements agenthandlers.SessionReader using the task repository. // This allows git handlers to look up session metadata (like base commit SHA) from the database. type sessionReaderAdapter struct { @@ -184,6 +195,34 @@ func provideGateway( gateways.RegisterTaskNotifications(ctx, eventBus, gateway.Hub, log) gateways.RegisterUserNotifications(ctx, eventBus, gateway.Hub, log) + // Route session focus/subscription transitions from the hub into the + // lifecycle manager so it can push poll-mode changes to agentctl. + if lifecycleMgr != nil { + gateway.Hub.AddSessionModeListener(func(sessionID string, mode gateways.SessionMode) { + lifecycleMgr.HandleSessionMode(sessionID, lifecycle.WorkspacePollMode(mode)) + }) + // Expose hub's live mode state to lifecycle so it can query on + // execution-ready and proactively push the right mode even when + // gateway events fired before the execution was registered. + lifecycleMgr.SetSessionModeQuerier(&hubModeQuerierAdapter{hub: gateway.Hub}) + } + + // Broadcast session poll mode transitions to all WS clients. Uses Broadcast + // (not BroadcastToSession) because focus events fire before subscribe events + // on page load, so BroadcastToSession misses the focused-but-not-yet- + // subscribed client. Volume is low (debounced down-transitions) and clients + // filter by session_id in the payload. + gateway.Hub.AddSessionModeListener(func(sessionID string, mode gateways.SessionMode) { + msg, err := ws.NewNotification(ws.ActionSessionPollModeChanged, map[string]interface{}{ + "session_id": sessionID, + "poll_mode": string(mode), + }) + if err != nil { + return + } + gateway.Hub.Broadcast(msg) + }) + notificationSvc := notificationservice.NewService(notificationRepo, taskRepo, gateway.Hub, log) notificationCtrl := notificationcontroller.NewController(notificationSvc) if eventBus != nil { diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index 49805e27fb..33ba18027f 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -69,6 +69,19 @@ func provideOrchestrator( orchestratorSvc.SetTurnService(newTurnServiceAdapter(taskSvc)) + // Publish task.updated when the first session is marked primary so the + // frontend receives primary_session_id for newly created tasks. + orchestratorSvc.SetOnPrimarySessionSet(func(ctx context.Context, taskID, _ string) { + task, err := taskRepo.GetTask(ctx, taskID) + if err != nil { + log.Warn("failed to get task for primary session event", + zap.String("task_id", taskID), + zap.Error(err)) + return + } + taskSvc.PublishTaskUpdated(ctx, task) + }) + // Wire workflow step getter for prompt building if workflowSvc != nil { orchestratorSvc.SetWorkflowStepGetter(&orchestratorWorkflowStepGetterAdapter{svc: workflowSvc}) diff --git a/apps/backend/internal/agent/lifecycle/manager.go b/apps/backend/internal/agent/lifecycle/manager.go index a8631b4ba3..65ea57e232 100644 --- a/apps/backend/internal/agent/lifecycle/manager.go +++ b/apps/backend/internal/agent/lifecycle/manager.go @@ -82,6 +82,10 @@ type Manager struct { remoteStatusBySession map[string]*RemoteStatus stopCh chan struct{} wg sync.WaitGroup + + // pollAggregator routes hub session-mode events to agentctl. See + // manager_subscription.go. + pollAggregator *workspacePollAggregator } // NewManager creates a new lifecycle manager. @@ -160,6 +164,8 @@ func NewManager( // Set session manager dependencies for full orchestration sessionManager.SetDependencies(eventPublisher, mgr.streamManager, executionStore, historyManager) + mgr.pollAggregator = newWorkspacePollAggregator(mgr) + if executorRegistry != nil { mgr.logger.Info("initialized with runtimes", zap.Int("count", len(executorRegistry.List()))) } @@ -167,6 +173,27 @@ func NewManager( return mgr } +// HandleSessionMode routes a session-level mode transition (from the gateway +// hub) into the per-workspace aggregator, which pushes the resulting +// workspace-effective mode to agentctl. See manager_subscription.go. +func (m *Manager) HandleSessionMode(sessionID string, mode WorkspacePollMode) { + if m.pollAggregator == nil { + return + } + m.pollAggregator.HandleSessionMode(sessionID, mode) +} + +// flushCachedPollMode pushes any session mode the gateway cached before this +// execution was ready. Closes the pre-execution-focus race where the frontend +// sent session.focus during execution startup and the cached mode never +// reached agentctl. No-op when nothing was cached. +func (m *Manager) flushCachedPollMode(sessionID string) { + if m.pollAggregator == nil { + return + } + m.pollAggregator.FlushSessionMode(sessionID) +} + // SetWorktreeManager sets the worktree manager for Git worktree isolation. // // This must be called before launching agents if Git worktree support is enabled in the runtime. diff --git a/apps/backend/internal/agent/lifecycle/manager_startup.go b/apps/backend/internal/agent/lifecycle/manager_startup.go index 4b0c527284..7a75b7a841 100644 --- a/apps/backend/internal/agent/lifecycle/manager_startup.go +++ b/apps/backend/internal/agent/lifecycle/manager_startup.go @@ -237,6 +237,11 @@ func (m *Manager) waitForAgentctlReady(execution *AgentExecution) { zap.String("execution_id", execution.ID), zap.Duration("duration", elapsed)) } + // Flush any session mode the gateway cached before this execution was + // ready (pre-execution-focus race). Without this, agentctl stays in its + // default slow poll mode even though the frontend already sent focus, + // and git state updates take up to 30s to reach the UI. + m.flushCachedPollMode(execution.SessionID) // Use the timeout context for event publishing instead of a fresh Background context m.eventPublisher.PublishAgentctlEvent(ctx, events.AgentctlReady, execution, "") } diff --git a/apps/backend/internal/agent/lifecycle/manager_subscription.go b/apps/backend/internal/agent/lifecycle/manager_subscription.go new file mode 100644 index 0000000000..65d923f45e --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/manager_subscription.go @@ -0,0 +1,283 @@ +package lifecycle + +import ( + "context" + "sync" + "time" + + "go.uber.org/zap" +) + +// WorkspacePollMode mirrors process.PollMode for the lifecycle layer. Defined +// here as a string to avoid importing the agentctl process package (and the +// surface area that comes with it). Values must stay aligned with +// process.PollMode{Fast,Slow,Paused}. +type WorkspacePollMode string + +const ( + WorkspacePollModeFast WorkspacePollMode = "fast" + WorkspacePollModeSlow WorkspacePollMode = "slow" + WorkspacePollModePaused WorkspacePollMode = "paused" +) + +// rank orders modes from coldest to hottest so we can pick the max across +// sessions sharing a workspace. paused < slow < fast. +func (m WorkspacePollMode) rank() int { + switch m { + case WorkspacePollModeFast: + return 2 + case WorkspacePollModeSlow: + return 1 + default: + return 0 + } +} + +// pushPollModeTimeout caps how long a single push to agentctl is allowed to +// block before we move on. Keeps the listener responsive even if a single +// agentctl is unreachable. +const pushPollModeTimeout = 5 * time.Second + +// SessionModeQuerier is implemented by the gateway hub. It lets the lifecycle +// manager look up the current effective mode for a session, so when an +// execution becomes ready we can proactively push the right mode even when +// the gateway's subscribe/focus events fired before the execution existed. +type SessionModeQuerier interface { + GetSessionMode(sessionID string) WorkspacePollMode +} + +// workspacePollAggregator tracks the per-session mode contributions for each +// workspace and pushes the effective workspace mode to agentctl when it changes. +type workspacePollAggregator struct { + mgr *Manager + hubQry SessionModeQuerier // nil until gateway wires it in + mu sync.Mutex + // sessionModes maps sessionID -> latest mode reported by the gateway. + sessionModes map[string]WorkspacePollMode + // lastPushed maps workspacePath -> last mode we sent to agentctl. Used to + // suppress duplicate pushes when workspace-level mode is unchanged. + lastPushed map[string]WorkspacePollMode + // workspaceSessions is a reverse index: workspacePath -> set of sessionIDs + // known to belong to that workspace. Lets recordAndCompute iterate only + // sessions in the affected workspace instead of scanning all sessionModes. + workspaceSessions map[string]map[string]bool +} + +// newWorkspacePollAggregator wires an aggregator to the lifecycle manager. +func newWorkspacePollAggregator(mgr *Manager) *workspacePollAggregator { + return &workspacePollAggregator{ + mgr: mgr, + sessionModes: make(map[string]WorkspacePollMode), + lastPushed: make(map[string]WorkspacePollMode), + workspaceSessions: make(map[string]map[string]bool), + } +} + +// HandleSessionMode is the entry point called by the gateway hub when a +// session's effective UI mode transitions. Resolves the session to its +// workspace, aggregates with sibling sessions in the same workspace, and pushes +// the workspace-level effective mode to agentctl if it changed. +// +// Best-effort: errors are logged, never returned. The hub should not block on +// this (the call is debounced + computed off the hub critical path already). +// +// Pre-execution focus race: if a session.focus arrives before the lifecycle +// has created its execution, we cache the mode in sessionModes and return. The +// lifecycle manager calls FlushSessionMode once agentctl is ready, which +// resolves the race by pushing the cached mode retroactively. +func (a *workspacePollAggregator) HandleSessionMode(sessionID string, mode WorkspacePollMode) { + execution, exists := a.mgr.GetExecutionBySessionID(sessionID) + if !exists { + // No execution yet — cache the mode so the next event can still + // aggregate correctly. If the mode is paused there's nothing to + // remember, so drop any prior entry to keep the map bounded. + a.mu.Lock() + if mode == WorkspacePollModePaused { + delete(a.sessionModes, sessionID) + } else { + a.sessionModes[sessionID] = mode + } + a.mu.Unlock() + return + } + if execution.WorkspacePath == "" { + return + } + + workspacePath, effective, changed := a.recordAndCompute(sessionID, mode, execution.WorkspacePath) + if !changed { + return + } + + a.pushAsync(execution, workspacePath, effective) +} + +// setSessionModeLocked updates sessionModes and the workspaceSessions reverse +// index for a single session. Caller must hold a.mu. +func (a *workspacePollAggregator) setSessionModeLocked(sessionID string, mode WorkspacePollMode, workspacePath string) { + if mode == WorkspacePollModePaused { + delete(a.sessionModes, sessionID) + a.removeFromWorkspaceIndex(sessionID, workspacePath) + return + } + a.sessionModes[sessionID] = mode + if a.workspaceSessions[workspacePath] == nil { + a.workspaceSessions[workspacePath] = make(map[string]bool) + } + a.workspaceSessions[workspacePath][sessionID] = true +} + +// removeFromWorkspaceIndex drops a session from the reverse index, cleaning up +// the workspace key if empty. Caller must hold a.mu. +func (a *workspacePollAggregator) removeFromWorkspaceIndex(sessionID, workspacePath string) { + sessions := a.workspaceSessions[workspacePath] + if sessions == nil { + return + } + delete(sessions, sessionID) + if len(sessions) == 0 { + delete(a.workspaceSessions, workspacePath) + } +} + +// recordAndCompute updates the per-session mode for the given workspace and +// returns the new workspace-effective mode, plus whether it changed since the +// last push. We compute this under a single lock so concurrent transitions in +// the same workspace can't observe inconsistent intermediate state. +// +// When a session goes to paused we drop its sessionModes entry so the map +// doesn't grow unbounded over a long-running gateway. Same for lastPushed +// when the workspace itself becomes paused. +func (a *workspacePollAggregator) recordAndCompute(sessionID string, mode WorkspacePollMode, workspacePath string) (string, WorkspacePollMode, bool) { + a.mu.Lock() + defer a.mu.Unlock() + + a.setSessionModeLocked(sessionID, mode, workspacePath) + + // Compute effective mode using only sessions in this workspace (O(k) + // where k = sessions in workspace, not O(N) total sessions). + effective := WorkspacePollModePaused + for sid := range a.workspaceSessions[workspacePath] { + if m, ok := a.sessionModes[sid]; ok && m.rank() > effective.rank() { + effective = m + } + } + + prev, hadPrev := a.lastPushed[workspacePath] + if hadPrev && prev == effective { + return workspacePath, effective, false + } + // Skip pushing paused when there's no prior entry — agentctl defaults to + // slow, so sending paused to a workspace we've never pushed to is a no-op + // RPC (and slightly misleading: agentctl would drop from slow to paused + // even though the gateway never told it to go slow in the first place). + if !hadPrev && effective == WorkspacePollModePaused { + return workspacePath, effective, false + } + if effective == WorkspacePollModePaused { + delete(a.lastPushed, workspacePath) + } else { + a.lastPushed[workspacePath] = effective + } + return workspacePath, effective, true +} + +// pushAsync issues the SetWorkspacePollMode RPC to the agentctl client without +// blocking the caller (typically the hub's session-mode goroutine). +func (a *workspacePollAggregator) pushAsync(execution *AgentExecution, workspacePath string, mode WorkspacePollMode) { + client := execution.GetAgentCtlClient() + if client == nil { + return + } + go func() { + ctx, cancel := context.WithTimeout(context.Background(), pushPollModeTimeout) + defer cancel() + if err := client.SetWorkspacePollMode(ctx, string(mode)); err != nil { + a.mgr.logger.Warn("failed to push workspace poll mode", + zap.String("workspace", workspacePath), + zap.String("mode", string(mode)), + zap.Error(err)) + } + }() +} + +// FlushSessionMode resolves two races that leave agentctl in the wrong mode: +// +// 1. Pre-execution focus: the gateway sent focus before the execution was +// in executionStore. HandleSessionMode cached the mode but short-circuited +// without pushing. +// 2. Pre-ready push: the gateway sent focus after Add() but before agentctl's +// HTTP server was accepting connections. pushAsync fired the RPC, it +// failed (connection refused), but lastPushed was already updated so +// nothing retries. +// +// This is called from the lifecycle manager's waitForAgentctlReady success +// path — once we know agentctl is accepting RPCs, force-resend whatever +// workspace-effective mode we currently believe is correct. No-op if nothing +// was cached (no gateway events reached us before flush). +func (a *workspacePollAggregator) FlushSessionMode(sessionID string) { + execution, exists := a.mgr.GetExecutionBySessionID(sessionID) + if !exists || execution.WorkspacePath == "" { + return + } + + // Authoritative source: query the hub for the session's current mode. + // This closes the race where gateway events fired before the execution + // was registered (nothing cached) or while agentctl was still starting + // up (the RPC fired but connection-refused). + // + // Read hubQry under a.mu so the race detector doesn't flag the + // concurrent write in SetSessionModeQuerier. Copy to a local and + // release a.mu before calling into the hub (which acquires its own + // locks — holding a.mu across the call could invert lock order). + a.mu.Lock() + qry := a.hubQry + a.mu.Unlock() + + var sessionMode WorkspacePollMode + if qry != nil { + sessionMode = qry.GetSessionMode(sessionID) + } else { + // Fall back to cached value if no hub wired (e.g., tests). + a.mu.Lock() + cached, hasCached := a.sessionModes[sessionID] + a.mu.Unlock() + if !hasCached { + return + } + sessionMode = cached + } + + a.mu.Lock() + // Merge the queried mode with whatever we had cached, then recompute + // the workspace-level effective mode across all sessions in this + // workspace. + wp := execution.WorkspacePath + a.setSessionModeLocked(sessionID, sessionMode, wp) + effective := WorkspacePollModePaused + for sid := range a.workspaceSessions[wp] { + if m, ok := a.sessionModes[sid]; ok && m.rank() > effective.rank() { + effective = m + } + } + if effective == WorkspacePollModePaused { + delete(a.lastPushed, wp) + } else { + a.lastPushed[wp] = effective + } + a.mu.Unlock() + + if effective != WorkspacePollModePaused { + a.pushAsync(execution, execution.WorkspacePath, effective) + } +} + +// SetSessionModeQuerier lets the gateway inject itself so the aggregator can +// query the hub's live session mode state. +func (m *Manager) SetSessionModeQuerier(q SessionModeQuerier) { + if m.pollAggregator != nil { + m.pollAggregator.mu.Lock() + m.pollAggregator.hubQry = q + m.pollAggregator.mu.Unlock() + } +} diff --git a/apps/backend/internal/agent/lifecycle/manager_subscription_test.go b/apps/backend/internal/agent/lifecycle/manager_subscription_test.go new file mode 100644 index 0000000000..b9a8fedaa7 --- /dev/null +++ b/apps/backend/internal/agent/lifecycle/manager_subscription_test.go @@ -0,0 +1,290 @@ +package lifecycle + +import ( + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +// newTestManagerForAggregator builds a minimal Manager with just the bits the +// aggregator needs: an executionStore and a logger. We don't need the real +// dependency graph because the aggregator only reads execution metadata and +// dispatches to AgentExecution.GetAgentCtlClient (which we leave nil — pushAsync +// will then no-op, and the test inspects sessionModes/lastPushed directly). +func newTestManagerForAggregator(t *testing.T) *Manager { + t.Helper() + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + mgr := &Manager{ + logger: log, + executionStore: NewExecutionStore(), + } + mgr.pollAggregator = newWorkspacePollAggregator(mgr) + return mgr +} + +// addExecution attaches a session→workspace mapping so the aggregator can +// resolve it. agentctl is left nil; pushAsync becomes a no-op for these tests. +func addExecution(mgr *Manager, sessionID, workspacePath string) { + mgr.executionStore.Add(&AgentExecution{ + ID: "exec-" + sessionID, + SessionID: sessionID, + WorkspacePath: workspacePath, + }) +} + +func TestAggregator_SingleSessionPushesItsMode(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + + // Inspect the recorded last-pushed state to confirm the workspace mode. + mgr.pollAggregator.mu.Lock() + got := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + + if got != WorkspacePollModeFast { + t.Errorf("last pushed mode = %q, want fast", got) + } +} + +func TestAggregator_TwoSessionsSameWorkspace_TakesMax(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + addExecution(mgr, "s2", "/tmp/ws1") + + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeSlow) + mgr.pollAggregator.HandleSessionMode("s2", WorkspacePollModeFast) + + mgr.pollAggregator.mu.Lock() + got := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + + if got != WorkspacePollModeFast { + t.Errorf("with one fast and one slow session, expected fast; got %q", got) + } +} + +func TestAggregator_DowngradeAfterFastSessionLeaves(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + addExecution(mgr, "s2", "/tmp/ws1") + + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + mgr.pollAggregator.HandleSessionMode("s2", WorkspacePollModeSlow) + + // s1 unfocuses (drops to slow). Workspace should now be slow. + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeSlow) + + mgr.pollAggregator.mu.Lock() + got := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + + if got != WorkspacePollModeSlow { + t.Errorf("after fast session drops to slow, workspace mode = %q, want slow", got) + } +} + +func TestAggregator_AllSessionsPausedYieldsPaused(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + addExecution(mgr, "s2", "/tmp/ws1") + + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeSlow) + mgr.pollAggregator.HandleSessionMode("s2", WorkspacePollModeSlow) + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModePaused) + mgr.pollAggregator.HandleSessionMode("s2", WorkspacePollModePaused) + + mgr.pollAggregator.mu.Lock() + _, hasEntry := mgr.pollAggregator.lastPushed["/tmp/ws1"] + sessionEntries := len(mgr.pollAggregator.sessionModes) + mgr.pollAggregator.mu.Unlock() + + // When the workspace becomes paused we drop the lastPushed entry to keep + // the map bounded over a long-running gateway. The push itself happens via + // pushAsync (test inspects the bookkeeping, not the actual RPC). + if hasEntry { + t.Errorf("expected lastPushed entry for /tmp/ws1 to be deleted when workspace fully paused") + } + // Same cleanup applies to per-session entries. + if sessionEntries != 0 { + t.Errorf("expected sessionModes to be empty when all sessions paused, got %d entries", sessionEntries) + } +} + +func TestAggregator_NoExecutionIsNoOp(t *testing.T) { + mgr := newTestManagerForAggregator(t) + // No execution registered — call should not panic, should not push anything. + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + + mgr.pollAggregator.mu.Lock() + defer mgr.pollAggregator.mu.Unlock() + if len(mgr.pollAggregator.lastPushed) != 0 { + t.Errorf("expected no pushes when execution is absent, got %+v", mgr.pollAggregator.lastPushed) + } +} + +func TestAggregator_DifferentWorkspacesAreIndependent(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + addExecution(mgr, "s2", "/tmp/ws2") + + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + mgr.pollAggregator.HandleSessionMode("s2", WorkspacePollModeSlow) + + mgr.pollAggregator.mu.Lock() + gotWS1 := mgr.pollAggregator.lastPushed["/tmp/ws1"] + gotWS2 := mgr.pollAggregator.lastPushed["/tmp/ws2"] + mgr.pollAggregator.mu.Unlock() + + if gotWS1 != WorkspacePollModeFast { + t.Errorf("ws1 mode = %q, want fast", gotWS1) + } + if gotWS2 != WorkspacePollModeSlow { + t.Errorf("ws2 mode = %q, want slow", gotWS2) + } +} + +func TestAggregator_FlushSessionMode_AppliesCachedModeAfterExecutionReady(t *testing.T) { + mgr := newTestManagerForAggregator(t) + + // Gateway sends focus BEFORE execution is registered — mode is cached, + // not pushed. + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + + mgr.pollAggregator.mu.Lock() + _, pushedBeforeReady := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + if pushedBeforeReady { + t.Fatal("expected no lastPushed entry before execution exists") + } + + // Execution is created (simulates waitForAgentctlReady completing). + addExecution(mgr, "s1", "/tmp/ws1") + mgr.pollAggregator.FlushSessionMode("s1") + + mgr.pollAggregator.mu.Lock() + got := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + + if got != WorkspacePollModeFast { + t.Errorf("after FlushSessionMode, workspace mode = %q, want fast", got) + } +} + +func TestAggregator_FlushSessionMode_NoOpWhenNothingCached(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + + // No prior HandleSessionMode call — nothing cached. + mgr.pollAggregator.FlushSessionMode("s1") + + mgr.pollAggregator.mu.Lock() + _, pushed := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + if pushed { + t.Error("FlushSessionMode should not push anything when nothing cached") + } +} + +func TestAggregator_RedundantEventDoesNotRepush(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + + // Repeated identical events should be a no-op for the workspace mode. + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeFast) + + mgr.pollAggregator.mu.Lock() + got := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + + if got != WorkspacePollModeFast { + t.Errorf("after redundant events, workspace mode = %q, want fast", got) + } +} + +// mockSessionModeQuerier is a test double for SessionModeQuerier that returns +// a fixed mode for any session ID. +type mockSessionModeQuerier struct { + mode WorkspacePollMode +} + +func (m *mockSessionModeQuerier) GetSessionMode(_ string) WorkspacePollMode { + return m.mode +} + +func TestAggregator_FlushSessionMode_QueriesHubWhenWired(t *testing.T) { + mgr := newTestManagerForAggregator(t) + + // Wire a mock hub that always reports fast. + mock := &mockSessionModeQuerier{mode: WorkspacePollModeFast} + mgr.SetSessionModeQuerier(mock) + + // Gateway sends focus BEFORE execution exists — cached but not pushed. + mgr.pollAggregator.HandleSessionMode("s1", WorkspacePollModeSlow) + + mgr.pollAggregator.mu.Lock() + _, pushedBeforeReady := mgr.pollAggregator.lastPushed["/tmp/ws1"] + mgr.pollAggregator.mu.Unlock() + if pushedBeforeReady { + t.Fatal("expected no lastPushed entry before execution exists") + } + + // Execution becomes ready — FlushSessionMode should query the hub (fast), + // not use the cached value (slow). + addExecution(mgr, "s1", "/tmp/ws1") + mgr.pollAggregator.FlushSessionMode("s1") + + mgr.pollAggregator.mu.Lock() + got := mgr.pollAggregator.lastPushed["/tmp/ws1"] + cachedMode := mgr.pollAggregator.sessionModes["s1"] + mgr.pollAggregator.mu.Unlock() + + if got != WorkspacePollModeFast { + t.Errorf("FlushSessionMode should use hub-queried mode; lastPushed = %q, want fast", got) + } + if cachedMode != WorkspacePollModeFast { + t.Errorf("FlushSessionMode should update sessionModes to hub-queried mode; got %q, want fast", cachedMode) + } +} + +// TestAggregator_ConcurrentEvents asserts no races. We don't assert ordering +// — only that the aggregator survives concurrent calls without data races +// (run with -race). +func TestAggregator_ConcurrentEvents(t *testing.T) { + mgr := newTestManagerForAggregator(t) + addExecution(mgr, "s1", "/tmp/ws1") + addExecution(mgr, "s2", "/tmp/ws1") + + var wg sync.WaitGroup + const goroutines = 20 + wg.Add(goroutines) + modes := []WorkspacePollMode{WorkspacePollModeFast, WorkspacePollModeSlow, WorkspacePollModePaused} + for i := 0; i < goroutines; i++ { + go func(i int) { + defer wg.Done() + session := "s1" + if i%2 == 0 { + session = "s2" + } + mgr.pollAggregator.HandleSessionMode(session, modes[i%len(modes)]) + }(i) + } + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + select { + case <-done: + // no-op; we just want to be sure it returns + case <-time.After(2 * time.Second): + t.Fatal("concurrent HandleSessionMode calls did not return in time") + } +} diff --git a/apps/backend/internal/agent/settings/store/sqlite.go b/apps/backend/internal/agent/settings/store/sqlite.go index 7ebdf4ef69..e025adbc92 100644 --- a/apps/backend/internal/agent/settings/store/sqlite.go +++ b/apps/backend/internal/agent/settings/store/sqlite.go @@ -114,9 +114,10 @@ func (r *sqliteRepository) initSchema() error { } // migrateDropModelCheckConstraint recreates agent_profiles without the legacy -// CHECK(model != '') constraint. Existing databases created before the ACP-first -// migration carry this constraint, which prevents empty model values. New -// databases (created by the CREATE TABLE IF NOT EXISTS above) never have it. +// non-empty-model CHECK constraint. Existing databases created before the +// ACP-first migration carry this constraint, which prevents empty model +// values. New databases (created by the CREATE TABLE IF NOT EXISTS above) +// never have it. // // The migration is idempotent: it inspects sqlite_master for the CHECK keyword // and only proceeds when the constraint is present. diff --git a/apps/backend/internal/agentctl/client/workspace_state.go b/apps/backend/internal/agentctl/client/workspace_state.go new file mode 100644 index 0000000000..488a67b0f3 --- /dev/null +++ b/apps/backend/internal/agentctl/client/workspace_state.go @@ -0,0 +1,43 @@ +package client + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" +) + +// SetWorkspacePollMode tells the agentctl instance how aggressively to poll +// the workspace for git/file changes. The gateway calls this when a session's +// UI subscription/focus state changes — see lifecycle/manager_subscription.go. +// +// Best-effort: failures are returned but do not block the gateway. Callers +// should log and move on; the agentctl will still be polling in its previous +// mode (or its default). +func (c *Client) SetWorkspacePollMode(ctx context.Context, mode string) error { + body, err := json.Marshal(struct { + Mode string `json:"mode"` + }{Mode: mode}) + if err != nil { + return err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/api/v1/workspace/poll-mode", bytes.NewReader(body)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + respBody, _ := readResponseBody(resp) + return fmt.Errorf("set workspace poll mode failed with status %d: %s", resp.StatusCode, string(respBody)) + } + return nil +} diff --git a/apps/backend/internal/agentctl/client/workspace_state_test.go b/apps/backend/internal/agentctl/client/workspace_state_test.go new file mode 100644 index 0000000000..5af7078e8f --- /dev/null +++ b/apps/backend/internal/agentctl/client/workspace_state_test.go @@ -0,0 +1,103 @@ +package client + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + ws "github.com/kandev/kandev/pkg/websocket" +) + +// newHTTPOnlyClient builds a Client wired to a custom HTTP base URL without +// the WebSocket stream — sufficient for testing simple HTTP RPC methods. +func newHTTPOnlyClient(baseURL string) *Client { + return &Client{ + baseURL: baseURL, + httpClient: &http.Client{Timeout: 5 * time.Second}, + longRunningHTTPClient: &http.Client{Timeout: 5 * time.Minute}, + logger: newTestLogger(), + pendingRequests: make(map[string]chan *ws.Message), + } +} + +func TestSetWorkspacePollMode_PostsExpectedPayload(t *testing.T) { + type received struct { + Path string + Method string + ContentType string + Body struct { + Mode string `json:"mode"` + } + } + var got received + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got.Path = r.URL.Path + got.Method = r.Method + got.ContentType = r.Header.Get("Content-Type") + _ = json.NewDecoder(r.Body).Decode(&got.Body) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"mode":"slow"}`)) + })) + defer srv.Close() + + c := newHTTPOnlyClient(srv.URL) + + if err := c.SetWorkspacePollMode(context.Background(), "slow"); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if got.Path != "/api/v1/workspace/poll-mode" { + t.Errorf("path = %q, want /api/v1/workspace/poll-mode", got.Path) + } + if got.Method != http.MethodPost { + t.Errorf("method = %q, want POST", got.Method) + } + if got.ContentType != "application/json" { + t.Errorf("content-type = %q, want application/json", got.ContentType) + } + if got.Body.Mode != "slow" { + t.Errorf("body.mode = %q, want slow", got.Body.Mode) + } +} + +func TestSetWorkspacePollMode_ReturnsErrorOnNon2xx(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(`{"error":"bad mode"}`)) + })) + defer srv.Close() + + c := newHTTPOnlyClient(srv.URL) + + err := c.SetWorkspacePollMode(context.Background(), "turbo") + if err == nil { + t.Fatal("expected error for non-2xx response") + } + if !strings.Contains(err.Error(), "400") { + t.Errorf("error should mention status code 400, got: %v", err) + } +} + +func TestSetWorkspacePollMode_HonoursContextCancellation(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // Sleep longer than the context deadline. + time.Sleep(500 * time.Millisecond) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + c := newHTTPOnlyClient(srv.URL) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + err := c.SetWorkspacePollMode(ctx, "fast") + if err == nil { + t.Fatal("expected error from cancelled context") + } +} diff --git a/apps/backend/internal/agentctl/server/api/server.go b/apps/backend/internal/agentctl/server/api/server.go index a0711dfe03..7c4dcb1b05 100644 --- a/apps/backend/internal/agentctl/server/api/server.go +++ b/apps/backend/internal/agentctl/server/api/server.go @@ -87,6 +87,9 @@ func (s *Server) setupRoutes() { // Unified workspace stream (git status, files, shell) api.GET("/workspace/stream", s.handleWorkspaceStreamWS) + // Workspace state (poll mode driven by gateway focus signal) + api.POST("/workspace/poll-mode", s.handleSetPollMode) + // Workspace file operations (simple HTTP) api.GET("/workspace/tree", s.handleFileTree) api.GET("/workspace/file/content", s.handleFileContent) diff --git a/apps/backend/internal/agentctl/server/api/workspace_state.go b/apps/backend/internal/agentctl/server/api/workspace_state.go new file mode 100644 index 0000000000..ef6e0537b6 --- /dev/null +++ b/apps/backend/internal/agentctl/server/api/workspace_state.go @@ -0,0 +1,39 @@ +package api + +import ( + "net/http" + + "github.com/gin-gonic/gin" + "github.com/kandev/kandev/internal/agentctl/server/process" + "go.uber.org/zap" +) + +// SetPollModeRequest is the body for POST /api/v1/workspace/poll-mode. +type SetPollModeRequest struct { + Mode string `json:"mode"` +} + +// handleSetPollMode updates the workspace tracker's poll mode based on the +// gateway's view of UI subscription/focus state for sessions in this workspace. +// +// See plan: focus-gated git polling. The gateway computes a workspace-level +// mode (fast if any session focused, slow if any subscribed, paused otherwise) +// and pushes it here so the tracker can throttle expensive git scans. +func (s *Server) handleSetPollMode(c *gin.Context) { + var req SetPollModeRequest + if err := c.ShouldBindJSON(&req); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid request: " + err.Error()}) + return + } + + mode := process.PollMode(req.Mode) + if !mode.IsValid() { + c.JSON(http.StatusBadRequest, gin.H{"error": "invalid mode: must be one of fast, slow, paused"}) + return + } + + s.procMgr.GetWorkspaceTracker().SetPollMode(mode) + s.logger.Debug("workspace poll mode updated", zap.String("mode", req.Mode)) + + c.JSON(http.StatusOK, gin.H{"mode": req.Mode}) +} diff --git a/apps/backend/internal/agentctl/server/api/workspace_state_test.go b/apps/backend/internal/agentctl/server/api/workspace_state_test.go new file mode 100644 index 0000000000..b5bc897ae9 --- /dev/null +++ b/apps/backend/internal/agentctl/server/api/workspace_state_test.go @@ -0,0 +1,86 @@ +package api + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/kandev/kandev/internal/agentctl/server/process" +) + +func TestHandleSetPollMode_AcceptsValidModes(t *testing.T) { + cases := []struct { + mode string + want process.PollMode + }{ + {"fast", process.PollModeFast}, + {"slow", process.PollModeSlow}, + {"paused", process.PollModePaused}, + } + for _, tc := range cases { + t.Run(tc.mode, func(t *testing.T) { + s := newTestServer(t) + + body, _ := json.Marshal(SetPollModeRequest{Mode: tc.mode}) + req := httptest.NewRequest(http.MethodPost, "/api/v1/workspace/poll-mode", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + s.router.ServeHTTP(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d (body: %s)", w.Code, w.Body.String()) + } + + if got := s.procMgr.GetWorkspaceTracker().GetPollMode(); got != tc.want { + t.Errorf("after request mode=%q, tracker mode = %q, want %q", tc.mode, got, tc.want) + } + }) + } +} + +func TestHandleSetPollMode_RejectsInvalidMode(t *testing.T) { + s := newTestServer(t) + + body, _ := json.Marshal(SetPollModeRequest{Mode: "turbo"}) + req := httptest.NewRequest(http.MethodPost, "/api/v1/workspace/poll-mode", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + s.router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400 for invalid mode, got %d", w.Code) + } +} + +func TestHandleSetPollMode_RejectsInvalidBody(t *testing.T) { + s := newTestServer(t) + + req := httptest.NewRequest(http.MethodPost, "/api/v1/workspace/poll-mode", bytes.NewReader([]byte(`not json`))) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + s.router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400 for invalid body, got %d", w.Code) + } +} + +func TestHandleSetPollMode_EmptyMode(t *testing.T) { + s := newTestServer(t) + + body, _ := json.Marshal(SetPollModeRequest{Mode: ""}) + req := httptest.NewRequest(http.MethodPost, "/api/v1/workspace/poll-mode", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + s.router.ServeHTTP(w, req) + + if w.Code != http.StatusBadRequest { + t.Errorf("expected 400 for empty mode, got %d", w.Code) + } +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go b/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go new file mode 100644 index 0000000000..06a5d14919 --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go @@ -0,0 +1,28 @@ +package process + +import ( + "context" + "os" + "os/exec" +) + +// gitOptionalLocksOff is the env var git reads to skip "optional" locks, i.e. +// the index refresh lock that `git status` and friends take to update stat +// info. The workspace tracker polls git read-only, but without this flag even +// those reads can briefly take `.git/index.lock`, racing with concurrent user +// operations (stage, commit) that need it for writes — in tight-loop polling +// this manifests as sporadic "Unable to create '.../index.lock': File exists" +// failures from user commands. +// +// See: https://git-scm.com/docs/git#Documentation/git.txt-codeGITOPTIONALLOCKSltbooleangtcode +const gitOptionalLocksOff = "GIT_OPTIONAL_LOCKS=0" + +// pollingGitCommand builds an exec.Cmd for a polling git invocation. It sets +// the workspace directory and disables optional locks so the background poll +// loop doesn't contend with user-initiated git operations. +func (wt *WorkspaceTracker) pollingGitCommand(ctx context.Context, args ...string) *exec.Cmd { + cmd := exec.CommandContext(ctx, "git", args...) + cmd.Dir = wt.workDir + cmd.Env = append(os.Environ(), gitOptionalLocksOff) + return cmd +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_diff.go b/apps/backend/internal/agentctl/server/process/workspace_git_diff.go index 69d39727ec..3329f0cedf 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_diff.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_diff.go @@ -30,10 +30,10 @@ const ( // binaryCheckSize is how many bytes to inspect for null bytes to detect binary files. binaryCheckSize = 8 * 1024 // 8 KB - diffSkipReasonTooLarge = "too_large" - diffSkipReasonBinary = "binary" - diffSkipReasonTruncated = "truncated" - diffSkipReasonBudgetExceeded = "budget_exceeded" + diffSkipReasonTooLarge = "too_large" + diffSkipReasonBinary = "binary" + diffSkipReasonTruncated = "truncated" + diffSkipReasonBudgetExceeded = "budget_exceeded" ) // enrichWithDiffData adds diff information (additions, deletions, diff content) to file info diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_poll.go b/apps/backend/internal/agentctl/server/process/workspace_git_poll.go index dd06c34e52..7f757d7d06 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_poll.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_poll.go @@ -14,7 +14,10 @@ import ( ) // pollGitChanges periodically checks for git changes (commits, branch switches, staging) -// This catches manual git operations done via shell that file watching might miss +// This catches manual git operations done via shell that file watching might miss. +// +// The tick interval is driven by PollMode (set via SetPollMode), same as +// monitorLoop — see workspace_monitor.go for the wake-up channel design. func (wt *WorkspaceTracker) pollGitChanges(ctx context.Context) { defer wt.wg.Done() @@ -26,8 +29,11 @@ func (wt *WorkspaceTracker) pollGitChanges(ctx context.Context) { return } - ticker := time.NewTicker(wt.gitPollInterval) - defer ticker.Stop() + timer := time.NewTimer(0) + if !timer.Stop() { + <-timer.C + } + defer timer.Stop() // Initialize cached HEAD SHA, branch name, and index hash wt.gitStateMu.Lock() @@ -37,33 +43,66 @@ func (wt *WorkspaceTracker) pollGitChanges(ctx context.Context) { wt.gitStateMu.Unlock() wt.logger.Info("git polling started", - zap.Duration("interval", wt.gitPollInterval), + zap.Duration("fast_interval", wt.gitPollInterval), + zap.String("initial_mode", string(wt.GetPollMode())), zap.String("initial_head", wt.cachedHeadSHA), zap.String("initial_branch", wt.cachedBranchName)) var consecutiveFailures int for { + _, gitPoll, _ := wt.pollIntervals(wt.GetPollMode()) + timer.Reset(gitPoll) + select { case <-ctx.Done(): return case <-wt.stopCh: return - case <-ticker.C: - // Skip this tick if the previous cycle is still running (prevents process pile-up - // when git commands take longer than the poll interval on large repos). - if !atomic.CompareAndSwapInt32(&wt.gitPollRunning, 0, 1) { - continue + case <-wt.gitPollModeChanged: + if stop := wt.handleGitPollModeChange(ctx, timer, &consecutiveFailures); stop { + return } - - stop := wt.gitPollTick(ctx, &consecutiveFailures) - if stop { + case <-timer.C: + if stop := wt.handleGitPollTimerTick(ctx, &consecutiveFailures); stop { return } } } } +// handleGitPollModeChange responds to a SetPollMode wake-up: drains the timer +// and runs an immediate scan if we just transitioned into fast mode. Returns +// true if pollGitChanges should exit. +func (wt *WorkspaceTracker) handleGitPollModeChange(ctx context.Context, timer *time.Timer, consecutiveFailures *int) bool { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + if wt.GetPollMode() != PollModeFast { + return false + } + if !atomic.CompareAndSwapInt32(&wt.gitPollRunning, 0, 1) { + return false + } + return wt.gitPollTick(ctx, consecutiveFailures) +} + +// handleGitPollTimerTick is the body of a regular timer-driven tick. In paused +// mode it's a no-op (we still wake periodically as a safety net). Returns true +// if pollGitChanges should exit. +func (wt *WorkspaceTracker) handleGitPollTimerTick(ctx context.Context, consecutiveFailures *int) bool { + if _, _, paused := wt.pollIntervals(wt.GetPollMode()); paused { + return false + } + if !atomic.CompareAndSwapInt32(&wt.gitPollRunning, 0, 1) { + return false + } + return wt.gitPollTick(ctx, consecutiveFailures) +} + // gitPollTick runs a single git poll cycle: checks for manual git operations // (commits, branch switches, staging). Returns true if the loop should stop. // The deferred flag reset ensures gitPollRunning is cleared even on panic. @@ -127,8 +166,7 @@ func (wt *WorkspaceTracker) getCurrentBranchName(ctx context.Context) string { // already handled by monitorLoop via git ls-files. This avoids the expensive // directory traversal that --untracked-files=all performs on large repos. func (wt *WorkspaceTracker) getGitStatusHash(ctx context.Context) string { - cmd := exec.CommandContext(ctx, "git", "status", "--porcelain", "--untracked-files=no") - cmd.Dir = wt.workDir + cmd := wt.pollingGitCommand(ctx, "status", "--porcelain", "--untracked-files=no") out, err := cmd.Output() if err != nil { return "" diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_status.go b/apps/backend/internal/agentctl/server/process/workspace_git_status.go index 2cfac6208f..602f1cbccf 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_status.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_status.go @@ -178,9 +178,10 @@ func (wt *WorkspaceTracker) getAheadBehindCounts(ctx context.Context, update *ty // parseGitStatusOutput runs git status --porcelain and populates the file lists and map. func (wt *WorkspaceTracker) parseGitStatusOutput(ctx context.Context, update *types.GitStatusUpdate) error { - // --untracked-files=all shows all files in untracked directories, not just the directory name - statusCmd := exec.CommandContext(ctx, "git", "status", "--porcelain", "--untracked-files=all") - statusCmd.Dir = wt.workDir + // --untracked-files=all shows all files in untracked directories, not just the directory name. + // GIT_OPTIONAL_LOCKS=0 (via pollingGitCommand) prevents the background poll loop from taking + // .git/index.lock, which would race with concurrent user-initiated git operations. + statusCmd := wt.pollingGitCommand(ctx, "status", "--porcelain", "--untracked-files=all") statusOut, err := statusCmd.Output() if err != nil { return err diff --git a/apps/backend/internal/agentctl/server/process/workspace_monitor.go b/apps/backend/internal/agentctl/server/process/workspace_monitor.go index 977764de60..a3042c68bd 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_monitor.go +++ b/apps/backend/internal/agentctl/server/process/workspace_monitor.go @@ -5,7 +5,6 @@ import ( "context" "fmt" "os" - "os/exec" "path/filepath" "strings" "sync/atomic" @@ -33,6 +32,10 @@ const maxConsecutiveGitFailures = 5 // monitorLoop polls for file changes using file mtimes and quick git checks. // This is more efficient than fsnotify on macOS and avoids file descriptor exhaustion. +// +// The tick interval is driven by the current PollMode (set via SetPollMode): +// fast (default 2s) for focused sessions, slow (30s) for sidebar-visible +// sessions, paused (no-op tick) when nothing is watching. func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) { defer wt.wg.Done() @@ -42,49 +45,120 @@ func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) { if wt.gitIndexPath == "" { wt.logger.Warn("no valid git repository found, file monitor not polling", zap.String("workDir", wt.workDir)) + // Close the initialScanDone channel even on early exit so callers + // waiting on <-wt.initialScanDone don't block forever. + select { + case <-wt.initialScanDone: + default: + close(wt.initialScanDone) + } return } - ticker := time.NewTicker(wt.filePollInterval) - defer ticker.Stop() + // Start with a stopped timer; we reset it explicitly per-iteration based + // on the current poll mode. Resettable timer (vs fixed Ticker) lets us + // react instantly when the mode changes — critical for the user-visible + // lag when a task gains focus. + timer := time.NewTimer(0) + if !timer.Stop() { + <-timer.C + } + defer timer.Stop() - // Initial update + // Initial update — done unconditionally so the first focus-grab isn't + // stuck waiting for the first tick. Skipped only in fully-paused initial + // state? No: even paused workspaces benefit from a one-time scan so + // transitioning to slow/fast later finds a populated cache. wt.updateGitStatus(ctx) wt.updateFiles(ctx) // Cache the last known state (ignore error on initial fetch) lastState, _ := wt.getWorkspaceState(ctx) - wt.logger.Info("file polling started", zap.Duration("interval", wt.filePollInterval)) + // Signal that the initial scan is done — tests wait on this instead of + // sleeping to avoid races with the goroutine's first getWorkspaceState. + select { + case <-wt.initialScanDone: + // already closed (shouldn't happen, but be safe) + default: + close(wt.initialScanDone) + } + + wt.logger.Info("file polling started", + zap.Duration("fast_interval", wt.filePollInterval), + zap.String("initial_mode", string(wt.GetPollMode()))) var consecutiveFailures int for { + filePoll, _, _ := wt.pollIntervals(wt.GetPollMode()) + timer.Reset(filePoll) + select { case <-ctx.Done(): return case <-wt.stopCh: return - case <-ticker.C: - // Skip this tick if the previous cycle is still running (prevents process pile-up - // when git commands take longer than the poll interval on large repos). - if !atomic.CompareAndSwapInt32(&wt.monitorRunning, 0, 1) { - continue + case <-wt.monitorModeChanged: + if stop := wt.handleMonitorModeChange(ctx, timer, &lastState, &consecutiveFailures); stop { + return } - - stop := wt.monitorTick(ctx, &lastState, &consecutiveFailures) - if stop { + case <-timer.C: + if stop := wt.handleMonitorTimerTick(ctx, &lastState, &consecutiveFailures); stop { return } } } } +// handleMonitorModeChange responds to a SetPollMode-triggered wake-up: drains +// any pending timer and, if we just transitioned into fast mode, runs an +// immediate scan so the user sees fresh state without waiting for the next tick. +// Returns true if monitorLoop should exit. +func (wt *WorkspaceTracker) handleMonitorModeChange(ctx context.Context, timer *time.Timer, lastState *workspaceState, consecutiveFailures *int) bool { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + if wt.GetPollMode() != PollModeFast { + return false + } + if !atomic.CompareAndSwapInt32(&wt.monitorRunning, 0, 1) { + return false + } + return wt.monitorTick(ctx, lastState, consecutiveFailures) +} + +// handleMonitorTimerTick is the body of a regular timer-driven tick. In paused +// mode the body is a no-op (we still wake periodically to check the mode in +// case the change-signal was missed). Returns true if monitorLoop should exit. +func (wt *WorkspaceTracker) handleMonitorTimerTick(ctx context.Context, lastState *workspaceState, consecutiveFailures *int) bool { + if _, _, paused := wt.pollIntervals(wt.GetPollMode()); paused { + return false + } + // Skip this tick if the previous cycle is still running (prevents process + // pile-up when git commands take longer than the poll interval on large repos). + if !atomic.CompareAndSwapInt32(&wt.monitorRunning, 0, 1) { + return false + } + return wt.monitorTick(ctx, lastState, consecutiveFailures) +} + // monitorTick runs a single monitor cycle: checks workspace state and triggers // updates if changes are detected. Returns true if the loop should stop. // The deferred flag reset ensures monitorRunning is cleared even on panic. func (wt *WorkspaceTracker) monitorTick(ctx context.Context, lastState *workspaceState, consecutiveFailures *int) bool { - defer atomic.StoreInt32(&wt.monitorRunning, 0) + defer func() { + atomic.StoreInt32(&wt.monitorRunning, 0) + // Signal that the tick completed — tests wait on this instead of + // spinning on the monitorRunning atomic flag. + select { + case wt.tickDone <- struct{}{}: + default: + } + }() currentState, err := wt.getWorkspaceState(ctx) if err != nil { @@ -153,8 +227,7 @@ func (wt *WorkspaceTracker) getWorkspaceState(ctx context.Context) (workspaceSta // again because the index blob hash stays constant and the worktree hash // is always shown as 0000000 (not computed). To detect subsequent changes // to dirty files, we also include the mtime of each dirty file. - cmd := exec.CommandContext(gitCtx, "git", "diff-files", "--name-only") - cmd.Dir = wt.workDir + cmd := wt.pollingGitCommand(gitCtx, "diff-files", "--name-only") var stderrBuf bytes.Buffer cmd.Stderr = &stderrBuf out, err := cmd.Output() @@ -207,8 +280,7 @@ func (wt *WorkspaceTracker) buildDirtyFilesID(diffFilesOutput string) string { // Returns an error if the git command fails (e.g., timeout, index lock). func (wt *WorkspaceTracker) getUntrackedFilesID(ctx context.Context) (string, error) { // Get list of untracked files (excluding ignored) - cmd := exec.CommandContext(ctx, "git", "ls-files", "--others", "--exclude-standard") - cmd.Dir = wt.workDir + cmd := wt.pollingGitCommand(ctx, "ls-files", "--others", "--exclude-standard") var stderrBuf bytes.Buffer cmd.Stderr = &stderrBuf out, err := cmd.Output() diff --git a/apps/backend/internal/agentctl/server/process/workspace_monitor_test.go b/apps/backend/internal/agentctl/server/process/workspace_monitor_test.go index fc338c2454..8b0f14349c 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_monitor_test.go +++ b/apps/backend/internal/agentctl/server/process/workspace_monitor_test.go @@ -17,6 +17,9 @@ func TestWorkspaceTracker_StopsWhenWorkDirDeleted(t *testing.T) { log := newTestLogger(t) wt := NewWorkspaceTracker(repoDir, log) wt.gitPollInterval = 100 * time.Millisecond + // Default mode is slow (30s) — set fast so the test exercises real polling + // cadence rather than sitting on a 30s timer. + wt.SetPollMode(PollModeFast) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -93,6 +96,7 @@ func TestWorkspaceTracker_StopsWhenGitBroken(t *testing.T) { // Use fast poll intervals so the test completes quickly wt.filePollInterval = 50 * time.Millisecond wt.gitPollInterval = 50 * time.Millisecond + wt.SetPollMode(PollModeFast) ctx, cancel := context.WithCancel(context.Background()) defer cancel() diff --git a/apps/backend/internal/agentctl/server/process/workspace_overlap_test.go b/apps/backend/internal/agentctl/server/process/workspace_overlap_test.go index 499106ade7..30d8ec3773 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_overlap_test.go +++ b/apps/backend/internal/agentctl/server/process/workspace_overlap_test.go @@ -295,6 +295,8 @@ func TestOverlapGuard_ResetsAfterNormalCycle(t *testing.T) { wt := NewWorkspaceTracker(repoDir, log) wt.filePollInterval = 50 * time.Millisecond wt.gitPollInterval = 50 * time.Millisecond + // Default mode is slow (30s) — set fast so the loops actually tick at 50ms. + wt.SetPollMode(PollModeFast) wt.Start(context.Background()) @@ -325,6 +327,8 @@ func TestOverlapGuard_WorkDirDeletedResetsFlag(t *testing.T) { wt := NewWorkspaceTracker(repoDir, log) wt.filePollInterval = 50 * time.Millisecond wt.gitPollInterval = 50 * time.Millisecond + // Default mode is slow (30s) — set fast so the loops actually tick at 50ms. + wt.SetPollMode(PollModeFast) wt.Start(context.Background()) defer wt.Stop() diff --git a/apps/backend/internal/agentctl/server/process/workspace_poll_mode.go b/apps/backend/internal/agentctl/server/process/workspace_poll_mode.go new file mode 100644 index 0000000000..cfbc07a362 --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/workspace_poll_mode.go @@ -0,0 +1,107 @@ +package process + +import ( + "time" + + "go.uber.org/zap" +) + +// PollMode controls how aggressively WorkspaceTracker polls for workspace changes. +// +// The mode is driven by the gateway based on UI subscription/focus state for the +// session(s) backed by this workspace. See plan: focus-gated git polling. +type PollMode string + +const ( + // PollModeFast is the default fast polling rate used when a UI client is + // actively focused on a session in this workspace (task details page, task + // panel modal). Matches the historical polling rate prior to the gating change. + PollModeFast PollMode = "fast" + + // PollModeSlow is used when sessions in this workspace are subscribed but not + // focused — typically a sidebar card showing a diff badge. Polls infrequently + // so the badge can stay roughly fresh without burning CPU on every retained + // task worktree. + PollModeSlow PollMode = "slow" + + // PollModePaused suppresses git scans entirely. The polling goroutines stay + // alive but their tick bodies are no-ops, so the next "fast" or "slow" mode + // change resumes work within one tick interval (no goroutine restart). + PollModePaused PollMode = "paused" +) + +// IsValid returns true if m is one of the known poll modes. +func (m PollMode) IsValid() bool { + switch m { + case PollModeFast, PollModeSlow, PollModePaused: + return true + default: + return false + } +} + +// Slow-mode interval for both monitorLoop (untracked-file scan) and +// pollGitChanges (HEAD/branch/index check). 30 seconds is the default. +// Smaller = fresher sidebar badges but more CPU; larger = staler badges. +const defaultSlowPollInterval = 30 * time.Second + +// pausedTickInterval is how often paused-mode timers wake up to check whether +// the mode has changed. The body is a no-op so this is purely a "wake and look +// at the mode" cadence. Kept relatively long because real wake-ups also happen +// via SetPollMode pushing on pollModeChanged. +const pausedTickInterval = 60 * time.Second + +// pollIntervals returns the file-monitor and git-poll intervals for a mode, +// plus whether the mode is paused (in which case the body should skip git work). +func (wt *WorkspaceTracker) pollIntervals(mode PollMode) (filePoll, gitPoll time.Duration, paused bool) { + switch mode { + case PollModeFast: + return wt.filePollInterval, wt.gitPollInterval, false + case PollModeSlow: + return defaultSlowPollInterval, defaultSlowPollInterval, false + case PollModePaused: + return pausedTickInterval, pausedTickInterval, true + default: + // Unknown mode: behave like slow (safe fallback). + return defaultSlowPollInterval, defaultSlowPollInterval, false + } +} + +// GetPollMode returns the current poll mode. +func (wt *WorkspaceTracker) GetPollMode() PollMode { + wt.pollModeMu.RLock() + defer wt.pollModeMu.RUnlock() + return wt.pollMode +} + +// SetPollMode updates the poll mode. Calling with the current mode is a no-op. +// Transitioning into PollModeFast (from any other mode) wakes both polling loops +// immediately so the focused user sees fresh git state without waiting up to +// the slow-poll interval for the next tick. +func (wt *WorkspaceTracker) SetPollMode(mode PollMode) { + if !mode.IsValid() { + wt.logger.Warn("ignoring invalid poll mode", zap.String("mode", string(mode))) + return + } + + wt.pollModeMu.Lock() + prev := wt.pollMode + if prev == mode { + wt.pollModeMu.Unlock() + return + } + wt.pollMode = mode + wt.pollModeMu.Unlock() + + // Wake both polling loops. Each has its own buffered(1) channel because + // a single shared channel would let whichever loop selects first steal + // the signal, leaving the other loop blocked on its old timer interval + // (potentially 30s) before noticing the mode change. Sends are non-blocking: + // if the buffer is full, a notification is already pending which is fine. + for _, ch := range []chan struct{}{wt.monitorModeChanged, wt.gitPollModeChanged} { + select { + case ch <- struct{}{}: + default: + } + } +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_poll_mode_loop_test.go b/apps/backend/internal/agentctl/server/process/workspace_poll_mode_loop_test.go new file mode 100644 index 0000000000..c7131d05ca --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/workspace_poll_mode_loop_test.go @@ -0,0 +1,220 @@ +package process + +import ( + "context" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentctl/types" +) + +// waitForMonitorIdle waits for the current monitorTick to complete by reading +// the tickDone channel. Falls back to checking the atomic flag if no tick is +// in progress. Fails the test after 5 seconds to avoid hanging on stuck git. +func waitForMonitorIdle(t *testing.T, wt *WorkspaceTracker) { + t.Helper() + // Drain any stale signal from a previous tick so we don't return early + // on a buffered value that doesn't correspond to the current tick. + select { + case <-wt.tickDone: + default: + } + // Re-check: if no tick is in progress, we're done. + if atomic.LoadInt32(&wt.monitorRunning) == 0 { + return + } + select { + case <-wt.tickDone: + case <-time.After(5 * time.Second): + t.Fatal("waitForMonitorIdle: tick did not complete within 5s") + } +} + +// drainStream consumes everything currently buffered in a workspace stream subscription +// without blocking. Used to clear the initial snapshot before the test starts asserting. +func drainStream(sub types.WorkspaceStreamSubscriber) { + for { + select { + case <-sub: + default: + return + } + } +} + +// waitForFileChangeNotification reads from the subscription until it sees a +// FileChange notification or the timeout fires. +func waitForFileChangeNotification(sub types.WorkspaceStreamSubscriber, timeout time.Duration) bool { + deadline := time.NewTimer(timeout) + defer deadline.Stop() + for { + select { + case msg, ok := <-sub: + if !ok { + return false + } + if msg.Type == types.WorkspaceMessageTypeFileChange { + return true + } + case <-deadline.C: + return false + } + } +} + +// TestMonitorLoop_PausedSuppressesNotifications verifies that no file-change +// notifications are emitted when the tracker is in PollModePaused, even when +// real changes happen in the workspace. +func TestMonitorLoop_PausedSuppressesNotifications(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + // Speed up the would-be-tick interval so the test runs fast. In paused + // mode the timer fires at pausedTickInterval (60s) which is too long to + // wait, but in fast mode it would fire at 50ms — proving paused suppresses. + wt.filePollInterval = 50 * time.Millisecond + wt.gitPollInterval = 50 * time.Millisecond + + wt.SetPollMode(PollModePaused) + wt.Start(context.Background()) + defer wt.Stop() + + sub := wt.SubscribeWorkspaceStream() + defer wt.UnsubscribeWorkspaceStream(sub) + drainStream(sub) + + // Create an untracked file — would normally trigger a notification on the + // next tick. In paused mode no monitor tick runs, so no notification. + if err := os.WriteFile(filepath.Join(repoDir, "new_file.txt"), []byte("hello"), 0o644); err != nil { + t.Fatalf("failed to write file: %v", err) + } + + // Wait long enough that several fast-mode ticks would have fired, plus + // margin for scheduling. If paused short-circuit works, no notification. + if got := waitForFileChangeNotification(sub, 500*time.Millisecond); got { + t.Error("expected no file-change notification in paused mode, got one") + } +} + +// TestMonitorLoop_TransitionToFastTriggersImmediateScan verifies that switching +// from a non-fast mode into fast does not wait for the next ticker — the loop +// runs an immediate scan so the user sees fresh state right away. +func TestMonitorLoop_TransitionToFastTriggersImmediateScan(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + // Make the timer interval irrelevant by using a long fast interval — we're + // proving the immediate-scan-on-mode-change path, not the timer. + wt.filePollInterval = 30 * time.Second + wt.gitPollInterval = 30 * time.Second + + wt.SetPollMode(PollModePaused) + wt.Start(context.Background()) + defer wt.Stop() + + // Subscribe + drain the initial snapshot. Subscribing first ensures we + // don't miss notifications fired between Start and our first wait. + sub := wt.SubscribeWorkspaceStream() + defer wt.UnsubscribeWorkspaceStream(sub) + + // Wait for monitorLoop's initial scan to complete and enter its select. + // Without this, the file we write below could land before lastState is + // captured, and the immediate-scan tick would not see any state change. + <-wt.initialScanDone + drainStream(sub) + + // Create a change so the immediate scan finds something to notify about. + if err := os.WriteFile(filepath.Join(repoDir, "trigger.txt"), []byte("x"), 0o644); err != nil { + t.Fatalf("failed to write file: %v", err) + } + + // Switch to fast — the loop should wake on pollModeChanged and run a tick + // without waiting 30s for the timer. + wt.SetPollMode(PollModeFast) + + if got := waitForFileChangeNotification(sub, 2*time.Second); !got { + t.Error("expected immediate file-change notification after transitioning to fast mode") + } +} + +// TestMonitorLoop_FastPolls verifies that fast mode actually polls at its +// configured interval (smoke test for the resettable-timer path). +func TestMonitorLoop_FastPolls(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + wt.filePollInterval = 50 * time.Millisecond + wt.gitPollInterval = 50 * time.Millisecond + + wt.SetPollMode(PollModeFast) + wt.Start(context.Background()) + defer wt.Stop() + + sub := wt.SubscribeWorkspaceStream() + defer wt.UnsubscribeWorkspaceStream(sub) + + // Wait for monitorLoop's initial state capture before writing. + <-wt.initialScanDone + drainStream(sub) + + if err := os.WriteFile(filepath.Join(repoDir, "tick.txt"), []byte("x"), 0o644); err != nil { + t.Fatalf("failed to write file: %v", err) + } + + if got := waitForFileChangeNotification(sub, 1*time.Second); !got { + t.Error("expected file-change notification within 1s in fast mode") + } +} + +// TestMonitorLoop_FastToPausedStopsPolling verifies that switching from fast to +// paused stops emitting notifications for subsequent changes. +func TestMonitorLoop_FastToPausedStopsPolling(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + wt.filePollInterval = 50 * time.Millisecond + wt.gitPollInterval = 50 * time.Millisecond + + wt.SetPollMode(PollModeFast) + wt.Start(context.Background()) + defer wt.Stop() + + sub := wt.SubscribeWorkspaceStream() + defer wt.UnsubscribeWorkspaceStream(sub) + + // Wait for monitorLoop's initial state capture. + <-wt.initialScanDone + drainStream(sub) + + // Confirm fast mode works first. + if err := os.WriteFile(filepath.Join(repoDir, "before_pause.txt"), []byte("x"), 0o644); err != nil { + t.Fatalf("failed to write: %v", err) + } + if !waitForFileChangeNotification(sub, 1*time.Second) { + t.Fatal("setup: expected fast mode to emit notification before pausing") + } + + // Pause and wait for any in-flight tick to complete before draining. + wt.SetPollMode(PollModePaused) + waitForMonitorIdle(t, wt) + drainStream(sub) + + // New change should NOT generate a notification in paused mode. + if err := os.WriteFile(filepath.Join(repoDir, "after_pause.txt"), []byte("x"), 0o644); err != nil { + t.Fatalf("failed to write: %v", err) + } + if waitForFileChangeNotification(sub, 500*time.Millisecond) { + t.Error("expected no notification after switching to paused mode") + } +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_poll_mode_test.go b/apps/backend/internal/agentctl/server/process/workspace_poll_mode_test.go new file mode 100644 index 0000000000..9eb1510bed --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/workspace_poll_mode_test.go @@ -0,0 +1,185 @@ +package process + +import ( + "testing" + "time" +) + +func TestPollMode_IsValid(t *testing.T) { + cases := []struct { + mode PollMode + valid bool + }{ + {PollModeFast, true}, + {PollModeSlow, true}, + {PollModePaused, true}, + {PollMode(""), false}, + {PollMode("turbo"), false}, + } + for _, tc := range cases { + if got := tc.mode.IsValid(); got != tc.valid { + t.Errorf("PollMode(%q).IsValid() = %v, want %v", tc.mode, got, tc.valid) + } + } +} + +func TestNewWorkspaceTracker_DefaultsToFast(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + // Fresh agentctl instances start in fast mode — the gateway pushes + // slow/paused once it determines no client is actively watching. + // Defaulting to fast avoids a startup window where workspace changes + // would go undetected for up to 30s. + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + if got := wt.GetPollMode(); got != PollModeFast { + t.Errorf("default poll mode = %q, want %q", got, PollModeFast) + } +} + +func TestSetPollMode_NoOpOnSameMode(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + + // Drain any signal that might already be in the channels + for _, ch := range []chan struct{}{wt.monitorModeChanged, wt.gitPollModeChanged} { + select { + case <-ch: + default: + } + } + + // Setting the same mode (fast — the default) should not push a signal to either channel + wt.SetPollMode(PollModeFast) + for name, ch := range map[string]chan struct{}{ + "monitor": wt.monitorModeChanged, + "gitPoll": wt.gitPollModeChanged, + } { + select { + case <-ch: + t.Errorf("SetPollMode with current mode should not signal %s channel", name) + case <-time.After(20 * time.Millisecond): + // expected: no signal + } + } +} + +func TestSetPollMode_SignalsOnTransition(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + // Drain both channels + for _, ch := range []chan struct{}{wt.monitorModeChanged, wt.gitPollModeChanged} { + select { + case <-ch: + default: + } + } + + // Start at slow so fast is a real transition (default is fast). + wt.SetPollMode(PollModeSlow) + for _, ch := range []chan struct{}{wt.monitorModeChanged, wt.gitPollModeChanged} { + select { + case <-ch: + default: + } + } + wt.SetPollMode(PollModeFast) + // Both loops must wake — each channel should receive a signal + for name, ch := range map[string]chan struct{}{ + "monitor": wt.monitorModeChanged, + "gitPoll": wt.gitPollModeChanged, + } { + select { + case <-ch: + // expected + case <-time.After(50 * time.Millisecond): + t.Fatalf("SetPollMode transition did not signal %s channel", name) + } + } + + if got := wt.GetPollMode(); got != PollModeFast { + t.Errorf("after SetPollMode(fast) GetPollMode = %q, want %q", got, PollModeFast) + } +} + +func TestSetPollMode_RejectsInvalidMode(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + wt.SetPollMode(PollMode("nonsense")) + // Default is fast; invalid mode should not mutate. + if got := wt.GetPollMode(); got != PollModeFast { + t.Errorf("invalid mode mutated state: GetPollMode = %q, want unchanged %q", got, PollModeFast) + } +} + +func TestPollIntervals(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + + cases := []struct { + mode PollMode + wantFile time.Duration + wantGit time.Duration + wantPaused bool + }{ + {PollModeFast, DefaultFilePollInterval, DefaultGitPollInterval, false}, + {PollModeSlow, defaultSlowPollInterval, defaultSlowPollInterval, false}, + {PollModePaused, pausedTickInterval, pausedTickInterval, true}, + } + for _, tc := range cases { + t.Run(string(tc.mode), func(t *testing.T) { + file, git, paused := wt.pollIntervals(tc.mode) + if file != tc.wantFile { + t.Errorf("file interval = %v, want %v", file, tc.wantFile) + } + if git != tc.wantGit { + t.Errorf("git interval = %v, want %v", git, tc.wantGit) + } + if paused != tc.wantPaused { + t.Errorf("paused = %v, want %v", paused, tc.wantPaused) + } + }) + } +} + +// TestSetPollMode_SignalChannelDoesNotBlock verifies that calling SetPollMode +// repeatedly without anyone reading the channel does not deadlock the caller. +// The channel is buffered(1) and the send is non-blocking. +func TestSetPollMode_SignalChannelDoesNotBlock(t *testing.T) { + isolateTestGitEnv(t) + repoDir, cleanup := setupTestRepo(t) + defer cleanup() + + wt := NewWorkspaceTracker(repoDir, newTestLogger(t)) + + done := make(chan struct{}) + go func() { + // Alternate modes so each call is a real transition. With no reader, + // only the first send fills the buffer; subsequent sends drop on the floor. + modes := []PollMode{PollModeFast, PollModeSlow, PollModeFast, PollModePaused, PollModeFast} + for _, m := range modes { + wt.SetPollMode(m) + } + close(done) + }() + + select { + case <-done: + // expected: SetPollMode never blocks + case <-time.After(500 * time.Millisecond): + t.Fatal("SetPollMode blocked when channel buffer was full") + } +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_tracker.go b/apps/backend/internal/agentctl/server/process/workspace_tracker.go index 7853ba0096..2e6524dcfc 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_tracker.go +++ b/apps/backend/internal/agentctl/server/process/workspace_tracker.go @@ -47,10 +47,22 @@ type WorkspaceTracker struct { workspaceStreamSubscribers map[types.WorkspaceStreamSubscriber]struct{} workspaceSubMu sync.RWMutex - // Polling intervals + // Polling intervals (used in PollModeFast) filePollInterval time.Duration gitPollInterval time.Duration + // pollMode is the current rate at which the polling loops scan git state. + // Default at construction is PollModeSlow — safe fallback before the gateway + // pushes a focus signal. Mutate via SetPollMode (never directly) so loops + // receive the wake-up notification on transitions. + pollMode PollMode + pollModeMu sync.RWMutex + // One channel per loop because we need both loops to wake up on a mode + // change. A single shared channel would let the first reader steal the + // signal so only one loop wakes. + monitorModeChanged chan struct{} // buffered(1); read by monitorLoop + gitPollModeChanged chan struct{} // buffered(1); read by pollGitChanges + // Overlap guards: prevent tick pile-up when git commands take longer than the poll interval. monitorRunning int32 // atomic; 1 if monitorLoop tick is in progress gitPollRunning int32 // atomic; 1 if pollGitChanges tick is in progress @@ -60,12 +72,14 @@ type WorkspaceTracker struct { updateMu sync.Mutex // Control - stopCh chan struct{} - wg sync.WaitGroup - started bool - stopOnce sync.Once - cancelCtx context.Context // Cancellable context for killing in-flight git commands on Stop - cancelFunc context.CancelFunc // Cancel function called during Stop + stopCh chan struct{} + wg sync.WaitGroup + started bool + stopOnce sync.Once + initialScanDone chan struct{} // closed after monitorLoop's first getWorkspaceState; tests wait on it + tickDone chan struct{} // buffered(1); signalled after each monitorTick completes; tests wait on it + cancelCtx context.Context // Cancellable context for killing in-flight git commands on Stop + cancelFunc context.CancelFunc // Cancel function called during Stop } // NewWorkspaceTracker creates a new workspace tracker @@ -84,9 +98,22 @@ func NewWorkspaceTracker(workDir string, log *logger.Logger) *WorkspaceTracker { workspaceStreamSubscribers: make(map[types.WorkspaceStreamSubscriber]struct{}), filePollInterval: DefaultFilePollInterval, gitPollInterval: DefaultGitPollInterval, - stopCh: make(chan struct{}), - cancelCtx: ctx, - cancelFunc: cancel, + // Default to fast polling — matches pre-PR behavior so newly-created + // agentctl instances don't have a startup window where changes go + // undetected for up to 30s. The gateway pushes slow/paused once it + // knows no client is actively watching this workspace; until then, + // fast is the safe default (a freshly-spawned instance was always + // about to be used by someone, historically). Retained-task CPU + // savings still apply because those instances eventually receive a + // slow or paused mode push. + pollMode: PollModeFast, + monitorModeChanged: make(chan struct{}, 1), + gitPollModeChanged: make(chan struct{}, 1), + stopCh: make(chan struct{}), + initialScanDone: make(chan struct{}), + tickDone: make(chan struct{}, 1), + cancelCtx: ctx, + cancelFunc: cancel, } } diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index 8c76502405..d203384629 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -36,6 +36,7 @@ type Client struct { send chan []byte subscriptions map[string]bool // Task IDs this client is subscribed to sessionSubscriptions map[string]bool // Session IDs this client is subscribed to + sessionFocus map[string]bool // Session IDs this client has focused (a strict subset of subscriptions, conceptually — see hub_session_mode.go) userSubscriptions map[string]bool // User IDs this client is subscribed to mu sync.RWMutex closed bool @@ -51,6 +52,7 @@ func NewClient(id string, conn *websocket.Conn, hub *Hub, log *logger.Logger) *C send: make(chan []byte, 256), subscriptions: make(map[string]bool), sessionSubscriptions: make(map[string]bool), + sessionFocus: make(map[string]bool), userSubscriptions: make(map[string]bool), logger: log.WithFields(zap.String("client_id", id)), } @@ -120,6 +122,12 @@ func (c *Client) handleMessage(ctx context.Context, msg *ws.Message) { case ws.ActionSessionUnsubscribe: c.handleSessionUnsubscribe(msg) return + case ws.ActionSessionFocus: + c.handleSessionFocus(msg) + return + case ws.ActionSessionUnfocus: + c.handleSessionUnfocus(msg) + return case ws.ActionUserSubscribe: c.handleUserSubscribe(msg) return @@ -317,6 +325,49 @@ func (c *Client) handleSessionUnsubscribe(msg *ws.Message) { c.sendMessage(resp) } +// handleSessionFocus handles session.focus — marks the session as actively +// viewed by this client, lifting backend polling to fast mode for the workspace. +func (c *Client) handleSessionFocus(msg *ws.Message) { + var req SessionSubscribeRequest + if err := msg.ParsePayload(&req); err != nil { + c.sendError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + return + } + if req.SessionID == "" { + c.sendError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) + return + } + c.hub.FocusSession(c, req.SessionID) + + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "session_id": req.SessionID, + }) + c.sendMessage(resp) +} + +// handleSessionUnfocus handles session.unfocus — releases the focus mark for +// this client. The session falls back to slow mode (still subscribed) or +// paused (no subscribers), with a debounce to absorb tab churn. +func (c *Client) handleSessionUnfocus(msg *ws.Message) { + var req SessionSubscribeRequest + if err := msg.ParsePayload(&req); err != nil { + c.sendError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil) + return + } + if req.SessionID == "" { + c.sendError(msg.ID, msg.Action, ws.ErrorCodeValidation, "session_id is required", nil) + return + } + c.hub.UnfocusSession(c, req.SessionID) + + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "success": true, + "session_id": req.SessionID, + }) + c.sendMessage(resp) +} + // sendMessage sends a message to the client func (c *Client) sendMessage(msg *ws.Message) { data, err := json.Marshal(msg) diff --git a/apps/backend/internal/gateway/websocket/hub.go b/apps/backend/internal/gateway/websocket/hub.go index e654bd6974..56c89efb2e 100644 --- a/apps/backend/internal/gateway/websocket/hub.go +++ b/apps/backend/internal/gateway/websocket/hub.go @@ -39,6 +39,10 @@ type Hub struct { // Optional provider for session data on subscription (e.g., git status) sessionDataProvider SessionDataProvider + // sessionMode tracks per-session focus state and fires listeners when + // effective mode (paused/slow/fast) transitions. See hub_session_mode.go. + sessionMode *sessionModeTracker + mu sync.RWMutex logger *logger.Logger } @@ -54,6 +58,7 @@ func NewHub(dispatcher *ws.Dispatcher, log *logger.Logger) *Hub { unregister: make(chan *Client), broadcast: make(chan *ws.Message, 256), dispatcher: dispatcher, + sessionMode: newSessionModeTracker(), logger: log.WithFields(zap.String("component", "ws_hub")), } } @@ -84,25 +89,29 @@ func (h *Hub) Run(ctx context.Context) { } } -// closeAllClients closes all client connections +// closeAllClients closes all client connections. +// Cancels any pending debounced session-mode transitions so timers don't fire +// after shutdown and call into listeners with stale state. func (h *Hub) closeAllClients() { h.mu.Lock() - defer h.mu.Unlock() - for client := range h.clients { client.closeSend() delete(h.clients, client) } h.taskSubscribers = make(map[string]map[*Client]bool) h.sessionSubscribers = make(map[string]map[*Client]bool) + h.sessionMode.focusByClient = make(map[string]map[*Client]bool) + h.mu.Unlock() + + h.stopAllPendingTransitions() } // removeClient removes a client from the hub func (h *Hub) removeClient(client *Client) { h.mu.Lock() - defer h.mu.Unlock() if _, ok := h.clients[client]; !ok { + h.mu.Unlock() h.logger.Debug("Client unregistered", zap.String("client_id", client.ID)) return } @@ -114,16 +123,46 @@ func (h *Hub) removeClient(client *Client) { for taskID := range client.subscriptions { removeClientFromSubscriberMap(h.taskSubscribers, taskID, client) } + // Capture session IDs that need mode recomputation after we drop the lock. + // Disconnect can change mode either way: removing the last subscriber drops + // to paused, removing the last focuser drops fast → slow. + affectedSessions := make([]string, 0, len(client.sessionSubscriptions)+len(client.sessionFocus)) for sessionID := range client.sessionSubscriptions { removeClientFromSubscriberMap(h.sessionSubscribers, sessionID, client) + affectedSessions = append(affectedSessions, sessionID) + } + for sessionID := range client.sessionFocus { + removeClientFromSubscriberMap(h.sessionMode.focusByClient, sessionID, client) + affectedSessions = append(affectedSessions, sessionID) } for userID := range client.userSubscriptions { removeClientFromSubscriberMap(h.userSubscribers, userID, client) } + h.mu.Unlock() + + for _, sessionID := range dedupStrings(affectedSessions) { + h.recomputeSessionMode(sessionID) + } h.logger.Debug("Client unregistered", zap.String("client_id", client.ID)) } +// dedupStrings returns the input with duplicates removed, preserving order. +// Used to call recomputeSessionMode at most once per affected session when a +// client is both subscribed and focused. +func dedupStrings(in []string) []string { + seen := make(map[string]struct{}, len(in)) + out := make([]string, 0, len(in)) + for _, s := range in { + if _, ok := seen[s]; ok { + continue + } + seen[s] = struct{}{} + out = append(out, s) + } + return out +} + // removeClientFromSubscriberMap removes a client from a subscriber map entry, // deleting the entry entirely when no subscribers remain. func removeClientFromSubscriberMap(subscribers map[string]map[*Client]bool, key string, client *Client) { @@ -261,24 +300,23 @@ func (h *Hub) SubscribeToTask(client *Client, taskID string) { // SubscribeToSession subscribes a client to session notifications func (h *Hub) SubscribeToSession(client *Client, sessionID string) { h.mu.Lock() - defer h.mu.Unlock() - if _, ok := h.sessionSubscribers[sessionID]; !ok { h.sessionSubscribers[sessionID] = make(map[*Client]bool) } h.sessionSubscribers[sessionID][client] = true client.sessionSubscriptions[sessionID] = true + h.mu.Unlock() h.logger.Debug("Client subscribed to session", zap.String("client_id", client.ID), zap.String("session_id", sessionID)) + + h.recomputeSessionMode(sessionID) } // UnsubscribeFromSession unsubscribes a client from session notifications func (h *Hub) UnsubscribeFromSession(client *Client, sessionID string) { h.mu.Lock() - defer h.mu.Unlock() - delete(client.sessionSubscriptions, sessionID) if clients, ok := h.sessionSubscribers[sessionID]; ok { delete(clients, client) @@ -286,6 +324,9 @@ func (h *Hub) UnsubscribeFromSession(client *Client, sessionID string) { delete(h.sessionSubscribers, sessionID) } } + h.mu.Unlock() + + h.recomputeSessionMode(sessionID) } // SubscribeToUser subscribes a client to user notifications diff --git a/apps/backend/internal/gateway/websocket/hub_session_mode.go b/apps/backend/internal/gateway/websocket/hub_session_mode.go new file mode 100644 index 0000000000..9444c86de7 --- /dev/null +++ b/apps/backend/internal/gateway/websocket/hub_session_mode.go @@ -0,0 +1,306 @@ +package websocket + +import ( + "sync" + "time" + + "go.uber.org/zap" +) + +// SessionMode is the desired polling intensity for a session derived from UI state. +// Mirrors the agentctl-side process.PollMode, but defined here so the gateway +// doesn't import agentctl/process types. +type SessionMode string + +const ( + // SessionModePaused: no clients subscribed to this session. + SessionModePaused SessionMode = "paused" + // SessionModeSlow: clients subscribed (e.g. sidebar diff badge) but none focused. + SessionModeSlow SessionMode = "slow" + // SessionModeFast: at least one client has the session focused (task details page, panel). + SessionModeFast SessionMode = "fast" +) + +// SessionModeListener is invoked when a session's effective mode transitions. +// Up-transitions (towards fast) fire immediately; down-transitions fire after +// a debounce window so quick tab churn doesn't tear down + restart polling. +type SessionModeListener func(sessionID string, mode SessionMode) + +// downTransitionDebounce is how long the hub waits before notifying listeners +// of a down-transition (fast → slow, slow → paused). Tunable; 5s catches the +// common "open / close / reopen" pattern without leaving CPU on for too long. +const downTransitionDebounce = 5 * time.Second + +// sessionModeTracker holds the focus map, debounce timers, and listener list. +// +// Locking: focusByClient is protected by Hub.mu (for consistency with +// sessionSubscribers). All other fields are protected by sessionModeTracker.mu. +type sessionModeTracker struct { + // One client set per session that has currently focused it. Separate from + // sessionSubscribers so a client can be "subscribed but not focused" (the + // sidebar case) — the sets evolve independently. + // Protected by Hub.mu, NOT by sessionModeTracker.mu. + focusByClient map[string]map[*Client]bool + + // Last known mode per session, used to suppress redundant listener calls. + lastMode map[string]SessionMode + + // Pending debounced down-transitions (sessionID → timer that will fire + // the listener if not cancelled by a re-up). + pendingDownTransitions map[string]*time.Timer + + // Listeners to invoke on transition. Multiple listeners are supported but + // in practice there's one (lifecycle manager). + listeners []SessionModeListener + + // debounce is the delay for down-transitions (fast→slow, slow→paused). + // Defaults to downTransitionDebounce (5s). Tests can shorten it via + // setDebounceForTest to avoid 5+ seconds of wall-clock sleep. + debounce time.Duration + + mu sync.Mutex +} + +func newSessionModeTracker() *sessionModeTracker { + return &sessionModeTracker{ + focusByClient: make(map[string]map[*Client]bool), + lastMode: make(map[string]SessionMode), + pendingDownTransitions: make(map[string]*time.Timer), + debounce: downTransitionDebounce, + } +} + +// setDebounceForTest allows tests to use a short debounce duration instead of +// waiting 5+ real seconds per test. +func (h *Hub) setDebounceForTest(d time.Duration) { + h.sessionMode.mu.Lock() + h.sessionMode.debounce = d + h.sessionMode.mu.Unlock() +} + +// AddSessionModeListener registers a callback for session mode transitions. +// Listeners are called from arbitrary goroutines; they should be fast. +func (h *Hub) AddSessionModeListener(l SessionModeListener) { + h.sessionMode.mu.Lock() + h.sessionMode.listeners = append(h.sessionMode.listeners, l) + h.sessionMode.mu.Unlock() +} + +// FocusSession marks a session as focused by the given client. Causes the +// session mode to transition to fast (immediately). +func (h *Hub) FocusSession(client *Client, sessionID string) { + if sessionID == "" { + return + } + h.mu.Lock() + if _, ok := h.sessionMode.focusByClient[sessionID]; !ok { + h.sessionMode.focusByClient[sessionID] = make(map[*Client]bool) + } + h.sessionMode.focusByClient[sessionID][client] = true + client.sessionFocus[sessionID] = true + h.mu.Unlock() + + h.logger.Debug("client focused session", + zap.String("client_id", client.ID), + zap.String("session_id", sessionID)) + + h.recomputeSessionMode(sessionID) +} + +// UnfocusSession removes the focus mark for the given client. +func (h *Hub) UnfocusSession(client *Client, sessionID string) { + if sessionID == "" { + return + } + h.mu.Lock() + if clients, ok := h.sessionMode.focusByClient[sessionID]; ok { + delete(clients, client) + if len(clients) == 0 { + delete(h.sessionMode.focusByClient, sessionID) + } + } + delete(client.sessionFocus, sessionID) + h.mu.Unlock() + + h.logger.Debug("client unfocused session", + zap.String("client_id", client.ID), + zap.String("session_id", sessionID)) + + h.recomputeSessionMode(sessionID) +} + +// computeSessionModeLocked returns the effective mode for a session given the +// current subscribers and focus sets. Caller must hold h.mu (read or write). +func (h *Hub) computeSessionModeLocked(sessionID string) SessionMode { + if len(h.sessionMode.focusByClient[sessionID]) > 0 { + return SessionModeFast + } + if len(h.sessionSubscribers[sessionID]) > 0 { + return SessionModeSlow + } + return SessionModePaused +} + +// GetSessionMode returns the current effective mode for a session (fast if any +// client focused, slow if any subscribed, paused otherwise). Used by the +// lifecycle manager when an execution becomes ready — it queries the hub's +// live state and pushes to agentctl, closing the race where the hub's mode +// transitions fired before the execution was registered. +func (h *Hub) GetSessionMode(sessionID string) SessionMode { + h.mu.RLock() + defer h.mu.RUnlock() + return h.computeSessionModeLocked(sessionID) +} + +// recomputeSessionMode is called after any state change that could affect a +// session's mode (subscribe, unsubscribe, focus, unfocus, client disconnect). +// Up-transitions notify immediately; down-transitions are debounced. +// +// Lock order: sessionMode.mu before h.mu (RLock). This is also the order used +// by the timer callback in scheduleDownTransition, which lets the callback +// re-read the latest mode atomically with respect to other recomputeSessionMode +// calls (otherwise a TOCTOU window between "read latest" and "compare to +// lastMode" can let a concurrent up-transition be silently overwritten). +func (h *Hub) recomputeSessionMode(sessionID string) { + h.sessionMode.mu.Lock() + h.mu.RLock() + current := h.computeSessionModeLocked(sessionID) + h.mu.RUnlock() + + prev, hadPrev := h.sessionMode.lastMode[sessionID] + + h.logger.Debug("recomputeSessionMode", + zap.String("session_id", sessionID), + zap.String("current", string(current)), + zap.String("prev", string(prev)), + zap.Bool("had_prev", hadPrev), + zap.Int("listener_count", len(h.sessionMode.listeners))) + + // Always cancel any pending debounced down-transition — either we're + // transitioning to a new mode (it's stale), or we're back at the same mode + // it was scheduled to drop us to (it's redundant). If we kept the old + // timer alive, a slow→fast→slow→fast within the debounce window would + // leave the original slow timer pending and fire after the user had + // already re-focused. + // + // When we cancel a pending timer, also clear lastMode so the cancelled + // transition doesn't leave stale state. Without this, a page refresh + // (disconnect → paused debounce → reconnect within 5s) would see + // prev=fast, current=fast → "no change", and never broadcast the mode + // to the new client. + if t, ok := h.sessionMode.pendingDownTransitions[sessionID]; ok { + t.Stop() + delete(h.sessionMode.pendingDownTransitions, sessionID) + delete(h.sessionMode.lastMode, sessionID) + prev = "" + hadPrev = false + } + + if hadPrev && prev == current { + h.logger.Debug("recomputeSessionMode: no change, skipping", + zap.String("session_id", sessionID)) + h.sessionMode.mu.Unlock() + return + } + + if isUpTransition(prev, current) { + // Up-transitions fire immediately so the user sees fresh data right away. + h.sessionMode.lastMode[sessionID] = current + listeners := h.snapshotListenersLocked() + h.logger.Debug("recomputeSessionMode: up-transition, firing listeners", + zap.String("session_id", sessionID), + zap.String("from", string(prev)), + zap.String("to", string(current)), + zap.Int("listener_count", len(listeners))) + h.sessionMode.mu.Unlock() + h.fireListeners(listeners, sessionID, current) + return + } + + // Down-transition: debounce so quick tab churn doesn't tear down + restart. + h.logger.Debug("recomputeSessionMode: down-transition, scheduling debounce", + zap.String("session_id", sessionID), + zap.String("from", string(prev)), + zap.String("to", string(current))) + h.sessionMode.pendingDownTransitions[sessionID] = time.AfterFunc(h.sessionMode.debounce, func() { + h.fireDebouncedDownTransition(sessionID) + }) + h.sessionMode.mu.Unlock() +} + +// fireDebouncedDownTransition is called by the debounce timer. Acquires +// sessionMode.mu first (matching recomputeSessionMode's lock order) so that +// reading latest and comparing/updating lastMode happens atomically — no +// concurrent recomputeSessionMode can sneak an up-transition between the two +// reads. +func (h *Hub) fireDebouncedDownTransition(sessionID string) { + h.sessionMode.mu.Lock() + // Drop the timer pointer so a future scheduling can replace it cleanly. + delete(h.sessionMode.pendingDownTransitions, sessionID) + + h.mu.RLock() + latest := h.computeSessionModeLocked(sessionID) + h.mu.RUnlock() + + prevAtFire, hadPrev := h.sessionMode.lastMode[sessionID] + if hadPrev && prevAtFire == latest { + h.sessionMode.mu.Unlock() + return + } + + if latest == SessionModePaused { + // Session is fully idle — drop the entry so the maps don't grow + // unbounded over a long-running gateway. Next event re-adds. + delete(h.sessionMode.lastMode, sessionID) + } else { + h.sessionMode.lastMode[sessionID] = latest + } + listeners := h.snapshotListenersLocked() + h.sessionMode.mu.Unlock() + h.fireListeners(listeners, sessionID, latest) +} + +// stopAllPendingTransitions cancels every pending debounce timer and clears +// the tracking maps. Called from closeAllClients during shutdown so timers +// don't outlive the hub and fire stale events into listeners. +func (h *Hub) stopAllPendingTransitions() { + h.sessionMode.mu.Lock() + defer h.sessionMode.mu.Unlock() + for id, t := range h.sessionMode.pendingDownTransitions { + t.Stop() + delete(h.sessionMode.pendingDownTransitions, id) + } + h.sessionMode.lastMode = make(map[string]SessionMode) +} + +// snapshotListenersLocked returns a copy of the listener slice. Caller holds h.sessionMode.mu. +func (h *Hub) snapshotListenersLocked() []SessionModeListener { + out := make([]SessionModeListener, len(h.sessionMode.listeners)) + copy(out, h.sessionMode.listeners) + return out +} + +func (h *Hub) fireListeners(listeners []SessionModeListener, sessionID string, mode SessionMode) { + for _, l := range listeners { + l(sessionID, mode) + } +} + +// isUpTransition returns true if newMode represents a higher polling intensity +// than oldMode. Order: paused < slow < fast. +func isUpTransition(oldMode, newMode SessionMode) bool { + return modeRank(newMode) > modeRank(oldMode) +} + +func modeRank(m SessionMode) int { + switch m { + case SessionModeFast: + return 2 + case SessionModeSlow: + return 1 + case SessionModePaused: + return 0 + default: + return 0 + } +} diff --git a/apps/backend/internal/gateway/websocket/hub_session_mode_test.go b/apps/backend/internal/gateway/websocket/hub_session_mode_test.go new file mode 100644 index 0000000000..401fa4c0fc --- /dev/null +++ b/apps/backend/internal/gateway/websocket/hub_session_mode_test.go @@ -0,0 +1,308 @@ +package websocket + +import ( + "sync" + "testing" + "testing/synctest" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +func newTestHub(t *testing.T) *Hub { + t.Helper() + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + h := NewHub(nil, log) + // Short debounce so tests don't block for 5+ real seconds. + h.setDebounceForTest(50 * time.Millisecond) + // Cancel any pending debounce timers when the test exits — otherwise the + // timer can fire after t.Run returns and race the next test. + t.Cleanup(func() { + h.stopAllPendingTransitions() + }) + return h +} + +func newTestClient(id string) *Client { + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"}) + return &Client{ + ID: id, + send: make(chan []byte, 16), + subscriptions: map[string]bool{}, + sessionSubscriptions: map[string]bool{}, + sessionFocus: map[string]bool{}, + userSubscriptions: map[string]bool{}, + logger: log, + } +} + +// modeRecorder collects mode transitions for assertion. Synchronised so tests +// can read what's been recorded so far without races. +type modeRecorder struct { + mu sync.Mutex + events []modeEvent + cond *sync.Cond +} + +type modeEvent struct { + sessionID string + mode SessionMode +} + +func newModeRecorder() *modeRecorder { + r := &modeRecorder{} + r.cond = sync.NewCond(&r.mu) + return r +} + +func (r *modeRecorder) listener() SessionModeListener { + return func(sessionID string, mode SessionMode) { + r.mu.Lock() + r.events = append(r.events, modeEvent{sessionID, mode}) + r.cond.Broadcast() + r.mu.Unlock() + } +} + +// waitForCount blocks until the recorder has at least n events or the timeout fires. +func (r *modeRecorder) waitForCount(n int, timeout time.Duration) []modeEvent { + deadline := time.Now().Add(timeout) + r.mu.Lock() + defer r.mu.Unlock() + for len(r.events) < n { + remaining := time.Until(deadline) + if remaining <= 0 { + break + } + // Use a separate goroutine to wake cond on timeout. + timer := time.AfterFunc(remaining, func() { + r.mu.Lock() + r.cond.Broadcast() + r.mu.Unlock() + }) + r.cond.Wait() + timer.Stop() + } + out := make([]modeEvent, len(r.events)) + copy(out, r.events) + return out +} + +func TestHub_SubscribeFiresSlow(t *testing.T) { + h := newTestHub(t) + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c := newTestClient("c1") + h.SubscribeToSession(c, "sess-1") + + got := rec.waitForCount(1, time.Second) + if len(got) != 1 || got[0].sessionID != "sess-1" || got[0].mode != SessionModeSlow { + t.Errorf("expected one slow event for sess-1, got %+v", got) + } +} + +func TestHub_FocusFiresFastImmediately(t *testing.T) { + h := newTestHub(t) + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c := newTestClient("c1") + h.SubscribeToSession(c, "sess-1") // slow + h.FocusSession(c, "sess-1") // fast + + got := rec.waitForCount(2, time.Second) + if len(got) < 2 { + t.Fatalf("expected at least 2 events, got %+v", got) + } + if got[1].mode != SessionModeFast { + t.Errorf("expected second event to be fast, got %v", got[1]) + } +} + +func TestHub_UnfocusDebouncesDownTransition(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + h := newTestHub(t) + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c := newTestClient("c1") + h.SubscribeToSession(c, "sess-1") + h.FocusSession(c, "sess-1") + + // Let all goroutines settle so the two synchronous listener calls complete. + synctest.Wait() + + rec.mu.Lock() + if len(rec.events) != 2 { + t.Fatalf("setup: expected 2 events, got %d", len(rec.events)) + } + rec.mu.Unlock() + + // Unfocus — schedules a debounced down-transition timer. synctest's fake + // clock doesn't advance until we ask, so the timer cannot fire. + h.UnfocusSession(c, "sess-1") + synctest.Wait() + + rec.mu.Lock() + gotCount := len(rec.events) + rec.mu.Unlock() + if gotCount != 2 { + t.Errorf("down-transition fired immediately; expected 2 events, got %d (%+v)", gotCount, rec.events) + } + }) +} + +func TestHub_RefocusWithinDebounceCancelsDownTransition(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + h := newTestHub(t) + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c := newTestClient("c1") + h.SubscribeToSession(c, "sess-1") + h.FocusSession(c, "sess-1") + synctest.Wait() + + rec.mu.Lock() + if len(rec.events) != 2 { + t.Fatalf("setup: expected 2 events, got %d", len(rec.events)) + } + rec.mu.Unlock() + + // Unfocus then re-focus quickly — cancelling the pending down-transition + // timer clears lastMode, so the re-focus fires a fresh up-transition + // (3rd event: fast). This is intentional: it ensures new clients after a + // page refresh receive the current mode immediately. + h.UnfocusSession(c, "sess-1") + h.FocusSession(c, "sess-1") + synctest.Wait() + + // Advance fake clock past the debounce window to prove the timer was + // cancelled and won't fire a down-transition. + time.Sleep(100 * time.Millisecond) + synctest.Wait() + + rec.mu.Lock() + got := rec.events + rec.mu.Unlock() + if len(got) != 3 { + t.Errorf("expected 3 events (slow, fast, fast); got %d: %+v", len(got), got) + } + if len(got) == 3 && got[2].mode != SessionModeFast { + t.Errorf("expected 3rd event to be fast (re-focus); got %v", got[2].mode) + } + }) +} + +func TestHub_UnsubscribeFromOnlySubscriberFiresPaused(t *testing.T) { + h := newTestHub(t) + // Override debounce via a faster path: rely on the existing listener and + // just wait the debounce period in the test. + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c := newTestClient("c1") + h.SubscribeToSession(c, "sess-1") // slow + rec.waitForCount(1, time.Second) + + h.UnsubscribeFromSession(c, "sess-1") + // Wait test debounce (50ms) plus margin. + got := rec.waitForCount(2, 1*time.Second) + + if len(got) < 2 { + t.Fatalf("expected paused event after debounce, got %+v", got) + } + if got[1].mode != SessionModePaused { + t.Errorf("second event mode = %v, want paused", got[1].mode) + } +} + +func TestHub_DisconnectCleansBothMaps(t *testing.T) { + h := newTestHub(t) + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c := newTestClient("c1") + // Manually register the client (no real WS connection in tests). + h.mu.Lock() + h.clients[c] = true + h.mu.Unlock() + + h.SubscribeToSession(c, "sess-1") + h.FocusSession(c, "sess-1") + rec.waitForCount(2, time.Second) + + h.removeClient(c) + + h.mu.RLock() + subscriberCount := len(h.sessionSubscribers["sess-1"]) + focusCount := len(h.sessionMode.focusByClient["sess-1"]) + h.mu.RUnlock() + + if subscriberCount != 0 { + t.Errorf("after disconnect subscriber count = %d, want 0", subscriberCount) + } + if focusCount != 0 { + t.Errorf("after disconnect focus count = %d, want 0", focusCount) + } +} + +func TestHub_MultipleClientsKeepWorkspaceFast(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + h := newTestHub(t) + rec := newModeRecorder() + h.AddSessionModeListener(rec.listener()) + + c1 := newTestClient("c1") + c2 := newTestClient("c2") + + h.SubscribeToSession(c1, "sess-1") + h.FocusSession(c1, "sess-1") + h.SubscribeToSession(c2, "sess-1") + h.FocusSession(c2, "sess-1") + synctest.Wait() + + rec.mu.Lock() + startCount := len(rec.events) + rec.mu.Unlock() + + // One client unfocuses — the other still has focus, so mode should stay fast. + h.UnfocusSession(c1, "sess-1") + synctest.Wait() + + // Advance fake clock past the debounce window. + time.Sleep(100 * time.Millisecond) + synctest.Wait() + + rec.mu.Lock() + got := len(rec.events) + rec.mu.Unlock() + if got != startCount { + t.Errorf("expected no new events while another client still focused; got %d new (events: %+v)", got-startCount, rec.events) + } + }) +} + +func TestIsUpTransition(t *testing.T) { + cases := []struct { + old, new SessionMode + want bool + }{ + {SessionModePaused, SessionModeSlow, true}, + {SessionModePaused, SessionModeFast, true}, + {SessionModeSlow, SessionModeFast, true}, + {SessionModeFast, SessionModeSlow, false}, + {SessionModeSlow, SessionModePaused, false}, + {SessionModeFast, SessionModePaused, false}, + {SessionModeFast, SessionModeFast, false}, // not a transition + } + for _, tc := range cases { + if got := isUpTransition(tc.old, tc.new); got != tc.want { + t.Errorf("isUpTransition(%v, %v) = %v, want %v", tc.old, tc.new, got, tc.want) + } + } +} diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index 45c82493cb..9be774f367 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -339,6 +339,11 @@ type AgentStartFailedFunc func(ctx context.Context, taskID, sessionID, agentExec // Useful for creating user-facing status messages tied to launch errors. type LaunchFailedFunc func(ctx context.Context, taskID, sessionID string, err error) +// PrimarySessionSetFunc is called when the first session for a task is marked +// primary. This lets the orchestrator publish a task.updated event so the +// frontend receives the primary_session_id. +type PrimarySessionSetFunc func(ctx context.Context, taskID, sessionID string) + // ExecutorTypeCapabilities provides behavioral queries about executor types. // Implemented by the lifecycle manager using its backend registry. type ExecutorTypeCapabilities interface { @@ -377,6 +382,9 @@ type Executor struct { // to emit user-friendly guidance for known failure patterns. onLaunchFailed LaunchFailedFunc + // Callback when the first session for a task is marked primary. + onPrimarySessionSet PrimarySessionSetFunc + // Per-session locks to prevent concurrent resume/launch operations on the same session. // This prevents race conditions when the backend restarts and multiple resume requests // arrive simultaneously (e.g., from frontend auto-resume). @@ -450,6 +458,13 @@ func (e *Executor) SetOnAgentStartFailed(fn AgentStartFailedFunc) { e.onAgentStartFailed = fn } +// SetOnPrimarySessionSet sets a callback for when the first session for a task +// is marked primary. This publishes a task.updated event so the frontend +// receives primary_session_id. +func (e *Executor) SetOnPrimarySessionSet(fn PrimarySessionSetFunc) { + e.onPrimarySessionSet = fn +} + // SetOnLaunchFailed sets a callback for launch failures that happen before // the agent process has started. func (e *Executor) SetOnLaunchFailed(fn LaunchFailedFunc) { diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index d57f5d4dd9..26ec7eafce 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -282,6 +282,9 @@ func (e *Executor) PrepareSession(ctx context.Context, task *v1.Task, agentProfi zap.String("session_id", sessionID), zap.Error(err)) } + if e.onPrimarySessionSet != nil { + e.onPrimarySessionSet(ctx, task.ID, sessionID) + } } e.logger.Info("session entry created", diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index 14082e9f18..6927e906da 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -59,6 +59,36 @@ func TestPrepareSession_Success(t *testing.T) { } } +func TestPrepareSession_InvokesPrimarySessionCallback(t *testing.T) { + repo := newMockRepository() + agentManager := &mockAgentManager{} + exec := newTestExecutor(t, agentManager, repo) + + var callbackTaskID, callbackSessionID string + exec.SetOnPrimarySessionSet(func(_ context.Context, taskID, sessionID string) { + callbackTaskID = taskID + callbackSessionID = sessionID + }) + + task := &v1.Task{ + ID: "task-456", + WorkspaceID: "workspace-456", + Title: "Callback Test", + } + + sessionID, err := exec.PrepareSession(context.Background(), task, "profile-1", "executor-1", "", "step-1") + if err != nil { + t.Fatalf("PrepareSession failed: %v", err) + } + + if callbackTaskID != task.ID { + t.Errorf("callback taskID = %q, want %q", callbackTaskID, task.ID) + } + if callbackSessionID != sessionID { + t.Errorf("callback sessionID = %q, want %q", callbackSessionID, sessionID) + } +} + func TestPrepareSession_NoAgentProfileID(t *testing.T) { repo := newMockRepository() agentManager := &mockAgentManager{} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index a171e32343..bbcba1ce26 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -348,6 +348,13 @@ func (s *Service) SetMessageCreator(mc MessageCreator) { s.messageCreator = mc } +// SetOnPrimarySessionSet sets a callback on the executor for when the first session +// of a task is marked primary. Used to publish a task.updated event so the frontend +// receives the primary_session_id. +func (s *Service) SetOnPrimarySessionSet(fn executor.PrimarySessionSetFunc) { + s.executor.SetOnPrimarySessionSet(fn) +} + // SetRepoCloner sets the repository cloner and updater on the executor, enabling automatic // cloning of provider-backed repositories (e.g. from a GitHub URL) when they are launched // for local/worktree execution and have no local path yet. diff --git a/apps/backend/internal/task/service/service_events.go b/apps/backend/internal/task/service/service_events.go index edfb5eef75..a9c92c5166 100644 --- a/apps/backend/internal/task/service/service_events.go +++ b/apps/backend/internal/task/service/service_events.go @@ -14,6 +14,13 @@ import ( "github.com/kandev/kandev/internal/task/models" ) +// PublishTaskUpdated publishes a task.updated event for the given task. +// Used when task metadata changes (e.g., primary session assignment) that +// don't go through the normal UpdateTask path. +func (s *Service) PublishTaskUpdated(ctx context.Context, task *models.Task) { + s.publishTaskEvent(ctx, events.TaskUpdated, task, nil) +} + // publishTaskEvent publishes task events to the event bus func (s *Service) publishTaskEvent(ctx context.Context, eventType string, task *models.Task, oldState *v1.TaskState) { if s.eventBus == nil { diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index ddc5d81f0d..ee37e6c92c 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -131,8 +131,13 @@ const ( ActionTaskUnsubscribe = "task.unsubscribe" ActionSessionSubscribe = "session.subscribe" ActionSessionUnsubscribe = "session.unsubscribe" - ActionUserSubscribe = "user.subscribe" - ActionUserUnsubscribe = "user.unsubscribe" + // Focus signals are layered on top of subscriptions to indicate which + // session the user is actively viewing (task details page or task panel), + // vs merely subscribed (sidebar diff badges). Drives backend polling tier. + ActionSessionFocus = "session.focus" + ActionSessionUnfocus = "session.unfocus" + ActionUserSubscribe = "user.subscribe" + ActionUserUnsubscribe = "user.unsubscribe" // Message actions ActionMessageAdd = "message.add" @@ -180,6 +185,7 @@ const ( ActionSessionSetMode = "session.set_mode" ActionSessionTodosUpdated = "session.todos_updated" ActionSessionPromptUsage = "session.prompt_usage" + ActionSessionPollModeChanged = "session.poll_mode_changed" ActionInputRequested = "input.requested" ActionRepositoryCreated = "repository.created" ActionRepositoryUpdated = "repository.updated" diff --git a/apps/web/components/task/task-item.tsx b/apps/web/components/task/task-item.tsx index 819f9c2221..5319874423 100644 --- a/apps/web/components/task/task-item.tsx +++ b/apps/web/components/task/task-item.tsx @@ -10,7 +10,10 @@ import { import { PRTaskIcon } from "@/components/github/pr-task-icon"; import { useAppStore } from "@/components/state-provider"; import { cn } from "@/lib/utils"; +import { DEBUG_UI } from "@/lib/config"; import type { TaskState, TaskSessionState } from "@/lib/types/http"; +import type { SessionPollMode } from "@/lib/state/slices/session-runtime/types"; +import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; import { RemoteCloudTooltip } from "./remote-cloud-tooltip"; import { classifyTask } from "./task-classify"; import { ScrollOnOverflow } from "@kandev/ui/scroll-on-overflow"; @@ -106,20 +109,48 @@ function TaskStateIcon({ ); } +const POLL_MODE_CONFIG: Record = + { + fast: { letter: "F", color: "text-emerald-500", label: "focused, 2s polling" }, + slow: { letter: "S", color: "text-yellow-500", label: "subscribed, 30s polling" }, + paused: { letter: "P", color: "text-muted-foreground/40", label: "no subscribers" }, + }; + function TaskItemStatsRow({ updatedAt, prInfo, + primarySessionId, }: { updatedAt?: string; prInfo?: { number: number; state: string }; + primarySessionId?: string | null; }) { - if (!updatedAt && !prInfo) return null; + const pollMode = useAppStore((s) => + DEBUG_UI && primarySessionId ? (s.sessionPollMode.bySessionId[primarySessionId] ?? null) : null, + ); + + if (!updatedAt && !prInfo && !pollMode) return null; + + const modeConfig = pollMode ? POLL_MODE_CONFIG[pollMode] : null; + return ( {updatedAt && ( {formatRelativeTime(updatedAt)} )} {prInfo && #{prInfo.number}} + {modeConfig && ( + + + + {modeConfig.letter} + + + + Git poll: {pollMode} ({modeConfig.label}) + + + )} ); } @@ -211,7 +242,7 @@ function TaskItemContent({ {repositories.join(" · ")} )} - + ); } diff --git a/apps/web/components/task/task-page-content.tsx b/apps/web/components/task/task-page-content.tsx index 6b68154e60..a2cc01f930 100644 --- a/apps/web/components/task/task-page-content.tsx +++ b/apps/web/components/task/task-page-content.tsx @@ -13,6 +13,7 @@ import { useRepositories } from "@/hooks/domains/workspace/use-repositories"; import { useSessionAgent } from "@/hooks/domains/session/use-session-agent"; import { useSessionResumption } from "@/hooks/domains/session/use-session-resumption"; import { useSessionAgentctl } from "@/hooks/domains/session/use-session-agentctl"; +import { useTaskFocus } from "@/hooks/domains/session/use-task-focus"; import { useAppStore } from "@/components/state-provider"; import { useTaskSessions } from "@/hooks/use-task-sessions"; import { fetchTask } from "@/lib/api"; @@ -602,6 +603,9 @@ export function TaskPageContent({ const resumption = useSessionResumption(task?.id ?? null, effectiveSessionId); const merged = useMergedAgentState(agent, resumption, sessionPanel, effectiveSessionId, task); const archivedValue = useMemo(() => buildArchivedValue(task, repository), [task, repository]); + // Mark this session as actively focused so the backend lifts polling to fast. + // Sidebar cards subscribe but never focus, so they stay on the cheap slow tier. + useTaskFocus(effectiveSessionId); useEffect(() => { queueMicrotask(() => setIsMounted(true)); diff --git a/apps/web/e2e/tests/session/session-focus-signal.spec.ts b/apps/web/e2e/tests/session/session-focus-signal.spec.ts new file mode 100644 index 0000000000..b6ef7fd982 --- /dev/null +++ b/apps/web/e2e/tests/session/session-focus-signal.spec.ts @@ -0,0 +1,64 @@ +import { test, expect } from "../../fixtures/test-base"; +import { SessionPage } from "../../pages/session-page"; + +/** + * Verifies the session.focus WS protocol added for focus-gated git polling. + * + * - Sidebar cards subscribe but do NOT focus -> only get slow polling. + * - Opening a task details page sends session.focus -> backend lifts to fast. + * + * We assert at the WS frame level (cheaper, more deterministic than asserting + * polling cadence in agentctl). + * + * Note: unfocus (cleanup on unmount) is covered by the useTaskFocus unit test. + * E2E can't reliably test it because Playwright's page.goto is a full-page + * navigation that tears down the WS before React effects clean up. + */ +test.describe("Session focus signal", () => { + test("task details page sends subscribe and focus", async ({ testPage, apiClient, seedData }) => { + test.setTimeout(60_000); + + // Capture sent WS frames from the moment the page opens. + const sentFrames: string[] = []; + testPage.on("websocket", (ws) => { + ws.on("framesent", (event) => { + const data = typeof event.payload === "string" ? event.payload : event.payload?.toString(); + if (data) sentFrames.push(data); + }); + }); + + const countFrames = (action: string) => + sentFrames.filter((f) => f.includes(`"action":"${action}"`)).length; + + const task = await apiClient.createTask(seedData.workspaceId, "Focus Signal Task", { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + agent_profile_id: seedData.agentProfileId, + }); + + // Visit the task page — this triggers useTaskFocus(sessionId) on mount. + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.idleInput()).toBeVisible({ timeout: 45_000 }); + + // Poll for the expected WS frames. The page mount + auto-start + WS + // connect sequence races, so useTaskFocus may fire a few hundred ms + // after the agent goes idle. expect.poll handles this gracefully. + await expect + .poll(() => countFrames("session.subscribe"), { + message: "expected at least one session.subscribe frame", + timeout: 10_000, + }) + .toBeGreaterThan(0); + + await expect + .poll(() => countFrames("session.focus"), { + message: "expected at least one session.focus frame from task page", + timeout: 10_000, + }) + .toBeGreaterThan(0); + }); +}); diff --git a/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts b/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts index 9b92043a67..1df19ac578 100644 --- a/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts +++ b/apps/web/hooks/domains/kanban/use-all-workflow-snapshots.ts @@ -82,9 +82,27 @@ export function useAllWorkflowSnapshots(workspaceId: string | null) { })); const stepIds = new Set(steps.map((s) => s.id)); + // Preserve runtime fields (e.g., primarySessionId) from existing + // snapshot tasks when the fresh API response omits them. The backend + // uses omitempty for session fields, so a task whose session was just + // created may not have primary_session_id in the snapshot response + // yet, even though a WS task.updated event already delivered it. + const existingSnapshot = store.getState().kanbanMulti.snapshots[wf.id]; + const existingById = new Map((existingSnapshot?.tasks ?? []).map((t) => [t.id, t])); + const tasks: KanbanTask[] = snapshot.tasks .filter((task) => !task.is_ephemeral) // Filter out ephemeral tasks (e.g., quick chat) - .map((task) => mapSnapshotTask(task, stepIds)) + .map((task) => { + const mapped = mapSnapshotTask(task, stepIds); + if (!mapped) return null; + const existing = existingById.get(mapped.id); + if (existing) { + mapped.primarySessionId = mapped.primarySessionId || existing.primarySessionId; + mapped.primarySessionState = + mapped.primarySessionState || existing.primarySessionState; + } + return mapped; + }) .filter((t): t is KanbanTask => t !== null); setWorkflowSnapshot(wf.id, { diff --git a/apps/web/hooks/domains/session/use-session-state.test.ts b/apps/web/hooks/domains/session/use-session-state.test.ts index 18079848ac..7348a50785 100644 --- a/apps/web/hooks/domains/session/use-session-state.test.ts +++ b/apps/web/hooks/domains/session/use-session-state.test.ts @@ -35,7 +35,7 @@ import { useSessionState } from "./use-session-state"; const createMockSession = ( id: string, taskId: string, - state: TaskSession["state"] = "IDLE", + state: TaskSession["state"] = "CREATED", ): TaskSession => ({ id, diff --git a/apps/web/hooks/domains/session/use-task-focus.test.ts b/apps/web/hooks/domains/session/use-task-focus.test.ts new file mode 100644 index 0000000000..b5e9dda467 --- /dev/null +++ b/apps/web/hooks/domains/session/use-task-focus.test.ts @@ -0,0 +1,65 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { renderHook } from "@testing-library/react"; + +const mockFocusSession = vi.fn(); +const mockUnsubscribe = vi.fn(); +let mockConnectionStatus = "connected"; + +vi.mock("@/components/state-provider", () => ({ + useAppStore: (selector: (state: { connection: { status: string } }) => unknown) => + selector({ connection: { status: mockConnectionStatus } }), +})); + +vi.mock("@/lib/ws/connection", () => ({ + getWebSocketClient: () => ({ + focusSession: mockFocusSession, + }), +})); + +import { useTaskFocus } from "./use-task-focus"; + +describe("useTaskFocus", () => { + beforeEach(() => { + vi.clearAllMocks(); + mockFocusSession.mockReturnValue(mockUnsubscribe); + mockConnectionStatus = "connected"; + }); + + it("calls focusSession on mount when connected", () => { + renderHook(() => useTaskFocus("sess-1")); + expect(mockFocusSession).toHaveBeenCalledWith("sess-1"); + }); + + it("calls the returned cleanup on unmount", () => { + const { unmount } = renderHook(() => useTaskFocus("sess-1")); + expect(mockUnsubscribe).not.toHaveBeenCalled(); + unmount(); + expect(mockUnsubscribe).toHaveBeenCalled(); + }); + + it("does nothing when no sessionId is provided", () => { + renderHook(() => useTaskFocus(null)); + expect(mockFocusSession).not.toHaveBeenCalled(); + }); + + it("does nothing when WebSocket is not connected", () => { + mockConnectionStatus = "connecting"; + renderHook(() => useTaskFocus("sess-1")); + expect(mockFocusSession).not.toHaveBeenCalled(); + }); + + it("re-focuses when sessionId changes", () => { + const { rerender } = renderHook(({ id }: { id: string | null }) => useTaskFocus(id), { + initialProps: { id: "sess-1" }, + }); + expect(mockFocusSession).toHaveBeenCalledTimes(1); + expect(mockFocusSession).toHaveBeenLastCalledWith("sess-1"); + expect(mockUnsubscribe).not.toHaveBeenCalled(); + + rerender({ id: "sess-2" }); + // Cleanup of previous focus should fire, then a new focus. + expect(mockUnsubscribe).toHaveBeenCalledTimes(1); + expect(mockFocusSession).toHaveBeenCalledTimes(2); + expect(mockFocusSession).toHaveBeenLastCalledWith("sess-2"); + }); +}); diff --git a/apps/web/hooks/domains/session/use-task-focus.ts b/apps/web/hooks/domains/session/use-task-focus.ts new file mode 100644 index 0000000000..a724bc90e6 --- /dev/null +++ b/apps/web/hooks/domains/session/use-task-focus.ts @@ -0,0 +1,30 @@ +import { useEffect } from "react"; +import { useAppStore } from "@/components/state-provider"; +import { getWebSocketClient } from "@/lib/ws/connection"; + +/** + * Marks a session as actively focused by the user — the task details page or + * the task panel modal. The backend uses this signal (separate from + * subscription) to lift the workspace's git polling into fast mode. + * + * Subscriptions remain unchanged: they say "I want updates"; focus says + * "I'm currently looking at this". Sidebar cards subscribe but never focus, + * so they get slow polling. The page that actually shows the task focuses, + * lifting that workspace to fast. + * + * Multiple components can call useTaskFocus(sameSessionId) safely — the WS + * client ref-counts focus calls so only one focus signal is sent backend-side. + */ +export function useTaskFocus(sessionId: string | null | undefined) { + const connectionStatus = useAppStore((state) => state.connection.status); + + useEffect(() => { + if (!sessionId) return; + if (connectionStatus !== "connected") return; + + const client = getWebSocketClient(); + if (!client) return; + + return client.focusSession(sessionId); + }, [sessionId, connectionStatus]); +} diff --git a/apps/web/lib/state/default-state.ts b/apps/web/lib/state/default-state.ts new file mode 100644 index 0000000000..2fbfab47c5 --- /dev/null +++ b/apps/web/lib/state/default-state.ts @@ -0,0 +1,155 @@ +import { + defaultKanbanState, + defaultWorkspaceState, + defaultSettingsState, + defaultSessionState, + defaultSessionRuntimeState, + defaultUIState, + defaultGitHubState, +} from "./slices"; + +export const defaultState = { + kanban: defaultKanbanState.kanban, + kanbanMulti: defaultKanbanState.kanbanMulti, + workflows: defaultKanbanState.workflows, + tasks: defaultKanbanState.tasks, + workspaces: defaultWorkspaceState.workspaces, + repositories: defaultWorkspaceState.repositories, + repositoryBranches: defaultWorkspaceState.repositoryBranches, + repositoryScripts: defaultWorkspaceState.repositoryScripts, + executors: defaultSettingsState.executors, + settingsAgents: defaultSettingsState.settingsAgents, + agentDiscovery: defaultSettingsState.agentDiscovery, + availableAgents: defaultSettingsState.availableAgents, + agentProfiles: defaultSettingsState.agentProfiles, + editors: defaultSettingsState.editors, + prompts: defaultSettingsState.prompts, + secrets: defaultSettingsState.secrets, + notificationProviders: defaultSettingsState.notificationProviders, + settingsData: defaultSettingsState.settingsData, + userSettings: defaultSettingsState.userSettings, + messages: defaultSessionState.messages, + turns: defaultSessionState.turns, + taskSessions: defaultSessionState.taskSessions, + taskSessionsByTask: defaultSessionState.taskSessionsByTask, + sessionAgentctl: defaultSessionState.sessionAgentctl, + worktrees: defaultSessionState.worktrees, + sessionWorktreesBySessionId: defaultSessionState.sessionWorktreesBySessionId, + pendingModel: defaultSessionState.pendingModel, + activeModel: defaultSessionState.activeModel, + taskPlans: defaultSessionState.taskPlans, + queue: defaultSessionState.queue, + terminal: defaultSessionRuntimeState.terminal, + shell: defaultSessionRuntimeState.shell, + processes: defaultSessionRuntimeState.processes, + gitStatus: defaultSessionRuntimeState.gitStatus, + environmentIdBySessionId: defaultSessionRuntimeState.environmentIdBySessionId, + sessionCommits: defaultSessionRuntimeState.sessionCommits, + contextWindow: defaultSessionRuntimeState.contextWindow, + agents: defaultSessionRuntimeState.agents, + availableCommands: defaultSessionRuntimeState.availableCommands, + sessionMode: defaultSessionRuntimeState.sessionMode, + userShells: defaultSessionRuntimeState.userShells, + prepareProgress: defaultSessionRuntimeState.prepareProgress, + sessionTodos: defaultSessionRuntimeState.sessionTodos, + agentCapabilities: defaultSessionRuntimeState.agentCapabilities, + sessionModels: defaultSessionRuntimeState.sessionModels, + promptUsage: defaultSessionRuntimeState.promptUsage, + sessionPollMode: defaultSessionRuntimeState.sessionPollMode, + githubStatus: defaultGitHubState.githubStatus, + taskPRs: defaultGitHubState.taskPRs, + prWatches: defaultGitHubState.prWatches, + reviewWatches: defaultGitHubState.reviewWatches, + previewPanel: defaultUIState.previewPanel, + rightPanel: defaultUIState.rightPanel, + diffs: defaultUIState.diffs, + connection: defaultUIState.connection, + mobileKanban: defaultUIState.mobileKanban, + mobileSession: defaultUIState.mobileSession, + chatInput: defaultUIState.chatInput, + documentPanel: defaultUIState.documentPanel, + systemHealth: defaultUIState.systemHealth, + quickChat: defaultUIState.quickChat, + sessionFailureNotification: defaultUIState.sessionFailureNotification, + bottomTerminal: defaultUIState.bottomTerminal, +}; + +export type DefaultState = typeof defaultState; + +export function mergeInitialState(initialState?: Partial): DefaultState { + if (!initialState) return defaultState; + + return { + ...defaultState, + ...initialState, + // Ensure nested objects are properly merged + kanban: { ...defaultState.kanban, ...initialState.kanban }, + kanbanMulti: { ...defaultState.kanbanMulti, ...initialState.kanbanMulti }, + workflows: { ...defaultState.workflows, ...initialState.workflows }, + tasks: { ...defaultState.tasks, ...initialState.tasks }, + workspaces: { ...defaultState.workspaces, ...initialState.workspaces }, + repositories: { ...defaultState.repositories, ...initialState.repositories }, + repositoryBranches: { ...defaultState.repositoryBranches, ...initialState.repositoryBranches }, + repositoryScripts: { ...defaultState.repositoryScripts, ...initialState.repositoryScripts }, + executors: { ...defaultState.executors, ...initialState.executors }, + settingsAgents: { ...defaultState.settingsAgents, ...initialState.settingsAgents }, + agentDiscovery: { ...defaultState.agentDiscovery, ...initialState.agentDiscovery }, + availableAgents: { ...defaultState.availableAgents, ...initialState.availableAgents }, + agentProfiles: { ...defaultState.agentProfiles, ...initialState.agentProfiles }, + editors: { ...defaultState.editors, ...initialState.editors }, + prompts: { ...defaultState.prompts, ...initialState.prompts }, + secrets: { ...defaultState.secrets, ...initialState.secrets }, + notificationProviders: { + ...defaultState.notificationProviders, + ...initialState.notificationProviders, + }, + settingsData: { ...defaultState.settingsData, ...initialState.settingsData }, + userSettings: { ...defaultState.userSettings, ...initialState.userSettings }, + messages: { ...defaultState.messages, ...initialState.messages }, + turns: { ...defaultState.turns, ...initialState.turns }, + taskSessions: { ...defaultState.taskSessions, ...initialState.taskSessions }, + taskSessionsByTask: { ...defaultState.taskSessionsByTask, ...initialState.taskSessionsByTask }, + sessionAgentctl: { ...defaultState.sessionAgentctl, ...initialState.sessionAgentctl }, + worktrees: { ...defaultState.worktrees, ...initialState.worktrees }, + sessionWorktreesBySessionId: { + ...defaultState.sessionWorktreesBySessionId, + ...initialState.sessionWorktreesBySessionId, + }, + pendingModel: { ...defaultState.pendingModel, ...initialState.pendingModel }, + activeModel: { ...defaultState.activeModel, ...initialState.activeModel }, + taskPlans: { ...defaultState.taskPlans, ...initialState.taskPlans }, + queue: { ...defaultState.queue, ...initialState.queue }, + terminal: { ...defaultState.terminal, ...initialState.terminal }, + shell: { ...defaultState.shell, ...initialState.shell }, + processes: { ...defaultState.processes, ...initialState.processes }, + gitStatus: { ...defaultState.gitStatus, ...initialState.gitStatus }, + sessionCommits: { ...defaultState.sessionCommits, ...initialState.sessionCommits }, + contextWindow: { ...defaultState.contextWindow, ...initialState.contextWindow }, + agents: { ...defaultState.agents, ...initialState.agents }, + sessionMode: { ...defaultState.sessionMode, ...initialState.sessionMode }, + userShells: { ...defaultState.userShells, ...initialState.userShells }, + prepareProgress: { ...defaultState.prepareProgress, ...initialState.prepareProgress }, + sessionTodos: { ...defaultState.sessionTodos, ...initialState.sessionTodos }, + agentCapabilities: { ...defaultState.agentCapabilities, ...initialState.agentCapabilities }, + sessionModels: { ...defaultState.sessionModels, ...initialState.sessionModels }, + promptUsage: { ...defaultState.promptUsage, ...initialState.promptUsage }, + sessionPollMode: { ...defaultState.sessionPollMode, ...initialState.sessionPollMode }, + githubStatus: { ...defaultState.githubStatus, ...initialState.githubStatus }, + taskPRs: { ...defaultState.taskPRs, ...initialState.taskPRs }, + prWatches: { ...defaultState.prWatches, ...initialState.prWatches }, + reviewWatches: { ...defaultState.reviewWatches, ...initialState.reviewWatches }, + previewPanel: { ...defaultState.previewPanel, ...initialState.previewPanel }, + rightPanel: { ...defaultState.rightPanel, ...initialState.rightPanel }, + diffs: { ...defaultState.diffs, ...initialState.diffs }, + connection: { ...defaultState.connection, ...initialState.connection }, + mobileKanban: { ...defaultState.mobileKanban, ...initialState.mobileKanban }, + mobileSession: { ...defaultState.mobileSession, ...initialState.mobileSession }, + chatInput: { ...defaultState.chatInput, ...initialState.chatInput }, + documentPanel: { ...defaultState.documentPanel, ...initialState.documentPanel }, + systemHealth: { ...defaultState.systemHealth, ...initialState.systemHealth }, + quickChat: { ...defaultState.quickChat, ...initialState.quickChat }, + sessionFailureNotification: + initialState.sessionFailureNotification ?? defaultState.sessionFailureNotification, + bottomTerminal: { ...defaultState.bottomTerminal, ...initialState.bottomTerminal }, + }; +} diff --git a/apps/web/lib/state/slices/session-runtime/migrate-env-keyed-data.test.ts b/apps/web/lib/state/slices/session-runtime/migrate-env-keyed-data.test.ts index d51b7df3cf..f2fb08b9fa 100644 --- a/apps/web/lib/state/slices/session-runtime/migrate-env-keyed-data.test.ts +++ b/apps/web/lib/state/slices/session-runtime/migrate-env-keyed-data.test.ts @@ -14,11 +14,11 @@ type CombinedSlice = SessionRuntimeSlice & SessionSlice; function makeCombinedStore() { return create()( - immer((set, get, api) => ({ + immer((set) => ({ // eslint-disable-next-line @typescript-eslint/no-explicit-any - ...createSessionSlice(set as any), + ...(createSessionSlice as any)(set), // eslint-disable-next-line @typescript-eslint/no-explicit-any - ...createSessionRuntimeSlice(set as any, get as any, api as any), + ...(createSessionRuntimeSlice as any)(set), })), ); } diff --git a/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts b/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts index 0217735f8c..b0109d0bad 100644 --- a/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts +++ b/apps/web/lib/state/slices/session-runtime/session-runtime-slice.ts @@ -2,6 +2,7 @@ import type { StateCreator } from "zustand"; import type { SessionRuntimeSlice, SessionRuntimeSliceState, + SessionPollMode, GitStatusEntry, FileInfo, } from "./types"; @@ -89,6 +90,7 @@ export const defaultSessionRuntimeState: SessionRuntimeSliceState = { sessionTodos: { bySessionId: {} }, userShells: { byEnvironmentId: {}, loading: {}, loaded: {} }, prepareProgress: { bySessionId: {} }, + sessionPollMode: { bySessionId: {} }, }; type ImmerSet = Parameters[0]; @@ -182,6 +184,10 @@ function buildUserShellActions(set: ImmerSet) { (s) => s.terminalId !== terminalId, ); }), + setSessionPollMode: (sessionId: string, mode: SessionPollMode) => + set((draft) => { + draft.sessionPollMode.bySessionId[sessionId] = mode; + }), }; } diff --git a/apps/web/lib/state/slices/session-runtime/types.ts b/apps/web/lib/state/slices/session-runtime/types.ts index 2bd3e52fd6..b27961192d 100644 --- a/apps/web/lib/state/slices/session-runtime/types.ts +++ b/apps/web/lib/state/slices/session-runtime/types.ts @@ -255,6 +255,12 @@ export type SessionTodosState = { bySessionId: Record; }; +export type SessionPollMode = "fast" | "slow" | "paused"; + +export type SessionPollModeState = { + bySessionId: Record; +}; + export type SessionRuntimeSliceState = { terminal: TerminalState; shell: ShellState; @@ -273,6 +279,7 @@ export type SessionRuntimeSliceState = { sessionTodos: SessionTodosState; userShells: UserShellsState; prepareProgress: PrepareProgressState; + sessionPollMode: SessionPollModeState; }; export type SessionRuntimeSliceActions = { @@ -322,6 +329,7 @@ export type SessionRuntimeSliceActions = { setUserShellsLoading: (sessionId: string, loading: boolean) => void; addUserShell: (sessionId: string, shell: UserShellInfo) => void; removeUserShell: (sessionId: string, terminalId: string) => void; + setSessionPollMode: (sessionId: string, mode: SessionPollMode) => void; }; export type SessionRuntimeSlice = SessionRuntimeSliceState & SessionRuntimeSliceActions; diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index 14c4b0d1a4..15c8048b83 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -16,6 +16,7 @@ import type { ReviewWatch as GitHubReviewWatch, } from "@/lib/types/github"; import type { SystemHealthResponse } from "@/lib/types/health"; +import { mergeInitialState } from "./default-state"; import { createKanbanSlice, createWorkspaceSlice, @@ -62,6 +63,7 @@ import type { SessionModelEntry, ConfigOptionEntry, PromptUsageEntry, + SessionPollMode, TodoEntry, UserShellInfo, } from "./slices/session-runtime/types"; @@ -187,6 +189,7 @@ export type AppState = { agentCapabilities: (typeof defaultSessionRuntimeState)["agentCapabilities"]; sessionModels: (typeof defaultSessionRuntimeState)["sessionModels"]; promptUsage: (typeof defaultSessionRuntimeState)["promptUsage"]; + sessionPollMode: (typeof defaultSessionRuntimeState)["sessionPollMode"]; // GitHub slice githubStatus: (typeof defaultGitHubState)["githubStatus"]; @@ -407,152 +410,11 @@ export type AppState = { setUserShellsLoading: (sessionId: string, loading: boolean) => void; addUserShell: (sessionId: string, shell: UserShellInfo) => void; removeUserShell: (sessionId: string, terminalId: string) => void; + setSessionPollMode: (sessionId: string, mode: SessionPollMode) => void; }; export type AppStore = ReturnType; -const defaultState = { - kanban: defaultKanbanState.kanban, - kanbanMulti: defaultKanbanState.kanbanMulti, - workflows: defaultKanbanState.workflows, - tasks: defaultKanbanState.tasks, - workspaces: defaultWorkspaceState.workspaces, - repositories: defaultWorkspaceState.repositories, - repositoryBranches: defaultWorkspaceState.repositoryBranches, - repositoryScripts: defaultWorkspaceState.repositoryScripts, - executors: defaultSettingsState.executors, - settingsAgents: defaultSettingsState.settingsAgents, - agentDiscovery: defaultSettingsState.agentDiscovery, - availableAgents: defaultSettingsState.availableAgents, - agentProfiles: defaultSettingsState.agentProfiles, - editors: defaultSettingsState.editors, - prompts: defaultSettingsState.prompts, - secrets: defaultSettingsState.secrets, - notificationProviders: defaultSettingsState.notificationProviders, - settingsData: defaultSettingsState.settingsData, - userSettings: defaultSettingsState.userSettings, - messages: defaultSessionState.messages, - turns: defaultSessionState.turns, - taskSessions: defaultSessionState.taskSessions, - taskSessionsByTask: defaultSessionState.taskSessionsByTask, - sessionAgentctl: defaultSessionState.sessionAgentctl, - worktrees: defaultSessionState.worktrees, - sessionWorktreesBySessionId: defaultSessionState.sessionWorktreesBySessionId, - pendingModel: defaultSessionState.pendingModel, - activeModel: defaultSessionState.activeModel, - taskPlans: defaultSessionState.taskPlans, - queue: defaultSessionState.queue, - terminal: defaultSessionRuntimeState.terminal, - shell: defaultSessionRuntimeState.shell, - processes: defaultSessionRuntimeState.processes, - gitStatus: defaultSessionRuntimeState.gitStatus, - environmentIdBySessionId: defaultSessionRuntimeState.environmentIdBySessionId, - sessionCommits: defaultSessionRuntimeState.sessionCommits, - contextWindow: defaultSessionRuntimeState.contextWindow, - agents: defaultSessionRuntimeState.agents, - availableCommands: defaultSessionRuntimeState.availableCommands, - sessionMode: defaultSessionRuntimeState.sessionMode, - userShells: defaultSessionRuntimeState.userShells, - prepareProgress: defaultSessionRuntimeState.prepareProgress, - sessionTodos: defaultSessionRuntimeState.sessionTodos, - agentCapabilities: defaultSessionRuntimeState.agentCapabilities, - sessionModels: defaultSessionRuntimeState.sessionModels, - promptUsage: defaultSessionRuntimeState.promptUsage, - githubStatus: defaultGitHubState.githubStatus, - taskPRs: defaultGitHubState.taskPRs, - prWatches: defaultGitHubState.prWatches, - reviewWatches: defaultGitHubState.reviewWatches, - previewPanel: defaultUIState.previewPanel, - rightPanel: defaultUIState.rightPanel, - diffs: defaultUIState.diffs, - connection: defaultUIState.connection, - mobileKanban: defaultUIState.mobileKanban, - mobileSession: defaultUIState.mobileSession, - chatInput: defaultUIState.chatInput, - documentPanel: defaultUIState.documentPanel, - systemHealth: defaultUIState.systemHealth, - quickChat: defaultUIState.quickChat, - sessionFailureNotification: defaultUIState.sessionFailureNotification, - bottomTerminal: defaultUIState.bottomTerminal, -}; - -function mergeInitialState(initialState?: Partial): typeof defaultState { - if (!initialState) return defaultState; - - return { - ...defaultState, - ...initialState, - // Ensure nested objects are properly merged - kanban: { ...defaultState.kanban, ...initialState.kanban }, - kanbanMulti: { ...defaultState.kanbanMulti, ...initialState.kanbanMulti }, - workflows: { ...defaultState.workflows, ...initialState.workflows }, - tasks: { ...defaultState.tasks, ...initialState.tasks }, - workspaces: { ...defaultState.workspaces, ...initialState.workspaces }, - repositories: { ...defaultState.repositories, ...initialState.repositories }, - repositoryBranches: { ...defaultState.repositoryBranches, ...initialState.repositoryBranches }, - repositoryScripts: { ...defaultState.repositoryScripts, ...initialState.repositoryScripts }, - executors: { ...defaultState.executors, ...initialState.executors }, - settingsAgents: { ...defaultState.settingsAgents, ...initialState.settingsAgents }, - agentDiscovery: { ...defaultState.agentDiscovery, ...initialState.agentDiscovery }, - availableAgents: { ...defaultState.availableAgents, ...initialState.availableAgents }, - agentProfiles: { ...defaultState.agentProfiles, ...initialState.agentProfiles }, - editors: { ...defaultState.editors, ...initialState.editors }, - prompts: { ...defaultState.prompts, ...initialState.prompts }, - secrets: { ...defaultState.secrets, ...initialState.secrets }, - notificationProviders: { - ...defaultState.notificationProviders, - ...initialState.notificationProviders, - }, - settingsData: { ...defaultState.settingsData, ...initialState.settingsData }, - userSettings: { ...defaultState.userSettings, ...initialState.userSettings }, - messages: { ...defaultState.messages, ...initialState.messages }, - turns: { ...defaultState.turns, ...initialState.turns }, - taskSessions: { ...defaultState.taskSessions, ...initialState.taskSessions }, - taskSessionsByTask: { ...defaultState.taskSessionsByTask, ...initialState.taskSessionsByTask }, - sessionAgentctl: { ...defaultState.sessionAgentctl, ...initialState.sessionAgentctl }, - worktrees: { ...defaultState.worktrees, ...initialState.worktrees }, - sessionWorktreesBySessionId: { - ...defaultState.sessionWorktreesBySessionId, - ...initialState.sessionWorktreesBySessionId, - }, - pendingModel: { ...defaultState.pendingModel, ...initialState.pendingModel }, - activeModel: { ...defaultState.activeModel, ...initialState.activeModel }, - taskPlans: { ...defaultState.taskPlans, ...initialState.taskPlans }, - queue: { ...defaultState.queue, ...initialState.queue }, - terminal: { ...defaultState.terminal, ...initialState.terminal }, - shell: { ...defaultState.shell, ...initialState.shell }, - processes: { ...defaultState.processes, ...initialState.processes }, - gitStatus: { ...defaultState.gitStatus, ...initialState.gitStatus }, - sessionCommits: { ...defaultState.sessionCommits, ...initialState.sessionCommits }, - contextWindow: { ...defaultState.contextWindow, ...initialState.contextWindow }, - agents: { ...defaultState.agents, ...initialState.agents }, - sessionMode: { ...defaultState.sessionMode, ...initialState.sessionMode }, - userShells: { ...defaultState.userShells, ...initialState.userShells }, - prepareProgress: { ...defaultState.prepareProgress, ...initialState.prepareProgress }, - sessionTodos: { ...defaultState.sessionTodos, ...initialState.sessionTodos }, - agentCapabilities: { ...defaultState.agentCapabilities, ...initialState.agentCapabilities }, - sessionModels: { ...defaultState.sessionModels, ...initialState.sessionModels }, - promptUsage: { ...defaultState.promptUsage, ...initialState.promptUsage }, - githubStatus: { ...defaultState.githubStatus, ...initialState.githubStatus }, - taskPRs: { ...defaultState.taskPRs, ...initialState.taskPRs }, - prWatches: { ...defaultState.prWatches, ...initialState.prWatches }, - reviewWatches: { ...defaultState.reviewWatches, ...initialState.reviewWatches }, - previewPanel: { ...defaultState.previewPanel, ...initialState.previewPanel }, - rightPanel: { ...defaultState.rightPanel, ...initialState.rightPanel }, - diffs: { ...defaultState.diffs, ...initialState.diffs }, - connection: { ...defaultState.connection, ...initialState.connection }, - mobileKanban: { ...defaultState.mobileKanban, ...initialState.mobileKanban }, - mobileSession: { ...defaultState.mobileSession, ...initialState.mobileSession }, - chatInput: { ...defaultState.chatInput, ...initialState.chatInput }, - documentPanel: { ...defaultState.documentPanel, ...initialState.documentPanel }, - systemHealth: { ...defaultState.systemHealth, ...initialState.systemHealth }, - quickChat: { ...defaultState.quickChat, ...initialState.quickChat }, - sessionFailureNotification: - initialState.sessionFailureNotification ?? defaultState.sessionFailureNotification, - bottomTerminal: { ...defaultState.bottomTerminal, ...initialState.bottomTerminal }, - }; -} - export function createAppStore(initialState?: Partial) { const merged = mergeInitialState(initialState); @@ -617,6 +479,7 @@ export function createAppStore(initialState?: Partial) { agentCapabilities: merged.agentCapabilities, sessionModels: merged.sessionModels, promptUsage: merged.promptUsage, + sessionPollMode: merged.sessionPollMode, githubStatus: merged.githubStatus, taskPRs: merged.taskPRs, prWatches: merged.prWatches, diff --git a/apps/web/lib/types/backend.ts b/apps/web/lib/types/backend.ts index f5ccab5c25..0c1949c412 100644 --- a/apps/web/lib/types/backend.ts +++ b/apps/web/lib/types/backend.ts @@ -37,6 +37,7 @@ export type BackendMessageType = | "session.models_updated" | "session.todos_updated" | "session.prompt_usage" + | "session.poll_mode_changed" | "executor.created" | "executor.updated" | "executor.deleted" @@ -539,6 +540,10 @@ export type BackendMessageMap = { "session.models_updated": BackendMessage<"session.models_updated", SessionModelsPayload>; "session.todos_updated": BackendMessage<"session.todos_updated", SessionTodosPayload>; "session.prompt_usage": BackendMessage<"session.prompt_usage", SessionPromptUsagePayload>; + "session.poll_mode_changed": BackendMessage< + "session.poll_mode_changed", + { session_id: string; poll_mode: string } + >; "executor.created": BackendMessage<"executor.created", ExecutorPayload>; "executor.updated": BackendMessage<"executor.updated", ExecutorPayload>; "executor.deleted": BackendMessage<"executor.deleted", ExecutorPayload>; diff --git a/apps/web/lib/ws/client.ts b/apps/web/lib/ws/client.ts index 7d7a35e411..f11bc8d043 100644 --- a/apps/web/lib/ws/client.ts +++ b/apps/web/lib/ws/client.ts @@ -40,6 +40,11 @@ export class WebSocketClient { private intentionalClose = false; private subscriptions = new Map(); private sessionSubscriptions = new Map(); + // Ref-counted focus signals: a session can be focused by both the task panel + // and the task details page if both are mounted. Backend wakes its workspace + // tracker into fast-poll mode while any client has focus, falling back to + // slow when the count reaches 0 (debounced server-side). + private sessionFocusCounts = new Map(); private userSubscriptionCount = 0; constructor( @@ -147,6 +152,7 @@ export class WebSocketClient { const currentCount = this.sessionSubscriptions.get(sessionId) ?? 0; const nextCount = currentCount + 1; this.sessionSubscriptions.set(sessionId, nextCount); + if (this.status === "open" && nextCount === 1) { this.send({ id: generateUUID(), @@ -158,6 +164,40 @@ export class WebSocketClient { return () => this.unsubscribeSession(sessionId); } + focusSession(sessionId: string) { + const currentCount = this.sessionFocusCounts.get(sessionId) ?? 0; + const nextCount = currentCount + 1; + this.sessionFocusCounts.set(sessionId, nextCount); + if (this.status === "open" && nextCount === 1) { + this.send({ + id: generateUUID(), + type: "request", + action: "session.focus", + payload: { session_id: sessionId }, + }); + } + return () => this.unfocusSession(sessionId); + } + + unfocusSession(sessionId: string) { + const currentCount = this.sessionFocusCounts.get(sessionId); + if (!currentCount) return; + const nextCount = currentCount - 1; + if (nextCount <= 0) { + this.sessionFocusCounts.delete(sessionId); + if (this.status === "open") { + this.send({ + id: generateUUID(), + type: "request", + action: "session.unfocus", + payload: { session_id: sessionId }, + }); + } + return; + } + this.sessionFocusCounts.set(sessionId, nextCount); + } + subscribeUser() { this.userSubscriptionCount += 1; if (this.status === "open" && this.userSubscriptionCount === 1) { @@ -193,6 +233,7 @@ export class WebSocketClient { const currentCount = this.sessionSubscriptions.get(sessionId); if (!currentCount) return; const nextCount = currentCount - 1; + if (nextCount <= 0) { this.sessionSubscriptions.delete(sessionId); if (this.status === "open") { @@ -354,6 +395,14 @@ export class WebSocketClient { payload: { session_id: sessionId }, }); }); + this.sessionFocusCounts.forEach((_count, sessionId) => { + this.send({ + id: generateUUID(), + type: "request", + action: "session.focus", + payload: { session_id: sessionId }, + }); + }); if (this.userSubscriptionCount > 0) { this.send({ id: generateUUID(), diff --git a/apps/web/lib/ws/handlers/session-poll-mode.ts b/apps/web/lib/ws/handlers/session-poll-mode.ts new file mode 100644 index 0000000000..b3220b57ad --- /dev/null +++ b/apps/web/lib/ws/handlers/session-poll-mode.ts @@ -0,0 +1,17 @@ +import type { StoreApi } from "zustand"; +import type { AppState } from "@/lib/state/store"; +import type { WsHandlers } from "@/lib/ws/handlers/types"; +import type { SessionPollMode } from "@/lib/state/slices/session-runtime/types"; + +const VALID_MODES = new Set(["fast", "slow", "paused"]); + +export function registerSessionPollModeHandlers(store: StoreApi): WsHandlers { + return { + "session.poll_mode_changed": (message) => { + const { session_id, poll_mode } = message.payload; + if (!session_id || !poll_mode) return; + if (!VALID_MODES.has(poll_mode as SessionPollMode)) return; + store.getState().setSessionPollMode(session_id, poll_mode as SessionPollMode); + }, + }; +} diff --git a/apps/web/lib/ws/handlers/tasks.ts b/apps/web/lib/ws/handlers/tasks.ts index 68bda80c34..a6815256b4 100644 --- a/apps/web/lib/ws/handlers/tasks.ts +++ b/apps/web/lib/ws/handlers/tasks.ts @@ -7,7 +7,16 @@ import { useContextFilesStore } from "@/lib/state/context-files-store"; type KanbanTask = KanbanState["tasks"][number]; +/** Falls back to existing value when incoming is null, undefined, or empty string. */ function withFallback(value: T | null | undefined, fallback: T | undefined): T | undefined { + return value || fallback; +} + +/** Falls back only on null/undefined -- preserves 0 and other falsy non-string values. */ +function withNullishFallback( + value: T | null | undefined, + fallback: T | undefined, +): T | undefined { return value ?? fallback; } @@ -33,7 +42,7 @@ function buildNullableFields( repositoryId: withFallback(payload.repository_id, existing?.repositoryId), primarySessionId: withFallback(payload.primary_session_id, existing?.primarySessionId), primarySessionState: withFallback(payload.primary_session_state, existing?.primarySessionState), - sessionCount: withFallback(payload.session_count, existing?.sessionCount), + sessionCount: withNullishFallback(payload.session_count, existing?.sessionCount), reviewStatus: withFallback(payload.review_status, existing?.reviewStatus), primaryExecutorId: withFallback(payload.primary_executor_id, existing?.primaryExecutorId), primaryExecutorType: withFallback(payload.primary_executor_type, existing?.primaryExecutorType), diff --git a/apps/web/lib/ws/router.ts b/apps/web/lib/ws/router.ts index 778d42ed67..4961f2c2e3 100644 --- a/apps/web/lib/ws/router.ts +++ b/apps/web/lib/ws/router.ts @@ -4,6 +4,7 @@ import { registerAgentsHandlers } from "@/lib/ws/handlers/agents"; import { registerTaskSessionHandlers } from "@/lib/ws/handlers/agent-session"; import { registerAvailableCommandsHandlers } from "@/lib/ws/handlers/available-commands"; import { registerSessionModeHandlers } from "@/lib/ws/handlers/session-mode"; +import { registerSessionPollModeHandlers } from "@/lib/ws/handlers/session-poll-mode"; import { registerAgentCapabilitiesHandlers } from "@/lib/ws/handlers/agent-capabilities"; import { registerSessionModelsHandlers } from "@/lib/ws/handlers/session-models"; import { registerSessionTodosHandlers } from "@/lib/ws/handlers/session-todos"; @@ -43,6 +44,7 @@ export function registerWsHandlers(store: StoreApi) { ...registerTaskSessionHandlers(store), ...registerAvailableCommandsHandlers(store), ...registerSessionModeHandlers(store), + ...registerSessionPollModeHandlers(store), ...registerAgentCapabilitiesHandlers(store), ...registerSessionModelsHandlers(store), ...registerSessionTodosHandlers(store),