diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go index 51e243efb8..11c30c219a 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go @@ -47,6 +47,14 @@ const ( // allow rather than a tight RPC deadline. const wakeupPromptTimeout = 30 * time.Minute +// notifQueueCapacity sizes the buffered channel that feeds the update +// worker. 4096 covers any realistic session/load replay burst (the failure +// case that motivated this was ~5-10k notifications spread over several +// seconds, well within the worker's sub-microsecond drain rate). Set big +// enough that the SDK's 1024-slot upstream queue stays near-empty under +// burst, small enough that worst-case memory is bounded. +const notifQueueCapacity = 4096 + // AgentInfo contains information about the connected agent. type AgentInfo struct { Name string `json:"name"` @@ -83,6 +91,26 @@ type Adapter struct { // Update channel updatesCh chan AgentEvent + // notifQueue decouples the ACP SDK's notification goroutine from the + // per-notification processing we do in handleACPUpdate. The SDK's inbound + // channel is a 1024-slot non-blocking send — if our handler ever stalls + // (slow disk, GC, downstream backpressure on updatesCh), the SDK closes + // the connection. By front-loading a buffered channel + dedicated worker, + // the SDK only ever sees nanosecond-scale enqueues. workerWg lets Close + // wait for in-flight notifications to drain before tearing down updatesCh. + // + // Items are notifWork so the queue can also carry barrier-sync requests + // from syncNotifQueue — the worker closes the barrier's channel in FIFO + // order, which lets sendPrompt wait for queued text chunks before it + // emits EventTypeComplete. + // + // Drained-by-cancel, not by close: Close cancels lifetimeCtx (which the + // worker selects on) but never closes notifQueue, so any items queued + // after Close are dropped on the floor rather than re-delivered. Close + // is terminal — fine for shutdown but worth knowing when reading the loop. + notifQueue chan notifWork + workerWg sync.WaitGroup + // Permission handler permissionHandler PermissionHandler @@ -196,6 +224,7 @@ func NewAdapter(cfg *shared.Config, log *logger.Logger) *Adapter { agentID: cfg.AgentID, normalizer: NewNormalizer(cfg.AgentID), updatesCh: make(chan AgentEvent, 100), + notifQueue: make(chan notifWork, notifQueueCapacity), activeToolCalls: make(map[string]*streams.NormalizedPayload), activeMonitors: make(map[string]map[string]string), pendingWakeups: make(map[string]*pendingWakeup), @@ -206,6 +235,12 @@ func NewAdapter(cfg *shared.Config, log *logger.Logger) *Adapter { lifetimeCancel: cancel, } a.wakeup = newWakeupScheduler(l, a.fireWakeup) + // Start the update worker before returning so any caller that connects + // the SDK (Initialize, or any future direct wiring) always has a live + // consumer for notifQueue. Moving this out of Initialize closes a latent + // footgun where a future caller bypassing Initialize would silently fill + // the queue and lose notifications. + a.startUpdateWorker() return a } @@ -241,11 +276,11 @@ func (a *Adapter) Initialize(ctx context.Context) error { a.logger.Info("initializing ACP adapter", zap.String("workdir", a.cfg.WorkDir)) - // Create ACP client with update handler that converts to AgentEvent + // Create ACP client with update handler that enqueues for the worker. a.acpClient = acpclient.NewClient( acpclient.WithLogger(a.logger.Zap()), acpclient.WithWorkspaceRoot(a.cfg.WorkDir), - acpclient.WithUpdateHandler(a.handleACPUpdate), + acpclient.WithUpdateHandler(a.enqueueACPUpdate), acpclient.WithPermissionHandler(a.handlePermissionRequest), ) @@ -367,17 +402,19 @@ func (a *Adapter) sendUpdate(event AgentEvent) { // Close releases resources held by the adapter. func (a *Adapter) Close() error { a.mu.Lock() - defer a.mu.Unlock() - if a.closed { + a.mu.Unlock() return nil } a.closed = true + a.mu.Unlock() a.logger.Info("closing ACP adapter") // Stop any pending ScheduleWakeup timer so it doesn't fire after close, // and cancel the lifetime context so any in-flight wakeup prompt aborts. + // Cancelling lifetimeCtx also unblocks enqueueACPUpdate senders and + // signals the update worker to exit. if a.wakeup != nil { a.wakeup.cancel() } @@ -385,6 +422,11 @@ func (a *Adapter) Close() error { a.lifetimeCancel() } + // Wait for the update worker to exit before closing updatesCh. + // handleACPUpdate may call sendUpdate, so updatesCh must remain open + // until the worker is gone. + a.workerWg.Wait() + // Clean up any saved attachments a.attachMgr.Cleanup() diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go index 022fe95955..cda1357101 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go @@ -150,6 +150,21 @@ func (a *Adapter) sendPrompt(ctx context.Context, message string, attachments [] return err } + // Drain queued ACP notifications before running the post-prompt sweeps and + // the complete event. The worker (now async) may still hold final frames + // the agent emitted right before its prompt response — the final text + // chunk, the terminal monitor_end tool_call_update, the registration + // frame for a Monitor whose events haven't yet been routed. Without this + // barrier: + // - cancelActiveToolCalls / sweepMonitorsOnPromptEnd race the worker + // for activeToolCalls and activeMonitors. If the worker hasn't added + // a Monitor yet when the sweep takes the map, subsequent monitor_event + // text frames find no tracking and drop their events on the floor. + // - The complete event emitted to updatesCh outruns the final text chunk, + // so the downstream buffer flush yields empty and the turn persists as + // had_output=false even when the agent did produce text. + a.syncNotifQueue() + // Cancel any tool calls still in-flight (e.g. a denied permission leaves the // tool_call without a terminal status update from the agent). a.cancelActiveToolCalls(sessionID) diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go index 3f487eac4a..7664376386 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go @@ -10,20 +10,107 @@ import ( "go.uber.org/zap" ) -// handleACPUpdate converts ACP SessionNotification to protocol-agnostic AgentEvent. -func (a *Adapter) handleACPUpdate(n acp.SessionNotification) { - // Marshal once for both debug logging and tracing - rawData, _ := json.Marshal(n) +// notifWork is the item type carried on notifQueue. Exactly one of the two +// fields is populated: notif for a real SDK notification (the common case), +// sync for a barrier posted by syncNotifQueue. The worker closes sync when +// it pops the barrier off the queue, releasing the waiter once everything +// queued ahead of it has been processed. +type notifWork struct { + notif acp.SessionNotification + sync chan struct{} +} - // Log raw event for debugging - if len(rawData) > 0 { - shared.LogRawEvent(shared.ProtocolACP, a.agentID, string(n.SessionId), "session_notification", rawData) +// enqueueACPUpdate is the SDK-facing notification handler. It pushes the +// notification onto notifQueue and returns immediately, so the ACP SDK's +// receive loop is never blocked by our per-item processing (json.Marshal, +// debug log I/O, monitor capture, downstream sendUpdate). The actual work +// runs on runUpdateWorker. +// +// The select honours lifetimeCtx so a Close in flight unblocks any sender +// that would otherwise stall on a full queue. Sending blocks (rather than +// dropping) when the queue is full: with the handleACPUpdate fast path the +// worker drains in nanoseconds-to-microseconds, so a full 4096-slot queue +// means we're in a catastrophic stall and slowing the SDK is preferable to +// silently losing notifications. +func (a *Adapter) enqueueACPUpdate(n acp.SessionNotification) { + select { + case <-a.lifetimeCtx.Done(): + return + case a.notifQueue <- notifWork{notif: n}: } +} - // During session/load, suppress history replay notifications. - // ACP agents stream the entire conversation history during load, which should not - // be emitted as new message events to avoid duplicating messages in the UI. - // We suppress: message chunks, thinking chunks, tool calls, and tool updates. +// syncNotifQueue blocks until every notifWork enqueued before this call has +// been processed by the worker. It works by posting a barrier item onto the +// same FIFO queue and waiting for the worker to close it. +// +// Motivation: the worker is async, so after conn.Prompt() returns the final +// text chunks emitted by the agent right before the prompt response may +// still be sitting in notifQueue. If sendPrompt emits EventTypeComplete +// immediately, the downstream "turn complete" handler flushes the message +// buffer before those chunks land, the agent's text is dropped, and the +// turn persists as had_output=false. Calling this before the complete emit +// guarantees the chunks are delivered to updatesCh first. +// +// No caller-context honored: returning before the barrier closes would +// reintroduce the very race this primitive exists to prevent (sweeps and +// EventTypeComplete running while the worker still has queued frames). +// Only adapter shutdown via lifetimeCtx can release the wait early. +func (a *Adapter) syncNotifQueue() { + ch := make(chan struct{}) + select { + case <-a.lifetimeCtx.Done(): + return + case a.notifQueue <- notifWork{sync: ch}: + } + select { + case <-a.lifetimeCtx.Done(): + case <-ch: + } +} + +// startUpdateWorker spawns the goroutine that drains notifQueue and calls +// handleACPUpdate for each notification. Called exactly once from +// NewAdapter so the queue always has a consumer for the adapter's +// lifetime — do not call again (a second invocation would Add(1) to +// workerWg and spawn a second reader, breaking the single-worker FIFO +// guarantee). Close waits for the worker to exit via workerWg before +// closing updatesCh. +func (a *Adapter) startUpdateWorker() { + a.workerWg.Add(1) + go a.runUpdateWorker() +} + +// runUpdateWorker is the worker loop. It exits when lifetimeCtx is cancelled +// (by Close). A single worker preserves FIFO ordering across notifications, +// matching the SDK's own serial delivery contract. +func (a *Adapter) runUpdateWorker() { + defer a.workerWg.Done() + for { + select { + case <-a.lifetimeCtx.Done(): + return + case item := <-a.notifQueue: + if item.sync != nil { + close(item.sync) + continue + } + a.handleACPUpdate(item.notif) + } + } +} + +// handleACPUpdate converts ACP SessionNotification to protocol-agnostic AgentEvent. +// Runs synchronously on the update worker goroutine; do not call from the SDK's +// notification path (use enqueueACPUpdate instead). Unit tests invoke this +// directly to exercise the conversion logic without spinning up the worker. +func (a *Adapter) handleACPUpdate(n acp.SessionNotification) { + // Fast path during session/load: history-replay notifications can arrive as + // a burst large enough to overflow the ACP SDK's 1024-deep notification + // queue if the per-item handler is slow. Check the loading flag first and + // skip json.Marshal + LogRawEvent for replay notifications we'd suppress + // anyway. We still capture the Plan + Monitor state needed to reconcile + // after replay. a.mu.RLock() isLoading := a.isLoadingSession a.mu.RUnlock() @@ -51,12 +138,16 @@ func (a *Adapter) handleACPUpdate(n acp.SessionNotification) { if u.AgentMessageChunk != nil || u.UserMessageChunk != nil || u.AgentThoughtChunk != nil || u.ToolCall != nil || u.ToolCallUpdate != nil || u.Plan != nil || u.CurrentModeUpdate != nil || u.ConfigOptionUpdate != nil { - a.logger.Debug("suppressing history replay notification during session load", - zap.String("session_id", string(n.SessionId))) return } } + // Marshal once for both debug logging and tracing. + rawData, _ := json.Marshal(n) + if len(rawData) > 0 { + shared.LogRawEvent(shared.ProtocolACP, a.agentID, string(n.SessionId), "session_notification", rawData) + } + sessionID := string(n.SessionId) event := a.convertNotification(n) diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go new file mode 100644 index 0000000000..f359ed14a3 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/load_burst_test.go @@ -0,0 +1,381 @@ +package acp + +import ( + "context" + "errors" + "fmt" + "io" + "testing" + "time" + + acp "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +// burstAgent is a minimal acp.Agent stub used by the load-replay regression +// test. None of these methods are exercised — we only need a peer that holds +// the connection open while we push notifications agent → client. +type burstAgent struct{} + +func (burstAgent) Initialize(context.Context, acp.InitializeRequest) (acp.InitializeResponse, error) { + return acp.InitializeResponse{}, nil +} + +func (burstAgent) Authenticate(context.Context, acp.AuthenticateRequest) (acp.AuthenticateResponse, error) { + return acp.AuthenticateResponse{}, nil +} +func (burstAgent) Cancel(context.Context, acp.CancelNotification) error { return nil } +func (burstAgent) CloseSession(context.Context, acp.CloseSessionRequest) (acp.CloseSessionResponse, error) { + return acp.CloseSessionResponse{}, nil +} + +func (burstAgent) ListSessions(context.Context, acp.ListSessionsRequest) (acp.ListSessionsResponse, error) { + return acp.ListSessionsResponse{}, nil +} + +func (burstAgent) NewSession(context.Context, acp.NewSessionRequest) (acp.NewSessionResponse, error) { + return acp.NewSessionResponse{}, nil +} +func (burstAgent) Prompt(context.Context, acp.PromptRequest) (acp.PromptResponse, error) { + return acp.PromptResponse{}, nil +} + +func (burstAgent) ResumeSession(context.Context, acp.ResumeSessionRequest) (acp.ResumeSessionResponse, error) { + return acp.ResumeSessionResponse{}, nil +} + +func (burstAgent) SetSessionConfigOption(context.Context, acp.SetSessionConfigOptionRequest) (acp.SetSessionConfigOptionResponse, error) { + return acp.SetSessionConfigOptionResponse{}, nil +} + +func (burstAgent) SetSessionMode(context.Context, acp.SetSessionModeRequest) (acp.SetSessionModeResponse, error) { + return acp.SetSessionModeResponse{}, nil +} + +// burstClient is a minimal acp.Client stub that forwards SessionUpdate +// notifications to the adapter's real handler. +type burstClient struct { + onUpdate func(acp.SessionNotification) +} + +func (b *burstClient) SessionUpdate(_ context.Context, n acp.SessionNotification) error { + b.onUpdate(n) + return nil +} + +func (*burstClient) ReadTextFile(context.Context, acp.ReadTextFileRequest) (acp.ReadTextFileResponse, error) { + return acp.ReadTextFileResponse{}, errors.New("not implemented") +} + +func (*burstClient) WriteTextFile(context.Context, acp.WriteTextFileRequest) (acp.WriteTextFileResponse, error) { + return acp.WriteTextFileResponse{}, errors.New("not implemented") +} + +func (*burstClient) RequestPermission(context.Context, acp.RequestPermissionRequest) (acp.RequestPermissionResponse, error) { + return acp.RequestPermissionResponse{}, errors.New("not implemented") +} + +func (*burstClient) CreateTerminal(context.Context, acp.CreateTerminalRequest) (acp.CreateTerminalResponse, error) { + return acp.CreateTerminalResponse{}, errors.New("not implemented") +} + +func (*burstClient) KillTerminal(context.Context, acp.KillTerminalRequest) (acp.KillTerminalResponse, error) { + return acp.KillTerminalResponse{}, errors.New("not implemented") +} + +func (*burstClient) TerminalOutput(context.Context, acp.TerminalOutputRequest) (acp.TerminalOutputResponse, error) { + return acp.TerminalOutputResponse{}, errors.New("not implemented") +} + +func (*burstClient) ReleaseTerminal(context.Context, acp.ReleaseTerminalRequest) (acp.ReleaseTerminalResponse, error) { + return acp.ReleaseTerminalResponse{}, errors.New("not implemented") +} + +func (*burstClient) WaitForTerminalExit(context.Context, acp.WaitForTerminalExitRequest) (acp.WaitForTerminalExitResponse, error) { + return acp.WaitForTerminalExitResponse{}, errors.New("not implemented") +} + +// newBurstPair wires a real acp.ClientSideConnection / AgentSideConnection +// pair over in-memory pipes. The client's SessionUpdate forwards to the +// supplied handler. Cleanup is registered on the test. +func newBurstPair(t *testing.T, onUpdate func(acp.SessionNotification)) (*acp.ClientSideConnection, *acp.AgentSideConnection) { + t.Helper() + + c2aR, c2aW := io.Pipe() + a2cR, a2cW := io.Pipe() + + clientConn := acp.NewClientSideConnection(&burstClient{onUpdate: onUpdate}, c2aW, a2cR) + agentConn := acp.NewAgentSideConnection(burstAgent{}, a2cW, c2aR) + + t.Cleanup(func() { + _ = c2aR.Close() + _ = c2aW.Close() + _ = a2cR.Close() + _ = a2cW.Close() + }) + + return clientConn, agentConn +} + +// TestLoadReplayBurst_HandlesLargeReplay is a regression test for the +// "notification queue overflow" failure we hit on a 304-exchange auggie +// session load. The acp-go-sdk exposes a hardcoded 1024-slot inbound channel +// (defaultMaxQueuedNotifications) and shuts down the connection when the +// single-goroutine consumer falls behind. Before the adapter_updates.go +// fast-path, json.Marshal + LogRawEvent on every replayed notification was +// slow enough that the session/load burst would back the queue up. +// +// This test wires a real acp.ClientSideConnection / AgentSideConnection pair +// over io.Pipes, sets isLoadingSession=true on the adapter, and pushes a +// large burst of replay notifications agent → client. It then sends an +// AvailableCommandsUpdate (which is intentionally NOT suppressed during +// load) as a sentinel and asserts: +// 1. the connection is still alive after the burst (overflow would have +// closed it), +// 2. the sentinel flows through the un-suppressed path, +// 3. the last Plan from the burst is captured for post-load re-emit. +// +// Note: in-memory io.Pipe is synchronous, so the producer is naturally +// gated by the consumer drain rate; an overflow here would mean the +// handler is hanging, not just slow. The throughput-sensitive scenario +// is exercised by BenchmarkLoadReplayHandler below. +func TestLoadReplayBurst_HandlesLargeReplay(t *testing.T) { + const notificationBurstSize = 20000 + const sentinelCmd = "burst-sentinel" + + a := newTestAdapter() + a.mu.Lock() + a.isLoadingSession = true + a.mu.Unlock() + + // Exercise the production path: SDK → enqueueACPUpdate → notifQueue → + // runUpdateWorker → handleACPUpdate. NewAdapter starts the worker; we + // just rely on Close to drain it. + t.Cleanup(func() { _ = a.Close() }) + + clientConn, agentConn := newBurstPair(t, a.enqueueACPUpdate) + + const sessionID = "burst-session" + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + // Push the burst of replay notifications. These would all be suppressed + // by the adapter (AgentMessageChunk + ToolCall + Plan are on the suppress + // list) — the point is whether the SDK's 1024-deep queue stays drained. + for i := 0; i < notificationBurstSize; i++ { + if err := agentConn.SessionUpdate(ctx, makeReplayNotification(sessionID, i)); err != nil { + t.Fatalf("SessionUpdate at i=%d failed: %v", i, err) + } + } + + // Sentinel: AvailableCommandsUpdate passes through during load, so it + // will land on updatesCh once the consumer drains the queue. + if err := agentConn.SessionUpdate(ctx, acp.SessionNotification{ + SessionId: acp.SessionId(sessionID), + Update: acp.SessionUpdate{ + AvailableCommandsUpdate: &acp.SessionAvailableCommandsUpdate{ + AvailableCommands: []acp.AvailableCommand{ + {Name: sentinelCmd, Description: "burst sentinel"}, + }, + }, + }, + }); err != nil { + t.Fatalf("sentinel SessionUpdate failed: %v", err) + } + + // The sentinel-wait loop below is the authoritative overflow check: the + // SDK tears down both sides if its inbound queue ever stalls, so a closed + // Done() observed there proves overflow. A non-blocking peek here would + // race the async teardown and could produce a false PASS. + + // Wait for the sentinel to flow through to the adapter's updates channel. + deadline := time.After(15 * time.Second) + var got *AgentEvent + for got == nil { + select { + case ev := <-a.updatesCh: + if ev.Type == streams.EventTypeAvailableCommands { + got = &ev + } + case <-clientConn.Done(): + t.Fatal("client connection closed while waiting for sentinel — overflow") + case <-deadline: + t.Fatal("timeout waiting for sentinel AvailableCommands event after replay burst") + } + } + + if len(got.AvailableCommands) != 1 || got.AvailableCommands[0].Name != sentinelCmd { + t.Fatalf("unexpected sentinel payload: %+v", got) + } + + // Plan was sent during the burst and should have been captured (last + // Plan seen during load is stashed in loadReplayPlan for re-emit). + a.mu.RLock() + captured := a.loadReplayPlan + a.mu.RUnlock() + if captured == nil { + t.Fatal("expected loadReplayPlan to be captured during replay burst") + } +} + +// makeReplayNotification builds a representative replay notification of one +// of the suppressed kinds (AgentMessageChunk / ToolCall / Plan). Shared by +// the burst test above and the benchmarks below. +func makeReplayNotification(sessionID string, i int) acp.SessionNotification { + var update acp.SessionUpdate + switch i % 3 { + case 0: + update = acp.SessionUpdate{ + AgentMessageChunk: &acp.SessionUpdateAgentMessageChunk{ + Content: acp.TextBlock(fmt.Sprintf("replay chunk %d", i)), + }, + } + case 1: + update = acp.SessionUpdate{ + ToolCall: &acp.SessionUpdateToolCall{ + ToolCallId: acp.ToolCallId(fmt.Sprintf("tc-%d", i)), + Title: "replay tool call", + }, + } + case 2: + update = acp.SessionUpdate{ + Plan: &acp.SessionUpdatePlan{ + Entries: []acp.PlanEntry{ + {Content: fmt.Sprintf("plan entry %d", i), Status: "in_progress"}, + }, + }, + } + } + return acp.SessionNotification{ + SessionId: acp.SessionId(sessionID), + Update: update, + } +} + +// BenchmarkHandleACPUpdate_LoadSuppressed measures the per-notification cost +// of the suppressed-during-load fast path. This is the hot loop during +// session/load replay, and the path that previously did json.Marshal + +// LogRawEvent on every notification before being short-circuited. +func BenchmarkHandleACPUpdate_LoadSuppressed(b *testing.B) { + a := newTestAdapter() + b.Cleanup(func() { _ = a.Close() }) + a.mu.Lock() + a.isLoadingSession = true + a.mu.Unlock() + + notes := make([]acp.SessionNotification, 256) + for i := range notes { + notes[i] = makeReplayNotification("bench-session", i) + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + a.handleACPUpdate(notes[i%len(notes)]) + } +} + +// BenchmarkHandleACPUpdate_NormalPath measures the per-notification cost of +// the non-loading path (json.Marshal + LogRawEvent + convertNotification + +// updatesCh send). Drain updatesCh in a background goroutine so we don't +// stall on the unbuffered/buffered send. +func BenchmarkHandleACPUpdate_NormalPath(b *testing.B) { + a := newTestAdapter() + b.Cleanup(func() { _ = a.Close() }) + + done := make(chan struct{}) + go func() { + for { + select { + case <-a.updatesCh: + case <-done: + return + } + } + }() + defer close(done) + + notes := make([]acp.SessionNotification, 256) + for i := range notes { + notes[i] = makeReplayNotification("bench-session", i) + } + + b.ResetTimer() + for i := 0; i < b.N; i++ { + a.handleACPUpdate(notes[i%len(notes)]) + } +} + +// TestUpdateWorker_FIFOAndDrainOnClose verifies the SDK-decoupling worker: +// 1. enqueueACPUpdate hands notifications to a single worker that processes +// them in FIFO order (preserving the SDK's serial-delivery contract); +// 2. Close cancels the worker via lifetimeCtx, workerWg.Wait returns, and +// updatesCh is then safely closed — no goroutine leak, no panic. +// +// We don't poll the queue length directly; the contract we care about is +// "the events arrive on updatesCh in order, and Close completes". +func TestUpdateWorker_FIFOAndDrainOnClose(t *testing.T) { + const burst = 32 + + a := newTestAdapter() + + // Push a burst of AvailableCommands notifications (one-shot events that + // land on updatesCh and carry an identifier we can assert order on). + for i := 0; i < burst; i++ { + a.enqueueACPUpdate(acp.SessionNotification{ + SessionId: acp.SessionId("s1"), + Update: acp.SessionUpdate{ + AvailableCommandsUpdate: &acp.SessionAvailableCommandsUpdate{ + AvailableCommands: []acp.AvailableCommand{ + {Name: fmt.Sprintf("cmd-%d", i)}, + }, + }, + }, + }) + } + + // Drain updatesCh until we've seen the full ordered burst. + deadline := time.After(5 * time.Second) + for i := 0; i < burst; i++ { + select { + case ev := <-a.updatesCh: + if ev.Type != streams.EventTypeAvailableCommands { + t.Fatalf("event %d: unexpected type %q", i, ev.Type) + } + if len(ev.AvailableCommands) != 1 { + t.Fatalf("event %d: expected 1 command, got %d", i, len(ev.AvailableCommands)) + } + want := fmt.Sprintf("cmd-%d", i) + if got := ev.AvailableCommands[0].Name; got != want { + t.Fatalf("event %d: out-of-order: got %q want %q (worker is supposed to be FIFO)", i, got, want) + } + case <-deadline: + t.Fatalf("timeout after %d/%d events — worker stalled", i, burst) + } + } + + // Close must return quickly and the worker goroutine must exit. + closed := make(chan error, 1) + go func() { closed <- a.Close() }() + select { + case err := <-closed: + if err != nil { + t.Fatalf("Close returned error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Close did not return — worker likely leaked") + } + + // Post-Close enqueues are no-ops (lifetimeCtx is cancelled). Must not + // block, must not panic. + done := make(chan struct{}) + go func() { + a.enqueueACPUpdate(acp.SessionNotification{SessionId: "s1"}) + close(done) + }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("enqueueACPUpdate blocked after Close — lifetimeCtx guard missing") + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/sync_barrier_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/sync_barrier_test.go new file mode 100644 index 0000000000..0bcf5b0b85 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/sync_barrier_test.go @@ -0,0 +1,49 @@ +package acp + +import ( + "fmt" + "testing" + + "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +// TestSyncNotifQueue_BlocksUntilQueueDrained verifies the barrier-sync +// primitive used by sendPrompt before emitting EventTypeComplete: every +// notification enqueued before syncNotifQueue is processed (and visible on +// updatesCh) by the time the call returns. +func TestSyncNotifQueue_BlocksUntilQueueDrained(t *testing.T) { + const burst = 50 + + a := newTestAdapter() + t.Cleanup(func() { _ = a.Close() }) + + for i := 0; i < burst; i++ { + a.enqueueACPUpdate(acp.SessionNotification{ + SessionId: acp.SessionId("s1"), + Update: acp.SessionUpdate{ + AvailableCommandsUpdate: &acp.SessionAvailableCommandsUpdate{ + AvailableCommands: []acp.AvailableCommand{ + {Name: fmt.Sprintf("cmd-%d", i)}, + }, + }, + }, + }) + } + + a.syncNotifQueue() + + events := drainEvents(a) + if len(events) != burst { + t.Fatalf("expected %d events on updatesCh after sync, got %d", burst, len(events)) + } + for i, ev := range events { + if ev.Type != streams.EventTypeAvailableCommands { + t.Fatalf("event %d: unexpected type %q", i, ev.Type) + } + want := fmt.Sprintf("cmd-%d", i) + if len(ev.AvailableCommands) != 1 || ev.AvailableCommands[0].Name != want { + t.Fatalf("event %d: got %+v, want command %q", i, ev.AvailableCommands, want) + } + } +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go b/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go index 06a5d14919..f1ae245da7 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_cmd.go @@ -4,6 +4,8 @@ import ( "context" "os" "os/exec" + + "github.com/kandev/kandev/internal/common/subproc" ) // gitOptionalLocksOff is the env var git reads to skip "optional" locks, i.e. @@ -26,3 +28,44 @@ func (wt *WorkspaceTracker) pollingGitCommand(ctx context.Context, args ...strin cmd.Env = append(os.Environ(), gitOptionalLocksOff) return cmd } + +// gitCmdContext builds a per-command timeout-bounded ctx and an exec.Cmd +// for a git invocation. When polling is true, the cmd is built via +// pollingGitCommand (which sets GIT_OPTIONAL_LOCKS=0) so the background +// poll loop doesn't contend with user-initiated git operations on the +// index lock. The returned cancel must be deferred by the caller to +// release the timer once the subprocess exits. +func (wt *WorkspaceTracker) gitCmdContext(ctx context.Context, polling bool, args ...string) (context.Context, context.CancelFunc, *exec.Cmd) { + cctx, cancel := context.WithTimeout(ctx, gitCommandTimeout) + if polling { + return cctx, cancel, wt.pollingGitCommand(cctx, args...) + } + cmd := exec.CommandContext(cctx, "git", args...) + cmd.Dir = wt.workDir + return cctx, cancel, cmd +} + +// runGitOutput runs a git command with a per-command timeout and returns its +// stdout. The derived ctx ensures cancellation SIGKILLs the subprocess and +// releases its throttle slot even when the outer poll ctx is long-lived. +func (wt *WorkspaceTracker) runGitOutput(ctx context.Context, args ...string) ([]byte, error) { + cctx, cancel, cmd := wt.gitCmdContext(ctx, false, args...) + defer cancel() + return subproc.RunGitOutput(cctx, cmd) +} + +// runGit is runGitOutput's no-stdout sibling for verify-style probes where +// only the exit code matters. +func (wt *WorkspaceTracker) runGit(ctx context.Context, args ...string) error { + cctx, cancel, cmd := wt.gitCmdContext(ctx, false, args...) + defer cancel() + return subproc.RunGit(cctx, cmd) +} + +// runPollingGitOutput is runGitOutput's polling sibling — see +// gitCmdContext for the GIT_OPTIONAL_LOCKS rationale. +func (wt *WorkspaceTracker) runPollingGitOutput(ctx context.Context, args ...string) ([]byte, error) { + cctx, cancel, cmd := wt.gitCmdContext(ctx, true, args...) + defer cancel() + return subproc.RunGitOutput(cctx, cmd) +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_cmd_test.go b/apps/backend/internal/agentctl/server/process/workspace_git_cmd_test.go new file mode 100644 index 0000000000..1d68a6ce26 --- /dev/null +++ b/apps/backend/internal/agentctl/server/process/workspace_git_cmd_test.go @@ -0,0 +1,127 @@ +//go:build !windows + +package process + +import ( + "context" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/subproc" +) + +// TestRunGitOutput_ReleasesThrottleOnTimeout is the regression test for the +// "empty Changes panel after freeze" symptom (see workspace_monitor.go +// gitCommandTimeout). It proves that when a git invocation wedges past +// gitCommandTimeout, runGitOutput SIGKILLs the subprocess and releases its +// gitThrottle slot so queued callers can still make progress. +// +// Mechanism: install a 'git' shim on $PATH that sleeps for an hour, shrink +// gitCommandTimeout to 200ms and the throttle cap to 2, then spawn 6 +// goroutines all calling runGitOutput. Without the per-command timeout the +// 4 queued goroutines would block on Acquire forever (the test would hang +// past its bound). With the fix each batch of 2 times out, releases, and +// the next pair Acquires — total ≈ 200ms × 3 batches = 600ms. +func TestRunGitOutput_ReleasesThrottleOnTimeout(t *testing.T) { + prev := gitCommandTimeout + gitCommandTimeout = 200 * time.Millisecond + t.Cleanup(func() { gitCommandTimeout = prev }) + + restoreCap := subproc.Git().SetCapForTest(2) + t.Cleanup(restoreCap) + + shimDir := installSleepGitShim(t) + t.Setenv("PATH", shimDir+string(os.PathListSeparator)+os.Getenv("PATH")) + + _, wt := setupTestDir(t) + + const N = 6 + var wg sync.WaitGroup + wg.Add(N) + done := make(chan struct{}) + start := time.Now() + for i := 0; i < N; i++ { + go func() { + defer wg.Done() + _, _ = wt.runGitOutput(context.Background(), "status") + }() + } + go func() { wg.Wait(); close(done) }() + + select { + case <-done: + elapsed := time.Since(start) + // 6 calls through cap=2 with ~200ms each ⇒ ~600ms. 5s is a generous + // upper bound that still catches a leaked slot (which would hang + // indefinitely and only be observed via the select timeout below). + if elapsed > 5*time.Second { + t.Fatalf("runGitOutput took %v for %d calls through cap=2; throttle slot may have leaked", elapsed, N) + } + case <-time.After(5 * time.Second): + t.Fatal("runGitOutput hung past 5s — throttle slot leaked on per-command timeout") + } +} + +// TestRunGitOutput_PollingVariantReleasesThrottleOnTimeout mirrors the +// regression above for runPollingGitOutput, which is the entry point used +// by the actual workspace poll loop. Same setup, same expectations. +func TestRunGitOutput_PollingVariantReleasesThrottleOnTimeout(t *testing.T) { + prev := gitCommandTimeout + gitCommandTimeout = 200 * time.Millisecond + t.Cleanup(func() { gitCommandTimeout = prev }) + + restoreCap := subproc.Git().SetCapForTest(2) + t.Cleanup(restoreCap) + + shimDir := installSleepGitShim(t) + t.Setenv("PATH", shimDir+string(os.PathListSeparator)+os.Getenv("PATH")) + + _, wt := setupTestDir(t) + + const N = 6 + var wg sync.WaitGroup + wg.Add(N) + done := make(chan struct{}) + start := time.Now() + for i := 0; i < N; i++ { + go func() { + defer wg.Done() + _, _ = wt.runPollingGitOutput(context.Background(), "status") + }() + } + go func() { wg.Wait(); close(done) }() + + select { + case <-done: + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Fatalf("runPollingGitOutput took %v for %d calls through cap=2; throttle slot may have leaked", elapsed, N) + } + case <-time.After(5 * time.Second): + t.Fatal("runPollingGitOutput hung past 5s — throttle slot leaked on per-command timeout") + } +} + +// installSleepGitShim writes an executable shell script named 'git' into a +// fresh tmpdir that blocks for an hour, and returns the dir. Callers must +// prepend it to $PATH so exec.Command("git", ...) picks it up before the +// real binary. The long sleep guarantees that any successful completion +// must come from the per-command timeout SIGKILLing the subprocess. +// +// `exec sleep 3600` (instead of `sleep 3600`) makes sleep replace sh as +// the process image, so SIGKILL on the exec.Cmd's pid lands directly on +// sleep. Otherwise sh would die, leave sleep orphaned still holding the +// stdout/stderr pipes, and cmd.Output() would block waiting for pipe +// closure for the full hour — making the test flaky. +func installSleepGitShim(t *testing.T) string { + t.Helper() + dir := t.TempDir() + shim := filepath.Join(dir, "git") + const script = "#!/bin/sh\nexec sleep 3600\n" + if err := os.WriteFile(shim, []byte(script), 0o755); err != nil { + t.Fatalf("write shim: %v", err) + } + return dir +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_diff.go b/apps/backend/internal/agentctl/server/process/workspace_git_diff.go index 0cae50cd85..7a94626c9f 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_diff.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_diff.go @@ -37,31 +37,43 @@ const ( diffSkipReasonBudgetExceeded = "budget_exceeded" ) -// enrichWithDiffData adds diff information (additions, deletions, diff content) to file info -func (wt *WorkspaceTracker) enrichWithDiffData(ctx context.Context, update *types.GitStatusUpdate) { +// enrichWithDiffData adds diff information (additions, deletions, diff content) +// to file info. The prior status is threaded through so per-file diff data can +// be carried forward when a git command times out — without it a single bad +// poll would wipe the diffs panel for files that genuinely still have changes. +// +// Carry-forward runs once at the end, after every enrichment phase has had a +// chance to populate fresh data. Running it inside enrichWithUnstagedDiff would +// pre-fill fi.Diff for staged-only files, and enrichWithStagedDiff's +// `if fileInfo.Diff == ""` guard would then skip the fresh staged diff fetch. +func (wt *WorkspaceTracker) enrichWithDiffData(ctx context.Context, update *types.GitStatusUpdate, prior types.GitStatusUpdate) { // Always diff against HEAD for unstaged/staged content so that files committed // locally (but not yet pushed) show only their uncommitted changes rather than // the entire file as new. The remote branch is only relevant for ahead/behind counts. - wt.enrichWithUnstagedDiff(ctx, update, "HEAD") - wt.enrichWithStagedDiff(ctx, update, "HEAD") + wt.enrichWithUnstagedDiff(ctx, update, "HEAD", prior) + wt.enrichWithStagedDiff(ctx, update, "HEAD", prior) wt.enrichUntrackedFileDiffs(ctx, update) + carryForwardFileDiffs(update, prior) } // enrichWithBranchDiff computes the total additions/deletions for the entire branch // vs the merge-base, covering committed + staged + unstaged changes in one pass. // Untracked file line counts (already computed) are added on top. // The result is stored in BranchAdditions/BranchDeletions for the sidebar display. -func (wt *WorkspaceTracker) enrichWithBranchDiff(ctx context.Context, update *types.GitStatusUpdate) { +// +// On numstat failure the totals are carried forward from prior when HEAD +// hasn't moved, mirroring the per-command carry-forward used elsewhere in +// getGitStatus to avoid a transient git timeout clearing the sidebar count. +func (wt *WorkspaceTracker) enrichWithBranchDiff(ctx context.Context, update *types.GitStatusUpdate, prior types.GitStatusUpdate) { if update.BaseCommit == "" { return } // git diff --numstat covers committed + staged + unstaged changes. - numstatCmd := exec.CommandContext(ctx, "git", "diff", "--numstat", update.BaseCommit) - numstatCmd.Dir = wt.workDir - numstatOut, err := subproc.RunGitOutput(ctx, numstatCmd) + numstatOut, err := wt.runGitOutput(ctx, "diff", "--numstat", update.BaseCommit) if err != nil { - wt.logger.Debug("enrichWithBranchDiff: numstat failed", zap.Error(err)) + wt.logger.Debug("enrichWithBranchDiff: numstat failed, carrying forward", zap.Error(err)) + carryBranchDiff(update, prior) return } @@ -101,6 +113,83 @@ func totalDiffBytes(update *types.GitStatusUpdate) int64 { return total } +// carryBranchDiff copies prior.BranchAdditions/BranchDeletions onto update when +// neither HEAD nor BaseCommit has moved. A new HEAD or a re-pointed merge-base +// invalidates the cached totals — both are inputs to `git diff --numstat +// `, so a change in either makes the prior count stale. +func carryBranchDiff(update *types.GitStatusUpdate, prior types.GitStatusUpdate) { + if prior.HeadCommit == "" || prior.HeadCommit != update.HeadCommit { + return + } + if prior.BaseCommit != update.BaseCommit { + return + } + update.BranchAdditions = prior.BranchAdditions + update.BranchDeletions = prior.BranchDeletions +} + +// carryForwardFileDiffs copies per-file diff content, additions, deletions, and +// skip reason from prior.Files into update.Files for files still present in +// update.Files. Used when a `git diff --numstat` call (the gate that drives +// per-file diff enrichment) fails: the porcelain status already reported which +// files changed; we just can't compute fresh diffs this poll, so we keep the +// last known diff data visible until the next successful poll. HEAD is +// required to be unchanged so we don't show stale diffs after a reset/commit. +func carryForwardFileDiffs(update *types.GitStatusUpdate, prior types.GitStatusUpdate) { + if prior.HeadCommit == "" || prior.HeadCommit != update.HeadCommit { + return + } + for path, fi := range update.Files { + // A non-empty Diff means an earlier enrichment pass already populated + // this entry (e.g. unstaged numstat succeeded, then staged numstat + // failed). Leave the freshly-computed entry alone — only fill in + // completely-missing diff data. + if fi.Diff != "" { + continue + } + // A skip reason set this poll (budget_exceeded, too_large, binary) + // is an intentional decision to not surface diff content; restoring + // a prior diff would defeat that skip and could show stale content + // for a file that has since grown past the size or budget cap. + if fi.DiffSkipReason != "" { + continue + } + priorFi, ok := prior.Files[path] + if !ok || priorFi.Diff == "" { + continue + } + fi.Diff = priorFi.Diff + fi.DiffSkipReason = priorFi.DiffSkipReason + if fi.Additions == 0 { + fi.Additions = priorFi.Additions + } + if fi.Deletions == 0 { + fi.Deletions = priorFi.Deletions + } + update.Files[path] = fi + } +} + +// carryForwardFileDiff fills fi.Diff/DiffSkipReason from prior for a single +// file when HEAD matches and prior had diff content. Used when numstat +// succeeded (so Additions/Deletions are fresh) but the per-file capDiffOutput +// call returned empty — typically a 10s gitCommandTimeout on a large diff, +// throttle starvation, or a broken pipe. Without this, a single slow file +// would blank the diff content in the UI even though we have a usable cached +// copy. Returns the (possibly updated) FileInfo so callers can re-store it. +func carryForwardFileDiff(fi types.FileInfo, filePath string, update *types.GitStatusUpdate, prior types.GitStatusUpdate) types.FileInfo { + if prior.HeadCommit == "" || prior.HeadCommit != update.HeadCommit { + return fi + } + priorFi, ok := prior.Files[filePath] + if !ok || priorFi.Diff == "" { + return fi + } + fi.Diff = priorFi.Diff + fi.DiffSkipReason = priorFi.DiffSkipReason + return fi +} + // capDiffOutput runs a git diff command and returns at most maxDiffOutputSize bytes. // Returns the output string and whether it was truncated. // @@ -110,14 +199,16 @@ func totalDiffBytes(update *types.GitStatusUpdate) int64 { // pipe-based reads. Slot is acquired before Start; if Start fails we // release immediately, else release runs after Wait. func capDiffOutput(ctx context.Context, workDir string, args ...string) (string, bool) { - cmd := exec.CommandContext(ctx, "git", args...) + cctx, cancel := context.WithTimeout(ctx, gitCommandTimeout) + defer cancel() + cmd := exec.CommandContext(cctx, "git", args...) cmd.Dir = workDir stdout, err := cmd.StdoutPipe() if err != nil { return "", false } - release, err := subproc.Git().Acquire(ctx) + release, err := subproc.Git().Acquire(cctx) if err != nil { _ = stdout.Close() return "", false @@ -169,13 +260,17 @@ func resolveNumstatPath(numstatPath string) string { } // enrichWithUnstagedDiff populates additions/deletions and diff content for files -// with unstaged changes by comparing the worktree against baseRef. -func (wt *WorkspaceTracker) enrichWithUnstagedDiff(ctx context.Context, update *types.GitStatusUpdate, baseRef string) { - numstatCmd := exec.CommandContext(ctx, "git", "diff", "--numstat", baseRef) - numstatCmd.Dir = wt.workDir - numstatOut, err := subproc.RunGitOutput(ctx, numstatCmd) +// with unstaged changes by comparing the worktree against baseRef. On numstat +// failure the function returns early and leaves the carry-forward to +// enrichWithDiffData — preventing a single git timeout from blanking the diffs +// panel for files whose porcelain status still shows them as changed, while +// keeping the staged phase free to populate fresh data for staged-only files. +func (wt *WorkspaceTracker) enrichWithUnstagedDiff(ctx context.Context, update *types.GitStatusUpdate, baseRef string, prior types.GitStatusUpdate) { + numstatOut, err := wt.runGitOutput(ctx, "diff", "--numstat", baseRef) if err != nil { - wt.logger.Debug("failed to get numstat", zap.Error(err)) + // Carry-forward happens once at the end of enrichWithDiffData so the + // staged phase can still populate fresh diffs for staged-only files. + wt.logger.Debug("enrichWithUnstagedDiff: numstat failed", zap.Error(err)) return } @@ -221,6 +316,8 @@ func (wt *WorkspaceTracker) enrichWithUnstagedDiff(ctx context.Context, update * if truncated { fileInfo.DiffSkipReason = diffSkipReasonTruncated } + } else { + fileInfo = carryForwardFileDiff(fileInfo, filePath, update, prior) } update.Files[filePath] = fileInfo @@ -228,15 +325,18 @@ func (wt *WorkspaceTracker) enrichWithUnstagedDiff(ctx context.Context, update * } // enrichWithStagedDiff populates additions/deletions and diff content for staged files -// that have no additional unstaged changes, using git diff --cached. -func (wt *WorkspaceTracker) enrichWithStagedDiff(ctx context.Context, update *types.GitStatusUpdate, baseRef string) { +// that have no additional unstaged changes, using git diff --cached. On --cached +// numstat failure the function returns early; carry-forward happens once at the +// end of enrichWithDiffData (same rationale as enrichWithUnstagedDiff). +func (wt *WorkspaceTracker) enrichWithStagedDiff(ctx context.Context, update *types.GitStatusUpdate, baseRef string, prior types.GitStatusUpdate) { // For staged files that don't have unstaged changes, we need to get the diff from the index. // The first diff (git diff baseRef) shows worktree vs baseRef, but if a file is staged // and has no additional unstaged changes, its diff won't appear there. - stagedCmd := exec.CommandContext(ctx, "git", "diff", "--cached", "--numstat", baseRef) - stagedCmd.Dir = wt.workDir - stagedOut, err := subproc.RunGitOutput(ctx, stagedCmd) + stagedOut, err := wt.runGitOutput(ctx, "diff", "--cached", "--numstat", baseRef) if err != nil { + // Carry-forward happens at the end of enrichWithDiffData; running it + // here would mask the unstaged phase's fresh data. + wt.logger.Debug("enrichWithStagedDiff: --cached numstat failed", zap.Error(err)) return } @@ -283,6 +383,8 @@ func (wt *WorkspaceTracker) enrichWithStagedDiff(ctx context.Context, update *ty if truncated { fileInfo.DiffSkipReason = diffSkipReasonTruncated } + } else { + fileInfo = carryForwardFileDiff(fileInfo, filePath, update, prior) } } update.Files[filePath] = fileInfo diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_diff_test.go b/apps/backend/internal/agentctl/server/process/workspace_git_diff_test.go index d7858a79e7..f7ef9fa561 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_diff_test.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_diff_test.go @@ -273,7 +273,7 @@ func TestEnrichStagedDiff_RenamedFile(t *testing.T) { }, } - wt.enrichWithStagedDiff(context.Background(), update, "HEAD") + wt.enrichWithStagedDiff(context.Background(), update, "HEAD", streams.GitStatusUpdate{}) fi := update.Files["renamed.txt"] if fi.Diff == "" { @@ -283,3 +283,271 @@ func TestEnrichStagedDiff_RenamedFile(t *testing.T) { t.Error("expected non-zero additions or deletions for renamed+modified file") } } + +// TestCarryBranchDiff covers carry-forward of branch additions/deletions when +// the `git diff --numstat ` call fails — without it, a transient +// timeout would clear the sidebar's branch-totals count. +func TestCarryBranchDiff(t *testing.T) { + head := "abc123" + base := "base000" + tests := []struct { + name string + prior streams.GitStatusUpdate + updateHead string + updateBase string + wantAdditions int + wantDeletions int + }{ + { + name: "same head and base preserves totals", + prior: streams.GitStatusUpdate{HeadCommit: head, BaseCommit: base, BranchAdditions: 42, BranchDeletions: 7}, + updateHead: head, + updateBase: base, + wantAdditions: 42, + wantDeletions: 7, + }, + { + name: "different head drops totals", + prior: streams.GitStatusUpdate{HeadCommit: head, BaseCommit: base, BranchAdditions: 42, BranchDeletions: 7}, + updateHead: "def456", + updateBase: base, + wantAdditions: 0, + wantDeletions: 0, + }, + { + name: "different base drops totals", + prior: streams.GitStatusUpdate{HeadCommit: head, BaseCommit: base, BranchAdditions: 42, BranchDeletions: 7}, + updateHead: head, + updateBase: "newbase", + wantAdditions: 0, + wantDeletions: 0, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + update := &streams.GitStatusUpdate{HeadCommit: tt.updateHead, BaseCommit: tt.updateBase} + carryBranchDiff(update, tt.prior) + if update.BranchAdditions != tt.wantAdditions || update.BranchDeletions != tt.wantDeletions { + t.Errorf("additions/deletions = %d/%d, want %d/%d", + update.BranchAdditions, update.BranchDeletions, tt.wantAdditions, tt.wantDeletions) + } + }) + } +} + +// TestCarryForwardFileDiffs covers per-file diff carry-forward. The numstat +// command is the gate that drives per-file enrichment; when it fails we keep +// the prior diff content visible (as long as HEAD hasn't moved) instead of +// clearing the diff panel on a transient timeout. +func TestCarryForwardFileDiffs(t *testing.T) { + head := "abc123" + priorDiff := "diff --git a/foo b/foo\n+new line\n" + + t.Run("preserves diff additions deletions when head matches", func(t *testing.T) { + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff, Additions: 5, Deletions: 2, DiffSkipReason: ""}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go"}, + }, + } + carryForwardFileDiffs(update, prior) + got := update.Files["foo.go"] + if got.Diff != priorDiff { + t.Errorf("Diff = %q, want %q", got.Diff, priorDiff) + } + if got.Additions != 5 || got.Deletions != 2 { + t.Errorf("Additions/Deletions = %d/%d, want 5/2", got.Additions, got.Deletions) + } + }) + + t.Run("skips when head moved", func(t *testing.T) { + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff, Additions: 5, Deletions: 2}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: "different", + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go"}, + }, + } + carryForwardFileDiffs(update, prior) + got := update.Files["foo.go"] + if got.Diff != "" || got.Additions != 0 || got.Deletions != 0 { + t.Errorf("expected zeroed FileInfo on head mismatch, got %+v", got) + } + }) + + t.Run("skips files not in prior", func(t *testing.T) { + prior := streams.GitStatusUpdate{HeadCommit: head, Files: map[string]streams.FileInfo{}} + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{"new.go": {Path: "new.go"}}, + } + carryForwardFileDiffs(update, prior) + got := update.Files["new.go"] + if got.Diff != "" { + t.Errorf("expected no carry-forward for new file, got Diff=%q", got.Diff) + } + }) + + t.Run("does not overwrite already-populated update entry", func(t *testing.T) { + newDiff := "diff --git a/foo b/foo\n+brand new\n" + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff, Additions: 5, Deletions: 2}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: newDiff, Additions: 1, Deletions: 0}, + }, + } + carryForwardFileDiffs(update, prior) + got := update.Files["foo.go"] + if got.Diff != newDiff { + t.Errorf("Diff overwritten; got %q, want %q", got.Diff, newDiff) + } + if got.Additions != 1 || got.Deletions != 0 { + t.Errorf("counts overwritten; got %d/%d, want 1/0", got.Additions, got.Deletions) + } + }) + + t.Run("respects skip reason set this poll", func(t *testing.T) { + skipReasons := []string{ + diffSkipReasonBudgetExceeded, + diffSkipReasonTooLarge, + diffSkipReasonBinary, + } + for _, reason := range skipReasons { + t.Run(reason, func(t *testing.T) { + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff, Additions: 5, Deletions: 2}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", DiffSkipReason: reason}, + }, + } + carryForwardFileDiffs(update, prior) + got := update.Files["foo.go"] + if got.Diff != "" { + t.Errorf("Diff carried forward despite skip reason %q; got %q", reason, got.Diff) + } + if got.DiffSkipReason != reason { + t.Errorf("DiffSkipReason = %q, want %q", got.DiffSkipReason, reason) + } + }) + } + }) +} + +func TestCarryForwardFileDiff(t *testing.T) { + head := "abc123" + priorDiff := "diff --git a/foo b/foo\n+kept line\n" + + t.Run("fills empty Diff from prior when head matches", func(t *testing.T) { + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff, DiffSkipReason: "truncated"}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + // numstat already populated additions/deletions; capDiffOutput then returned "". + "foo.go": {Path: "foo.go", Additions: 7, Deletions: 3}, + }, + } + fi := update.Files["foo.go"] + fi = carryForwardFileDiff(fi, "foo.go", update, prior) + if fi.Diff != priorDiff { + t.Errorf("Diff = %q, want %q", fi.Diff, priorDiff) + } + if fi.DiffSkipReason != "truncated" { + t.Errorf("DiffSkipReason = %q, want %q", fi.DiffSkipReason, "truncated") + } + if fi.Additions != 7 || fi.Deletions != 3 { + t.Errorf("counts should be untouched, got %d/%d, want 7/3", fi.Additions, fi.Deletions) + } + }) + + t.Run("no-op when head moved", func(t *testing.T) { + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: "different", + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go"}, + }, + } + fi := carryForwardFileDiff(update.Files["foo.go"], "foo.go", update, prior) + if fi.Diff != "" { + t.Errorf("expected no carry-forward on head mismatch, got Diff=%q", fi.Diff) + } + }) + + t.Run("no-op when file absent from prior", func(t *testing.T) { + prior := streams.GitStatusUpdate{HeadCommit: head, Files: map[string]streams.FileInfo{}} + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{"new.go": {Path: "new.go"}}, + } + fi := carryForwardFileDiff(update.Files["new.go"], "new.go", update, prior) + if fi.Diff != "" { + t.Errorf("expected no carry-forward for new file, got Diff=%q", fi.Diff) + } + }) + + t.Run("no-op when prior diff empty", func(t *testing.T) { + prior := streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Additions: 1, Deletions: 1}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{"foo.go": {Path: "foo.go"}}, + } + fi := carryForwardFileDiff(update.Files["foo.go"], "foo.go", update, prior) + if fi.Diff != "" || fi.DiffSkipReason != "" { + t.Errorf("expected no carry-forward when prior Diff empty, got %+v", fi) + } + }) + + t.Run("no-op when prior head empty", func(t *testing.T) { + prior := streams.GitStatusUpdate{ + Files: map[string]streams.FileInfo{ + "foo.go": {Path: "foo.go", Diff: priorDiff}, + }, + } + update := &streams.GitStatusUpdate{ + HeadCommit: head, + Files: map[string]streams.FileInfo{"foo.go": {Path: "foo.go"}}, + } + fi := carryForwardFileDiff(update.Files["foo.go"], "foo.go", update, prior) + if fi.Diff != "" { + t.Errorf("expected no carry-forward on empty prior head, got Diff=%q", fi.Diff) + } + }) +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_status.go b/apps/backend/internal/agentctl/server/process/workspace_git_status.go index 6c946fa7f3..afbf3f62b1 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_status.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_status.go @@ -2,13 +2,11 @@ package process import ( "context" - "os/exec" "strconv" "strings" "time" "github.com/kandev/kandev/internal/agentctl/types" - "github.com/kandev/kandev/internal/common/subproc" "go.uber.org/zap" ) @@ -76,7 +74,12 @@ func (wt *WorkspaceTracker) GetGitStatus(ctx context.Context, fresh bool) (types return status, nil } -// getGitStatus retrieves the current git status +// getGitStatus retrieves the current git status. Secondary commands +// (ahead/behind counts, diff enrichment) that fail or time out carry their +// values forward from the prior cached status rather than overwriting them +// with zero. The two primary commands (branch info and `git status +// --porcelain`) still propagate errors upward; on those, updateGitStatus +// skips the cache write entirely so the prior state stays visible to the UI. func (wt *WorkspaceTracker) getGitStatus(ctx context.Context) (types.GitStatusUpdate, error) { update := types.GitStatusUpdate{ Timestamp: time.Now(), @@ -99,47 +102,50 @@ func (wt *WorkspaceTracker) getGitStatus(ctx context.Context) (types.GitStatusUp return update, nil } + // Snapshot the prior cache once so per-command carry-forward sees a + // consistent view even if a concurrent RefreshGitStatus replaces + // currentStatus mid-poll. + wt.mu.RLock() + prior := wt.currentStatus + wt.mu.RUnlock() + if err := wt.getGitBranchInfo(ctx, &update); err != nil { return update, err } - wt.getAheadBehindCounts(ctx, &update) + wt.getAheadBehindCounts(ctx, &update, prior) if err := wt.parseGitStatusOutput(ctx, &update); err != nil { return update, err } // Enrich file info with diff data (additions, deletions, and actual diff content) - wt.enrichWithDiffData(ctx, &update) + wt.enrichWithDiffData(ctx, &update, prior) // Compute full branch totals vs merge-base (committed + staged + unstaged + untracked) - wt.enrichWithBranchDiff(ctx, &update) + wt.enrichWithBranchDiff(ctx, &update, prior) return update, nil } // getGitBranchInfo populates branch, remote branch, head commit, and base commit fields. +// Each command runs under a per-command timeout via the runGit* helpers so a +// single wedged git invocation cannot pin the shared throttle slot. func (wt *WorkspaceTracker) getGitBranchInfo(ctx context.Context, update *types.GitStatusUpdate) error { // Get current branch - branchCmd := exec.CommandContext(ctx, "git", "rev-parse", "--abbrev-ref", "HEAD") - branchCmd.Dir = wt.workDir - branchOut, err := subproc.RunGitOutput(ctx, branchCmd) + branchOut, err := wt.runGitOutput(ctx, "rev-parse", "--abbrev-ref", "HEAD") if err != nil { return err } update.Branch = strings.TrimSpace(string(branchOut)) // Get remote branch - remoteCmd := exec.CommandContext(ctx, "git", "rev-parse", "--abbrev-ref", "@{upstream}") - remoteCmd.Dir = wt.workDir - if remoteOut, err := subproc.RunGitOutput(ctx, remoteCmd); err == nil { + if remoteOut, err := wt.runGitOutput(ctx, "rev-parse", "--abbrev-ref", "@{upstream}"); err == nil { update.RemoteBranch = strings.TrimSpace(string(remoteOut)) } // Get current HEAD commit SHA - headCmd := exec.CommandContext(ctx, "git", "rev-parse", "HEAD") - headCmd.Dir = wt.workDir - if headOut, err := subproc.RunGitOutput(ctx, headCmd); err == nil { + if headOut, err := wt.runGitOutput(ctx, "rev-parse", "HEAD"); err == nil { update.HeadCommit = strings.TrimSpace(string(headOut)) } @@ -148,9 +154,7 @@ func (wt *WorkspaceTracker) getGitBranchInfo(ctx context.Context, update *types. // so we show all changes this branch introduces compared to the main development line. var baseBranch string for _, candidate := range []string{"origin/main", "origin/master", "main", "master"} { - checkCmd := exec.CommandContext(ctx, "git", "rev-parse", "--verify", candidate) - checkCmd.Dir = wt.workDir - if err := subproc.RunGit(ctx, checkCmd); err == nil { + if err := wt.runGit(ctx, "rev-parse", "--verify", candidate); err == nil { baseBranch = candidate break } @@ -158,9 +162,7 @@ func (wt *WorkspaceTracker) getGitBranchInfo(ctx context.Context, update *types. if baseBranch != "" { // Use merge-base to find common ancestor between current branch and base branch. // This is correct even when the branch has diverged from main. - mergeBaseCmd := exec.CommandContext(ctx, "git", "merge-base", baseBranch, "HEAD") - mergeBaseCmd.Dir = wt.workDir - if mergeBaseOut, err := subproc.RunGitOutput(ctx, mergeBaseCmd); err == nil { + if mergeBaseOut, err := wt.runGitOutput(ctx, "merge-base", baseBranch, "HEAD"); err == nil { update.BaseCommit = strings.TrimSpace(string(mergeBaseOut)) } } @@ -168,44 +170,63 @@ func (wt *WorkspaceTracker) getGitBranchInfo(ctx context.Context, update *types. return nil } -// getAheadBehindCounts populates the Ahead/Behind fields relative to the base branch -// (origin/main or origin/master). Always compares against the base branch rather than -// the remote tracking branch, because after a rebase the tracking branch has stale -// commit SHAs that produce inflated counts. -func (wt *WorkspaceTracker) getAheadBehindCounts(ctx context.Context, update *types.GitStatusUpdate) { +// getAheadBehindCounts populates the Ahead/Behind fields relative to the base +// branch (origin/main or origin/master). Always compares against the base +// branch rather than the remote tracking branch, because after a rebase the +// tracking branch has stale commit SHAs that produce inflated counts. +// +// If either the compareRef lookup or the count command fails (e.g. timeout +// under throttle saturation, transient FS hang), Ahead/Behind are carried +// forward from prior when HEAD hasn't moved. Otherwise a single bad poll +// would silently overwrite the cached counts with 0/0 and hide a legitimate +// "Pull N" / "Push N" indicator in the UI. +func (wt *WorkspaceTracker) getAheadBehindCounts(ctx context.Context, update *types.GitStatusUpdate, prior types.GitStatusUpdate) { // Always compare against the base branch (origin/main or origin/master). // Using the remote tracking branch (origin/) gives wrong counts // after rebase because rebased commits have new SHAs. var compareRef string for _, candidate := range []string{"origin/main", "origin/master"} { - checkCmd := exec.CommandContext(ctx, "git", "rev-parse", "--verify", candidate) - checkCmd.Dir = wt.workDir - if err := subproc.RunGit(ctx, checkCmd); err == nil { + if err := wt.runGit(ctx, "rev-parse", "--verify", candidate); err == nil { compareRef = candidate break } } if compareRef == "" { + carryAheadBehind(update, prior) return } - countCmd := exec.CommandContext(ctx, "git", "rev-list", "--left-right", "--count", update.Branch+"..."+compareRef) - countCmd.Dir = wt.workDir - if countOut, err := subproc.RunGitOutput(ctx, countCmd); err == nil { - parts := strings.Fields(string(countOut)) - if len(parts) == 2 { - update.Ahead, _ = strconv.Atoi(parts[0]) - update.Behind, _ = strconv.Atoi(parts[1]) - } + countOut, err := wt.runGitOutput(ctx, "rev-list", "--left-right", "--count", update.Branch+"..."+compareRef) + if err != nil { + wt.logger.Debug("getAheadBehindCounts: rev-list failed, carrying forward", zap.Error(err)) + carryAheadBehind(update, prior) + return } + parts := strings.Fields(string(countOut)) + if len(parts) != 2 { + carryAheadBehind(update, prior) + return + } + update.Ahead, _ = strconv.Atoi(parts[0]) + update.Behind, _ = strconv.Atoi(parts[1]) +} + +// carryAheadBehind copies prior.Ahead/Behind onto update when HEAD hasn't +// moved. A new HEAD invalidates prior counts (the user committed, pulled, or +// reset), so we leave update zeroed in that case. +func carryAheadBehind(update *types.GitStatusUpdate, prior types.GitStatusUpdate) { + if prior.HeadCommit == "" || prior.HeadCommit != update.HeadCommit { + return + } + update.Ahead = prior.Ahead + update.Behind = prior.Behind } // 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. - // 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 := subproc.RunGitOutput(ctx, statusCmd) + // GIT_OPTIONAL_LOCKS=0 (via runPollingGitOutput) prevents the background poll loop from + // taking .git/index.lock, which would race with concurrent user-initiated git operations. + statusOut, err := wt.runPollingGitOutput(ctx, "status", "--porcelain", "--untracked-files=all") if err != nil { return err } diff --git a/apps/backend/internal/agentctl/server/process/workspace_git_status_test.go b/apps/backend/internal/agentctl/server/process/workspace_git_status_test.go index 2237301b18..f48dbcc054 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_git_status_test.go +++ b/apps/backend/internal/agentctl/server/process/workspace_git_status_test.go @@ -5,6 +5,8 @@ import ( "os" "path/filepath" "testing" + + "github.com/kandev/kandev/internal/agentctl/types" ) func TestUnquoteGitPath(t *testing.T) { @@ -208,3 +210,49 @@ func mapKeys[V any](m map[string]V) []string { } return keys } + +// TestCarryAheadBehind covers the carry-forward fallback used when the +// ahead/behind git command fails (timeout, missing upstream). The contract: +// preserve prior counts when HEAD is unchanged, leave them zero when HEAD +// moved (the prior counts are stale by definition) or when prior is empty. +func TestCarryAheadBehind(t *testing.T) { + head := "abc123" + tests := []struct { + name string + prior types.GitStatusUpdate + updateHead string + wantAhead int + wantBehind int + }{ + { + name: "same head preserves counts", + prior: types.GitStatusUpdate{HeadCommit: head, Ahead: 1, Behind: 3}, + updateHead: head, + wantAhead: 1, + wantBehind: 3, + }, + { + name: "different head drops counts", + prior: types.GitStatusUpdate{HeadCommit: head, Ahead: 1, Behind: 3}, + updateHead: "def456", + wantAhead: 0, + wantBehind: 0, + }, + { + name: "empty prior head no-op", + prior: types.GitStatusUpdate{Ahead: 9, Behind: 9}, + updateHead: head, + wantAhead: 0, + wantBehind: 0, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + update := &types.GitStatusUpdate{HeadCommit: tt.updateHead} + carryAheadBehind(update, tt.prior) + if update.Ahead != tt.wantAhead || update.Behind != tt.wantBehind { + t.Errorf("ahead/behind = %d/%d, want %d/%d", update.Ahead, update.Behind, tt.wantAhead, tt.wantBehind) + } + }) + } +} diff --git a/apps/backend/internal/agentctl/server/process/workspace_monitor.go b/apps/backend/internal/agentctl/server/process/workspace_monitor.go index 9fed11af14..087fdabfbf 100644 --- a/apps/backend/internal/agentctl/server/process/workspace_monitor.go +++ b/apps/backend/internal/agentctl/server/process/workspace_monitor.go @@ -20,9 +20,19 @@ import ( // causing "too many open files" errors in large workspaces. const DefaultFilePollInterval = 2 * time.Second -// gitCommandTimeout is the maximum time to wait for git commands during polling. -// This prevents the monitor loop from hanging if git is blocked (e.g., index lock). -const gitCommandTimeout = 10 * time.Second +// gitCommandTimeout is the maximum time to wait for any single git subprocess +// the workspace tracker spawns (monitor loop quick-status, full status refresh, +// diff enrichment). Without a per-command bound, a wedged invocation (FS hang, +// index lock, EDR scan) would inherit only the tracker's lifetime ctx and pin +// its gitThrottle slot for the lifetime of the session, blocking subsequent +// poll cycles and producing the "empty changes panel after freeze" symptom. +// 10s is generous for normal completion (typical poll commands finish in +// <100ms) while still freeing the slot in time for the next poll cycle. +// +// A var (not const) so the regression test in workspace_git_cmd_test.go can +// shrink it to milliseconds and prove that timed-out invocations release +// their gitThrottle slot. Production code must not mutate it. +var gitCommandTimeout = 10 * time.Second // maxConsecutiveGitFailures is the number of consecutive git command failures // before the polling loop gives up and stops. At 2-second intervals, 5 failures