feat: interrupt busy child task turn on parent message_task_kandev (#1653)

apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go

@@ -180,6 +180,18 @@ func (m *Manager) CancelAgent(ctx context.Context, executionID string) error {
 // exit and close promptFinished, and marks the execution ready so the workflow
 // leaves the running state. Returns ErrCancelEscalated so callers know the ACP
 // cancel did not get a clean acknowledgement.
+//
+// The ready-event publish below is dispatched asynchronously (asyncPublish=true
+// on markReadyEventWithContext), not inline: every caller that can reach this
+// escalation path (Service.CancelAgent, and cancelAgentSilent's callers —
+// QueueAndInterruptForPeerMessage, retryClarificationAfterCancel,
+// PauseForClarificationInput) does so while still holding sessionID's
+// per-session cancelInFlightGuard, and the in-memory event bus delivers to
+// the orchestrator's handleAgentReady (a queue subscriber for this event
+// type) *synchronously*, on this same goroutine. An inline publish would
+// have handleAgentReady try to re-acquire that same guard reentrantly and
+// deadlock forever on the non-reentrant sync.Mutex. See
+// markReadyEventWithContext's doc comment for the full explanation.
 func (m *Manager) escalateStuckCancel(ctx context.Context, execution *AgentExecution, ch <-chan struct{}) error {
 	m.logger.Warn("timed out waiting for in-flight prompt to finish after cancel; escalating",
 		zap.String("execution_id", execution.ID),
@@ -208,7 +220,7 @@ func (m *Manager) escalateStuckCancel(ctx context.Context, execution *AgentExecu
 		// breaks the next PromptAgent call.
 	}
 
-	if err := m.MarkReady(execution.ID); err != nil {
+	if err := m.markReadyEventWithContext(context.Background(), execution.ID, events.AgentReady, true); err != nil {
 		m.logger.Warn("failed to mark execution ready after cancel escalation",
 			zap.String("execution_id", execution.ID),
 			zap.Error(err))
@@ -1094,13 +1106,13 @@ func (m *Manager) MarkReady(executionID string) error {
 //
 // Publishes events.AgentBootReady. Returns error if execution not found.
 func (m *Manager) MarkBootReady(executionID string) error {
-	return m.markReadyEventWithContext(context.Background(), executionID, events.AgentBootReady)
+	return m.markReadyEventWithContext(context.Background(), executionID, events.AgentBootReady, false)
 }
 
 // markReadyEvent is the shared body of MarkReady / MarkBootReady — both flip
 // the execution to the Ready status and publish their respective event type.
 func (m *Manager) markReadyEvent(executionID, eventType string) error {
-	return m.markReadyEventWithContext(context.Background(), executionID, eventType)
+	return m.markReadyEventWithContext(context.Background(), executionID, eventType, false)
 }
 
 func (m *Manager) markBootReadyFromFailed(ctx context.Context, executionID string) error {
@@ -1111,10 +1123,30 @@ func (m *Manager) markBootReadyFromFailed(ctx context.Context, executionID strin
 	if execution.Status != v1.AgentStatusFailed {
 		return nil
 	}
-	return m.markReadyEventWithContext(ctx, executionID, events.AgentBootReady)
+	return m.markReadyEventWithContext(ctx, executionID, events.AgentBootReady, false)
 }
 
-func (m *Manager) markReadyEventWithContext(ctx context.Context, executionID, eventType string) error {
+// markReadyEventWithContext flips executionID to Ready and publishes
+// eventType. When asyncPublish is false (MarkReady, MarkBootReady, and
+// markBootReadyFromFailed all pass false — unchanged behavior), the publish
+// happens inline before this returns, matching the event bus's synchronous,
+// order-preserving delivery to registered subscribers.
+//
+// When asyncPublish is true (escalateStuckCancel only — see its own doc
+// comment), the publish is dispatched on its own goroutine instead. That
+// caller is reached from Service.CancelAgent (and cancelAgentSilent's other
+// guard-holding callers) while still holding sessionID's per-session
+// cancelInFlightGuard: the in-memory event bus delivers to a queue
+// subscription's handler *synchronously*, on the publishing goroutine, and
+// the orchestrator's handleAgentReady is registered on exactly such a queue
+// subscription for this event type. An inline publish here would have
+// handleAgentReady try to re-acquire that same guard from the very same
+// goroutine that is already holding it — sync.Mutex is not reentrant, so
+// that blocks forever. Deferring the publish lets the guard-holding caller
+// finish (and release the guard) first; handleAgentReady's own blocking
+// acquire then proceeds normally on its own goroutine, reloads the session,
+// finds it past this turn, and safely no-ops instead of deadlocking.
+func (m *Manager) markReadyEventWithContext(ctx context.Context, executionID, eventType string, asyncPublish bool) error {
 	execution, exists := m.executionStore.Get(executionID)
 	if !exists {
 		return fmt.Errorf("execution %q not found", executionID)
@@ -1135,7 +1167,11 @@ func (m *Manager) markReadyEventWithContext(ctx context.Context, executionID, ev
 		zap.String("execution_id", executionID),
 		zap.String("event_type", eventType))
 
-	m.eventPublisher.PublishAgentEvent(ctx, eventType, execution)
+	if asyncPublish {
+		go m.eventPublisher.PublishAgentEvent(context.Background(), eventType, execution)
+	} else {
+		m.eventPublisher.PublishAgentEvent(ctx, eventType, execution)
+	}
 	return nil
 }
 

apps/backend/internal/agent/runtime/lifecycle/manager_interaction_cancel_test.go

@@ -2,6 +2,7 @@ package lifecycle
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -9,6 +10,7 @@ import (
 
 	agentctlClient "github.com/kandev/kandev/internal/agent/runtime/agentctl"
 	"github.com/kandev/kandev/internal/events"
+	"github.com/kandev/kandev/internal/events/bus"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 	ws "github.com/kandev/kandev/pkg/websocket"
 )
@@ -55,6 +57,9 @@ func TestManager_CancelAgent_EscalatesWhenAgentHangs(t *testing.T) {
 	}
 
 	mgr := newTestManager(t)
+	mockBus, ok := mgr.eventBus.(*MockEventBus)
+	require.True(t, ok)
+	mockBus.Notify = make(chan struct{}, 4)
 
 	promptFinished := make(chan struct{})
 	exec := &AgentExecution{
@@ -103,8 +108,16 @@ func TestManager_CancelAgent_EscalatesWhenAgentHangs(t *testing.T) {
 	require.Equal(t, v1.AgentStatusReady, updated.Status,
 		"execution must be marked ready after cancel escalation so the workflow can proceed")
 
-	mockBus, ok := mgr.eventBus.(*MockEventBus)
-	require.True(t, ok)
+	// The AgentReady publish is dispatched asynchronously (escalateStuckCancel's
+	// asyncPublish=true — see markReadyEventWithContext's doc comment for why).
+	// Wait on Notify rather than reading PublishedEvents immediately: the
+	// channel receive establishes happens-before with the publishing
+	// goroutine's append, so the read below is race-free.
+	select {
+	case <-mockBus.Notify:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for the async AgentReady publish")
+	}
 	var sawReady bool
 	for _, ev := range mockBus.PublishedEvents {
 		if ev.Type == events.AgentReady {
@@ -193,3 +206,140 @@ func TestManager_CancelAgent_EscalationCleanupSurvivesCtxCancel(t *testing.T) {
 	default:
 	}
 }
+
+// TestManager_CancelAgent_EscalationDoesNotDeadlockOnReentrantReadySubscriber
+// pins the #1653 E2E CI regression: escalateStuckCancel calls MarkReady,
+// which publishes events.AgentReady. In production the orchestrator's
+// handleAgentReady is registered as a *queue* subscriber for this event, and
+// (per the centralized cancelInFlightGuard work on PR #1653) tries to
+// re-acquire the very same per-session guard Service.CancelAgent still holds
+// at this exact point in the call chain — on the *same goroutine*, since the
+// in-memory event bus delivers to queue subscriptions synchronously (see
+// bus.MemoryEventBus.publishToQueue's "deliver synchronously to preserve
+// ordering" comment). A synchronous inline publish here would have that
+// reentrant Lock() call block forever on the non-reentrant mutex, and
+// CancelAgent would never return — exactly what the E2E "a message queued
+// during a running turn is delivered when the turn is paused" test caught.
+//
+// This test cannot reproduce the real cross-package reentrancy (it would
+// need the real bus.MemoryEventBus plus a real orchestrator.Service), so it
+// simulates the hazard directly: OnPublish blocks on a mutex the test holds
+// for CancelAgent's entire duration, standing in for the guard a reentrant
+// handleAgentReady would try (and fail) to acquire. If escalateStuckCancel
+// published inline, CancelAgent itself would call OnPublish and deadlock on
+// its own goroutine. Dispatching asynchronously (asyncPublish=true) lets
+// CancelAgent return promptly regardless of how long the subscriber blocks.
+func TestManager_CancelAgent_EscalationDoesNotDeadlockOnReentrantReadySubscriber(t *testing.T) {
+	prevWait := cancelWaitTimeout
+	prevEsc := cancelEscalationTimeout
+	cancelWaitTimeout = 20 * time.Millisecond
+	cancelEscalationTimeout = 20 * time.Millisecond
+	t.Cleanup(func() {
+		cancelWaitTimeout = prevWait
+		cancelEscalationTimeout = prevEsc
+	})
+
+	mock := newMockAgentServer(t)
+	t.Cleanup(func() { mock.server.Close() })
+	mock.handler = func(msg ws.Message) *ws.Message {
+		if msg.Action == "agent.cancel" {
+			resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{
+				"success": true,
+			})
+			return resp
+		}
+		return mock.defaultHandler(msg)
+	}
+
+	client := createTestClient(t, mock.server.URL)
+	t.Cleanup(client.Close)
+
+	streamCtx, streamCancel := context.WithCancel(context.Background())
+	t.Cleanup(streamCancel)
+	require.NoError(t, client.StreamUpdates(streamCtx, func(_ agentctlClient.AgentEvent) {}, nil, nil))
+	select {
+	case <-mock.wsConnected:
+	case <-time.After(2 * time.Second):
+		t.Fatal("mock server did not see WS connection")
+	}
+
+	mgr := newTestManager(t)
+	mockBus, ok := mgr.eventBus.(*MockEventBus)
+	require.True(t, ok)
+
+	// reentrantGuard stands in for the per-session cancelInFlightGuard that
+	// the real orchestrator.Service.CancelAgent holds for its own entire
+	// call — held here for this whole test, so any synchronous, same-
+	// goroutine attempt to acquire it (an inline publish reaching a
+	// reentrant handleAgentReady) would block forever.
+	var reentrantGuard sync.Mutex
+	reentrantGuard.Lock()
+	subscriberAttempted := make(chan struct{}, 1)
+	subscriberFinished := make(chan struct{})
+	mockBus.OnPublish = func(subject string, _ *bus.Event) {
+		if subject != events.AgentReady {
+			return
+		}
+		select {
+		case subscriberAttempted <- struct{}{}:
+		default:
+		}
+		reentrantGuard.Lock()
+		reentrantGuard.Unlock() //nolint:staticcheck // deliberately mirrors a reentrant acquire-then-release
+		close(subscriberFinished)
+	}
+
+	promptFinished := make(chan struct{})
+	exec := &AgentExecution{
+		ID:             "exec-cancel-reentrant",
+		TaskID:         "task-1",
+		SessionID:      "session-1",
+		Status:         v1.AgentStatusRunning,
+		WorkspacePath:  "/workspace",
+		agentctl:       client,
+		promptDoneCh:   make(chan PromptCompletionSignal, 1),
+		promptFinished: promptFinished,
+	}
+	require.NoError(t, mgr.executionStore.Add(exec))
+
+	go func() {
+		<-exec.promptDoneCh
+		close(promptFinished)
+	}()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	cancelDone := make(chan error, 1)
+	go func() {
+		cancelDone <- mgr.CancelAgent(ctx, exec.ID)
+	}()
+
+	select {
+	case err := <-cancelDone:
+		require.ErrorIs(t, err, ErrCancelEscalated,
+			"CancelAgent must return promptly even though its own escalation-triggered AgentReady publish is still blocked on a reentrant acquire")
+	case <-time.After(1 * time.Second):
+		t.Fatal("CancelAgent deadlocked waiting for its own escalation-triggered ready event to be handled — asyncPublish regression")
+	}
+
+	// The simulated subscriber must have at least been reached (proving the
+	// event really was published), even though it's still blocked.
+	select {
+	case <-subscriberAttempted:
+	case <-time.After(1 * time.Second):
+		t.Fatal("expected the AgentReady publish to reach the simulated subscriber")
+	}
+
+	reentrantGuard.Unlock()
+
+	// Wait for the simulated subscriber goroutine to actually finish before
+	// returning — otherwise it's still running (blocked, then unblocking)
+	// when the test exits, making outcomes scheduler-dependent and risking
+	// a goleak false positive.
+	select {
+	case <-subscriberFinished:
+	case <-time.After(1 * time.Second):
+		t.Fatal("simulated AgentReady subscriber did not finish after the guard was released")
+	}
+}

apps/backend/internal/agent/runtime/lifecycle/manager_test.go

@@ -138,10 +138,30 @@ func (m *MockDockerClient) ListContainers(ctx context.Context, labels map[string
 // MockEventBus implements bus.EventBus for testing
 type MockEventBus struct {
 	PublishedEvents []*bus.Event
+	// Notify, when non-nil, receives a value after each Publish call appends
+	// its event — lets a test synchronize with an async publish (e.g.
+	// escalateStuckCancel's asyncPublish=true path spawns Publish on its own
+	// goroutine) via a channel receive instead of racing on unsynchronized
+	// access to PublishedEvents. The send blocks, so set it with enough
+	// buffer for however many events the test expects, or drain it promptly.
+	Notify chan struct{}
+	// OnPublish, when non-nil, runs synchronously inside Publish itself
+	// (before appending/notifying) — lets a test simulate a slow or
+	// (deliberately, for deadlock regression tests) blocking subscriber,
+	// standing in for a real queue subscription's synchronous handler
+	// (see bus.MemoryEventBus.Publish/publishToQueue's "deliver
+	// synchronously to preserve ordering" comment).
+	OnPublish func(subject string, event *bus.Event)
 }
 
 func (m *MockEventBus) Publish(ctx context.Context, subject string, event *bus.Event) error {
+	if m.OnPublish != nil {
+		m.OnPublish(subject, event)
+	}
 	m.PublishedEvents = append(m.PublishedEvents, event)
+	if m.Notify != nil {
+		m.Notify <- struct{}{}
+	}
 	return nil
 }
 

apps/backend/internal/mcp/handlers/handlers.go

@@ -89,6 +89,17 @@ type SessionLauncher interface {
 	ResumeTaskSession(ctx context.Context, taskID, sessionID string) (*executor.TaskExecution, error)
 	ProcessOnTurnStart(ctx context.Context, taskID, sessionID string) error
 	GetMessageQueue() *messagequeue.Service
+	// QueueAndInterruptForPeerMessage atomically queues prompt for sessionID
+	// then interrupts the session's in-flight turn to dispatch it right
+	// away, bypassing FIFO order. Used only by queueThenInterruptTaskMessage
+	// when the message_task_kandev sender is the target task's parent (see
+	// handleMessageTask) — see the orchestrator implementation's doc
+	// comment for why "queue" and "interrupt" must be one atomic call
+	// rather than two separate steps. The returned bool reports whether
+	// this call actually dispatched the message immediately; callers must
+	// not report "sent" when it's false — the message is still only
+	// queued, to be delivered later by whichever drain gets to it.
+	QueueAndInterruptForPeerMessage(ctx context.Context, taskID, sessionID, prompt string, metadata map[string]interface{}) (*messagequeue.QueuedMessage, bool, error)
 }
 
 // MessageQueuer queues a prompt message for delivery to a session on its next turn.
@@ -1544,6 +1555,7 @@ func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.
 		Prompt          string `json:"prompt"`
 		SenderTaskID    string `json:"sender_task_id"`
 		SenderSessionID string `json:"sender_session_id"`
+		DeliveryMode    string `json:"delivery_mode"`
 	}
 	if err := json.Unmarshal(msg.Payload, &req); err != nil {
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "Invalid payload: "+err.Error(), nil)
@@ -1560,6 +1572,10 @@ func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.
 	if req.SenderTaskID == req.TaskID {
 		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation, "task cannot send a message to itself", nil)
 	}
+	if req.DeliveryMode != "" && req.DeliveryMode != deliveryModeQueued && req.DeliveryMode != deliveryModeInterrupt {
+		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeValidation,
+			fmt.Sprintf("delivery_mode must be %q or %q", deliveryModeQueued, deliveryModeInterrupt), nil)
+	}
 
 	// Sender lookup is global, not workspace-scoped: cross-workspace agent
 	// messaging is intentionally allowed (badge URL handles cross-workspace
@@ -1575,8 +1591,11 @@ func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.
 	// task_id (e.g. a truncated UUID prefix) reports "task not found" instead
 	// of the misleading "no primary session" error from the session lookup.
 	// This is purely an existence check — GetTask returns a wrapped
-	// ErrTaskNotFound on no-rows, never (nil, nil).
-	if _, err := h.taskSvc.GetTask(ctx, req.TaskID); err != nil {
+	// ErrTaskNotFound on no-rows, never (nil, nil). The loaded task's
+	// ParentID also tells us whether the sender is the target's parent,
+	// which decides interrupt eligibility below.
+	targetTask, err := h.taskSvc.GetTask(ctx, req.TaskID)
+	if err != nil {
 		if errors.Is(err, taskrepo.ErrTaskNotFound) {
 			return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound,
 				"target task not found: "+req.TaskID+" (pass the full task UUID, not a truncated prefix)", nil)
@@ -1598,7 +1617,27 @@ func (h *Handlers) handleMessageTask(ctx context.Context, msg *ws.Message) (*ws.
 	prompt := h.appendPromptReferenceExpansionContext(ctx, req.Prompt)
 	wrappedPrompt, senderMeta := wrapAgentMessage(prompt, senderTask, req.SenderSessionID)
 
-	result, err := h.dispatchTaskMessage(ctx, req.TaskID, session, wrappedPrompt, senderMeta)
+	// Interrupt intent is explicit, never inferred: a parent/child
+	// relationship alone no longer drives interruption (see
+	// dispatchTaskMessage's interruptIfBusy parameter and
+	// queueThenInterruptTaskMessage). The server still authorizes
+	// delivery_mode="interrupt" only when the sender is the target's
+	// direct parent — the same relationship check as before, now gating
+	// an explicit request instead of driving an implicit one. A non-parent
+	// sender explicitly requesting "interrupt" is hard-rejected rather than
+	// silently downgraded to "queued": a silent downgrade would misreport
+	// what happened and hide caller misuse instead of telling the caller
+	// its request was rejected. Omitted or "queued" keeps the default
+	// queue-and-wait behavior documented on message_task_kandev, even for
+	// a parent sender.
+	isParentToChild := targetTask.ParentID != "" && targetTask.ParentID == senderTask.ID
+	wantsInterrupt := req.DeliveryMode == deliveryModeInterrupt
+	if wantsInterrupt && !isParentToChild {
+		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeForbidden,
+			`delivery_mode="interrupt" is only allowed when the sender is the target task's direct parent`, nil)
+	}
+
+	result, err := h.dispatchTaskMessage(ctx, req.TaskID, session, wrappedPrompt, senderMeta, wantsInterrupt)
 	if err != nil {
 		var qfErr *queueFullDispatchError
 		if errors.As(err, &qfErr) {
@@ -1844,9 +1883,44 @@ const (
 	keyPosition         = "position"
 )
 
+// taskMessageStatusSent is the taskMessageDispatchResult.status value used
+// when a message_task_kandev prompt was dispatched immediately rather than
+// left in the FIFO queue for later delivery — either because the target
+// session was idle (promptWithAutoResume) or because a parent's interrupt
+// just dispatched it (queueThenInterruptTaskMessage). Extracted to satisfy
+// goconst's repeated-string rule; see dispatchTaskMessage's doc comment for
+// the full status enum ("queued", "sent", "started").
+const taskMessageStatusSent = "sent"
+
+// taskMessageStatusQueued is the taskMessageDispatchResult.status value
+// used when a message_task_kandev prompt is left in the FIFO queue for
+// later delivery rather than dispatched immediately. Extracted alongside
+// taskMessageStatusSent to satisfy goconst's repeated-string rule.
+const taskMessageStatusQueued = "queued"
+
+// deliveryModeQueued and deliveryModeInterrupt are the two accepted values
+// for message_task_kandev's delivery_mode request parameter — distinct
+// from taskMessageStatusQueued/taskMessageStatusSent above, which describe
+// the *response* status enum instead. "queued" happens to be a shared
+// literal between the two concepts (a delivery_mode request value and a
+// dispatch-result status value), but they are different fields on
+// different structs (request vs. response) and are kept as separate
+// constants so a change to one enum's spelling can't silently drift the
+// other.
+const (
+	deliveryModeQueued    = "queued"
+	deliveryModeInterrupt = "interrupt"
+)
+
+// queuedEntryID is set only by queueTaskMessage (the message queue's
+// returned entry id) so queueThenInterruptTaskMessage can target that exact
+// entry when interrupting instead of the FIFO head — see
+// InterruptForPeerMessage's doc comment. Never serialized to the wire; the
+// MCP response only reads status/sessionID (see handleMessageTask).
 type taskMessageDispatchResult struct {
-	status    string
-	sessionID string
+	status        string
+	sessionID     string
+	queuedEntryID string
 }
 
 type taskMessageReviewRollback struct {
@@ -2016,7 +2090,7 @@ func cloneTaskMessagePendingMove(move *messagequeue.PendingMove) *messagequeue.P
 // row (sender_task_id, sender_task_title, sender_session_id when called from
 // handleMessageTask). It is propagated to all three delivery paths so the
 // receiving task's chat displays the sender badge consistently.
-func (h *Handlers) dispatchTaskMessage(ctx context.Context, taskID string, session *models.TaskSession, prompt string, metadata map[string]interface{}) (taskMessageDispatchResult, error) {
+func (h *Handlers) dispatchTaskMessage(ctx context.Context, taskID string, session *models.TaskSession, prompt string, metadata map[string]interface{}, interruptIfBusy bool) (taskMessageDispatchResult, error) {
 	if h.sessionLauncher == nil {
 		return taskMessageDispatchResult{}, errors.New("orchestrator not available")
 	}
@@ -2026,6 +2100,9 @@ func (h *Handlers) dispatchTaskMessage(ctx context.Context, taskID string, sessi
 		return taskMessageDispatchResult{}, fmt.Errorf("session is %s — cannot send message", session.State)
 
 	case models.TaskSessionStateRunning, models.TaskSessionStateStarting:
+		if interruptIfBusy {
+			return h.queueThenInterruptTaskMessage(ctx, taskID, session, prompt, metadata)
+		}
 		return h.queueTaskMessage(ctx, taskID, session, prompt, metadata)
 
 	default:
@@ -2104,12 +2181,19 @@ func (h *Handlers) shouldStartTaskMessageSession(ctx context.Context, session *m
 	return err == nil && running != nil && running.Status == models.ExecutorRunningStatusPrepared
 }
 
+// queueTaskMessage appends prompt to the target session's FIFO message
+// queue for delivery on its next turn boundary (normal turn completion, or
+// an explicit interrupt — see queueThenInterruptTaskMessage). Returns the
+// queued entry's id via taskMessageDispatchResult.queuedEntryID so callers
+// that need to target this exact entry (rather than the FIFO head) can do
+// so later.
 func (h *Handlers) queueTaskMessage(ctx context.Context, taskID string, session *models.TaskSession, prompt string, metadata map[string]interface{}) (taskMessageDispatchResult, error) {
 	queue := h.sessionLauncher.GetMessageQueue()
 	if queue == nil {
 		return taskMessageDispatchResult{}, errors.New("message queue not available")
 	}
-	if _, err := queue.QueueMessageWithMetadata(ctx, session.ID, taskID, prompt, "", messagequeue.QueuedByAgent, false, nil, metadata); err != nil {
+	queued, err := queue.QueueMessageWithMetadata(ctx, session.ID, taskID, prompt, "", messagequeue.QueuedByAgent, false, nil, metadata)
+	if err != nil {
 		if errors.Is(err, messagequeue.ErrQueueFull) {
 			status := queue.GetStatus(ctx, session.ID)
 			return taskMessageDispatchResult{}, &queueFullDispatchError{
@@ -2122,7 +2206,73 @@ func (h *Handlers) queueTaskMessage(ctx context.Context, taskID string, session
 		return taskMessageDispatchResult{}, fmt.Errorf("failed to queue message: %w", err)
 	}
 	h.publishQueueStatusEvent(ctx, session.ID, queue)
-	return taskMessageDispatchResult{status: "queued", sessionID: session.ID}, nil
+	return taskMessageDispatchResult{status: taskMessageStatusQueued, sessionID: session.ID, queuedEntryID: queued.ID}, nil
+}
+
+// queueThenInterruptTaskMessage atomically queues prompt for a running/
+// starting session and interrupts its current turn so the message is
+// delivered right away instead of waiting for the turn to end naturally.
+// Used only when the sender is the target's parent task (see
+// handleMessageTask); regular peer messages keep the default queue-and-wait
+// behavior documented on message_task_kandev via queueTaskMessage.
+//
+// "Queue" and "interrupt" must be one atomic orchestrator call
+// (QueueAndInterruptForPeerMessage), not queueTaskMessage followed by a
+// separate interrupt call: between an insert becoming visible and a later,
+// separate interrupt call claiming the session's cancel lock, the child's
+// turn could complete naturally and the orchestrator's normal FIFO drain
+// could grab the just-queued entry and start dispatching it as an ordinary
+// turn — only for the later interrupt's cancel to land on and kill that
+// very turn, orphaning the parent's message mid-delivery. See
+// QueueAndInterruptForPeerMessage's doc comment for the full race this
+// closes.
+//
+// The returned status reflects what actually happened: "sent" only when
+// the interrupt actually dispatched the message immediately (the returned
+// bool), or "queued" when the cancel-and-take step ran but genuinely
+// failed to dispatch anything (see cancelAndTakeForPeerMessage's doc
+// comment for that case — it does not include lock contention, since
+// QueueAndInterruptForPeerMessage always waits for the lock rather than
+// skipping a busy one).
+// A failure past the queue insert is deliberately NOT surfaced as an error to the caller —
+// the message is already safely persisted and will still be delivered by
+// the normal turn-completion drain, so the interrupt is purely a latency
+// optimization on top of that always-safe default. The calling agent has
+// no useful recovery action on a hard error here besides retrying
+// message_task_kandev, and retrying would enqueue a second copy of the
+// same message since queuing is not idempotent — reporting the accurate
+// "queued" status avoids inviting that duplicate. The failure is still
+// logged server-side for operators.
+func (h *Handlers) queueThenInterruptTaskMessage(ctx context.Context, taskID string, session *models.TaskSession, prompt string, metadata map[string]interface{}) (taskMessageDispatchResult, error) {
+	queued, dispatched, err := h.sessionLauncher.QueueAndInterruptForPeerMessage(ctx, taskID, session.ID, prompt, metadata)
+	if err != nil {
+		if errors.Is(err, messagequeue.ErrQueueFull) {
+			queue := h.sessionLauncher.GetMessageQueue()
+			status := queue.GetStatus(ctx, session.ID)
+			return taskMessageDispatchResult{}, &queueFullDispatchError{
+				sessionID: session.ID,
+				queueSize: status.Count,
+				max:       status.Max,
+				entries:   status.Entries,
+			}
+		}
+		if queued == nil {
+			return taskMessageDispatchResult{}, fmt.Errorf("failed to queue and interrupt message: %w", err)
+		}
+		// The message is already safely queued (queued != nil); only the
+		// interrupt step failed — see the doc comment above for why that
+		// is not surfaced as an error to the caller.
+		h.logger.Warn("failed to interrupt child session's turn; message stays queued for normal drain",
+			zap.String("task_id", taskID),
+			zap.String("session_id", session.ID),
+			zap.Error(err))
+		return taskMessageDispatchResult{status: taskMessageStatusQueued, sessionID: session.ID, queuedEntryID: queued.ID}, nil
+	}
+	result := taskMessageDispatchResult{status: taskMessageStatusQueued, sessionID: session.ID, queuedEntryID: queued.ID}
+	if dispatched {
+		result.status = taskMessageStatusSent
+	}
+	return result, nil
 }
 
 func (h *Handlers) prepareSessionForTaskMessage(ctx context.Context, taskID string, session *models.TaskSession) (*models.TaskSession, error) {
@@ -2377,7 +2527,7 @@ func (h *Handlers) deleteRecordedUserMessage(ctx context.Context, message *model
 func (h *Handlers) promptWithAutoResume(ctx context.Context, taskID, sessionID, prompt string) (string, error) {
 	_, err := h.sessionLauncher.PromptTask(ctx, taskID, sessionID, prompt, "", false, nil, true)
 	if err == nil {
-		return "sent", nil
+		return taskMessageStatusSent, nil
 	}
 	if !errors.Is(err, executor.ErrExecutionNotFound) {
 		return "", fmt.Errorf("failed to send prompt: %w", err)
@@ -2393,7 +2543,7 @@ func (h *Handlers) promptWithAutoResume(ctx context.Context, taskID, sessionID,
 	if _, retryErr := h.sessionLauncher.PromptTask(ctx, taskID, sessionID, prompt, "", false, nil, true); retryErr != nil {
 		return "", fmt.Errorf("failed to send prompt after resume: %w", retryErr)
 	}
-	return "sent", nil
+	return taskMessageStatusSent, nil
 }
 
 // publishQueueStatusEvent fires a queue.status_changed event so the frontend

apps/backend/internal/mcp/handlers/handlers_test.go

@@ -1193,6 +1193,14 @@ func (m *mockSessionLauncher) ProcessOnTurnStart(context.Context, string, string
 }
 func (m *mockSessionLauncher) GetMessageQueue() *messagequeue.Service { return nil }
 
+// QueueAndInterruptForPeerMessage always reports a successful immediate
+// dispatch with a fake queued entry; tests exercising other outcomes
+// (failure, not-dispatched) use fakeOrchestrator in message_task_test.go
+// instead of this generic stub.
+func (m *mockSessionLauncher) QueueAndInterruptForPeerMessage(context.Context, string, string, string, map[string]interface{}) (*messagequeue.QueuedMessage, bool, error) {
+	return &messagequeue.QueuedMessage{ID: "mock-entry"}, true, nil
+}
+
 func TestAutoStartTask_DefaultsToWorktreeExecutor(t *testing.T) {
 	launcher := newMockSessionLauncher()
 	log := testLogger(t)

apps/backend/internal/mcp/handlers/message_task_test.go

@@ -36,11 +36,25 @@ type fakeOrchestrator struct {
 	resumeCalls       int
 	turnStartCalls    []turnStartCall
 	onTurnStart       func(context.Context, string, string) error
+	interruptCalls    []interruptCall
 
 	// Configurable: error returned by PromptTask. Cleared after first call so
 	// the retry-after-resume path can succeed on the second call.
 	promptErrFirst  error
 	startCreatedErr error
+	// interruptErr is returned by every InterruptForPeerMessage call — lets
+	// tests exercise the "interrupt failed, message must stay queued" path.
+	interruptErr error
+	// interruptSkippedNoError simulates InterruptForPeerMessage's busy-skip
+	// branch: returns (false, nil) — no error, but nothing was actually
+	// dispatched by this call — so tests can exercise the "status stays
+	// queued even though InterruptForPeerMessage succeeded" contract.
+	interruptSkippedNoError bool
+}
+
+// interruptCall records one InterruptForPeerMessage invocation.
+type interruptCall struct {
+	taskID, sessionID, entryID string
 }
 
 type promptCall struct {
@@ -107,6 +121,28 @@ func (f *fakeOrchestrator) ProcessOnTurnStart(ctx context.Context, taskID, sessi
 
 func (f *fakeOrchestrator) GetMessageQueue() *messagequeue.Service { return f.queue }
 
+// QueueAndInterruptForPeerMessage inserts prompt into the fake's real
+// message queue (so tests can assert on queue state via f.queue), then
+// reports the configured interruptErr, or (false, nil) when
+// interruptSkippedNoError simulates a busy/failed-take outcome — real
+// interrupt/drain behavior is exercised by the orchestrator-level
+// QueueAndInterruptForPeerMessage tests, not this fake.
+func (f *fakeOrchestrator) QueueAndInterruptForPeerMessage(ctx context.Context, taskID, sessionID, prompt string, metadata map[string]interface{}) (*messagequeue.QueuedMessage, bool, error) {
+	queued, err := f.queue.QueueMessageWithMetadata(ctx, sessionID, taskID, prompt, "", messagequeue.QueuedByAgent, false, nil, metadata)
+	if err != nil {
+		return nil, false, err
+	}
+
+	f.mu.Lock()
+	f.interruptCalls = append(f.interruptCalls, interruptCall{taskID: taskID, sessionID: sessionID, entryID: queued.ID})
+	f.mu.Unlock()
+
+	if f.interruptErr != nil {
+		return queued, false, f.interruptErr
+	}
+	return queued, !f.interruptSkippedNoError, nil
+}
+
 type fakePromptReferenceResolver struct {
 	expansions []promptservice.PromptReferenceExpansion
 	err        error
@@ -224,6 +260,45 @@ func seedTaskWithSession(t *testing.T, svc *service.Service, repo seedRepo, stat
 	return sender, target, loaded
 }
 
+// seedChildTaskWithSession is like seedTaskWithSession, but the target task
+// is created as a child of the sender task (ParentID = sender.ID) so callers
+// can exercise the parent -> child interrupt-on-message path. Returns
+// (parent/sender, child/target, target session).
+func seedChildTaskWithSession(t *testing.T, svc *service.Service, repo seedRepo, state models.TaskSessionState) (*models.Task, *models.Task, *models.TaskSession) {
+	t.Helper()
+	ctx := context.Background()
+	require.NoError(t, repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws-1", Name: "Test"}))
+	require.NoError(t, repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf-1", WorkspaceID: "ws-1", Name: "Board"}))
+	parent, err := svc.CreateTask(ctx, &service.CreateTaskRequest{
+		WorkspaceID: "ws-1",
+		WorkflowID:  "wf-1",
+		Title:       "Parent task",
+	})
+	require.NoError(t, err)
+	child, err := svc.CreateTask(ctx, &service.CreateTaskRequest{
+		WorkspaceID: "ws-1",
+		WorkflowID:  "wf-1",
+		Title:       "Child task",
+		ParentID:    parent.ID,
+	})
+	require.NoError(t, err)
+
+	sess := &models.TaskSession{
+		ID:             "sess-1",
+		TaskID:         child.ID,
+		AgentProfileID: "agent-profile-1",
+		IsPrimary:      true,
+		State:          models.TaskSessionStateCreated,
+	}
+	require.NoError(t, repo.CreateTaskSession(ctx, sess))
+	if state != models.TaskSessionStateCreated {
+		require.NoError(t, repo.UpdateTaskSessionState(ctx, sess.ID, state, ""))
+	}
+	loaded, err := svc.GetTaskSession(ctx, sess.ID)
+	require.NoError(t, err)
+	return parent, child, loaded
+}
+
 // senderPayload returns the standard payload shape sent by the MCP server
 // (agentctl injects sender_task_id and sender_session_id). Helper keeps test
 // bodies focused on the behaviour under test.
@@ -236,6 +311,15 @@ func senderPayload(targetTaskID, prompt, senderTaskID string) map[string]interfa
 	}
 }
 
+// senderPayloadWithMode is senderPayload plus an explicit delivery_mode —
+// used by tests that must opt into (or explicitly stay on) a specific
+// message_task_kandev delivery_mode rather than relying on the default.
+func senderPayloadWithMode(targetTaskID, prompt, senderTaskID, deliveryMode string) map[string]interface{} {
+	payload := senderPayload(targetTaskID, prompt, senderTaskID)
+	payload["delivery_mode"] = deliveryMode
+	return payload
+}
+
 func TestHandleMessageTask_MissingTaskID(t *testing.T) {
 	h := &Handlers{}
 	msg := makeWSMessage(t, ws.ActionMCPMessageTask, map[string]interface{}{
@@ -300,6 +384,222 @@ func TestHandleMessageTask_RunningSession_Queues(t *testing.T) {
 	assert.Equal(t, "sender-sess-1", entry.Metadata["sender_session_id"])
 	assert.Empty(t, orch.promptCalls)
 	assert.Empty(t, orch.startCreatedCalls)
+	// Unrelated senders (not the target's parent) must never trigger an
+	// interrupt — the default "queue and deliver at turn end" contract
+	// documented on message_task_kandev stays unchanged for them.
+	assert.Empty(t, orch.interruptCalls)
+}
+
+// TestHandleMessageTask_ParentToChildRunningSession_Interrupts pins the
+// steering contract: when the sender is the target's parent task AND
+// explicitly requests delivery_mode="interrupt", a running/starting target
+// is interrupted right after the message is queued, so the message is
+// delivered without waiting for the child's current turn to finish
+// naturally. Being the parent is necessary but not sufficient - the caller
+// must opt in; see TestHandleMessageTask_ParentToChildRunningSession_OmittedDeliveryMode_DoesNotInterrupt
+// for the (default) queued-and-wait behavior parents get otherwise. The
+// reported status is "sent" (not "queued") because the interrupt actually
+// dispatched it immediately - see queueThenInterruptTaskMessage's doc
+// comment. See InterruptForPeerMessage's doc comment for why this matters
+// for long-running children.
+func TestHandleMessageTask_ParentToChildRunningSession_Interrupts(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	parent, child, sess := seedChildTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, orch := newMessageTaskHandler(t, svc)
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayloadWithMode(child.ID, "stop and pivot to X", parent.ID, "interrupt"))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, "sent", payload["status"])
+
+	// The message was queued first exactly like any other message_task
+	// call...
+	status := orch.queue.GetStatus(context.Background(), sess.ID)
+	require.Equal(t, 1, status.Count)
+	assert.Contains(t, status.Entries[0].Content, "stop and pivot to X")
+
+	// ...but because the sender is the target's parent, the child's current
+	// turn is interrupted immediately instead of waiting for it to finish,
+	// targeting the exact entry id that was just queued (not just "whatever
+	// is at the FIFO head") — see InterruptForPeerMessage's doc comment.
+	require.Len(t, orch.interruptCalls, 1)
+	assert.Equal(t, child.ID, orch.interruptCalls[0].taskID)
+	assert.Equal(t, sess.ID, orch.interruptCalls[0].sessionID)
+	assert.Equal(t, status.Entries[0].ID, orch.interruptCalls[0].entryID)
+}
+
+// TestHandleMessageTask_ParentToChildRunningSession_InterruptFailure_KeepsMessageQueued
+// pins the failure contract: an interrupt failure must NOT surface as an MCP
+// error. The message was already safely persisted by queueTaskMessage and is
+// still delivered later by the normal turn-completion drain, so the
+// interrupt is only a latency optimization on top of that always-safe
+// default — surfacing a hard error here would just invite the calling agent
+// to retry message_task_kandev, which would enqueue a second copy of the
+// same message since queuing is not idempotent. The response instead reports
+// the accurate "queued" status (identical to the non-interrupt path), and
+// the queued message is left in place.
+func TestHandleMessageTask_ParentToChildRunningSession_InterruptFailure_KeepsMessageQueued(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	parent, child, sess := seedChildTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, orch := newMessageTaskHandler(t, svc)
+	orch.interruptErr = errors.New("cancel agent: agent manager unreachable")
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayloadWithMode(child.ID, "stop now", parent.ID, "interrupt"))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, "queued", payload["status"])
+
+	status := orch.queue.GetStatus(context.Background(), sess.ID)
+	require.Equal(t, 1, status.Count, "message must remain queued even though the interrupt failed")
+}
+
+// TestHandleMessageTask_ParentToChildRunningSession_InterruptSkipped_KeepsQueuedStatus
+// pins the busy-skip status contract: InterruptForPeerMessage returning
+// (false, nil) — no error, but nothing actually dispatched by this call,
+// e.g. because a concurrent cancel already owned the session — must still
+// report "queued" (not "sent"). Reporting "sent" here would tell the parent
+// its message was delivered immediately when it is still only sitting in
+// the queue for the in-flight cancel to deliver later.
+func TestHandleMessageTask_ParentToChildRunningSession_InterruptSkipped_KeepsQueuedStatus(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	parent, child, sess := seedChildTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, orch := newMessageTaskHandler(t, svc)
+	orch.interruptSkippedNoError = true
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayloadWithMode(child.ID, "stop now", parent.ID, "interrupt"))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, "queued", payload["status"])
+
+	status := orch.queue.GetStatus(context.Background(), sess.ID)
+	require.Equal(t, 1, status.Count, "message must remain queued when the interrupt was skipped")
+
+	// Confirm the interrupt path was actually entered, even though it
+	// reported the busy-skip outcome — otherwise this test would pass
+	// identically if the handler never called InterruptForPeerMessage at
+	// all, since the queued message and "queued" status look the same
+	// either way.
+	require.Len(t, orch.interruptCalls, 1)
+	assert.Equal(t, child.ID, orch.interruptCalls[0].taskID)
+	assert.Equal(t, sess.ID, orch.interruptCalls[0].sessionID)
+}
+
+// TestHandleMessageTask_ParentToChildRunningSession_OmittedDeliveryMode_DoesNotInterrupt
+// pins the new default: since delivery_mode now defaults to "queued", a
+// parent messaging a busy child *without* specifying delivery_mode no
+// longer interrupts — this is the behavior change from the previous round,
+// where any parent-to-child message always interrupted. The message is
+// still queued and delivered normally once the child's current turn ends.
+func TestHandleMessageTask_ParentToChildRunningSession_OmittedDeliveryMode_DoesNotInterrupt(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	parent, child, sess := seedChildTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, orch := newMessageTaskHandler(t, svc)
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayload(child.ID, "fyi, no rush", parent.ID))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, "queued", payload["status"])
+
+	status := orch.queue.GetStatus(context.Background(), sess.ID)
+	require.Equal(t, 1, status.Count)
+	assert.Contains(t, status.Entries[0].Content, "fyi, no rush")
+	assert.Empty(t, orch.interruptCalls, "omitted delivery_mode must default to queued, not interrupt, even from a parent sender")
+}
+
+// TestHandleMessageTask_ParentToChildRunningSession_ExplicitQueued_DoesNotInterrupt
+// is the explicit-value twin of the omitted-default case above: a parent
+// sender that explicitly passes delivery_mode="queued" gets exactly the
+// same queue-and-wait behavior as omitting it.
+func TestHandleMessageTask_ParentToChildRunningSession_ExplicitQueued_DoesNotInterrupt(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	parent, child, sess := seedChildTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, orch := newMessageTaskHandler(t, svc)
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayloadWithMode(child.ID, "fyi, no rush", parent.ID, "queued"))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+	assert.Equal(t, ws.MessageTypeResponse, resp.Type)
+
+	var payload map[string]interface{}
+	require.NoError(t, json.Unmarshal(resp.Payload, &payload))
+	assert.Equal(t, "queued", payload["status"])
+
+	status := orch.queue.GetStatus(context.Background(), sess.ID)
+	require.Equal(t, 1, status.Count)
+	assert.Empty(t, orch.interruptCalls, `explicit delivery_mode="queued" from a parent sender must not interrupt`)
+}
+
+// TestHandleMessageTask_NonParentSender_InterruptRequest_HardRejected pins
+// the authorization contract: delivery_mode="interrupt" is only ever
+// honored when the sender is the target's direct parent. A non-parent
+// (sibling/unrelated) sender explicitly requesting "interrupt" must get a
+// hard rejection — not a silent downgrade to "queued" — and the rejection
+// must have no side effect: nothing is queued, dispatched, or interrupted.
+// A silent downgrade would misreport what happened and hide caller misuse
+// instead of telling the caller its request was rejected.
+func TestHandleMessageTask_NonParentSender_InterruptRequest_HardRejected(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	sender, target, sess := seedTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, orch := newMessageTaskHandler(t, svc)
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayloadWithMode(target.ID, "stop now", sender.ID, "interrupt"))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	assertWSError(t, resp, ws.ErrorCodeForbidden)
+
+	var errPayload ws.ErrorPayload
+	require.NoError(t, json.Unmarshal(resp.Payload, &errPayload))
+	assert.Contains(t, errPayload.Message, "direct parent")
+
+	// No side effect from the rejected request: the target's queue stays
+	// empty and neither the interrupt nor any other dispatch path ran.
+	status := orch.queue.GetStatus(context.Background(), sess.ID)
+	assert.Equal(t, 0, status.Count, "a rejected interrupt request must not be queued as a side effect")
+	assert.Empty(t, orch.interruptCalls)
+	assert.Empty(t, orch.promptCalls)
+	assert.Empty(t, orch.startCreatedCalls)
+}
+
+// TestHandleMessageTask_InvalidDeliveryMode_Rejected pins plain input
+// validation: any delivery_mode value other than "queued"/"interrupt"
+// (or omitted) is rejected before any task/session lookup.
+func TestHandleMessageTask_InvalidDeliveryMode_Rejected(t *testing.T) {
+	svc, repo := newTestTaskService(t)
+	parent, child, _ := seedChildTaskWithSession(t, svc, repo, models.TaskSessionStateRunning)
+
+	h, _ := newMessageTaskHandler(t, svc)
+
+	msg := makeWSMessage(t, ws.ActionMCPMessageTask, senderPayloadWithMode(child.ID, "stop now", parent.ID, "immediately"))
+	resp, err := h.handleMessageTask(context.Background(), msg)
+	require.NoError(t, err)
+	assertWSError(t, resp, ws.ErrorCodeValidation)
 }
 
 func TestHandleMessageTask_AppendsPromptReferenceExpansions(t *testing.T) {

apps/backend/internal/mcp/server/handlers.go

@@ -243,6 +243,7 @@ func (s *Server) messageTaskHandler() server.ToolHandlerFunc {
 			"sender_task_id":    s.taskID,
 			"sender_session_id": s.sessionID,
 		}
+		copyOptionalStringArg(payload, req, "delivery_mode")
 		var result map[string]interface{}
 		if err := s.backend.RequestPayload(ctx, ws.ActionMCPMessageTask, payload, &result); err != nil {
 			return mcp.NewToolResultError(err.Error()), nil

apps/backend/internal/mcp/server/server.go

@@ -481,14 +481,19 @@ func (s *Server) registerKanbanTools() {
 Use this to communicate with a sibling task, a parent task, or any task you know the ID of — for example to ask a delegated subtask for clarification, hand it new context, or nudge a paused task forward.
 
 Behaviour by session state:
-- Running/starting: the message is queued and delivered when the current turn ends.
-- Idle (waiting for input or completed): the message is sent immediately as a new turn.
-- Created (not yet started): the agent is started with this message as its first prompt.
+- Running/starting: the message is queued and delivered when the current turn ends. Pass delivery_mode="interrupt" to instead interrupt the target's current turn immediately so the message is delivered without waiting — keeps a parent's steering/stop messages from piling up behind a long-running child's turn. Only honored when you are the target task's direct parent; a non-parent sender requesting "interrupt" gets a hard error instead of being silently downgraded to "queued".
+- Idle (waiting for input or completed): the message is sent immediately as a new turn (delivery_mode has no effect).
+- Created (not yet started): the agent is started with this message as its first prompt (delivery_mode has no effect).
 - Failed/cancelled: an error is returned (use create_task_kandev to start fresh).
 
 Returns the dispatch status: "queued", "sent", or "started".`),
 			mcp.WithString("task_id", mcp.Required(), mcp.Description("The target task's full UUID (not a truncated prefix)")),
 			mcp.WithString("prompt", mcp.Required(), mcp.Description("The message to deliver to the task's agent")),
+			mcp.WithString("delivery_mode",
+				mcp.Enum("queued", "interrupt"),
+				mcp.DefaultString("queued"),
+				mcp.Description(`How to deliver this message if the target is currently running/starting. "queued" (default): wait for the current turn to finish, like any other peer message. "interrupt": cancel the target's current turn now and deliver this message immediately instead — only allowed when you are the target task's direct parent; requesting "interrupt" as a non-parent is rejected with an error rather than silently queued.`),
+			),
 		),
 		s.wrapHandler("message_task_kandev", s.messageTaskHandler()),
 	)

apps/backend/internal/orchestrator/event_handlers_agent.go

@@ -235,6 +235,13 @@ func (s *Service) handleAgentBootReady(ctx context.Context, data watcher.AgentEv
 	// on the queue until the user manually sends another message. After the
 	// agent has booted and the session is back to WAITING_FOR_INPUT it's safe
 	// to dispatch any pending message.
+	// Claim the shared per-session lock first (see handleAgentReady's
+	// analogous claim for the full race this closes): a concurrent
+	// QueueAndInterruptForPeerMessage racing to deliver its own just-queued
+	// message must never have this drain steal it before the interrupt's
+	// own cancel+take runs. drainQueuedMessageForPromptableSession now owns
+	// that acquisition itself (blocking, plus its own promptability
+	// reload) — see its doc comment.
 	s.drainQueuedMessageForPromptableSession(ctx, data.SessionID)
 }
 
@@ -242,8 +249,39 @@ func (s *Service) handleAgentBootReady(ctx context.Context, data watcher.AgentEv
 // a prompt and is waiting for the next input. This is the *only* event that
 // should evaluate workflow on_turn_complete actions — boot signals route
 // through handleAgentBootReady instead.
+//
+// Acquires the per-session cancelInFlight guard *before* any
+// turn-completion/pending-move/on_turn_complete bookkeeping runs, not just
+// before the final queue-take decision (as it did before). Without this, a
+// ready event could pass the early checks, then a concurrent parent
+// interrupt (QueueAndInterruptForPeerMessage) could acquire the guard and
+// cancel-and-redispatch on this same session — starting a *new* turn — all
+// before this event reaches its own completeTurnForSession /
+// processOnTurnCompleteViaEngine, which would then wrongly complete and
+// evaluate on_turn_complete against that new turn instead of the one this
+// event actually reports the completion of, or apply a pending move while
+// the interrupt is still targeting the "old" session.
+//
+// The acquisition is a genuine blocking Lock (mirroring
+// QueueAndInterruptForPeerMessage's own precedent), not the previous
+// TryLock-and-skip: skipping here would leave this event's own
+// turn-completion/workflow bookkeeping undone forever — nothing else
+// performs it — and cancelAndTakeForPeerMessage's own "cancel failed but
+// already promptable" recovery explicitly relies on a *future* agent.ready
+// completing that bookkeeping once the guard frees up; this event may be
+// exactly that future ready event. Blocking here adds no new deadlock
+// risk: every existing guard holder already bounds its own hold time (see
+// QueueAndInterruptForPeerMessage's doc comment), and acquiring an
+// uncontended mutex via Lock costs the same as via TryLock, so the common
+// (non-racing) case is unaffected.
+//
+// Once the guard is held, the session and its active turn are re-validated
+// against the snapshot taken before waiting (session state, plus turn
+// identity via peekActiveTurnID when a TurnService is wired): if either
+// changed, a concurrent interrupt (or another turn entirely) has already
+// superseded this event, so it backs off without touching anything —
+// whatever superseded it owns that turn's own eventual completion.
 func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventData) {
-
 	if data.SessionID == "" {
 		s.logger.Warn("missing session_id for agent ready event",
 			zap.String("task_id", data.TaskID))
@@ -257,13 +295,6 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 			zap.String("agent_execution_id", data.AgentExecutionID))
 		return
 	}
-	if s.isCancelInFlight(data.SessionID) {
-		s.logger.Debug("ignoring agent.ready while cancel is in progress",
-			zap.String("task_id", data.TaskID),
-			zap.String("session_id", data.SessionID),
-			zap.String("agent_execution_id", data.AgentExecutionID))
-		return
-	}
 
 	session, err := s.repo.GetTaskSession(ctx, data.SessionID)
 	if err != nil {
@@ -286,6 +317,63 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 		return
 	}
 
+	// Snapshot which turn this event reports the completion of *before*
+	// contending for the guard — re-checked below once it's held. See the
+	// function doc comment for the race this closes.
+	turnAtEventFire, turnSnapshotErr := s.peekActiveTurnID(ctx, data.SessionID)
+
+	lock, release := s.acquireCancelInFlightGuard(data.SessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
+
+	// Re-validate now that the guard is held: a concurrent interrupt (or
+	// clarification recovery, or another drain) may have already resolved
+	// this exact session while this event waited. isSessionResetInProgress
+	// and the session state are re-checked in case either changed in that
+	// window; the turn-identity comparison additionally catches the
+	// specific case a plain state re-check can't — a *different* turn
+	// (e.g. one the interrupt cancelled-and-redispatched) that also
+	// happens to be RUNNING/STARTING.
+	if s.isSessionResetInProgress(data.SessionID) {
+		s.logger.Debug("stale agent.ready: session reset started while waiting for the guard",
+			zap.String("task_id", data.TaskID),
+			zap.String("session_id", data.SessionID))
+		return
+	}
+	session, err = s.repo.GetTaskSession(ctx, data.SessionID)
+	if err != nil {
+		s.logger.Warn("failed to reload session for agent.ready once the guard was held",
+			zap.String("task_id", data.TaskID),
+			zap.String("session_id", data.SessionID),
+			zap.Error(err))
+		return
+	}
+	if session.State != models.TaskSessionStateRunning && session.State != models.TaskSessionStateStarting {
+		s.logger.Debug("stale agent.ready: session no longer running/starting once the guard was held",
+			zap.String("task_id", data.TaskID),
+			zap.String("session_id", data.SessionID),
+			zap.String("session_state", string(session.State)))
+		return
+	}
+	if s.turnService != nil {
+		if turnSnapshotErr != nil {
+			s.logger.Warn("could not confirm this event's active turn before waiting for the guard; treating as stale rather than risk completing a possible successor turn",
+				zap.String("task_id", data.TaskID),
+				zap.String("session_id", data.SessionID),
+				zap.Error(turnSnapshotErr))
+			return
+		}
+		turnNow, turnNowErr := s.peekActiveTurnID(ctx, data.SessionID)
+		if turnNowErr != nil || turnNow != turnAtEventFire {
+			s.logger.Debug("stale agent.ready: active turn changed (or could not be reconfirmed) while waiting for the guard",
+				zap.String("task_id", data.TaskID),
+				zap.String("session_id", data.SessionID),
+				zap.Error(turnNowErr))
+			return
+		}
+	}
+
 	// A turn completed successfully — clear any transient retry budget so a
 	// later, unrelated provider overload starts its backoff fresh at attempt 1.
 	s.resetTransientRetry(data.SessionID)
@@ -327,12 +415,6 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 		return
 	}
 
-	// Check for queued messages when no workflow transition occurred.
-	queueStatus := s.messageQueue.GetStatus(ctx, data.SessionID)
-	s.logger.Info("checking for queued messages",
-		zap.String("session_id", data.SessionID),
-		zap.Int("count", queueStatus.Count))
-
 	// Passthrough sessions: deliver queued messages via PTY stdin instead of ACP.
 	if s.agentManager.IsPassthroughSession(ctx, data.SessionID) {
 		queuedMsg, exists := s.messageQueue.TakeQueued(ctx, data.SessionID)
@@ -349,43 +431,27 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 		return
 	}
 
-	queuedMsg, exists := s.messageQueue.TakeQueued(ctx, data.SessionID)
-	if !exists {
-		s.logger.Debug("no queued message to execute",
-			zap.String("session_id", data.SessionID))
-		return
-	}
-
-	// Skip if the queued message has empty content (might have been cleared accidentally)
-	if queuedMsg.Content == "" && len(queuedMsg.Attachments) == 0 {
-		s.logger.Warn("skipping empty queued message",
-			zap.String("session_id", data.SessionID),
-			zap.String("queue_id", queuedMsg.ID))
-
-		// Still publish status change to clear frontend state
-		s.publishQueueStatusEvent(ctx, data.SessionID)
-		return
-	}
-
-	s.logger.Info("auto-executing queued message",
-		zap.String("session_id", data.SessionID),
-		zap.String("task_id", queuedMsg.TaskID),
-		zap.String("queue_id", queuedMsg.ID))
-
-	// PromptTask rejects while session is RUNNING; queued follow-ups should be
-	// treated like fresh user input after turn completion.
-	s.updateTaskSessionState(ctx, data.TaskID, data.SessionID, models.TaskSessionStateWaitingForInput, "", false, session)
-
-	// Publish queue status changed event to notify frontend
-	s.publishQueueStatusEvent(ctx, data.SessionID)
-
-	// Execute the queued message asynchronously
-	go s.executeQueuedMessage(data.SessionID, queuedMsg)
+	// Check for queued messages when no workflow transition occurred. Uses
+	// the Locked variant directly: the guard above is held for this
+	// entire function now, not just this final step.
+	s.drainQueuedMessageForPromptableSessionLocked(ctx, data.SessionID)
 }
 
 func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messagequeue.QueuedMessage) {
 	promptCtx := context.Background() // Use a fresh context for async execution
 
+	// Safety net: guarantee the in-flight marker (set by
+	// dispatchTakenQueuedMessage before spawning this goroutine — see the
+	// Service.dispatchingQueued field doc comment) is cleared on every exit
+	// path, including the early reset-in-progress return below, PROVIDED
+	// this goroutine still owns it (compare-and-delete keyed on this
+	// entry's own ID) — the primary claim-and-clear happens deterministically
+	// inside promptTask's guarded claim step, right before setSessionRunning
+	// further down; this defer only catches paths that return before
+	// reaching it, or a losing claim that must not touch a newer dispatch's
+	// marker.
+	defer s.clearQueuedDispatchInFlightIfCurrent(queuedMsg.SessionID, queuedMsg.ID)
+
 	if s.isSessionResetInProgress(queuedMsg.SessionID) {
 		s.logger.Warn("queued message execution deferred due to context reset in progress",
 			zap.String("session_id", callerSessionID),
@@ -448,8 +514,28 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag
 		s.processOnTurnStartViaEngine(promptCtx, queuedMsg.TaskID, session)
 	}
 
-	_, err := s.PromptTask(promptCtx, queuedMsg.TaskID, queuedMsg.SessionID,
-		queuedMsg.Content, queuedMsg.Model, queuedMsg.PlanMode, attachments, false)
+	// Call the internal promptTask directly (not the public PromptTask
+	// wrapper), passing this entry's own ID as the claim token — see
+	// promptTask's doc comment and the Service.dispatchingQueued field doc
+	// comment for the guarded claim-then-mark-RUNNING step this enables,
+	// and for why a bare, unguarded check-then-clear would not be enough.
+	_, err := s.promptTask(promptCtx, queuedMsg.TaskID, queuedMsg.SessionID,
+		queuedMsg.Content, queuedMsg.Model, queuedMsg.PlanMode, attachments, false,
+		queuedMsg.ID)
+	if errors.Is(err, errQueuedDispatchSuperseded) {
+		// A newer dispatch for this same session (e.g. a second parent
+		// interrupt cancelling and re-taking while this one was still
+		// settling) won the claim first. This entry's content still
+		// matters — steering messages are not silently dropped — so
+		// requeue it instead of losing it; it will be delivered once the
+		// winning turn completes naturally.
+		s.logger.Info("queued message superseded by a newer dispatch for the same session before it could be prompted; requeueing",
+			zap.String("session_id", callerSessionID),
+			zap.String("task_id", queuedMsg.TaskID),
+			zap.String("queue_id", queuedMsg.ID))
+		s.requeueMessage(promptCtx, queuedMsg, "superseded-by-newer-dispatch")
+		return
+	}
 	if err != nil {
 		s.logger.Error("failed to execute queued message",
 			zap.String("session_id", callerSessionID),

apps/backend/internal/orchestrator/event_handlers_clarification.go

@@ -257,6 +257,17 @@ func (s *Service) retryClarificationAfterCancel(ctx context.Context, data clarif
 		zap.String("task_id", data.TaskID),
 		zap.String("session_id", data.SessionID))
 
+	// Claim the shared per-session guard across the whole cancel-then-retry
+	// sequence, not just the cancel call — see the Service.cancelInFlight
+	// field doc comment. Releasing between the cancel and the retry prompt
+	// would let a concurrent QueueAndInterruptForPeerMessage (or another
+	// drain) take-and-dispatch a queued entry in that gap, only for this
+	// retry's own PromptTask call to then prompt over top of it.
+	lock, release := s.acquireCancelInFlightGuard(data.SessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
+
 	if err := s.cancelAgentSilent(ctx, data.TaskID, data.SessionID); err != nil {
 		s.logger.Warn("cancel failed (agent likely dead), force-transitioning session state",
 			zap.String("session_id", data.SessionID),
@@ -323,6 +334,16 @@ func (s *Service) PauseForClarificationInput(ctx context.Context, sessionID stri
 	// same ask_user_question call. A duplicate cancel is safe: lifecycle returns
 	// ErrCancelEscalated/ErrNoExecutionForSession once the first pause wins, and
 	// completeTurnForSession is idempotent when there is no active turn left.
+	//
+	// Claim the shared per-session guard around the cancel itself — see the
+	// Service.cancelInFlight field doc comment: every cancel/take-and-dispatch
+	// decision for a session must serialize through this one guard, including
+	// this clarification-timeout cancel, or it can race a concurrent parent
+	// interrupt (or another drain) for the same session.
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
 	if err := s.cancelAgentSilent(writeCtx, session.TaskID, sessionID); err != nil {
 		return detached, err
 	}

apps/backend/internal/orchestrator/event_handlers_pending_move_test.go

@@ -854,8 +854,10 @@ func TestHandleAgentBootReady_DoesNotDrainWhileCancelInFlight(t *testing.T) {
 	); err != nil {
 		t.Fatalf("queue prompt: %v", err)
 	}
-	svc.cancelInFlight.Store("s1", struct{}{})
-	defer svc.cancelInFlight.Delete("s1")
+	lock, release := svc.acquireCancelInFlightGuard("s1")
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
 
 	svc.handleAgentBootReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"})
 

apps/backend/internal/orchestrator/event_handlers_queue_test.go

@@ -87,6 +87,7 @@ func TestExecuteQueuedMessage_RequeuesCancelReleaseFailure(t *testing.T) {
 		QueuedBy:  "test",
 	}
 
+	svc.markQueuedDispatchInFlight("s1", queuedMsg.ID)
 	svc.executeQueuedMessage("s1", queuedMsg)
 
 	status := svc.messageQueue.GetStatus(ctx, "s1")
@@ -140,6 +141,7 @@ func TestExecuteQueuedMessage_SkipsUserMessageWhenAlreadyRecorded(t *testing.T)
 		},
 	}
 
+	svc.markQueuedDispatchInFlight("s1", queuedMsg.ID)
 	svc.executeQueuedMessage("s1", queuedMsg)
 
 	if len(mc.userMessages) != 0 {
@@ -192,6 +194,7 @@ func TestExecuteQueuedMessage_RecordsCIAutomationPromptOnDrain(t *testing.T) {
 		},
 	}
 
+	svc.markQueuedDispatchInFlight("s1", queuedMsg.ID)
 	svc.executeQueuedMessage("s1", queuedMsg)
 
 	if len(mc.userMessages) != 1 {

apps/backend/internal/orchestrator/event_handlers_test.go

@@ -87,7 +87,13 @@ type mockTaskRepo struct {
 	tasks         map[string]*v1.Task
 	updatedStates map[string]v1.TaskState
 	stateWrites   map[string]int // per-task UpdateTaskState call count for dedup tests
-	getTaskErr    error          // if set, GetTask returns this error
+	// stateHistory records every state actually written per task, in order.
+	// updatedStates only keeps the latest value, which hides a transient
+	// write (e.g. a REVIEW write later overwritten by an async prompt
+	// dispatch's IN_PROGRESS write) — tests asserting a state was NEVER
+	// written at any point must check stateHistory, not updatedStates.
+	stateHistory map[string][]v1.TaskState
+	getTaskErr   error // if set, GetTask returns this error
 }
 
 func newMockTaskRepo() *mockTaskRepo {
@@ -95,6 +101,7 @@ func newMockTaskRepo() *mockTaskRepo {
 		tasks:         make(map[string]*v1.Task),
 		updatedStates: make(map[string]v1.TaskState),
 		stateWrites:   make(map[string]int),
+		stateHistory:  make(map[string][]v1.TaskState),
 	}
 }
 
@@ -119,6 +126,7 @@ func (m *mockTaskRepo) UpdateTaskState(_ context.Context, taskID string, state v
 	defer m.mu.Unlock()
 	m.updatedStates[taskID] = state
 	m.stateWrites[taskID]++
+	m.stateHistory[taskID] = append(m.stateHistory[taskID], state)
 	if t, ok := m.tasks[taskID]; ok {
 		t.State = state
 	}
@@ -140,6 +148,7 @@ func (m *mockTaskRepo) UpdateTaskStateIfCurrentIn(
 		}
 		m.updatedStates[taskID] = state
 		m.stateWrites[taskID]++
+		m.stateHistory[taskID] = append(m.stateHistory[taskID], state)
 		t.State = state
 		return true, nil
 	}
@@ -215,6 +224,11 @@ type mockAgentManager struct {
 	cancelAgentCalls   atomic.Int32
 	cancelAgentBlock   chan struct{}
 	cancelAgentEntered chan struct{}
+	// cancelAgentErr, when set, is returned by CancelAgent instead of nil —
+	// lets tests exercise callers that must react to a genuine cancel
+	// failure (as opposed to the tolerated ErrNoExecutionForSession /
+	// ErrCancelEscalated sentinels handled inside cancelAgentSilent).
+	cancelAgentErr error
 
 	// set_session_mode tracking (issue #1183). Records (sessionID, modeID) for
 	// every SetSessionModeBySessionID call. setSessionModeErr, when set, is
@@ -317,7 +331,7 @@ func (m *mockAgentManager) CancelAgent(_ context.Context, _ string) error {
 	if m.cancelAgentBlock != nil {
 		<-m.cancelAgentBlock
 	}
-	return nil
+	return m.cancelAgentErr
 }
 func (m *mockAgentManager) RespondToPermissionBySessionID(_ context.Context, _, _, _ string, _ bool) error {
 	return nil
@@ -720,7 +734,17 @@ func TestHandleAgentReadyGuards(t *testing.T) {
 		}
 	})
 
-	t.Run("ignores ready while cancel is in flight", func(t *testing.T) {
+	t.Run("waits for an in-flight cancel/interrupt then proceeds once it clears", func(t *testing.T) {
+		// This is the "no transition" side of the parent-interrupt vs.
+		// agent.ready race (see the reviewed PR #1653 discussion): once
+		// handleAgentReady acquires the guard *before* any bookkeeping, a
+		// concurrent holder no longer causes it to skip forever — it waits,
+		// then (since nothing about this turn actually changed while it
+		// waited) proceeds exactly as it would have uncontended, draining
+		// the queued message. TestHandleAgentReadyGuards_ConcurrentInterruptRaces
+		// below covers the cases where the guard holder *does* change
+		// something (a redispatched successor turn) that handleAgentReady
+		// must detect and back off from instead.
 		repo := setupTestRepo(t)
 		seedSession(t, repo, "t1", "s1", "step1")
 
@@ -728,23 +752,52 @@ func TestHandleAgentReadyGuards(t *testing.T) {
 		stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
 			ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0,
 		}
-		agentMgr := &mockAgentManager{isAgentRunning: true}
+		agentMgr := &mockAgentManager{isAgentRunning: true, repoForExecutionLookup: repo, promptDone: make(chan struct{})}
 		svc := createTestServiceWithAgent(repo, stepGetter, newMockTaskRepo(), agentMgr)
+		svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+		seedExecutorRunning(t, repo, "s1", "t1", "exec-1")
 
 		if _, err := svc.messageQueue.QueueMessage(ctx, "s1", "t1", "queued", "", messagequeue.QueuedByUser, false, nil); err != nil {
 			t.Fatalf("failed to queue message: %v", err)
 		}
-		svc.cancelInFlight.Store("s1", struct{}{})
-		defer svc.cancelInFlight.Delete("s1")
 
-		svc.handleAgentReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"})
+		lock, release := svc.acquireCancelInFlightGuard("s1")
+		lock.Lock()
+
+		readyDone := make(chan struct{})
+		go func() {
+			svc.handleAgentReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"})
+			close(readyDone)
+		}()
+
+		select {
+		case <-readyDone:
+			t.Fatal("handleAgentReady returned before the guard was released — it must block, not skip")
+		case <-time.After(100 * time.Millisecond):
+		}
 
 		status := svc.messageQueue.GetStatus(ctx, "s1")
 		if status.Count != 1 {
-			t.Fatalf("expected queued message to remain parked during cancel, count=%d entries=%+v", status.Count, status.Entries)
+			t.Fatalf("expected queued message to remain parked while the guard is held, count=%d entries=%+v", status.Count, status.Entries)
 		}
-		if len(agentMgr.capturedPrompts) != 0 {
-			t.Fatalf("expected no queued prompt dispatch during cancel, got %d prompts", len(agentMgr.capturedPrompts))
+
+		lock.Unlock()
+		release()
+
+		select {
+		case <-readyDone:
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for handleAgentReady to finish once the guard was released")
+		}
+
+		final := svc.messageQueue.GetStatus(ctx, "s1")
+		if final.Count != 0 {
+			t.Fatalf("expected handleAgentReady to drain the queued message once unblocked, count=%d entries=%+v", final.Count, final.Entries)
+		}
+		select {
+		case <-agentMgr.promptDone:
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for the queued message to be dispatched")
 		}
 	})
 
@@ -815,6 +868,196 @@ func TestHandleAgentReadyGuards(t *testing.T) {
 	// that makes it unnecessary.
 }
 
+// turnSnapshotSyncTurnService wraps repoTurnService so tests can
+// deterministically wait for handleAgentReady's pre-lock
+// peekActiveTurnID(sessionID) call to have actually captured the
+// *original* active turn before the test mutates it. Without this, the
+// goroutine in raceGuardAgainstTurnReplacement below is scheduler-
+// dependent: nothing guarantees handleAgentReady's first GetActiveTurn
+// call has actually run (vs. merely been scheduled) before the test
+// replaces the turn — a descheduled goroutine that captures the
+// *replacement* turn as its own baseline would make the "must detect
+// staleness" assertion pass for the wrong reason (no real race exercised)
+// instead of failing when the guard's revalidation regresses.
+// snapshotTaken closes once, on the first GetActiveTurn(sessionID) call
+// (handleAgentReady's pre-lock snapshot is always its first turn lookup).
+type turnSnapshotSyncTurnService struct {
+	*repoTurnService
+	sessionID     string
+	snapshotTaken chan struct{}
+	once          sync.Once
+}
+
+func (s *turnSnapshotSyncTurnService) GetActiveTurn(ctx context.Context, sessionID string) (*models.Turn, error) {
+	turn, err := s.repoTurnService.GetActiveTurn(ctx, sessionID)
+	if sessionID == s.sessionID {
+		s.once.Do(func() { close(s.snapshotTaken) })
+	}
+	return turn, err
+}
+
+// TestHandleAgentReadyGuards_ConcurrentInterruptRaces extends the
+// no-transition guard coverage above ("waits for an in-flight
+// cancel/interrupt then proceeds once it clears") with the two races
+// carlosflorencio's review explicitly called out as missing on PR #1653:
+// an actual on_turn_complete transition, and a pending move, each racing a
+// concurrent parent interrupt that cancels-and-redispatches the session
+// *before* handleAgentReady's blocked guard acquisition returns.
+//
+// Each subtest manually replaces the active turn (completeTurnForSession +
+// startTurnForSession) while holding the guard, mirroring exactly what a
+// real QueueAndInterruptForPeerMessage call does internally (cancel the
+// old turn, dispatch a new one) without needing to drive the full
+// mockAgentManager cancel/prompt round trip — the property under test is
+// specifically handleAgentReady's own post-guard revalidation, not the
+// interrupt's own dispatch mechanics (covered separately by the
+// TestQueueAndInterruptForPeerMessage_* suite in task_operations_test.go,
+// including TestQueueAndInterruptForPeerMessage_DoesNotCancelUnrelatedSuccessorTurn
+// which drives the same race through the public interrupt API instead).
+func TestHandleAgentReadyGuards_ConcurrentInterruptRaces(t *testing.T) {
+	ctx := context.Background()
+
+	// raceGuardAgainstTurnReplacement claims sessionID's cancelInFlight
+	// guard first (simulating an interrupt already in flight), starts
+	// handleAgentReady in a goroutine, deterministically waits for it to
+	// have snapshotted the *original* active turn (via
+	// turnSnapshotSyncTurnService, not a sleep) before replacing the
+	// active turn with a *different* one while still holding the guard
+	// (simulating the interrupt's own cancel-and-redispatch), then
+	// releases and waits for handleAgentReady to finish. Returns the turn
+	// that replaced the original. svc.turnService must already be a
+	// *turnSnapshotSyncTurnService for sessionID.
+	raceGuardAgainstTurnReplacement := func(t *testing.T, svc *Service, sessionID string) *models.Turn {
+		t.Helper()
+		turnSync, ok := svc.turnService.(*turnSnapshotSyncTurnService)
+		if !ok || turnSync.sessionID != sessionID {
+			t.Fatalf("test setup bug: svc.turnService must be a *turnSnapshotSyncTurnService for %q", sessionID)
+		}
+
+		lock, release := svc.acquireCancelInFlightGuard(sessionID)
+		lock.Lock()
+
+		readyDone := make(chan struct{})
+		go func() {
+			svc.handleAgentReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: sessionID})
+			close(readyDone)
+		}()
+
+		select {
+		case <-turnSync.snapshotTaken:
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for handleAgentReady to snapshot the original active turn")
+		}
+		select {
+		case <-readyDone:
+			t.Fatal("handleAgentReady returned before the guard was released — it must block on a concurrent interrupt")
+		default:
+		}
+
+		svc.completeTurnForSession(ctx, sessionID)
+		turnB, err := svc.turnService.StartTurn(ctx, sessionID)
+		if err != nil {
+			t.Fatalf("start replacement turn: %v", err)
+		}
+
+		lock.Unlock()
+		release()
+
+		select {
+		case <-readyDone:
+		case <-time.After(2 * time.Second):
+			t.Fatal("timed out waiting for handleAgentReady to finish once the guard was released")
+		}
+		return turnB
+	}
+
+	t.Run("actual on_turn_complete transition: stale ready must not apply it", func(t *testing.T) {
+		repo := setupTestRepo(t)
+		seedSession(t, repo, "t1", "s1", "step1")
+
+		stepGetter := newMockStepGetter()
+		stepGetter.steps["step1"] = &wfmodels.WorkflowStep{
+			ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0,
+			Events: wfmodels.StepEvents{
+				OnTurnComplete: []wfmodels.OnTurnCompleteAction{{Type: wfmodels.OnTurnCompleteMoveToNext}},
+			},
+		}
+		stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1}
+		taskRepo := newMockTaskRepo()
+		seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
+		svc := createTestService(repo, stepGetter, taskRepo)
+		svc.turnService = &turnSnapshotSyncTurnService{
+			repoTurnService: &repoTurnService{repo: repo},
+			sessionID:       "s1",
+			snapshotTaken:   make(chan struct{}),
+		}
+
+		if _, err := svc.turnService.StartTurn(ctx, "s1"); err != nil {
+			t.Fatalf("seed original turn: %v", err)
+		}
+
+		turnB := raceGuardAgainstTurnReplacement(t, svc, "s1")
+
+		active, err := svc.turnService.GetActiveTurn(ctx, "s1")
+		if err != nil {
+			t.Fatalf("get active turn: %v", err)
+		}
+		if active == nil || active.ID != turnB.ID {
+			t.Fatalf("expected the replacement turn %q to still be open, got %+v", turnB.ID, active)
+		}
+
+		updatedTask, err := repo.GetTask(ctx, "t1")
+		if err != nil {
+			t.Fatalf("get task: %v", err)
+		}
+		if updatedTask.WorkflowStepID != "step1" {
+			t.Fatalf("a stale ready event must not apply the on_turn_complete transition for a turn it no longer owns; workflow step moved to %q", updatedTask.WorkflowStepID)
+		}
+	})
+
+	t.Run("pending move: stale ready must leave it for the replacement turn", func(t *testing.T) {
+		repo := setupTestRepo(t)
+		seedSession(t, repo, "t1", "s1", "step1")
+
+		stepGetter := newMockStepGetter()
+		stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0}
+		stepGetter.steps["step2"] = &wfmodels.WorkflowStep{ID: "step2", WorkflowID: "wf1", Name: "Step 2", Position: 1}
+		taskRepo := newMockTaskRepo()
+		seedMockTaskState(taskRepo, "t1", v1.TaskStateInProgress)
+		svc := createTestService(repo, stepGetter, taskRepo)
+		svc.turnService = &turnSnapshotSyncTurnService{
+			repoTurnService: &repoTurnService{repo: repo},
+			sessionID:       "s1",
+			snapshotTaken:   make(chan struct{}),
+		}
+
+		if _, err := svc.turnService.StartTurn(ctx, "s1"); err != nil {
+			t.Fatalf("seed original turn: %v", err)
+		}
+		svc.messageQueue.SetPendingMove(ctx, "s1", &messagequeue.PendingMove{
+			TaskID: "t1", WorkflowID: "wf1", WorkflowStepID: "step2",
+		})
+
+		turnB := raceGuardAgainstTurnReplacement(t, svc, "s1")
+
+		active, err := svc.turnService.GetActiveTurn(ctx, "s1")
+		if err != nil {
+			t.Fatalf("get active turn: %v", err)
+		}
+		if active == nil || active.ID != turnB.ID {
+			t.Fatalf("expected the replacement turn %q to still be open, got %+v", turnB.ID, active)
+		}
+
+		move, exists := svc.messageQueue.GetPendingMove(ctx, "s1")
+		if !exists {
+			t.Fatal("a stale ready event must not consume the pending move meant for the replacement turn")
+		}
+		if move.WorkflowStepID != "step2" {
+			t.Fatalf("expected the pending move to still target step2, got %q", move.WorkflowStepID)
+		}
+	})
+}
+
 func TestExecuteQueuedMessage_RequeuesTransientPromptFailure(t *testing.T) {
 	ctx := context.Background()
 	repo := setupTestRepo(t)
@@ -847,6 +1090,7 @@ func TestExecuteQueuedMessage_RequeuesTransientPromptFailure(t *testing.T) {
 		QueuedBy:  "test",
 	}
 
+	svc.markQueuedDispatchInFlight("s1", queuedMsg.ID)
 	svc.executeQueuedMessage("s1", queuedMsg)
 
 	status := svc.messageQueue.GetStatus(ctx, "s1")
@@ -894,6 +1138,7 @@ func TestExecuteQueuedMessage_RequeuesCoalescedMessageWithOriginalSender(t *test
 		t.Fatalf("seed newer coalesced message: %v", err)
 	}
 
+	svc.markQueuedDispatchInFlight("s1", queuedMsg.ID)
 	svc.executeQueuedMessage("s1", queuedMsg)
 
 	status := svc.messageQueue.GetStatus(ctx, "s1")

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -1043,7 +1043,7 @@ func (s *Service) applyPendingMove(ctx context.Context, taskID, sessionID string
 		s.logger.Info("pending move target equals current step; skipping transition",
 			zap.String("task_id", taskID),
 			zap.String("step_id", fromStepID))
-		s.drainQueuedMessageForPromptableSession(ctx, sessionID)
+		s.drainQueuedMessageForPromptableSessionLocked(ctx, sessionID)
 		return
 	}
 
@@ -1139,17 +1139,79 @@ func (s *Service) syncTaskStateForPendingMove(ctx context.Context, taskID, fromS
 	s.publishTaskStateChanged(ctx, task, oldState)
 }
 
-// drainQueuedMessageForPromptableSession takes the next queued message and dispatches
-// it for execution. Callers must ensure the session is ready for input first.
+// drainQueuedMessageForPromptableSession acquires sessionID's cancelInFlight
+// guard and takes+dispatches the next queued message, blocking until any
+// concurrent cancel/interrupt/drain for the same session finishes first —
+// see the Service.cancelInFlight field doc comment for why every
+// take-and-dispatch decision must serialize through this one guard rather
+// than risk two callers racing to steal the same entry. Blocking (not
+// skipping on contention) matters here because, unlike
+// handleAgentReady/handleAgentBootReady's own take-decision (where a losing
+// side can rely on the winning side's take-and-dispatch, or a future turn's
+// own agent.ready, to eventually retry), callers of this function —
+// workflow on_enter branches, manual drain requests, CI automation — have
+// no other future trigger that would retry a skipped drain; skipping on
+// contention here could strand an already-queued message.
 //
-// Used by processOnEnter branches that don't auto-start the agent, manual
-// drain requests, and cancel recovery. Without this, the message is orphaned
-// because handleAgentReady only drains from a normal turn-end event.
+// Callers must ensure the session is ready for input *before* calling this
+// — exactly as before this was guarded — but must not have already claimed
+// the guard themselves; use drainQueuedMessageForPromptableSessionLocked
+// instead when the guard is already held (e.g. inside CancelAgent,
+// cancelAndTakeForPeerMessage, handleAgentBootReady, handleAgentReady,
+// applyPendingMove).
+//
+// Reloads the session and re-confirms promptability *after* acquiring the
+// guard rather than trusting the caller's own earlier check: callers like
+// processOnEnter typically call setSessionWaitingForInput and then this
+// function without holding the guard across both, so a concurrent
+// cancel/interrupt for the same session could land in between and this
+// call would otherwise blindly take a message for a session that turned
+// out to no longer be idle.
 func (s *Service) drainQueuedMessageForPromptableSession(ctx context.Context, sessionID string) bool {
-	if s.messageQueue == nil {
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
+
+	session, err := s.repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		s.logger.Warn("failed to reload session before drain",
+			zap.String("session_id", sessionID), zap.Error(err))
+		return false
+	}
+	if err := s.checkSessionPromptable(session.TaskID, sessionID, session.State); err != nil {
+		s.logger.Debug("skipping drain: session is not promptable once the guard is held",
+			zap.String("session_id", sessionID), zap.Error(err))
+		return false
+	}
+	return s.drainQueuedMessageForPromptableSessionLocked(ctx, sessionID)
+}
+
+// drainQueuedMessageForPromptableSessionLocked takes the next queued
+// message and dispatches it for execution. Callers must ensure the session
+// is ready for input first, AND must already hold sessionID's
+// cancelInFlight lock — this neither acquires nor releases it. Use the
+// public drainQueuedMessageForPromptableSession instead when the guard is
+// not already held.
+//
+// Backs off without taking anything when isQueuedDispatchInFlight reports
+// a different dispatch already handed off for this session — see the
+// Service.dispatchingQueued field doc comment for the double-dispatch
+// window this closes.
+func (s *Service) drainQueuedMessageForPromptableSessionLocked(ctx context.Context, sessionID string) bool {
+	if s.messageQueue == nil || s.isQueuedDispatchInFlight(sessionID) {
 		return false
 	}
 	queuedMsg, ok := s.messageQueue.TakeQueued(ctx, sessionID)
+	return s.dispatchTakenQueuedMessage(ctx, sessionID, queuedMsg, ok)
+}
+
+// dispatchTakenQueuedMessage publishes the queue-status update and dispatches
+// queuedMsg for execution, given the (message, ok) pair returned by a Take*
+// call on the message queue (TakeQueued or TakeQueuedEntry). Shared so
+// InterruptForPeerMessage's targeted-entry take gets the same empty-message
+// guard and dispatch behavior as the FIFO-head drain.
+func (s *Service) dispatchTakenQueuedMessage(ctx context.Context, sessionID string, queuedMsg *messagequeue.QueuedMessage, ok bool) bool {
 	if !ok || queuedMsg == nil {
 		return false
 	}
@@ -1160,6 +1222,12 @@ func (s *Service) drainQueuedMessageForPromptableSession(ctx context.Context, se
 			zap.String("queue_id", queuedMsg.ID))
 		return false
 	}
+	// Reserve entryID as sessionID's "dispatch in flight" token *before*
+	// handing off to the async goroutine — see the Service.dispatchingQueued
+	// field doc comment for why session.State alone isn't a reliable busy
+	// signal until executeQueuedMessage's own promptTask call reaches its
+	// guarded claim step, several DB round-trips later.
+	s.markQueuedDispatchInFlight(sessionID, queuedMsg.ID)
 	go s.executeQueuedMessage(sessionID, queuedMsg)
 	return true
 }

apps/backend/internal/orchestrator/messagequeue/repository.go

@@ -31,6 +31,14 @@ type Repository interface {
 	// session. Returns nil, nil if the queue is empty.
 	TakeHead(ctx context.Context, sessionID string) (*QueuedMessage, error)
 
+	// TakeByID atomically returns and deletes the entry identified by entryID
+	// for sessionID, regardless of its FIFO position. Unlike DeleteByID, it has
+	// no queued_by="agent" guard: the caller is internal orchestrator code
+	// (InterruptForPeerMessage) dispatching the specific entry that triggered
+	// its own interrupt, not a user-driven "remove queued message" request.
+	// Returns nil, nil if no entry matches (already taken or never existed).
+	TakeByID(ctx context.Context, sessionID, entryID string) (*QueuedMessage, error)
+
 	// UpdateContent replaces the content/attachments of an entry. The session
 	// scope (`AND session_id = ?`) is mandatory so a caller can't update an
 	// entry by guessing its UUID across sessions. If queuedBy is non-empty the

apps/backend/internal/orchestrator/messagequeue/repository_memory.go

@@ -154,6 +154,31 @@ func (r *memoryRepository) TakeHead(_ context.Context, sessionID string) (*Queue
 	return &out, nil
 }
 
+// TakeByID atomically returns and deletes the entry identified by entryID,
+// regardless of its FIFO position. Unlike DeleteByID, it has no
+// QueuedByAgent guard — see the Repository interface doc comment.
+func (r *memoryRepository) TakeByID(_ context.Context, sessionID, entryID string) (*QueuedMessage, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	list, ok := r.entries[sessionID]
+	if !ok {
+		return nil, nil
+	}
+	for i, m := range list {
+		if m.ID != entryID {
+			continue
+		}
+		r.entries[sessionID] = append(list[:i], list[i+1:]...)
+		if len(r.entries[sessionID]) == 0 {
+			delete(r.entries, sessionID)
+			delete(r.nextPosition, sessionID)
+		}
+		out := *m
+		return &out, nil
+	}
+	return nil, nil
+}
+
 func (r *memoryRepository) UpdateContent(_ context.Context, sessionID, entryID, content string, attachments []MessageAttachment, queuedBy string) error {
 	r.mu.Lock()
 	defer r.mu.Unlock()

apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go

@@ -384,6 +384,52 @@ func (r *sqliteRepository) TakeHead(ctx context.Context, sessionID string) (*Que
 	return msg, nil
 }
 
+// TakeByID atomically returns and deletes the entry identified by entryID,
+// regardless of its FIFO position. Mirrors TakeHead's race handling: if a
+// concurrent take already removed the row between the SELECT and DELETE,
+// RowsAffected()==0 is treated as "already taken" (nil, nil) rather than an
+// error, so two racing takers can never both dispatch the same entry. The
+// DELETE is also scoped by session_id (not just id) so a concurrent
+// TransferSession that reassigns this row to a different session between
+// the SELECT and DELETE can't have it removed out from under the new
+// session — same session-scope invariant DeleteByID/UpdateContent enforce.
+func (r *sqliteRepository) TakeByID(ctx context.Context, sessionID, entryID string) (*QueuedMessage, error) {
+	tx, err := r.db.BeginTxx(ctx, nil)
+	if err != nil {
+		return nil, fmt.Errorf("begin take tx: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	row := tx.QueryRowxContext(ctx, r.db.Rebind(`
+		SELECT id, session_id, task_id, position, content, model, plan_mode,
+		       attachments_json, metadata_json, queued_at, queued_by
+		FROM queued_messages
+		WHERE id = ? AND session_id = ?
+	`), entryID, sessionID)
+	msg, err := scanQueuedRow(row)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("take by id: %w", err)
+	}
+	res, err := tx.ExecContext(ctx, r.db.Rebind(`DELETE FROM queued_messages WHERE id = ? AND session_id = ?`), msg.ID, sessionID)
+	if err != nil {
+		return nil, fmt.Errorf("delete by id: %w", err)
+	}
+	affected, err := res.RowsAffected()
+	if err != nil {
+		return nil, fmt.Errorf("delete by id rows affected: %w", err)
+	}
+	if affected == 0 {
+		return nil, nil
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+	return msg, nil
+}
+
 func (r *sqliteRepository) UpdateContent(ctx context.Context, sessionID, entryID, content string, attachments []MessageAttachment, queuedBy string) error {
 	attachmentsJSON, err := marshalAttachments(attachments)
 	if err != nil {

apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go

@@ -249,6 +249,79 @@ func TestSQLiteRepository_DeleteByID(t *testing.T) {
 	}
 }
 
+// TestSQLiteRepository_TakeByID covers TakeByID's cross-session guard,
+// out-of-FIFO-order removal, idempotent re-take of an already-taken id, and
+// (unlike DeleteByID) that agent-authored entries are takeable.
+func TestSQLiteRepository_TakeByID(t *testing.T) {
+	repo := newTestSQLiteRepo(t)
+	ctx := context.Background()
+
+	first := &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "first", QueuedBy: "u"}
+	if err := repo.Insert(ctx, first, 0); err != nil {
+		t.Fatalf("insert first: %v", err)
+	}
+	second := &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "second", QueuedBy: "u"}
+	if err := repo.Insert(ctx, second, 0); err != nil {
+		t.Fatalf("insert second: %v", err)
+	}
+	third := &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "third", QueuedBy: "u"}
+	if err := repo.Insert(ctx, third, 0); err != nil {
+		t.Fatalf("insert third: %v", err)
+	}
+
+	// Cross-session take attempt: must not affect the row.
+	got, err := repo.TakeByID(ctx, "s-attacker", second.ID)
+	if err != nil {
+		t.Fatalf("cross-session take: %v", err)
+	}
+	if got != nil {
+		t.Errorf("expected nil for cross-session take, got %+v", got)
+	}
+	count, _ := repo.CountBySession(ctx, "s1")
+	if count != 3 {
+		t.Errorf("entries should survive cross-session take attempt, got count=%d", count)
+	}
+
+	// Taking the middle entry out of FIFO order must not disturb the others.
+	got, err = repo.TakeByID(ctx, "s1", second.ID)
+	if err != nil {
+		t.Fatalf("take: %v", err)
+	}
+	if got == nil || got.Content != "second" {
+		t.Fatalf("expected to take %q, got %+v", "second", got)
+	}
+	remaining, err := repo.ListBySession(ctx, "s1")
+	if err != nil {
+		t.Fatalf("list: %v", err)
+	}
+	if len(remaining) != 2 || remaining[0].Content != "first" || remaining[1].Content != "third" {
+		t.Fatalf("expected [first, third] to remain in order, got %+v", remaining)
+	}
+
+	// Taking the same id again finds nothing — it's already gone.
+	got, err = repo.TakeByID(ctx, "s1", second.ID)
+	if err != nil {
+		t.Fatalf("re-take: %v", err)
+	}
+	if got != nil {
+		t.Errorf("expected nil on re-take of already-taken id, got %+v", got)
+	}
+
+	// Unlike DeleteByID, TakeByID has no queued_by="agent" guard — the caller
+	// is internal orchestrator code dispatching its own queued entry.
+	agentMsg := &QueuedMessage{SessionID: "s1", TaskID: "t1", Content: "agent", QueuedBy: QueuedByAgent}
+	if err := repo.Insert(ctx, agentMsg, 0); err != nil {
+		t.Fatalf("insert agent message: %v", err)
+	}
+	got, err = repo.TakeByID(ctx, "s1", agentMsg.ID)
+	if err != nil {
+		t.Fatalf("take agent-authored entry: %v", err)
+	}
+	if got == nil || got.Content != "agent" {
+		t.Fatalf("expected to take agent-authored entry, got %+v", got)
+	}
+}
+
 func TestSQLiteRepository_DeleteAllBySession(t *testing.T) {
 	repo := newTestSQLiteRepo(t)
 	ctx := context.Background()

apps/backend/internal/orchestrator/messagequeue/service.go

@@ -164,6 +164,36 @@ func (s *Service) TakeQueued(ctx context.Context, sessionID string) (*QueuedMess
 	return msg, true
 }
 
+// TakeQueuedEntry atomically removes and returns the entry identified by
+// entryID, regardless of its FIFO position. Returns nil, false, nil when
+// the entry no longer exists (already drained or removed by a concurrent
+// path); returns a non-nil error on a genuine repository failure — the two
+// are deliberately distinguished so InterruptForPeerMessage can treat "not
+// found" (safe to fall back to a FIFO-head drain) differently from "the
+// repository call itself failed" (unsafe to fall back: a transient error
+// here says nothing about which entry is actually at the FIFO head, and
+// falling back anyway risks dispatching the wrong message while still
+// reporting "sent" for the parent's — see InterruptForPeerMessage). Used by
+// InterruptForPeerMessage to dispatch the specific message that triggered
+// the interrupt instead of whatever happens to be at the FIFO head.
+func (s *Service) TakeQueuedEntry(ctx context.Context, sessionID, entryID string) (*QueuedMessage, bool, error) {
+	msg, err := s.repo.TakeByID(ctx, sessionID, entryID)
+	if err != nil {
+		s.logger.Error("take by id failed",
+			zap.String("session_id", sessionID),
+			zap.String("entry_id", entryID),
+			zap.Error(err))
+		return nil, false, err
+	}
+	if msg == nil {
+		return nil, false, nil
+	}
+	s.logger.Info("message dequeued by id",
+		zap.String("session_id", sessionID),
+		zap.String("entry_id", msg.ID))
+	return msg, true, nil
+}
+
 // UpdateMessage replaces the content (and optionally attachments) of a queued
 // entry. The sessionID scope is mandatory — callers can't update an entry by
 // guessing its UUID across sessions. Returns ErrEntryNotFound when the entry

apps/backend/internal/orchestrator/messagequeue/service_test.go

@@ -212,6 +212,116 @@ func TestTakeQueued(t *testing.T) {
 	})
 }
 
+// TestTakeQueuedEntry covers TakeQueuedEntry: out-of-FIFO-order removal,
+// takeability of agent-authored entries (unlike RemoveEntry), the
+// not-found (nil, false, nil) shape for a missing or foreign-session id,
+// and — distinctly — a genuine repository error propagating as a non-nil
+// error rather than being collapsed into the not-found shape.
+func TestTakeQueuedEntry(t *testing.T) {
+	t.Run("removes and returns the targeted entry regardless of FIFO position", func(t *testing.T) {
+		svc := setupService(t)
+		ctx := context.Background()
+
+		_, err := svc.QueueMessage(ctx, "s", "t", "first", "", "u", false, nil)
+		require.NoError(t, err)
+		second, err := svc.QueueMessage(ctx, "s", "t", "second", "", "u", false, nil)
+		require.NoError(t, err)
+		_, err = svc.QueueMessage(ctx, "s", "t", "third", "", "u", false, nil)
+		require.NoError(t, err)
+
+		msg, ok, err := svc.TakeQueuedEntry(ctx, "s", second.ID)
+		require.NoError(t, err)
+		require.True(t, ok)
+		assert.Equal(t, "second", msg.Content)
+
+		// The other two entries are untouched and keep their relative order.
+		status := svc.GetStatus(ctx, "s")
+		require.Equal(t, 2, status.Count)
+		assert.Equal(t, "first", status.Entries[0].Content)
+		assert.Equal(t, "third", status.Entries[1].Content)
+
+		// Taking the same id again finds nothing — it's already gone.
+		_, ok, err = svc.TakeQueuedEntry(ctx, "s", second.ID)
+		require.NoError(t, err)
+		assert.False(t, ok)
+	})
+
+	t.Run("takes agent-authored entries unlike RemoveEntry", func(t *testing.T) {
+		svc := setupService(t)
+		ctx := context.Background()
+
+		agentEntry, err := svc.QueueMessageWithMetadata(ctx, "s", "t", "agent entry", "", QueuedByAgent, false, nil, nil)
+		require.NoError(t, err)
+
+		msg, ok, err := svc.TakeQueuedEntry(ctx, "s", agentEntry.ID)
+		require.NoError(t, err)
+		require.True(t, ok)
+		assert.Equal(t, "agent entry", msg.Content)
+	})
+
+	t.Run("returns false for a missing or foreign-session id", func(t *testing.T) {
+		svc := setupService(t)
+		ctx := context.Background()
+
+		victim, err := svc.QueueMessage(ctx, "s-victim", "t", "victim entry", "", "u", false, nil)
+		require.NoError(t, err)
+
+		_, ok, err := svc.TakeQueuedEntry(ctx, "s-victim", "missing-id")
+		require.NoError(t, err)
+		assert.False(t, ok)
+
+		_, ok, err = svc.TakeQueuedEntry(ctx, "s-attacker", victim.ID)
+		require.NoError(t, err)
+		assert.False(t, ok)
+
+		status := svc.GetStatus(ctx, "s-victim")
+		assert.Equal(t, 1, status.Count)
+	})
+
+	t.Run("propagates a genuine repository error instead of reporting not-found", func(t *testing.T) {
+		// A repository error (e.g. a transient DB failure) must not be
+		// collapsed into the same (nil, false) shape as a legitimate
+		// not-found — InterruptForPeerMessage treats the two differently
+		// (not-found falls back to a FIFO-head drain; an error propagates
+		// without a fallback, since the error says nothing about what is
+		// actually at the FIFO head).
+		log, err := logger.NewLogger(logger.LoggingConfig{
+			Level:      "error",
+			Format:     "console",
+			OutputPath: "stderr",
+		})
+		require.NoError(t, err)
+		wantErr := errors.New("db unavailable")
+		repo := &errInjectingRepository{Repository: NewMemoryRepository(), takeByIDErr: wantErr}
+		svc := NewService(repo, DefaultMaxPerSession, log)
+		ctx := context.Background()
+
+		msg, ok, err := svc.TakeQueuedEntry(ctx, "s", "some-id")
+		assert.Nil(t, msg)
+		assert.False(t, ok)
+		assert.ErrorIs(t, err, wantErr)
+	})
+}
+
+// errInjectingRepository wraps a Repository and returns a configured error
+// from TakeByID, letting tests exercise TakeQueuedEntry's error-propagation
+// path without needing a real repository failure.
+type errInjectingRepository struct {
+	Repository
+	takeByIDErr error
+}
+
+// TakeByID returns the configured error if set, otherwise delegates to the
+// embedded Repository so the remaining repository operations (Insert,
+// ListBySession, CountBySession, ...) still work against the real
+// underlying store.
+func (r *errInjectingRepository) TakeByID(ctx context.Context, sessionID, entryID string) (*QueuedMessage, error) {
+	if r.takeByIDErr != nil {
+		return nil, r.takeByIDErr
+	}
+	return r.Repository.TakeByID(ctx, sessionID, entryID)
+}
+
 func TestUpdateMessage(t *testing.T) {
 	t.Run("updates content and survives in list", func(t *testing.T) {
 		svc := setupService(t)

apps/backend/internal/orchestrator/service.go

@@ -373,6 +373,35 @@ type Service struct {
 	// Active turns map: sessionID -> turnID
 	activeTurns sync.Map
 
+	// dispatchingQueued tracks, per session, the entry ID of whichever
+	// queued message was most recently taken and handed off to the async
+	// executeQueuedMessage goroutine, but hasn't yet reached promptTask's
+	// own setSessionRunning call — the only point at which session.State
+	// itself starts correctly reporting "busy" for that dispatch. Without
+	// this, a second take-and-dispatch decision for the same session (a
+	// workflow drain, a manual drain, or another parent interrupt) could
+	// acquire the cancelInFlight guard in that gap, see the still-idle
+	// session.State, and dispatch a second, unrelated queued entry before
+	// the first dispatch's turn has even started.
+	//
+	// Stores the entry ID (string), not a bool: a newer dispatch for the
+	// same session (e.g. a second parent interrupt cancelling and
+	// re-taking) always supersedes an older one that's still settling —
+	// markQueuedDispatchInFlight unconditionally overwrites. Each
+	// goroutine must reconfirm it still owns *its own* entry ID
+	// (isCurrentQueuedDispatch) immediately before calling
+	// setSessionRunning, and may only clear the marker via a
+	// compare-and-delete keyed on that same entry ID
+	// (clearQueuedDispatchInFlightIfCurrent) — otherwise a superseded
+	// goroutine finishing late could delete a newer dispatch's marker out
+	// from under it, or proceed to prompt after it no longer owns the
+	// session. Non-cancelling take-and-dispatch paths
+	// (drainQueuedMessageForPromptableSessionLocked, takeIfPromptableLocked)
+	// instead use isQueuedDispatchInFlight, which only asks "is *anything*
+	// still settling" — they have no cancel to supersede it with, so any
+	// in-flight dispatch at all is reason enough to defer.
+	dispatchingQueued sync.Map
+
 	// taskRuntimeStateMu serializes task-state flips derived from session
 	// runtime state. Without it, a completion/cancel path can check for active
 	// sibling sessions just before another handler marks one RUNNING, then
@@ -403,11 +432,30 @@ type Service struct {
 	// activity before triggering fallback resume.
 	clarificationWatchdogTimeout time.Duration
 
-	// cancelInFlight tracks sessionIDs whose CancelAgent call is currently in
-	// progress. It deduplicates impatient retries from the UI and lets late
-	// agent.ready/boot_ready events from the cancelled turn return before they
-	// evaluate workflow transitions or drain queued messages.
-	cancelInFlight sync.Map
+	// cancelInFlightMu guards cancelInFlight's map structure (insert/
+	// lookup/remove of *cancelInFlightGuard entries) — held only briefly
+	// for that bookkeeping, never across a caller's actual per-session
+	// critical section (which is guarded by the entry's own mutex, handed
+	// to callers by acquireCancelInFlightGuard).
+	cancelInFlightMu sync.Mutex
+	// cancelInFlight holds one *cancelInFlightGuard per session with an
+	// active reference — an in-progress or waiting claim from CancelAgent
+	// (the user cancel button — TryLock, dedup impatient retries), the
+	// natural turn-completion/boot-ready drain decision in handleAgentReady
+	// / handleAgentBootReady (TryLock, skip if a cancel/interrupt owns it),
+	// or QueueAndInterruptForPeerMessage (blocking Lock — must wait rather
+	// than work around a busy lock with an unguarded insert; see its doc
+	// comment). All of these must go through the same per-session guard —
+	// a second, independent lock for any of them would defeat the mutual
+	// exclusion the others rely on to avoid racing each other's
+	// take-and-dispatch decision for the same session.
+	//
+	// Entries are reference-counted (acquireCancelInFlightGuard /
+	// releaseCancelInFlightGuard) and pruned once nobody holds a reference,
+	// so this stays bounded by concurrently-active sessions rather than
+	// growing by one permanent entry per session ever created over a
+	// long-lived backend's lifetime.
+	cancelInFlight map[string]*cancelInFlightGuard
 
 	// transientRetries tracks in-progress transient-provider-error (529
 	// Overloaded) retry loops. key: sessionID, value: *transientRetryEntry.
@@ -901,6 +949,106 @@ func (s *Service) getActiveTurnID(sessionID string) string {
 	return s.startTurnForSession(context.Background(), sessionID)
 }
 
+// peekActiveTurnID returns sessionID's currently active turn ID, or "" if
+// none, WITHOUT creating one when none exists — unlike getActiveTurnID,
+// this never calls startTurnForSession. Used by QueueAndInterruptForPeerMessage
+// and handleAgentReady to snapshot "which turn is this?" before contending
+// for the per-session cancelInFlight guard, then re-check it once the guard
+// is held to detect whether a *different* turn has started in the
+// meantime (e.g. a workflow transition auto-started a successor while the
+// caller waited) — see their doc comments for the races this closes.
+//
+// Deliberately reads through to turnService.GetActiveTurn (the DB) rather
+// than only the activeTurns in-memory cache: completeTurnForSession's own
+// doc comment notes the cache can drift or miss turns that exist only in
+// the DB (e.g. a lazily-started turn from service.CreateMessage), and
+// missing a genuinely-active successor here would defeat the safety check
+// callers rely on this for. Returns ("", nil) when turnService is not
+// wired — callers must treat that as "no turn identity can be
+// established" and fall back to their pre-existing (turn-unaware)
+// behavior, not as a confirmed empty turn.
+func (s *Service) peekActiveTurnID(ctx context.Context, sessionID string) (string, error) {
+	if sessionID == "" || s.turnService == nil {
+		return "", nil
+	}
+	turn, err := s.turnService.GetActiveTurn(ctx, sessionID)
+	if err != nil {
+		return "", err
+	}
+	if turn == nil {
+		return "", nil
+	}
+	return turn.ID, nil
+}
+
+// markQueuedDispatchInFlight records that entryID — a specific queued
+// message — has been taken and handed off to the async
+// executeQueuedMessage goroutine for sessionID. Unconditionally
+// overwrites any previous entry for the session: a newer dispatch always
+// supersedes an older one that's still settling. See the
+// Service.dispatchingQueued field doc comment for why this exists:
+// session.State alone doesn't reflect "busy" until that goroutine's own
+// promptTask call reaches setSessionRunning, several DB round-trips
+// later — and for why the token must be the specific entry ID, not a bare
+// bool.
+func (s *Service) markQueuedDispatchInFlight(sessionID, entryID string) {
+	if sessionID == "" {
+		return
+	}
+	s.dispatchingQueued.Store(sessionID, entryID)
+}
+
+// clearQueuedDispatchInFlightIfCurrent removes sessionID's in-flight
+// marker only if it still names entryID — a compare-and-delete so a
+// goroutine whose own dispatch has since been superseded by a newer one
+// (a different entryID overwrote the marker) can never clear the
+// *newer* dispatch's marker out from under it. Called by
+// executeQueuedMessage via defer so the marker is cleared on every exit
+// path (success, transient requeue, superseded, or lost/dropped message)
+// — but only when this goroutine is still the current owner.
+func (s *Service) clearQueuedDispatchInFlightIfCurrent(sessionID, entryID string) {
+	if sessionID == "" {
+		return
+	}
+	s.dispatchingQueued.CompareAndDelete(sessionID, entryID)
+}
+
+// isCurrentQueuedDispatch reports whether entryID is still the most
+// recently handed-off in-flight dispatch for sessionID — i.e. no *other*
+// dispatch has superseded it since markQueuedDispatchInFlight was called
+// for it. A goroutine that owns entryID must confirm this immediately
+// before calling setSessionRunning (see promptTask's claimDispatch
+// parameter); if it no longer owns the token, a different dispatch has
+// already taken over the session and this one must not proceed.
+func (s *Service) isCurrentQueuedDispatch(sessionID, entryID string) bool {
+	if sessionID == "" {
+		return false
+	}
+	v, ok := s.dispatchingQueued.Load(sessionID)
+	if !ok {
+		return false
+	}
+	current, _ := v.(string)
+	return current == entryID
+}
+
+// isQueuedDispatchInFlight reports whether *any* queued message is
+// currently in the handoff window for sessionID, regardless of which one
+// — see markQueuedDispatchInFlight. Checked by
+// drainQueuedMessageForPromptableSessionLocked and takeIfPromptableLocked
+// before either takes and directly dispatches another entry on the same
+// session without a cancel to supersede whatever is already settling; a
+// genuine cancel (cancelAndTakeForPeerMessage) is exempt — see its own
+// doc comment for why it may proceed regardless and simply overwrite the
+// token via markQueuedDispatchInFlight.
+func (s *Service) isQueuedDispatchInFlight(sessionID string) bool {
+	if sessionID == "" {
+		return false
+	}
+	_, ok := s.dispatchingQueued.Load(sessionID)
+	return ok
+}
+
 func (s *Service) setSessionResetInProgress(sessionID string, inProgress bool) {
 	if sessionID == "" {
 		return

apps/backend/internal/orchestrator/task_operations.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"go.uber.org/zap"
@@ -16,6 +17,7 @@ import (
 	"github.com/kandev/kandev/internal/agent/runtime/routingerr"
 	"github.com/kandev/kandev/internal/orchestrator/dto"
 	"github.com/kandev/kandev/internal/orchestrator/executor"
+	"github.com/kandev/kandev/internal/orchestrator/messagequeue"
 	"github.com/kandev/kandev/internal/orchestrator/queue"
 	"github.com/kandev/kandev/internal/sysprompt"
 	"github.com/kandev/kandev/internal/task/models"
@@ -48,6 +50,15 @@ var ErrSessionResetInProgress = errors.New("session reset in progress")
 // the two misleads the UI and any caller doing errors.Is checks.
 var ErrSessionNotPromptable = errors.New("session not promptable")
 
+// errQueuedDispatchSuperseded is returned by promptTask when a queued
+// message's dispatch loses its claim (see isCurrentQueuedDispatch) to a
+// newer dispatch for the same session — e.g. a second parent interrupt
+// cancelling and re-taking while the first dispatch's own goroutine was
+// still settling. Not surfaced to callers of the public PromptTask; only
+// executeQueuedMessage's internal call passes a claim token that can
+// trigger this.
+var errQueuedDispatchSuperseded = errors.New("queued dispatch superseded by a newer one for this session")
+
 var (
 	// Backend restart recovery can restore the session state before the ACP
 	// stream is promptable again. Keep this above CI's slow-start tail so a
@@ -2233,6 +2244,37 @@ func (s *Service) saveArchiveCommits(ctx context.Context, sessionID string, comm
 // If planMode is true, a plan mode prefix is prepended to the prompt.
 // Attachments (images) are passed through to the agent if provided.
 func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prompt string, model string, planMode bool, attachments []v1.MessageAttachment, dispatchOnly bool) (*PromptResult, error) {
+	return s.promptTask(ctx, taskID, sessionID, prompt, model, planMode, attachments, dispatchOnly, "")
+}
+
+// promptTask is PromptTask's implementation, taking an additional
+// claimEntryID parameter used only by executeQueuedMessage's internal
+// call. When non-empty, this acquires sessionID's cancelInFlight guard
+// around a "confirm I still own this dispatch, then mark the session
+// RUNNING" step, immediately before startTurnForSession and the
+// (potentially long-blocking) executor.Prompt call below it — see
+// isCurrentQueuedDispatch and the Service.dispatchingQueued field doc
+// comment for the race this closes.
+//
+// A bare (unguarded) "check the token, then separately call
+// setSessionRunning" would still let two settling dispatches for the same
+// session both pass their own check before either actually marks the
+// session RUNNING — the exact double-dispatch this mechanism exists to
+// prevent, just narrowed rather than closed. Serializing the check
+// *and* the mark through the same per-session guard used for every other
+// cancel/take-and-dispatch decision (see the Service.cancelInFlight field
+// doc comment) makes "exactly one dispatch wins" a property of mutual
+// exclusion instead of a racy read. The guard is held only for this
+// fast, DB-only step — released well before the long-running
+// executor.Prompt call — matching every other guard holder's
+// bounded-hold-time contract in this file.
+//
+// Every return path *before* this step (invalid session, not promptable,
+// ensureSessionRunning failure, a model switch) never claims anything —
+// those are covered by executeQueuedMessage's own deferred cleanup
+// instead (clearQueuedDispatchInFlightIfCurrent), which is safe since
+// none of them block on an agent turn.
+func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prompt string, model string, planMode bool, attachments []v1.MessageAttachment, dispatchOnly bool, claimEntryID string) (*PromptResult, error) {
 	s.logger.Debug("PromptTask called",
 		zap.String("task_id", taskID),
 		zap.String("session_id", sessionID),
@@ -2265,16 +2307,8 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
 		session = readySession
 	}
 
-	// Inject config context for config-mode sessions (dedicated settings chat, not plan mode)
-	effectivePrompt := prompt
-	if cm, ok := session.Metadata["config_mode"].(bool); ok && cm {
-		effectivePrompt = sysprompt.InjectConfigContext(sessionID, prompt)
-	}
-
-	// Inject plan mode prefix for follow-up messages in plan mode sessions.
-	if planMode {
-		effectivePrompt = sysprompt.InjectPlanMode(effectivePrompt)
-	}
+	// Apply config-mode and plan-mode prompt transforms.
+	effectivePrompt := s.effectivePromptForSession(sessionID, prompt, planMode, session)
 
 	// Ensure the agent process is actually running. After a lazy backend restart,
 	// the session may be in WAITING_FOR_INPUT but no agent process exists yet.
@@ -2294,13 +2328,7 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
 	if reloaded, err := s.repo.GetTaskSession(ctx, sessionID); err == nil && reloaded != nil {
 		session = reloaded
 		// Re-apply transforms in case metadata changed during ensureSessionRunning.
-		effectivePrompt = prompt
-		if cm, ok := session.Metadata["config_mode"].(bool); ok && cm {
-			effectivePrompt = sysprompt.InjectConfigContext(sessionID, prompt)
-		}
-		if planMode {
-			effectivePrompt = sysprompt.InjectPlanMode(effectivePrompt)
-		}
+		effectivePrompt = s.effectivePromptForSession(sessionID, prompt, planMode, session)
 	}
 
 	// Check if model switching is requested
@@ -2315,7 +2343,11 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
 	// the pre-injection prompt; PromptTask re-applies config/plan transforms.
 	s.rememberTurnPrompt(sessionID, prompt, model, planMode, attachments)
 
-	s.setSessionRunning(ctx, taskID, sessionID, session)
+	claimedSession, err := s.claimSessionRunningForPrompt(ctx, taskID, sessionID, claimEntryID, session)
+	if err != nil {
+		return nil, err
+	}
+	session = claimedSession
 	s.startTurnForSession(ctx, sessionID)
 
 	// Use context.WithoutCancel to prevent WebSocket request timeout from canceling the prompt.
@@ -2324,17 +2356,7 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
 	promptCtx := context.WithoutCancel(ctx)
 	result, err := s.executor.Prompt(promptCtx, taskID, sessionID, effectivePrompt, attachments, dispatchOnly, session)
 	if err != nil {
-		if resumedForPrompt && errors.Is(err, executor.ErrExecutionNotFound) {
-			s.logger.Warn("prompt after lazy resume hit missing execution; falling back to fresh launch",
-				zap.String("task_id", taskID),
-				zap.String("session_id", sessionID))
-			if freshErr := s.fallbackFreshLaunchOnMissingExecution(ctx, taskID, sessionID, prompt, planMode, nil, attachments); freshErr == nil {
-				return &PromptResult{}, nil
-			} else {
-				err = freshErr
-			}
-		}
-		return nil, s.handlePromptError(ctx, taskID, sessionID, previousSessionState, err)
+		return s.handlePromptDispatchFailure(ctx, taskID, sessionID, prompt, planMode, resumedForPrompt, attachments, previousSessionState, err)
 	}
 	return &PromptResult{
 		StopReason:   result.StopReason,
@@ -2342,6 +2364,102 @@ func (s *Service) PromptTask(ctx context.Context, taskID, sessionID string, prom
 	}, nil
 }
 
+// handlePromptDispatchFailure handles a failed executor.Prompt call from
+// promptTask. If the failure is the specific "missing execution" error a
+// just-resumed session can hit (see promptTask's resumedForPrompt comment),
+// it falls back to a fresh launch instead of surfacing the error to the
+// caller. Otherwise — or if that fallback launch itself fails — it
+// delegates to handlePromptError for the caller-facing result.
+func (s *Service) handlePromptDispatchFailure(ctx context.Context, taskID, sessionID, prompt string, planMode, resumedForPrompt bool, attachments []v1.MessageAttachment, previousSessionState models.TaskSessionState, promptErr error) (*PromptResult, error) {
+	if resumedForPrompt && errors.Is(promptErr, executor.ErrExecutionNotFound) {
+		s.logger.Warn("prompt after lazy resume hit missing execution; falling back to fresh launch",
+			zap.String("task_id", taskID),
+			zap.String("session_id", sessionID))
+		if freshErr := s.fallbackFreshLaunchOnMissingExecution(ctx, taskID, sessionID, prompt, planMode, nil, attachments); freshErr == nil {
+			return &PromptResult{}, nil
+		} else {
+			promptErr = freshErr
+		}
+	}
+	return nil, s.handlePromptError(ctx, taskID, sessionID, previousSessionState, promptErr)
+}
+
+// effectivePromptForSession applies promptTask's config-mode and plan-mode
+// prompt transforms to the raw prompt, in that order, using session's
+// *current* Metadata. Each call always starts from the raw prompt argument
+// (never a previously-transformed result), so it is safe for promptTask to
+// call this twice — once on session load and again after a reload that
+// follows ensureSessionRunning, in case metadata changed in between —
+// without double-prepending either transform's system-prompt wrapper.
+func (s *Service) effectivePromptForSession(sessionID, prompt string, planMode bool, session *models.TaskSession) string {
+	effectivePrompt := prompt
+	if cm, ok := session.Metadata["config_mode"].(bool); ok && cm {
+		effectivePrompt = sysprompt.InjectConfigContext(sessionID, prompt)
+	}
+	if planMode {
+		effectivePrompt = sysprompt.InjectPlanMode(effectivePrompt)
+	}
+	return effectivePrompt
+}
+
+// claimSessionRunningForPrompt marks sessionID RUNNING immediately before
+// promptTask dispatches to the agent, and returns the session to use for
+// the rest of the call (possibly reloaded).
+//
+// When claimEntryID is empty (a direct, non-queued prompt) this is a bare,
+// unguarded transition — preserving PromptTask's pre-existing direct-call
+// behavior as-is; this branch takes no guard and makes no claim about
+// races with a concurrent queued-dispatch decision on the same session.
+//
+// When claimEntryID is set (a queued dispatch claiming its reserved token —
+// see the Service.dispatchingQueued field doc comment), the claim happens
+// under the same per-session cancelInFlightGuard every other cancel/take-
+// and-dispatch decision in this file serializes through. Re-verifying
+// ownership *and* promptability inside this same critical section,
+// immediately before writing RUNNING, closes a gap that checking the token
+// alone would leave open: nothing would otherwise stop this call from
+// blindly marking the session RUNNING over top of a state some other
+// guard-synchronized decision-maker (a cancel, another claim) already
+// changed since promptTask's much-earlier, unguarded checkSessionPromptable
+// call near its top. Reloading the session and re-running
+// checkSessionPromptable here makes "is it actually still safe to mark this
+// session RUNNING right now" an atomic fact, not a stale read.
+func (s *Service) claimSessionRunningForPrompt(ctx context.Context, taskID, sessionID, claimEntryID string, session *models.TaskSession) (*models.TaskSession, error) {
+	if claimEntryID == "" {
+		s.setSessionRunning(ctx, taskID, sessionID, session)
+		return session, nil
+	}
+
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
+
+	if !s.isCurrentQueuedDispatch(sessionID, claimEntryID) {
+		return nil, errQueuedDispatchSuperseded
+	}
+	freshSession, sessErr := s.repo.GetTaskSession(ctx, sessionID)
+	if sessErr != nil {
+		return nil, fmt.Errorf("reload session before claiming queued dispatch: %w", sessErr)
+	}
+	if freshSession == nil {
+		return nil, errQueuedDispatchSuperseded
+	}
+	if promptErr := s.checkSessionPromptable(taskID, sessionID, freshSession.State); promptErr != nil {
+		return nil, promptErr
+	}
+	s.setSessionRunning(ctx, taskID, sessionID, freshSession)
+	// Clear the token now, inside the same lock, rather than deferring to
+	// executeQueuedMessage's cleanup after this entire (potentially
+	// long-blocking) call returns: from this point session.State is the
+	// authoritative "busy" signal for this dispatch, so holding the marker
+	// any longer would only risk the natural agent.ready firing when *this*
+	// turn completes seeing it still set and skipping the *next* queued
+	// entry — see the Service.dispatchingQueued field doc comment.
+	s.clearQueuedDispatchInFlightIfCurrent(sessionID, claimEntryID)
+	return freshSession, nil
+}
+
 // checkSessionPromptable returns nil when the session's state accepts a new
 // prompt. RUNNING is rejected with ErrAgentPromptInProgress; any other
 // non-acceptable state (STARTING / CREATED / FAILED / CANCELLED) is rejected
@@ -2520,6 +2638,23 @@ func (s *Service) DrainQueuedMessage(ctx context.Context, sessionID string) (boo
 	if sessionID == "" {
 		return false, fmt.Errorf("session_id is required")
 	}
+	// Acquire the guard *before* checking promptability, not after: check-
+	// then-lock would leave a gap where a concurrent
+	// QueueAndInterruptForPeerMessage (or another drain) changes the
+	// session's state between this call's check and its take, letting a
+	// manual drain request race a parent interrupt for the same entry —
+	// see the Service.cancelInFlight field doc comment. A blocking Lock
+	// (not a TryLock skip) is deliberate: unlike handleAgentReady /
+	// handleAgentBootReady (where a losing side can rely on the winning
+	// side's own take-and-dispatch, or a future turn's own agent.ready, to
+	// retry), a manual drain request has no other future trigger —
+	// skipping here could strand an already-queued message instead of
+	// just reporting an accurate "not promptable right now".
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
+
 	session, err := s.repo.GetTaskSession(ctx, sessionID)
 	if err != nil {
 		return false, fmt.Errorf("failed to get session: %w", err)
@@ -2527,12 +2662,74 @@ func (s *Service) DrainQueuedMessage(ctx context.Context, sessionID string) (boo
 	if err := s.checkSessionPromptable(session.TaskID, sessionID, session.State); err != nil {
 		return false, err
 	}
-	return s.drainQueuedMessageForPromptableSession(ctx, sessionID), nil
+	return s.drainQueuedMessageForPromptableSessionLocked(ctx, sessionID), nil
 }
 
+// cancelInFlightGuard is a per-session mutex serializing cancel/interrupt/
+// queue-take decisions (see the Service.cancelInFlight field doc comment),
+// paired with a reference count so the registry can safely reclaim the
+// entry once every acquirer has released it — refs is guarded by
+// Service.cancelInFlightMu, never by mu itself, since mu can be held for
+// the caller's whole critical section while refs bookkeeping must stay
+// brief and independent of that.
+type cancelInFlightGuard struct {
+	mu   sync.Mutex
+	refs int
+}
+
+// acquireCancelInFlightGuard returns the shared per-session mutex guarding
+// cancel/interrupt/queue-take decisions for sessionID, creating one if it
+// doesn't exist yet and registering the caller's reference to it. Callers
+// MUST call the returned release func exactly once when done with the
+// mutex — whether or not they actually acquired it (e.g. a TryLock that
+// returned false still needs release to drop its reference). Pairing every
+// acquire with a release is what keeps cancelInFlight bounded by
+// concurrently-active sessions: once refs drops to zero the entry is
+// deleted from the map instead of accumulating one permanent entry per
+// session ever created.
+func (s *Service) acquireCancelInFlightGuard(sessionID string) (*sync.Mutex, func()) {
+	s.cancelInFlightMu.Lock()
+	guard, ok := s.cancelInFlight[sessionID]
+	if !ok {
+		guard = &cancelInFlightGuard{}
+		if s.cancelInFlight == nil {
+			s.cancelInFlight = make(map[string]*cancelInFlightGuard)
+		}
+		s.cancelInFlight[sessionID] = guard
+	}
+	guard.refs++
+	s.cancelInFlightMu.Unlock()
+
+	var released bool
+	release := func() {
+		if released {
+			return
+		}
+		released = true
+		s.cancelInFlightMu.Lock()
+		guard.refs--
+		if guard.refs == 0 {
+			delete(s.cancelInFlight, sessionID)
+		}
+		s.cancelInFlightMu.Unlock()
+	}
+	return &guard.mu, release
+}
+
+// isCancelInFlight reports whether sessionID's cancelInFlight guard is
+// currently held by someone else, without itself claiming it — a
+// TryLock-then-immediately-Unlock peek rather than a real acquisition. The
+// guard entry this creates to perform the peek is released (and pruned if
+// nothing else references it) before this returns, so a passive readiness
+// probe never leaves permanent state behind.
 func (s *Service) isCancelInFlight(sessionID string) bool {
-	_, ok := s.cancelInFlight.Load(sessionID)
-	return ok
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	if lock.TryLock() {
+		lock.Unlock()
+		return false
+	}
+	return true
 }
 
 // CancelAgent interrupts the current agent turn without terminating the process,
@@ -2553,12 +2750,21 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
 	// "Turn cancelled by user" message and races on turn cleanup — the second
 	// call's getActiveTurnID lazily starts a phantom turn after the first call
 	// already closed the real one.
-	if _, busy := s.cancelInFlight.LoadOrStore(sessionID, struct{}{}); busy {
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	if !lock.TryLock() {
 		s.logger.Debug("cancel already in flight; skipping duplicate",
 			zap.String("session_id", sessionID))
 		return nil
 	}
-	defer s.cancelInFlight.Delete(sessionID)
+	unlocked := false
+	unlockGuard := func() {
+		if !unlocked {
+			unlocked = true
+			lock.Unlock()
+		}
+	}
+	defer unlockGuard()
 
 	// Fetch session for state updates and message creation
 	session, err := s.repo.GetTaskSession(ctx, sessionID)
@@ -2635,29 +2841,253 @@ func (s *Service) CancelAgent(ctx context.Context, sessionID string) error {
 	// concurrent agent.complete event having already closed the turn.
 	s.completeTurnForSession(ctx, sessionID)
 
-	// Clear the duplicate-cancel guard early so drainQueuedMessageForPromptableSession
-	// can dispatch the queued prompt without being blocked by the in-flight sentinel.
-	// The defer at function entry remains as the error-path safety net.
-	s.cancelInFlight.Delete(sessionID)
-
 	// Deliver any message the operator queued while the turn was running
-	// (#1597 pause→resume recovery). On a normal cancel the agent emits
-	// complete(cancelled) → handleAgentReady → on_turn_complete, which drains
-	// the queue. But an escalated cancel (agent hung) or a dead-process cancel
-	// fires no agent.ready event, so nothing would drain it — leaving the
-	// operator's queued message stranded until a second manual send. Draining
-	// here after the turn is completed covers those paths. It is idempotent with
-	// the event-driven drain: drainQueuedMessageForPromptableSession pops via
-	// TakeQueued atomically, so a normal cancel that also fires handleAgentReady
-	// cannot double-deliver.
+	// (#1597 pause→resume recovery) *while still holding the guard* — this
+	// used to release the guard early and drain unguarded, which let a
+	// concurrent QueueAndInterruptForPeerMessage (or another drain) race
+	// this exact take-and-dispatch decision for the same session; every
+	// take-and-dispatch decision must serialize through this one guard
+	// (see the Service.cancelInFlight field doc comment). On a normal
+	// cancel the agent emits complete(cancelled) → handleAgentReady →
+	// on_turn_complete, which drains the queue. But an escalated cancel
+	// (agent hung) or a dead-process cancel fires no agent.ready event, so
+	// nothing would drain it — leaving the operator's queued message
+	// stranded until a second manual send. Draining here after the turn is
+	// completed covers those paths. It is idempotent with the event-driven
+	// drain: drainQueuedMessageForPromptableSessionLocked pops via
+	// TakeQueued atomically, so a normal cancel that also fires
+	// handleAgentReady cannot double-deliver.
 	if session != nil {
-		s.drainQueuedMessageForPromptableSession(ctx, sessionID)
+		s.drainQueuedMessageForPromptableSessionLocked(ctx, sessionID)
 	}
 
+	// Release the cancel/interrupt guard now that this session's
+	// cancel-and-drain decision is fully resolved. Both unlockGuard and
+	// release are idempotent, so the deferred calls above become safe
+	// no-op error-path safety nets once this fires.
+	unlockGuard()
+	release()
+
 	s.logger.Debug("agent turn cancelled", zap.String("session_id", sessionID))
 	return nil
 }
 
+// takeAndDispatchEntryLocked takes entryID from sessionID's queue and
+// dispatches it, falling back to draining the FIFO head only if the
+// targeted take doesn't find it (a benign not-found, e.g. already taken by
+// a concurrent path). A genuine repository error on the targeted take is
+// propagated instead of falling back — see cancelAndTakeForPeerMessage's
+// doc comment for why. The caller must already hold sessionID's
+// cancelInFlight lock; this neither acquires nor releases it.
+//
+// Deliberately does NOT check isQueuedDispatchInFlight itself: callers
+// that reach here have either just performed a cancel
+// (cancelAndTakeForPeerMessage) — which supersedes whatever another
+// dispatch was in the middle of settling, so a stale in-flight marker from
+// that dispatch must not block taking a *different* entry here — or have
+// already confirmed the session is genuinely idle (takeIfPromptableLocked,
+// which checks the marker itself before delegating here). See
+// drainQueuedMessageForPromptableSessionLocked for the *non-cancelling*
+// FIFO-head drain path, which does check the marker directly.
+func (s *Service) takeAndDispatchEntryLocked(ctx context.Context, sessionID, entryID string) (bool, error) {
+	if s.messageQueue != nil {
+		queuedMsg, ok, err := s.messageQueue.TakeQueuedEntry(ctx, sessionID, entryID)
+		if err != nil {
+			return false, fmt.Errorf("take targeted queued message: %w", err)
+		}
+		if s.dispatchTakenQueuedMessage(ctx, sessionID, queuedMsg, ok) {
+			return true, nil
+		}
+	}
+	return s.drainQueuedMessageForPromptableSessionLocked(ctx, sessionID), nil
+}
+
+// takeIfPromptableLocked takes and dispatches entryID only if sessionID is
+// currently promptable, without ever issuing a cancel — used by
+// QueueAndInterruptForPeerMessage when cancelling would risk hitting a
+// *different*, unrelated turn instead of the one the caller meant to
+// interrupt (see its active-turn revalidation doc comment). Returns
+// (false, nil), never an error, when the session turns out not to be
+// promptable — or when a *different* dispatch is already settling for
+// this session (isQueuedDispatchInFlight; see the Service.dispatchingQueued
+// field doc comment) — mirroring cancelAndTakeForPeerMessage's own "cancel
+// failed but already promptable" recovery except without ever attempting
+// the cancel. Unlike cancelAndTakeForPeerMessage's path, this never
+// cancels anything, so it has no way to supersede a settling dispatch the
+// way a genuine cancel would — it must defer to it instead. The caller
+// must already hold sessionID's cancelInFlight lock.
+func (s *Service) takeIfPromptableLocked(ctx context.Context, taskID, sessionID, entryID string) (bool, error) {
+	if s.isQueuedDispatchInFlight(sessionID) {
+		return false, nil
+	}
+	session, err := s.repo.GetTaskSession(ctx, sessionID)
+	if err != nil || session == nil || s.checkSessionPromptable(taskID, sessionID, session.State) != nil {
+		return false, nil
+	}
+	return s.takeAndDispatchEntryLocked(ctx, sessionID, entryID)
+}
+
+// cancelAndTakeForPeerMessage cancels sessionID's in-flight turn and takes
+// and dispatches entryID — the specific message that triggered this
+// interrupt — falling back to draining the FIFO head only if the targeted
+// take doesn't find it (a benign not-found, e.g. already taken by a
+// concurrent path). entryID must be non-empty: the only caller,
+// QueueAndInterruptForPeerMessage, always passes the entry it just
+// inserted. A genuine repository error on the targeted take is propagated
+// instead of falling back: an error says nothing about which entry is
+// actually at the FIFO head, and dispatching it anyway would risk
+// delivering the wrong message while the caller still reports "sent" for
+// the parent's.
+//
+// The caller must already hold sessionID's cancelInFlight lock; this
+// neither acquires nor releases it — see QueueAndInterruptForPeerMessage.
+//
+// The returned bool reports whether this call actually dispatched a
+// message; a false result means the caller's message is still only
+// queued, to be delivered later by whichever drain gets to it.
+//
+// Deliberately mirrors cancelAgentSilent + dispatchTakenQueuedMessage rather
+// than delegating to CancelAgent: the interrupt is an internal steering
+// signal from the parent, not a user action, so unlike the cancel button it
+// must not write a visible "Turn cancelled" message and must not move the
+// task to REVIEW (writeTaskReviewStateOnCancel).
+func (s *Service) cancelAndTakeForPeerMessage(ctx context.Context, taskID, sessionID, entryID string) (bool, error) {
+	if cancelErr := s.cancelAgentSilent(ctx, taskID, sessionID); cancelErr != nil {
+		// A genuine (non-tolerated — cancelAgentSilent already swallows "no
+		// active execution") cancel failure leaves session state untouched
+		// by this call: cancelAgentSilent returns before reaching its own
+		// updateTaskSessionState/completeTurnForSession reconciliation on
+		// that path. Every other decision-maker that could independently
+		// mark the session promptable — handleAgentReady,
+		// handleAgentBootReady, CancelAgent, clarification recovery — now
+		// serializes through this same cancelInFlight guard (see the
+		// Service.cancelInFlight field doc comment), so none of them can be
+		// racing this exact call while we hold it. This recheck is a
+		// defensive backstop for any not-yet-guarded completion path or a
+		// stale read further up the call chain: if the session turns out to
+		// already be promptable despite the cancel error, take-and-dispatch
+		// anyway instead of leaving the message stranded with nothing left
+		// to trigger it.
+		session, sessErr := s.repo.GetTaskSession(ctx, sessionID)
+		if sessErr != nil || session == nil || s.checkSessionPromptable(taskID, sessionID, session.State) != nil {
+			return false, fmt.Errorf("interrupt for peer message: %w", cancelErr)
+		}
+		s.logger.Warn("cancel failed but session is already promptable; taking queued message directly instead of relying on a future drain",
+			zap.String("task_id", taskID),
+			zap.String("session_id", sessionID),
+			zap.Error(cancelErr))
+	}
+	return s.takeAndDispatchEntryLocked(ctx, sessionID, entryID)
+}
+
+// QueueAndInterruptForPeerMessage atomically queues prompt for sessionID
+// and then interrupts the child's in-flight turn to deliver it immediately,
+// instead of waiting for the turn to end naturally. It is used only by
+// handleMessageTask (mcp/handlers) when the sender is the target task's
+// parent: a long-running child otherwise leaves its parent's control/steer
+// messages parked on the FIFO queue until the child's current turn finishes
+// on its own, and with several such children running in parallel the
+// per-session queue can fill up to messagequeue.DefaultMaxPerSession before
+// any of them are delivered.
+//
+// This must be one atomic operation rather than "queue, then separately
+// interrupt" (the original two-step split across the handlers/orchestrator
+// boundary): between an insert becoming visible and a later, separate
+// interrupt call acquiring the session's cancelInFlight lock, the child's
+// turn can complete naturally and handleAgentReady's normal FIFO drain can
+// grab the just-queued entry and start dispatching it as an ordinary turn —
+// only for the later interrupt's cancel to land on and kill that very turn,
+// orphaning the parent's message mid-delivery. Taking the lock before the
+// queue insert closes that: handleAgentReady and handleAgentBootReady's
+// take-decision claims (TryLock, not a passive peek) the exact same lock
+// before their own take, so neither can ever observe — and steal — an
+// entry this call is still in the process of queueing-and-claiming.
+//
+// The lock is a genuine, blocking claim (Lock, mirroring
+// executor.Executor.getSessionLock's precedent for per-session mutual
+// exclusion in this codebase) — not a try-once peek with a fallback
+// unguarded insert. Working around a busy lock with an unguarded insert
+// would leave that insert visible to nobody in particular: the current
+// holder may have already finished its own take-or-skip decision before
+// the insert lands, and nothing then guarantees a future drain (the
+// session can go idle with no further agent.ready to retry on). Every
+// existing holder already bounds its own hold time through an independent
+// mechanism (cancelAgentSilent's underlying agent-cancel escalation
+// timeout, or a single fast DB round-trip in the natural-drain paths), so
+// blocking here adds no new deadlock risk.
+//
+// Active-turn revalidation: after acquiring the lock and queueing the
+// message, this re-checks whether the turn active for sessionID is still
+// the one that was active when the caller decided to interrupt (snapshotted
+// via peekActiveTurnID *before* contending for the lock). If a *different*
+// turn is now active — e.g. a workflow on_turn_complete transition
+// auto-started a successor for this same session while this call waited
+// for the guard — cancelling now would kill that unrelated successor turn
+// instead of the one the parent meant to interrupt. In that case this
+// falls back to takeIfPromptableLocked (dispatch directly only if the
+// session happens to already be idle *and* no other dispatch is
+// concurrently settling for it, per isQueuedDispatchInFlight — see the
+// Service.dispatchingQueued field doc comment; otherwise leave the
+// message queued for whichever turn is actually running to drain
+// naturally) instead of ever calling cancelAndTakeForPeerMessage. The
+// turn-identity comparison only trusts a positively-confirmed match — both
+// snapshots read without error, both non-empty, and equal — before it
+// will risk a cancel; any error or mismatch (including "no turn observed
+// at all") takes the safe fallback. The one exception is turnService
+// being unset entirely: with no turn identity obtainable at all, this
+// preserves the exact unconditional-interrupt behavior every existing
+// (turn-unaware) caller and test exercises.
+//
+// Note this deliberately does *not* also gate on isQueuedDispatchInFlight
+// when the turn *is* confirmed unchanged: a cancel (via
+// cancelAndTakeForPeerMessage below) supersedes whatever another dispatch
+// for this same session was in the middle of settling — e.g. a second
+// parent message arriving right behind a first one whose own dispatch
+// hasn't yet reached PromptTask — so it must still be allowed to cancel
+// and take its own entry rather than being blocked by a marker the cancel
+// itself is about to invalidate.
+func (s *Service) QueueAndInterruptForPeerMessage(ctx context.Context, taskID, sessionID, prompt string, metadata map[string]interface{}) (*messagequeue.QueuedMessage, bool, error) {
+	if s.messageQueue == nil {
+		return nil, false, errors.New("message queue not available")
+	}
+
+	turnBeforeWait, turnPeekErr := s.peekActiveTurnID(ctx, sessionID)
+	if turnPeekErr != nil {
+		s.logger.Warn("failed to snapshot active turn before peer-message interrupt; will fall back to a promptable-only dispatch instead of risking a cancel of unrelated work",
+			zap.String("task_id", taskID),
+			zap.String("session_id", sessionID),
+			zap.Error(turnPeekErr))
+	}
+
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	defer release()
+	lock.Lock()
+	defer lock.Unlock()
+
+	queued, err := s.messageQueue.QueueMessageWithMetadata(ctx, sessionID, taskID, prompt, "", messagequeue.QueuedByAgent, false, nil, metadata)
+	if err != nil {
+		return nil, false, err
+	}
+	s.publishQueueStatusEvent(ctx, sessionID)
+
+	if s.turnService != nil {
+		turnNow, turnNowErr := s.peekActiveTurnID(ctx, sessionID)
+		confirmedUnchanged := turnPeekErr == nil && turnNowErr == nil && turnBeforeWait != "" && turnNow == turnBeforeWait
+		if !confirmedUnchanged {
+			s.logger.Warn("active turn could not be confirmed unchanged while waiting to interrupt; dispatching only if already idle instead of risking a cancel of unrelated work",
+				zap.String("task_id", taskID),
+				zap.String("session_id", sessionID),
+				zap.String("turn_before_wait", turnBeforeWait),
+				zap.Bool("turn_peek_error", turnPeekErr != nil || turnNowErr != nil))
+			dispatched, fallbackErr := s.takeIfPromptableLocked(ctx, taskID, sessionID, queued.ID)
+			return queued, dispatched, fallbackErr
+		}
+	}
+
+	dispatched, err := s.cancelAndTakeForPeerMessage(ctx, taskID, sessionID, queued.ID)
+	return queued, dispatched, err
+}
+
 // CompleteTask explicitly completes a task and stops all its agents
 func (s *Service) CompleteTask(ctx context.Context, taskID string) error {
 	s.logger.Info("completing task",

docs/specs/INDEX.md

@@ -51,6 +51,7 @@ Kandev's task model: documents, execution stages, labels, blocker escalation, su
 | [wip-limit-pull-system](tasks/wip-limit-pull-system.md) | building |
 | [multi-branch](tasks/multi-branch/spec.md) | shipped |
 | [quick-chat-expiration](tasks/quick-chat-expiration.md) | draft |
+| [parent-child-message-interrupt](tasks/parent-child-message-interrupt.md) | shipped |
 
 ## agents/ — agent governance
 

docs/specs/tasks/parent-child-message-interrupt.md

@@ -0,0 +1,227 @@
+---
+status: shipped
+created: 2026-07-13
+owner: cfl
+---
+
+# Parent-Child Message Interrupt
+
+## Why
+
+A parent task's `message_task_kandev` call to a busy child (session `RUNNING`/`STARTING`)
+is queued and delivered at the child's next turn boundary, like any other peer message.
+For a long-running child, that boundary can be minutes or hours away — so a parent's
+steer/stop message (`"stop and re-check X"`, `"the requirements changed"`) sits behind
+the child's current turn the whole time, with several concurrent children compounding the
+delay.
+
+An earlier round of this feature made interruption automatic: any parent-to-child message
+cancelled the child's in-flight turn and delivered immediately, purely because the sender
+happened to be the target's parent. Review pushback (PR #1653) rejected inferring interrupt
+intent from the relationship alone: not every parent message is a stop/steer command — a
+parent might send an FYI note that should wait its turn like anything else — so cancelling
+a turn must be a deliberate, explicit choice made per call, not a side effect of who is
+calling.
+
+## What
+
+### `delivery_mode` request parameter
+
+`message_task_kandev` takes an optional `delivery_mode` string: `"queued"` (default) or
+`"interrupt"`.
+
+- **`"queued"` (or omitted):** the message is appended to the target session's FIFO queue
+  and delivered when the target's current turn ends — the same behavior every other peer
+  message gets, regardless of whether the sender is the target's parent. This is a real
+  behavior change from the automatic-interrupt round: a parent messaging a busy child
+  without opting in no longer interrupts.
+- **`"interrupt"`:** the message is queued *and* the target's current turn is cancelled so
+  the message is delivered immediately instead of waiting for the turn to end naturally.
+
+`delivery_mode` only affects dispatch when the target session is `RUNNING` or `STARTING`.
+For every other session state (`WAITING_FOR_INPUT`, `COMPLETED`, `CREATED`) the message is
+delivered through its normal path (prompt-with-resume or session start) regardless of
+`delivery_mode` — there is no in-flight turn to interrupt.
+
+### Authorization: parent-only, hard-rejected otherwise
+
+`delivery_mode="interrupt"` is honored only when the sender is the target task's **direct**
+parent (`targetTask.ParentID == senderTask.ID`). This is the same relationship check the
+automatic-interrupt round used — it now gates an explicit request instead of driving an
+implicit one.
+
+A non-parent sender that explicitly requests `"interrupt"` gets a **hard rejection**, not a
+silent downgrade to `"queued"`:
+
+- Response: WS error, code `FORBIDDEN`, message containing `"direct parent"`.
+- The rejection has **no side effect** — nothing is queued, dispatched, or interrupted. The
+  authorization check runs before the queue insert, so a rejected request never reaches the
+  message queue.
+
+This is a deliberate design choice: silently reinterpreting a rejected `"interrupt"` request
+as `"queued"` would misreport what happened and hide caller misuse from the agent that made
+the request. The caller needs to know its request was rejected outright, not quietly
+reinterpreted — if it still wants the message delivered, it can resend with
+`delivery_mode="queued"` (or omit the field).
+
+Any `delivery_mode` value other than `"queued"`/`"interrupt"`/empty is rejected up front —
+code `VALIDATION_ERROR` — before any task or session lookup runs.
+
+### Status reporting
+
+The response `status` field reflects what actually happened, not just what was requested:
+
+- `"queued"`: the message is sitting in the FIFO queue, waiting for a turn boundary. This is
+  the outcome for every `delivery_mode="queued"` call, and also the outcome for an
+  `delivery_mode="interrupt"` call whose cancel-and-take step ran but did not end up
+  dispatching anything immediately (queued successfully, interrupt not needed or its
+  delivery attempt didn't land — see "Composition" below).
+- `"sent"`: `delivery_mode="interrupt"` actually cancelled the target's turn and dispatched
+  the message right away.
+- `"started"`: the target session was `CREATED` and got launched with this message as its
+  first prompt (`delivery_mode` has no effect on this path).
+
+A failure in the interrupt's cancel step, once the message is already safely queued, is
+never surfaced as an error to the caller: the message will still be delivered by the
+target's normal turn-completion drain, so the interrupt is purely a latency optimization on
+top of that always-safe default. The call reports `"queued"` in that case (an accurate
+description of the actual outcome) instead of an error the caller has no useful recovery
+action for — retrying would enqueue a duplicate, since queuing is not idempotent. The
+failure is still logged server-side.
+
+### Composition with the FIFO queue and turn-safety guarantees
+
+`delivery_mode="interrupt"` queues the message and cancels-and-takes it in one atomic
+orchestrator call, not two separate steps: queuing followed by a separate interrupt call
+would leave a window where the target's turn could complete naturally, the ordinary FIFO
+drain could pick up the just-queued entry and start dispatching it as a normal turn, and the
+interrupt's later cancel could then land on and kill that very turn — orphaning the parent's
+message mid-delivery. Atomicity here means the queue insert and the cancel-and-take decision
+happen while holding the same per-session serialization point described next.
+
+Every path that can cancel a session's active turn or take-and-dispatch its next queued
+message — the parent interrupt, a manual or workflow-triggered queue drain, CI-automation
+drains, and clarification-timeout recovery — serializes through one per-session guard. Two
+such operations targeting the same session never interleave: whichever acquires the guard
+first completes its cancel/take-and-dispatch before the other proceeds. This guarantees at
+most one dispatch per queued entry (no double dispatch) and no entry is ever silently
+dropped — but it does *not* guarantee that one of the two racing callers itself performs an
+immediate delivery. The loser backs off once it observes the winner already changed the
+session/turn state; depending on what the winner was, backing off means either "my message
+was already delivered by the other side" (the winner was a drain/recovery that delivered
+the very entry the loser was also trying to take) or "my message stays safely queued for a
+later natural drain" (the winner started a genuinely different, unrelated turn — e.g.
+clarification-timeout recovery — that the loser correctly does not cancel into; see
+`QueueAndInterruptForPeerMessage`'s "unrelated successor" handling). Either way, the entry
+is accounted for exactly once: dispatched, or left queued for its own future turn boundary.
+
+The guard is acquired *before* any turn-completion bookkeeping runs (not just around the
+final queue-drain decision), and — whenever the orchestrator has turn identity tracking
+wired — bookkeeping re-validates that the turn it is about to complete is still the same one
+that was active when the event fired, once the guard is actually held. A ready event that
+raced a parent interrupt and lost cannot complete (or transition the workflow for) a turn the
+interrupt already cancelled and replaced; it detects the turn changed out from under it and
+backs off instead, leaving that turn's own eventual completion to whatever superseded it.
+Conversely, if the interrupt's own cancel genuinely fails while a ready event was blocked
+behind the same guard, the message is not stranded: the still-pending ready event finds the
+turn exactly as it left it (not stale) and delivers the message itself through the ordinary
+FIFO drain once it is unblocked.
+
+## API
+
+Request (`message_task_kandev` tool call → `message_task` WS action):
+
+```json
+{
+  "task_id": "<target task id>",
+  "prompt": "<message text>",
+  "sender_task_id": "<calling agent's task id>",
+  "sender_session_id": "<calling agent's session id>",
+  "delivery_mode": "queued"
+}
+```
+
+- `task_id`, `prompt`, `sender_task_id` are required. `sender_session_id` is optional
+  (used for message attribution). `delivery_mode` is optional; omitted is equivalent to
+  `"queued"`.
+
+Response:
+
+```json
+{
+  "task_id": "<target task id>",
+  "session_id": "<target session id>",
+  "status": "queued"
+}
+```
+
+`status` is one of `"queued"`, `"sent"`, `"started"` — see "Status reporting" above.
+
+### Errors
+
+| Condition | Code | Notes |
+|---|---|---|
+| `delivery_mode` not `"queued"`/`"interrupt"`/empty | `VALIDATION_ERROR` | Checked before any task/session lookup. |
+| `delivery_mode="interrupt"` from a non-parent sender | `FORBIDDEN` | Message contains `"direct parent"`. No side effect — nothing queued or dispatched. |
+| Target session `FAILED`/`CANCELLED` | `INTERNAL_ERROR` | Unrelated to `delivery_mode`; same for every peer message. |
+| Target queue full | `CONFLICT` (queue-full error) | Unrelated to `delivery_mode`. |
+
+## Scenarios
+
+**GIVEN** a child task's session is `RUNNING`, **WHEN** its parent calls `message_task_kandev`
+without `delivery_mode` (or with `delivery_mode="queued"`), **THEN** the message is appended
+to the FIFO queue, the child's current turn is left running untouched, and the response
+reports `status="queued"`.
+
+**GIVEN** a child task's session is `RUNNING`, **WHEN** its parent calls `message_task_kandev`
+with `delivery_mode="interrupt"`, **THEN** the message is queued and the child's current
+turn is cancelled in the same atomic call, the message is dispatched immediately, and the
+response reports `status="sent"`.
+
+**GIVEN** a task's session is `RUNNING`, **WHEN** a non-parent (sibling, unrelated, or the
+task's own child) sender calls `message_task_kandev` with `delivery_mode="interrupt"`,
+**THEN** the call is rejected with `FORBIDDEN` and no message is queued, dispatched, or
+otherwise delivered — the sender must resend without `delivery_mode="interrupt"` to get the
+message delivered.
+
+**GIVEN** any sender, **WHEN** `message_task_kandev` is called with an unrecognized
+`delivery_mode` value (e.g. `"immediately"`), **THEN** the call is rejected with
+`VALIDATION_ERROR` before any task or session lookup runs.
+
+**GIVEN** a parent's `delivery_mode="interrupt"` call is queuing and taking its own targeted
+entry while a concurrent manual/workflow-triggered drain is already in flight for the same
+session (about to take a *different*, already-queued sibling entry), **WHEN** both
+operations resolve, **THEN** at most one dispatch happens for each entry — the interrupt
+delivers its own entry, and the drain, having lost the race, backs off without
+double-dispatching or dropping the sibling entry (which stays queued for a later drain).
+
+**GIVEN** a parent's `delivery_mode="interrupt"` call is racing clarification-timeout
+recovery for the same session, **WHEN** the recovery wins the guard and starts a fresh,
+unrelated successor turn before the interrupt acquires it, **THEN** the interrupt does not
+cancel that successor turn — it defers instead, and the parent's message stays safely
+queued for the recovered turn's own eventual natural drain rather than being dropped or
+causing a second, competing dispatch.
+
+**GIVEN** a parent's `delivery_mode="interrupt"` call cancels and redispatches a session
+through a brand new turn, **WHEN** a `agent.ready` event for the *original* (now-cancelled)
+turn was already in flight when the interrupt started, **THEN** that ready event detects the
+turn changed once it can proceed and backs off without completing or transitioning the
+workflow for the interrupt's new turn.
+
+**GIVEN** a parent's `delivery_mode="interrupt"` call's own cancel step genuinely fails
+(not one of the tolerated "already idle"/"no active execution" cases) while a ready event
+for that same, still-active turn is blocked behind the same guard, **WHEN** the ready event
+is unblocked, **THEN** it completes the turn normally and delivers the parent's message
+itself via the ordinary FIFO drain — the message is not left stranded.
+
+## Out of scope
+
+- Interrupting a session on behalf of anyone other than its direct parent (siblings,
+  grandparents, unrelated tasks) — not supported at any `delivery_mode` value.
+- Cancelling more than the target's current turn — a single `delivery_mode="interrupt"` call
+  affects exactly one in-flight turn on one session.
+- Retrying or deduplicating a failed interrupt attempt automatically — the caller decides
+  whether to resend, and a resend queues a distinct message (queuing is not idempotent).
+- A `delivery_mode` value that queues *without* the possibility of a later natural interrupt,
+  or that interrupts without also queuing — the two are always coupled (interrupt implies
+  queue-then-cancel-and-take as one atomic step).