perf: focus-gated git polling to reduce CPU on retained worktrees (#610)

This commit is contained in:
Carlos Florêncio
2026-04-18 11:33:04 +01:00
committed by GitHub
parent 06cb48b300
commit 5b04c9c675
50 changed files with 2930 additions and 212 deletions
+39
View File
@@ -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 {
+13
View File
@@ -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})
@@ -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.
@@ -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, "")
}
@@ -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()
}
}
@@ -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")
}
}
@@ -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.
@@ -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
}
@@ -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")
}
}
@@ -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)
@@ -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})
}
@@ -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)
}
}
@@ -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
}
@@ -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
@@ -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 ""
@@ -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
@@ -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()
@@ -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()
@@ -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()
@@ -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:
}
}
}
@@ -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")
}
}
@@ -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")
}
}
@@ -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,
}
}
@@ -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)
+49 -8
View File
@@ -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
@@ -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
}
}
@@ -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)
}
}
}
@@ -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) {
@@ -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",
@@ -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{}
@@ -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.
@@ -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 {
+8 -2
View File
@@ -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"
+33 -2
View File
@@ -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<SessionPollMode, { letter: string; color: string; label: string }> =
{
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 (
<span className="flex items-center gap-1.5 text-[11px]">
{updatedAt && (
<span className="text-muted-foreground/50">{formatRelativeTime(updatedAt)}</span>
)}
{prInfo && <span className="text-muted-foreground/50">#{prInfo.number}</span>}
{modeConfig && (
<Tooltip>
<TooltipTrigger asChild>
<span className={cn("font-mono text-[10px] font-semibold", modeConfig.color)}>
{modeConfig.letter}
</span>
</TooltipTrigger>
<TooltipContent side="right">
Git poll: {pollMode} ({modeConfig.label})
</TooltipContent>
</Tooltip>
)}
</span>
);
}
@@ -211,7 +242,7 @@ function TaskItemContent({
{repositories.join(" · ")}
</span>
)}
<TaskItemStatsRow updatedAt={updatedAt} prInfo={prInfo} />
<TaskItemStatsRow updatedAt={updatedAt} prInfo={prInfo} primarySessionId={primarySessionId} />
</div>
);
}
@@ -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));
@@ -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);
});
});
@@ -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, {
@@ -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,
@@ -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");
});
});
@@ -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]);
}
+155
View File
@@ -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>): 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 },
};
}
@@ -14,11 +14,11 @@ type CombinedSlice = SessionRuntimeSlice & SessionSlice;
function makeCombinedStore() {
return create<CombinedSlice>()(
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),
})),
);
}
@@ -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<typeof createSessionRuntimeSlice>[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;
}),
};
}
@@ -255,6 +255,12 @@ export type SessionTodosState = {
bySessionId: Record<string, TodoEntry[]>;
};
export type SessionPollMode = "fast" | "slow" | "paused";
export type SessionPollModeState = {
bySessionId: Record<string, SessionPollMode>;
};
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;
+5 -142
View File
@@ -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<typeof createAppStore>;
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<AppState>): 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<AppState>) {
const merged = mergeInitialState(initialState);
@@ -617,6 +479,7 @@ export function createAppStore(initialState?: Partial<AppState>) {
agentCapabilities: merged.agentCapabilities,
sessionModels: merged.sessionModels,
promptUsage: merged.promptUsage,
sessionPollMode: merged.sessionPollMode,
githubStatus: merged.githubStatus,
taskPRs: merged.taskPRs,
prWatches: merged.prWatches,
+5
View File
@@ -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>;
+49
View File
@@ -40,6 +40,11 @@ export class WebSocketClient {
private intentionalClose = false;
private subscriptions = new Map<string, number>();
private sessionSubscriptions = new Map<string, number>();
// 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<string, number>();
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(),
@@ -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<SessionPollMode>(["fast", "slow", "paused"]);
export function registerSessionPollModeHandlers(store: StoreApi<AppState>): 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);
},
};
}
+10 -1
View File
@@ -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<T>(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<T>(
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),
+2
View File
@@ -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<AppState>) {
...registerTaskSessionHandlers(store),
...registerAvailableCommandsHandlers(store),
...registerSessionModeHandlers(store),
...registerSessionPollModeHandlers(store),
...registerAgentCapabilitiesHandlers(store),
...registerSessionModelsHandlers(store),
...registerSessionTodosHandlers(store),