fix: resolve two cancel/clarification deadlocks that wedge sessions in RUNNING (#1680)

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) {

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) {

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
 }

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 {

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")