diff --git a/apps/backend/internal/agent/runtime/agentctl/agent.go b/apps/backend/internal/agent/runtime/agentctl/agent.go index 06931f72b4..5b4039893d 100644 --- a/apps/backend/internal/agent/runtime/agentctl/agent.go +++ b/apps/backend/internal/agent/runtime/agentctl/agent.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "sync" "time" "github.com/gorilla/websocket" @@ -334,6 +335,29 @@ func (c *Client) HasAgentStream() bool { } // readUpdatesStream is the read loop for the agent updates WebSocket stream. +// +// Agent events (message_chunk, tool_call, complete, ...) are NOT run inline on +// this loop. They are handed, in order, to a single dispatchAgentEvents worker +// goroutine through an unbounded reader-side queue. This keeps the read loop +// free to deliver request/response frames (e.g. the agent.cancel response) even +// while an event handler is blocked. +// +// Why this matters: handler → orchestrator.handleAgentReady acquires the +// per-session cancelInFlight guard. During a user cancel, Service.CancelAgent +// holds that guard while blocked in agentctl.Client.Cancel → sendStreamRequest, +// waiting for the agent.cancel response frame. If the `complete` event's +// handler ran inline here, this loop would block on the guard and never read +// the response frame the guard holder is waiting for — a cross-goroutine +// deadlock (both sides waited ~forever in production; see the goroutine dump on +// task 544afdae). Offloading handler execution lets the response frame land +// immediately, the cancel returns, the guard releases, and the deferred +// handleAgentReady then safely no-ops. +// +// The queue is unbounded on purpose: a fixed buffered channel would let a burst +// of events pile up behind a blocked handler and then backpressure the read +// loop on send, re-wedging response-frame delivery exactly as the inline +// version did. enqueue never blocks the read loop, so no volume of events can +// starve the cancel response. func (c *Client) readUpdatesStream( ctx context.Context, conn *websocket.Conn, @@ -342,11 +366,33 @@ func (c *Client) readUpdatesStream( onDisconnect func(err error), writeMessage func([]byte) error, ) { + // Ordered, single-worker dispatch preserves per-stream event ordering while + // decoupling handler execution from response-frame delivery. + events := newAgentEventQueue() + workerDone := make(chan struct{}) + go c.dispatchAgentEvents(handler, events, workerDone) + var lastErr error defer func() { - // Clean up pending requests before signaling disconnect + // Clean up pending requests BEFORE draining the worker. On a + // connection drop mid-cancel, a worker handler + // (orchestrator.handleAgentReady) can block acquiring the per-session + // cancelInFlight guard that an in-flight Service.CancelAgent holds + // while it waits inside sendStreamRequest for the agent.cancel + // response frame. cleanupPendingRequests closes that pending response + // channel, unblocking CancelAgent so it releases the guard, which lets + // the blocked handler finish and the worker exit. Draining first + // (<-workerDone) would wait on that handler forever, so cleanup could + // never run — the exact deadlock class this stream rework fixes, but on + // the disconnect path. c.cleanupPendingRequests() + // Stop the worker and wait for the in-flight handler to unwind before + // signaling disconnect, so the drain barrier semantics callers rely on + // (promptDoneCh signaling, status updates) still hold. + events.close() + <-workerDone + c.mu.Lock() c.agentStreamConn = nil c.mu.Unlock() @@ -403,10 +449,97 @@ func (c *Client) readUpdatesStream( } tracing.TraceAgentEvent(ctx, event.Type, event.SessionID, c.executionID, message) + // Hand off to the ordered worker rather than running handler inline. + // enqueue never blocks the read loop, so a burst of events behind a + // blocked handler can't backpressure delivery of response frames. + events.enqueue(event) + } +} + +// dispatchAgentEvents runs the agent-event handler sequentially for every event +// pushed onto the queue, preserving arrival order. It is the single consumer of +// the queue; readUpdatesStream closes the queue on teardown and waits on +// workerDone so an in-flight handler finishes before disconnect is signaled. +func (c *Client) dispatchAgentEvents(handler func(AgentEvent), events *agentEventQueue, done chan<- struct{}) { + defer close(done) + for { + event, ok := events.dequeue() + if !ok { + return + } handler(event) } } +// agentEventQueue is an unbounded FIFO queue decoupling the stream read loop +// (producer) from the ordered event-dispatch worker (consumer). enqueue never +// blocks, so no volume of events can backpressure the read loop and starve +// response-frame delivery. A single-slot notify channel wakes the worker +// without accumulating a signal per event. +type agentEventQueue struct { + mu sync.Mutex + items []AgentEvent + notify chan struct{} + closed bool +} + +func newAgentEventQueue() *agentEventQueue { + return &agentEventQueue{notify: make(chan struct{}, 1)} +} + +// enqueue appends an event and wakes the worker. It is a no-op after close. +func (q *agentEventQueue) enqueue(event AgentEvent) { + q.mu.Lock() + if q.closed { + q.mu.Unlock() + return + } + q.items = append(q.items, event) + q.mu.Unlock() + select { + case q.notify <- struct{}{}: + default: + } +} + +// close marks the queue closed and wakes the worker so it can drain any +// remaining items and then exit. +func (q *agentEventQueue) close() { + q.mu.Lock() + q.closed = true + q.mu.Unlock() + select { + case q.notify <- struct{}{}: + default: + } +} + +// dequeue returns the next event in FIFO order, blocking until one is available. +// It reports ok=false only once the queue is closed AND fully drained, so a +// close never discards already-enqueued events. +func (q *agentEventQueue) dequeue() (AgentEvent, bool) { + for { + q.mu.Lock() + if len(q.items) > 0 { + event := q.items[0] + // Zero the vacated slot before advancing so the evicted event's + // maps/slices/pointer fields (tool-call payloads can be large) + // become collectable now, not only when the whole backing array is + // freed at stream shutdown. + q.items[0] = AgentEvent{} + q.items = q.items[1:] + q.mu.Unlock() + return event, true + } + if q.closed { + q.mu.Unlock() + return AgentEvent{}, false + } + q.mu.Unlock() + <-q.notify + } +} + // dispatchMCPRequest handles an incoming MCP request message, dispatches it to the handler, // and writes the response back over the stream. func (c *Client) dispatchMCPRequest(ctx context.Context, msg ws.Message, mcpHandler MCPHandler, writeMessage func([]byte) error) { diff --git a/apps/backend/internal/agent/runtime/agentctl/agent_test.go b/apps/backend/internal/agent/runtime/agentctl/agent_test.go index 02297d4515..4e89792ef2 100644 --- a/apps/backend/internal/agent/runtime/agentctl/agent_test.go +++ b/apps/backend/internal/agent/runtime/agentctl/agent_test.go @@ -797,6 +797,192 @@ func TestStreamUpdates_DisconnectCleansPending(t *testing.T) { } } +// TestReadUpdatesStream_BlockedHandlerDoesNotStarveResponse is the regression +// test for the production stream-reader deadlock (task 544afdae). An agent +// event handler (handleAgentReady) blocked on the per-session cancelInFlight +// guard, and that guard's holder — Service.CancelAgent — was itself blocked in +// sendStreamRequest waiting for the agent.cancel response frame. Because the +// read loop used to run handlers inline, it never looped back to deliver that +// response, so both sides waited forever. +// +// This pins the fix: the read loop offloads event handling to an ordered worker +// goroutine, so even while a handler is blocked mid-event, a concurrent +// sendStreamRequest still receives its response frame. +func TestReadUpdatesStream_BlockedHandlerDoesNotStarveResponse(t *testing.T) { + upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }} + + // The server emits an agent event first, then answers the request that the + // test fires while that event's handler is still blocked. + emitEvent := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + + // Push an agent event whose handler will block on the client side. + <-emitEvent + eventData, _ := json.Marshal(AgentEvent{Type: "complete"}) + _ = conn.WriteMessage(websocket.TextMessage, eventData) + + for { + _, message, err := conn.ReadMessage() + if err != nil { + return + } + var msg ws.Message + if err := json.Unmarshal(message, &msg); err != nil { + continue + } + if msg.Type == ws.MessageTypeRequest { + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"success": true}) + data, _ := json.Marshal(resp) + _ = conn.WriteMessage(websocket.TextMessage, data) + } + } + })) + defer server.Close() + + handlerEntered := make(chan struct{}) + releaseHandler := make(chan struct{}) + log := newTestLogger() + c := &Client{ + baseURL: server.URL, + httpClient: &http.Client{Timeout: 5 * time.Second}, + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ctx := context.Background() + var once sync.Once + if err := c.StreamUpdates(ctx, func(_ AgentEvent) { + // Stand in for handleAgentReady blocking on the cancelInFlight guard. + once.Do(func() { close(handlerEntered) }) + <-releaseHandler + }, nil, nil); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer func() { + close(releaseHandler) + c.Close() + }() + + // Trigger the event and wait until its handler is blocked in-flight. + close(emitEvent) + select { + case <-handlerEntered: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the blocking event handler to be entered") + } + + // With the handler still blocked, a request must still get its response — + // the read loop is not wedged behind the handler. + reqCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + resp, err := c.sendStreamRequest(reqCtx, "agent.cancel", nil) + if err != nil { + t.Fatalf("sendStreamRequest was starved by the blocked event handler: %v", err) + } + if resp.Type != ws.MessageTypeResponse { + t.Fatalf("expected response type, got %q", resp.Type) + } +} + +// TestReadUpdatesStream_EventBurstDoesNotStarveResponse pins the fix for the +// bounded-queue regression (cubic review on task 544afdae): even if the first +// event's handler blocks on the cancelInFlight guard and a large burst of +// further events arrives behind it, the read loop must stay free to deliver the +// agent.cancel response frame. A fixed buffered channel would fill and +// backpressure the read loop on send, re-wedging the cancel; the unbounded +// reader-side queue must not. +func TestReadUpdatesStream_EventBurstDoesNotStarveResponse(t *testing.T) { + const burst = 1000 // comfortably beyond the old 256 buffer + + upgrader := websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return true }} + emitEvents := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + + <-emitEvents + // Blast a burst of events; the first handler will block on the client + // side, so all of these queue up behind it. + for i := 0; i < burst; i++ { + eventData, _ := json.Marshal(AgentEvent{Type: "message_chunk"}) + if err := conn.WriteMessage(websocket.TextMessage, eventData); err != nil { + return + } + } + + for { + _, message, err := conn.ReadMessage() + if err != nil { + return + } + var msg ws.Message + if err := json.Unmarshal(message, &msg); err != nil { + continue + } + if msg.Type == ws.MessageTypeRequest { + resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"success": true}) + data, _ := json.Marshal(resp) + _ = conn.WriteMessage(websocket.TextMessage, data) + } + } + })) + defer server.Close() + + handlerEntered := make(chan struct{}) + releaseHandler := make(chan struct{}) + log := newTestLogger() + c := &Client{ + baseURL: server.URL, + httpClient: &http.Client{Timeout: 5 * time.Second}, + logger: log, + pendingRequests: make(map[string]chan *ws.Message), + } + + ctx := context.Background() + var once sync.Once + if err := c.StreamUpdates(ctx, func(_ AgentEvent) { + // Only the first handler blocks; the rest queue up behind it. + once.Do(func() { + close(handlerEntered) + <-releaseHandler + }) + }, nil, nil); err != nil { + t.Fatalf("failed to connect: %v", err) + } + defer func() { + close(releaseHandler) + c.Close() + }() + + close(emitEvents) + select { + case <-handlerEntered: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the blocking event handler to be entered") + } + + // With the first handler blocked and a full burst queued behind it, a + // request must still get its response — the read loop is not wedged behind + // a full event buffer. + reqCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + resp, err := c.sendStreamRequest(reqCtx, "agent.cancel", nil) + if err != nil { + t.Fatalf("sendStreamRequest was starved by the event burst: %v", err) + } + if resp.Type != ws.MessageTypeResponse { + t.Fatalf("expected response type, got %q", resp.Type) + } +} + // --- Concurrent request tests --- func TestConcurrentRequests(t *testing.T) { diff --git a/apps/backend/internal/orchestrator/event_handlers_clarification.go b/apps/backend/internal/orchestrator/event_handlers_clarification.go index 8911f23e52..cffd955ccd 100644 --- a/apps/backend/internal/orchestrator/event_handlers_clarification.go +++ b/apps/backend/internal/orchestrator/event_handlers_clarification.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" "github.com/kandev/kandev/internal/task/models" ) @@ -257,23 +258,25 @@ func (s *Service) retryClarificationAfterCancel(ctx context.Context, data clarif zap.String("task_id", data.TaskID), zap.String("session_id", data.SessionID)) - if err := s.cancelAndDispatchClarificationRetry(ctx, data, prompt); err != nil { - s.logger.Error("failed to resume agent after cancel in clarification recovery", - zap.String("task_id", data.TaskID), - zap.String("session_id", data.SessionID), - zap.Error(err)) - return false - } - - s.logger.Info("successfully recovered stuck agent with clarification answer", - zap.String("task_id", data.TaskID), - zap.String("session_id", data.SessionID)) - return true -} - -// cancelAndDispatchClarificationRetry owns the guard through cancellation and -// retry acceptance, then releases it while the recovered agent turn runs. -func (s *Service) cancelAndDispatchClarificationRetry(ctx context.Context, data clarificationAnsweredData, prompt string) error { + // Claim the shared per-session guard across the cancel-then-hand-off + // sequence — see the Service.cancelInFlight field doc comment. Releasing + // between the cancel and the hand-off would let a concurrent + // QueueAndInterruptForPeerMessage (or another drain) take-and-dispatch a + // queued entry in that gap, only for this retry to then prompt over top of + // it. + // + // The retry prompt itself is NOT sent under the guard: dispatching it + // through the queue's take-and-dispatch path hands the (potentially + // long-blocking) executor.Prompt call to a background goroutine + // (executeQueuedMessage). Sending it inline here instead — the previous + // behavior — held the guard across executor.Prompt, which blocks for as + // long as a jammed agent takes to accept the prompt (observed: minutes, + // stuck inside an MCP call). While it blocked, the user's Cancel button — + // which TryLocks this same guard — was permanently starved, leaving the + // session unstoppable. markQueuedDispatchInFlight (inside + // dispatchTakenQueuedMessage) makes the session "busy" under the guard, so + // a concurrent interrupt/drain still backs off exactly as an inline retry + // would have. lock, release := s.acquireCancelInFlightGuard(data.SessionID) defer release() lock.Lock() @@ -285,16 +288,78 @@ func (s *Service) cancelAndDispatchClarificationRetry(ctx context.Context, data zap.Error(err)) // Force-revert session state so the retry prompt can proceed if revertErr := s.repo.UpdateTaskSessionState(ctx, data.SessionID, models.TaskSessionStateWaitingForInput, ""); revertErr != nil { - s.logger.Warn("failed to force-revert session state for clarification recovery", + s.logger.Error("failed to force-revert session state for clarification recovery", zap.String("session_id", data.SessionID), zap.Error(revertErr)) - return revertErr + return false } s.completeTurnForSession(ctx, data.SessionID) } - if _, err := s.PromptTask(ctx, data.TaskID, data.SessionID, prompt, "", false, nil, true); err != nil { - return err + if err := s.dispatchClarificationResumeLocked(ctx, data, prompt); err != nil { + if errors.Is(err, errClarificationResumeQueuedForDrain) { + // Not a failure: the answer is safely queued and a future drain + // (the next agent.ready) will dispatch it. Logging this at error + // level produced misleading "failed to resume agent" noise even + // though recovery still succeeds. + s.logger.Info("clarification answer queued; awaiting drain to dispatch", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID)) + return true + } + s.logger.Error("failed to resume agent after cancel in clarification recovery", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID), + zap.Error(err)) + return false + } + + s.logger.Info("recovered stuck agent; dispatching clarification answer", + zap.String("task_id", data.TaskID), + zap.String("session_id", data.SessionID)) + return true +} + +// errClarificationResumeQueuedForDrain signals that the clarification resume +// prompt was safely queued but not immediately dispatched (a concurrent +// dispatch was already settling for this session, so take-and-dispatch backed +// off). The entry stays in the queue and the next drain will pick it up, so +// this is a success case for recovery — not an error — and must not be logged +// as a failure. +var errClarificationResumeQueuedForDrain = errors.New("clarification resume queued for future drain") + +// dispatchClarificationResumeLocked enqueues the clarification resume prompt and +// hands it to the async take-and-dispatch path so the blocking executor.Prompt +// call runs off the cancelInFlight guard (see retryClarificationAfterCancel). +// The entry is tagged user_message_recorded so executeQueuedMessage does not +// insert a spurious user chat message for the system-built resume prompt. The +// caller MUST already hold sessionID's cancelInFlight lock. +// +// Returns nil when the entry was dispatched immediately, +// errClarificationResumeQueuedForDrain when it was safely queued for a future +// drain (a non-failure the caller must not treat as an error), and any other +// error when the resume genuinely could not be handed off. +func (s *Service) dispatchClarificationResumeLocked(ctx context.Context, data clarificationAnsweredData, prompt string) error { + if s.messageQueue == nil { + // The queue is the only hand-off path now that the retry prompt no + // longer runs inline under the guard. A nil queue is a wiring bug, not + // a runtime condition, so surface it with context rather than failing + // silently the way a bare false return did. + return fmt.Errorf("cannot resume clarification: message queue is not configured") + } + queued, err := s.messageQueue.QueueMessageWithMetadata( + ctx, data.SessionID, data.TaskID, prompt, "", messagequeue.QueuedByAgent, false, nil, + map[string]interface{}{metaKeyUserMessageRecorded: true}, + ) + if err != nil { + return fmt.Errorf("queue clarification resume prompt: %w", err) + } + dispatched, err := s.takeAndDispatchEntryLocked(ctx, data.SessionID, queued.ID) + if err != nil { + return fmt.Errorf("dispatch clarification resume prompt: %w", err) + } + if !dispatched { + return errClarificationResumeQueuedForDrain } return nil } diff --git a/apps/backend/internal/orchestrator/event_handlers_clarification_test.go b/apps/backend/internal/orchestrator/event_handlers_clarification_test.go index 4e0aa41250..f494271929 100644 --- a/apps/backend/internal/orchestrator/event_handlers_clarification_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_clarification_test.go @@ -4,12 +4,15 @@ import ( "context" "database/sql" "errors" + "fmt" "strings" + "sync" "testing" "time" "github.com/kandev/kandev/internal/agent/runtime/lifecycle" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/orchestrator/executor" "github.com/kandev/kandev/internal/task/models" wfmodels "github.com/kandev/kandev/internal/workflow/models" v1 "github.com/kandev/kandev/pkg/api/v1" @@ -603,6 +606,131 @@ func TestClarificationWatchdog_ExpiresAndClearsEntry(t *testing.T) { } } +// TestRetryClarificationAfterCancel_DoesNotStarveUserCancel is the regression +// test for the production hang where a clarification-timeout recovery left a +// session permanently unstoppable. retryClarificationAfterCancel used to send +// its retry prompt inline while holding the per-session cancelInFlight guard. +// executor.Prompt blocks until a jammed agent accepts the prompt (observed: +// minutes, stuck in an MCP call), so the guard stayed held the whole time — +// and every user Cancel-button click TryLocks that same guard, so it was +// starved and silently no-op'd ("cancel already in flight; skipping +// duplicate"), leaving the session stuck RUNNING forever. +// +// This pins the fix: the retry prompt is dispatched on a background goroutine +// off the guard, so even while that prompt is blocked in-flight, a concurrent +// user CancelAgent still acquires the guard and reaches agentManager.CancelAgent. +func TestRetryClarificationAfterCancel_DoesNotStarveUserCancel(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + retryPromptBlock := make(chan struct{}) + retryPromptEntered := make(chan struct{}) + agentMgr := &mockAgentManager{ + isAgentRunning: true, + repoForExecutionLookup: repo, + } + taskRepo := newMockTaskRepo() + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + + // The retry prompt (the only prompt this test dispatches) blocks in-flight, + // standing in for a jammed agent that never accepts the resume prompt. + var enteredOnce sync.Once + agentMgr.promptAgentFunc = func(context.Context, string, string, []v1.MessageAttachment, bool) (*executor.PromptResult, error) { + enteredOnce.Do(func() { close(retryPromptEntered) }) + <-retryPromptBlock + return &executor.PromptResult{}, nil + } + t.Cleanup(func() { close(retryPromptBlock) }) + + seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateRunning) + seedExecutorRunning(t, repo, "session1", "task1", "exec-1") + + // Kick off the clarification-timeout recovery. Its silent cancel succeeds + // (mock CancelAgent returns nil), then it hands the retry prompt to the + // async path and returns — releasing the guard — even though that prompt is + // still blocked inside PromptAgent. + recoveryDone := make(chan struct{}) + go func() { + svc.retryClarificationAfterCancel(ctx, clarificationAnsweredData{ + TaskID: "task1", SessionID: "session1", + }, "the clarification answer", fmt.Errorf("wrap: %w", ErrAgentPromptInProgress)) + close(recoveryDone) + }() + + select { + case <-retryPromptEntered: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for the async retry prompt to reach the agent") + } + select { + case <-recoveryDone: + case <-time.After(2 * time.Second): + t.Fatal("retryClarificationAfterCancel blocked on the in-flight retry prompt instead of releasing the guard") + } + + // The recovery's silent cancel was the first agent-cancel call. + if got := agentMgr.cancelAgentCalls.Load(); got != 1 { + t.Fatalf("expected exactly 1 agent cancel from recovery's silent cancel, got %d", got) + } + + // The user clicks Cancel while the retry prompt is still blocked in-flight. + // Before the fix this returned immediately as a starved no-op; now it must + // acquire the guard and actually reach agentManager.CancelAgent. + if err := svc.CancelAgent(ctx, "session1"); err != nil { + t.Fatalf("user CancelAgent returned error: %v", err) + } + if got := agentMgr.cancelAgentCalls.Load(); got != 2 { + t.Fatalf("user cancel was starved by a leaked guard: expected 2 agent cancel calls, got %d", got) + } +} + +// TestDispatchClarificationResumeLocked_ReturnPaths pins the two outcomes that +// are deterministically reachable at this seam — a genuine error (nil queue) +// and an immediate dispatch (nil) — so the caller can tell a real failure from +// success and log accordingly (bot review on PR #1680: the old bool return +// logged an alarming "failed to resume agent" error even when the answer was +// safely queued). +// +// The third outcome, errClarificationResumeQueuedForDrain, is not unit-testable +// here: takeAndDispatchEntryLocked always finds and dispatches the entry this +// function just queued, so the sentinel only arises when a concurrent take +// removes that entry first (targeted take misses) AND a rival dispatch is +// already settling (drainQueuedMessageForPromptableSessionLocked backs off on +// isQueuedDispatchInFlight). That race is driven end-to-end by +// TestQueueAndInterruptForPeerMessage_RacesClarificationTimeoutRecovery, which +// asserts the answer stays queued for the recovered turn's own natural drain. +func TestDispatchClarificationResumeLocked_ReturnPaths(t *testing.T) { + ctx := context.Background() + data := clarificationAnsweredData{TaskID: "t1", SessionID: "s1"} + + t.Run("nil message queue is a genuine error", func(t *testing.T) { + repo := setupTestRepo(t) + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), &mockAgentManager{}) + svc.messageQueue = nil + + err := svc.dispatchClarificationResumeLocked(ctx, data, "answer") + if err == nil { + t.Fatal("expected an error when the message queue is not configured") + } + if errors.Is(err, errClarificationResumeQueuedForDrain) { + t.Fatalf("nil queue must not be reported as the benign queued-for-drain case: %v", err) + } + }) + + t.Run("immediate dispatch returns nil", func(t *testing.T) { + repo := setupTestRepo(t) + agentMgr := &mockAgentManager{isAgentRunning: true, repoForExecutionLookup: repo} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr) + svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{}) + seedTaskAndSession(t, repo, "t1", "s1", models.TaskSessionStateWaitingForInput) + seedExecutorRunning(t, repo, "s1", "t1", "exec-1") + + if err := svc.dispatchClarificationResumeLocked(ctx, data, "answer"); err != nil { + t.Fatalf("expected nil on immediate dispatch, got %v", err) + } + }) +} + func countClarificationWatchdogs(svc *Service) int { count := 0 svc.clarificationWatchdogs.Range(func(_, _ interface{}) bool { diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 13537d4209..17d9c388d3 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -1709,9 +1709,12 @@ func TestQueueAndInterruptForPeerMessage_RacesClarificationTimeoutRecovery(t *te case <-time.After(100 * time.Millisecond): } - // Release clarification recovery's cancel; it completes, then it - // prompts its own retry synchronously while still holding the guard — - // starting a fresh turn for this session. + // Release clarification recovery's cancel; it completes, marks the + // session busy under the guard, then hands its retry prompt off to the + // async take-and-dispatch path and RELEASES the guard before that + // (potentially long-blocking) prompt runs — so a jammed agent can no + // longer starve the user's Cancel button. retryClarificationAfterCancel + // therefore returns promptly rather than blocking on executor.Prompt. close(agentMgr.cancelAgentBlock) select { case <-recoveryDone: @@ -1735,6 +1738,15 @@ func TestQueueAndInterruptForPeerMessage_RacesClarificationTimeoutRecovery(t *te // turn. assert.Equal(t, int32(1), agentMgr.cancelAgentCalls.Load()) + // The retry prompt is dispatched on the async executeQueuedMessage + // goroutine (off the guard), so wait for it to land rather than reading + // synchronously. + require.Eventually(t, func() bool { + agentMgr.mu.Lock() + defer agentMgr.mu.Unlock() + return len(agentMgr.capturedPrompts) == 1 + }, 2*time.Second, 10*time.Millisecond, "expected exactly one prompt — clarification recovery's retry") + agentMgr.mu.Lock() prompts := append([]string(nil), agentMgr.capturedPrompts...) agentMgr.mu.Unlock() @@ -1825,7 +1837,13 @@ func TestClarificationRecovery_ReleasesGuardAfterRetryDispatch(t *testing.T) { firstPrompt := <-promptAccepted secondPrompt := <-promptAccepted require.Equal(t, "clarification answer", firstPrompt.Prompt) - require.True(t, firstPrompt.DispatchOnly) + // The clarification retry no longer relies on dispatchOnly to avoid + // starving the guard: it is handed to the async take-and-dispatch path + // (executeQueuedMessage) on a background goroutine, so + // retryClarificationAfterCancel returns before executor.Prompt runs even + // though the queued dispatch itself keeps the normal completion-wait + // (dispatchOnly=false) behavior. + require.False(t, firstPrompt.DispatchOnly, "queue-dispatched retry keeps the normal completion-wait behavior; the guard is released via the async hand-off, not dispatchOnly") require.Equal(t, "parent steer", secondPrompt.Prompt) require.False(t, secondPrompt.DispatchOnly, "normal queued dispatch keeps its existing completion-wait behavior")