fix: avoid passthrough ready event deadlock (#3177) (#3181)

2026-08-31 12:27:18 +00:00
parent 6074f39a32
commit 1c7b5afce1
13 changed files with 800 additions and 23 deletions
@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"

 	"go.uber.org/zap"
@@ -25,27 +26,78 @@ import (

 var errPassthroughProcessReplaced = errors.New("passthrough process was replaced")

-// MarkPassthroughRunning marks a passthrough execution as running when user submits input.
-// This is called when Enter key is detected in the terminal handler.
-// It updates the execution status and publishes an AgentRunning event.
-func (m *Manager) MarkPassthroughRunning(sessionID string) error {
+// PreparePassthroughRunning marks a passthrough execution as running and
+// returns a one-shot callback that publishes the corresponding AgentRunning
+// event. Callers that hold a session serialization guard must invoke the
+// callback after releasing that guard because event subscribers may re-enter
+// it synchronously.
+func (m *Manager) PreparePassthroughRunning(sessionID string) (func(), error) {
 	execution, exists := m.executionStore.GetBySessionID(sessionID)
 	if !exists {
-		return fmt.Errorf("no agent execution found for session: %s", sessionID)
+		return nil, fmt.Errorf("no agent execution found for session: %s", sessionID)
 	}

-	if execution.PassthroughProcessID == "" {
-		return fmt.Errorf("session %s is not in passthrough mode", sessionID)
-	}
-
-	// Only publish if not already running (prevents duplicate events)
-	if execution.Status != v1.AgentStatusRunning {
-		if err := m.UpdateStatus(execution.ID, v1.AgentStatusRunning); err != nil {
-			return err
+	var payload AgentEventPayload
+	var updated *AgentExecution
+	var alreadyRunning bool
+	var validationErr error
+	if err := m.executionStore.WithLock(execution.ID, func(current *AgentExecution) {
+		// Revalidate the session mapping and passthrough mode while claiming the
+		// transition. The initial lookup only supplies the execution ID to lock.
+		if current.SessionID != sessionID {
+			validationErr = fmt.Errorf("no agent execution found for session: %s", sessionID)
+			return
+		}
+		if current.PassthroughProcessID == "" {
+			validationErr = fmt.Errorf("session %s is not in passthrough mode", sessionID)
+			return
 		}
-		m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution)
-	}

+		// Only publish if not already running (prevents duplicate events).
+		if current.Status == v1.AgentStatusRunning {
+			alreadyRunning = true
+			return
+		}
+		current.Status = v1.AgentStatusRunning
+		updated = current
+		// Capture the payload under the same lock as the status claim so a
+		// competing stop/failure cannot relabel the deferred event.
+		payload = newAgentEventPayload(current)
+	}); err != nil {
+		if errors.Is(err, ErrExecutionNotFound) {
+			return nil, fmt.Errorf("no agent execution found for session: %s", sessionID)
+		}
+		return nil, err
+	}
+	if validationErr != nil {
+		return nil, validationErr
+	}
+	if alreadyRunning {
+		return func() {}, nil
+	}
+	m.persistExecutorRunning(context.Background(), updated)
+
+	var publishOnce sync.Once
+	return func() {
+		publishOnce.Do(func() {
+			m.eventPublisher.publishAgentEventPayload(context.Background(), events.AgentRunning, payload)
+		})
+	}, nil
+}
+
+// MarkPassthroughRunning marks a passthrough execution as running when the user submits input
+// via the terminal handler. It is a convenience wrapper around PreparePassthroughRunning that
+// publishes the AgentRunning event immediately.
+//
+// Callers that hold the per-session cancellation guard (e.g. handleAgentReady) must use
+// PreparePassthroughRunning directly and defer the returned publisher until after the guard
+// is released, to avoid re-entering the mutex through the synchronous event subscriber.
+func (m *Manager) MarkPassthroughRunning(sessionID string) error {
+	publish, err := m.PreparePassthroughRunning(sessionID)
+	if err != nil {
+		return err
+	}
+	publish()
 	return nil
 }

@@ -16,6 +16,7 @@ import (
 	settingsmodels "github.com/kandev/kandev/internal/agent/settings/models"
 	agentctltypes "github.com/kandev/kandev/internal/agentctl/types"
 	"github.com/kandev/kandev/internal/events"
+	taskmodels "github.com/kandev/kandev/internal/task/models"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
 )

@@ -135,6 +136,152 @@ func TestMarkPassthroughRunningPublishesOnceAndGuards(t *testing.T) {
 		"an already-running execution must not publish a duplicate AgentRunning event")
 }

+func TestPreparePassthroughRunningDefersAndSnapshotsPublication(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	startedAt := time.Date(2026, 8, 31, 8, 0, 0, 0, time.UTC)
+	execution := &AgentExecution{
+		ID:                   "exec-pty",
+		RunID:                "run-before",
+		TaskID:               "task-before",
+		SessionID:            "session-pty",
+		AgentID:              "agent-before",
+		AgentProfileID:       "profile-before",
+		PassthroughProcessID: "pty-1",
+		Status:               v1.AgentStatusReady,
+		StartedAt:            startedAt,
+	}
+	require.NoError(t, mgr.executionStore.Add(execution))
+
+	preparer, ok := interface{}(mgr).(interface {
+		PreparePassthroughRunning(string) (func(), error)
+	})
+	require.True(t, ok, "the lifecycle manager must expose deferred passthrough preparation")
+	publish, err := preparer.PreparePassthroughRunning("session-pty")
+	require.NoError(t, err)
+	require.NotNil(t, publish)
+	require.Equal(t, v1.AgentStatusRunning, execution.Status)
+	require.Empty(t, eventBus.PublishedEvents, "preparation must not publish while the caller still owns its guard")
+
+	execution.RunID = "run-after"
+	execution.TaskID = "task-after"
+	execution.Status = v1.AgentStatusFailed
+
+	publish()
+	publish()
+
+	require.Len(t, eventBus.PublishedEvents, 1, "deferred publication must be one-shot")
+	payload, ok := eventBus.PublishedEvents[0].Event.Data.(AgentEventPayload)
+	require.True(t, ok)
+	require.Equal(t, "run-before", payload.RunID)
+	require.Equal(t, "task-before", payload.TaskID)
+	require.Equal(t, "session-pty", payload.SessionID)
+	require.Equal(t, string(v1.AgentStatusRunning), payload.Status)
+}
+
+type competingExecutionWriter struct {
+	mutate func()
+}
+
+func (w *competingExecutionWriter) GetExecutorRunningBySessionID(context.Context, string) (*taskmodels.ExecutorRunning, error) {
+	if w.mutate != nil {
+		w.mutate()
+	}
+	return nil, taskmodels.ErrExecutorRunningNotFound
+}
+
+func (*competingExecutionWriter) UpsertExecutorRunning(context.Context, *taskmodels.ExecutorRunning) error {
+	return nil
+}
+
+func (*competingExecutionWriter) DeleteExecutorRunningBySessionID(context.Context, string) error {
+	return nil
+}
+
+func (*competingExecutionWriter) RepairExecutorRunningDead(context.Context, string) error {
+	return nil
+}
+
+func TestPreparePassthroughRunningCapturesSnapshotBeforeCompetingMutation(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := &AgentExecution{
+		ID:                   "exec-pty",
+		RunID:                "run-before",
+		TaskID:               "task-before",
+		SessionID:            "session-pty",
+		PassthroughProcessID: "pty-1",
+		Status:               v1.AgentStatusReady,
+	}
+	require.NoError(t, mgr.executionStore.Add(execution))
+	mgr.SetExecutorRunningWriter(&competingExecutionWriter{
+		mutate: func() {
+			require.NoError(t, mgr.executionStore.WithLock(execution.ID, func(current *AgentExecution) {
+				current.RunID = "run-after"
+				current.TaskID = "task-after"
+				current.Status = v1.AgentStatusFailed
+			}))
+		},
+	})
+
+	publish, err := mgr.PreparePassthroughRunning(execution.SessionID)
+	require.NoError(t, err)
+	publish()
+
+	require.Len(t, eventBus.PublishedEvents, 1)
+	payload, ok := eventBus.PublishedEvents[0].Event.Data.(AgentEventPayload)
+	require.True(t, ok)
+	require.Equal(t, "run-before", payload.RunID)
+	require.Equal(t, "task-before", payload.TaskID)
+	require.Equal(t, string(v1.AgentStatusRunning), payload.Status)
+}
+
+func TestPreparePassthroughRunningClaimsTransitionOnceConcurrently(t *testing.T) {
+	mgr, eventBus := createTestManagerWithTracking()
+	execution := &AgentExecution{
+		ID:                   "exec-pty",
+		RunID:                "run-before",
+		TaskID:               "task-before",
+		SessionID:            "session-pty",
+		PassthroughProcessID: "pty-1",
+		Status:               v1.AgentStatusReady,
+	}
+	require.NoError(t, mgr.executionStore.Add(execution))
+
+	start := make(chan struct{})
+	results := make(chan struct {
+		publish func()
+		err     error
+	}, 2)
+	var wg sync.WaitGroup
+	for range 2 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			<-start
+			publish, err := mgr.PreparePassthroughRunning(execution.SessionID)
+			results <- struct {
+				publish func()
+				err     error
+			}{publish: publish, err: err}
+		}()
+	}
+	close(start)
+	wg.Wait()
+	close(results)
+
+	for result := range results {
+		require.NoError(t, result.err)
+		require.NotNil(t, result.publish)
+		result.publish()
+	}
+
+	require.Len(t, eventBus.PublishedEvents, 1,
+		"concurrent prepare calls must claim one Ready to Running transition")
+	payload, ok := eventBus.PublishedEvents[0].Event.Data.(AgentEventPayload)
+	require.True(t, ok)
+	require.Equal(t, "run-before", payload.RunID)
+	require.Equal(t, string(v1.AgentStatusRunning), payload.Status)
+}
+
 func TestPassthroughAccessorsRequirePassthroughSessionAndRunner(t *testing.T) {
 	ctx := context.Background()
 	mgr := newTestManager(t)
@@ -120,6 +120,7 @@ var _ interface {
 	OwnsPromptActivity(sessionID, executionID string, generation, activityEpoch uint64) bool
 	GetPromptActivityForSession(ctx context.Context, sessionID string) (executionID string, generation, activityEpoch uint64, lastActivityAt time.Time, err error)
 	CancelAgentForPrompt(ctx context.Context, sessionID, executionID string, generation, activityEpoch uint64) error
+	PreparePassthroughRunning(sessionID string) (func(), error)
 } = (*lifecycleAdapter)(nil)

 // newLifecycleAdapter creates a new lifecycle adapter
@@ -657,6 +658,10 @@ func (a *lifecycleAdapter) MarkPassthroughRunning(sessionID string) error {
 	return a.mgr.MarkPassthroughRunning(sessionID)
 }

+func (a *lifecycleAdapter) PreparePassthroughRunning(sessionID string) (func(), error) {
+	return a.mgr.PreparePassthroughRunning(sessionID)
+}
+
 func (a *lifecycleAdapter) PollRemoteStatusForRecords(ctx context.Context, records []executor.RemoteStatusPollRequest) {
 	lcRecords := make([]lifecycle.RemoteStatusPollRecord, len(records))
 	for i, r := range records {
@@ -453,6 +453,12 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 		}
 		s.executeQueuedMessageWithReservation(data.SessionID, deferredLifecycleDispatch, deferredLifecycleReservation)
 	}()
+	var deferredPassthroughRunning func()
+	defer func() {
+		if deferredPassthroughRunning != nil {
+			deferredPassthroughRunning()
+		}
+	}()

 	if s.isSessionResetInProgress(data.SessionID) {
 		s.logger.Debug("ignoring agent.ready while session reset is in progress",
@@ -724,7 +730,20 @@ func (s *Service) handleAgentReady(ctx context.Context, data watcher.AgentEventD
 			return
 		}
 		if queuedMsg.Content != "" {
-			if err := s.deliverPassthroughPrompt(ctx, data.SessionID, queuedMsg.Content); err != nil {
+			var err error
+			if preparer, ok := s.agentManager.(passthroughRunningPreparer); ok {
+				deferredPassthroughRunning, err = preparer.PreparePassthroughRunning(data.SessionID)
+				if err != nil {
+					s.logger.Warn("failed to prepare passthrough as running before queued prompt",
+						zap.String("session_id", data.SessionID),
+						zap.Error(err))
+					return
+				}
+				err = s.writePassthroughPrompt(ctx, data.SessionID, queuedMsg.Content)
+			} else {
+				err = s.deliverPassthroughPrompt(ctx, data.SessionID, queuedMsg.Content)
+			}
+			if err != nil {
 				s.logger.Warn("failed to deliver queued message to passthrough",
 					zap.String("session_id", data.SessionID),
 					zap.Error(err))
@@ -0,0 +1,128 @@
+package orchestrator
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/kandev/kandev/internal/orchestrator/watcher"
+	"github.com/kandev/kandev/internal/task/models"
+	"github.com/stretchr/testify/require"
+)
+
+type reentrantPassthroughAgentManager struct {
+	*mockAgentManager
+
+	markPassthroughRunningFunc    func(string)
+	preparePassthroughRunningFunc func(string) (func(), error)
+}
+
+func (m *reentrantPassthroughAgentManager) MarkPassthroughRunning(sessionID string) error {
+	if err := m.mockAgentManager.MarkPassthroughRunning(sessionID); err != nil {
+		return err
+	}
+	if m.markPassthroughRunningFunc != nil {
+		m.markPassthroughRunningFunc(sessionID)
+	}
+	return nil
+}
+
+func (m *reentrantPassthroughAgentManager) PreparePassthroughRunning(sessionID string) (func(), error) {
+	if m.preparePassthroughRunningFunc == nil {
+		return func() {}, nil
+	}
+	return m.preparePassthroughRunningFunc(sessionID)
+}
+
+func TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	baseManager := &mockAgentManager{isPassthrough: true}
+	agentManager := &reentrantPassthroughAgentManager{mockAgentManager: baseManager}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+
+	var runningEventCalls int
+	agentManager.markPassthroughRunningFunc = func(sessionID string) {
+		runningEventCalls++
+		svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: sessionID})
+	}
+	agentManager.preparePassthroughRunningFunc = func(sessionID string) (func(), error) {
+		return func() {
+			runningEventCalls++
+			svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: sessionID})
+		}, nil
+	}
+
+	_, err := svc.messageQueue.QueueMessage(ctx, "s1", "t1", "queued prompt", "", "test", false, nil)
+	require.NoError(t, err)
+
+	readyDone := make(chan struct{})
+	go func() {
+		svc.handleAgentReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"})
+		close(readyDone)
+	}()
+
+	select {
+	case <-readyDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleAgentReady deadlocked while delivering a passthrough queued prompt")
+	}
+
+	require.Equal(t, 1, runningEventCalls)
+	require.Empty(t, baseManager.markPassthroughCalls, "guarded delivery must use the deferred lifecycle capability")
+	require.Len(t, baseManager.passthroughStdinCalls, 1)
+	require.Equal(t, "queued prompt\r", baseManager.passthroughStdinCalls[0].Data)
+	require.Zero(t, svc.messageQueue.GetStatus(ctx, "s1").Count)
+
+	updatedSession, err := repo.GetTaskSession(ctx, "s1")
+	require.NoError(t, err)
+	require.Equal(t, models.TaskSessionStateRunning, updatedSession.State)
+}
+
+func TestHandleAgentReady_PassthroughQueuedMessagePublishesAfterWriteFailure(t *testing.T) {
+	ctx := context.Background()
+	repo := setupTestRepo(t)
+	seedSession(t, repo, "t1", "s1", "step1")
+
+	baseManager := &mockAgentManager{
+		isPassthrough:       true,
+		passthroughStdinErr: errors.New("pty write failed"),
+	}
+	agentManager := &reentrantPassthroughAgentManager{mockAgentManager: baseManager}
+	svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentManager)
+
+	var runningEventCalls int
+	agentManager.markPassthroughRunningFunc = func(sessionID string) {
+		runningEventCalls++
+		svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: sessionID})
+	}
+	agentManager.preparePassthroughRunningFunc = func(sessionID string) (func(), error) {
+		return func() {
+			runningEventCalls++
+			svc.handleAgentRunning(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: sessionID})
+		}, nil
+	}
+
+	_, err := svc.messageQueue.QueueMessage(ctx, "s1", "t1", "queued prompt", "", "test", false, nil)
+	require.NoError(t, err)
+
+	readyDone := make(chan struct{})
+	go func() {
+		svc.handleAgentReady(ctx, watcher.AgentEventData{TaskID: "t1", SessionID: "s1"})
+		close(readyDone)
+	}()
+
+	select {
+	case <-readyDone:
+	case <-time.After(2 * time.Second):
+		t.Fatal("handleAgentReady deadlocked while publishing after a passthrough write failure")
+	}
+
+	require.Equal(t, 1, runningEventCalls, "the running event must still publish after the PTY write fails")
+	require.Empty(t, baseManager.markPassthroughCalls)
+	require.Len(t, baseManager.passthroughStdinCalls, 1)
+	require.Zero(t, svc.messageQueue.GetStatus(ctx, "s1").Count)
+}
@@ -3291,18 +3291,22 @@ func (s *Service) dispatchTakenQueuedMessage(ctx context.Context, sessionID stri
 	return true
 }

+type passthroughRunningPreparer interface {
+	PreparePassthroughRunning(sessionID string) (func(), error)
+}
+
 // deliverPassthroughPrompt writes a prompt to PTY stdin and marks the session as running.
 // Uses the per-agent PlanPassthroughStdinChunks so Claude's inter-chunk SubmitDelay is
 // honored here too (queued / workflow-auto-start path); other agents stay on the single
 // atomic write. Falls back to the simple "\r" append if config resolution fails so a
 // transient lookup error never silently swallows the prompt.
+//
+// Callers that already hold the per-session cancellation guard must use
+// PreparePassthroughRunning + writePassthroughPrompt instead (see handleAgentReady); this
+// function publishes agent.running synchronously and will re-enter the guard via the event
+// subscriber. Remaining callers: autoStartPassthroughPrompt (workflow auto-start) and the
+// legacy non-preparer fallback in handleAgentReady.
 func (s *Service) deliverPassthroughPrompt(ctx context.Context, sessionID, content string) error {
-	pt, cfgErr := s.agentManager.ResolvePassthroughConfig(ctx, sessionID)
-	if cfgErr != nil {
-		s.logger.Warn("failed to resolve passthrough config, falling back to \\r submit",
-			zap.String("session_id", sessionID),
-			zap.Error(cfgErr))
-	}
 	// Mark RUNNING before any writes so concurrent PromptTask / queued-message
 	// delivery is blocked by checkSessionPromptable during the inter-chunk
 	// SubmitDelay window (150ms for Claude). Mark error is non-fatal.
@@ -3311,6 +3315,19 @@ func (s *Service) deliverPassthroughPrompt(ctx context.Context, sessionID, conte
 			zap.String("session_id", sessionID),
 			zap.Error(err))
 	}
+	return s.writePassthroughPrompt(ctx, sessionID, content)
+}
+
+// writePassthroughPrompt writes a prompt to PTY stdin without changing runtime
+// state. The ready-event path uses this after preparing the running state so it
+// can publish the captured event after releasing the session guard.
+func (s *Service) writePassthroughPrompt(ctx context.Context, sessionID, content string) error {
+	pt, cfgErr := s.agentManager.ResolvePassthroughConfig(ctx, sessionID)
+	if cfgErr != nil {
+		s.logger.Warn("failed to resolve passthrough config, falling back to \\r submit",
+			zap.String("session_id", sessionID),
+			zap.Error(cfgErr))
+	}
 	if cfgErr != nil {
 		if err := s.agentManager.WritePassthroughStdin(ctx, sessionID, content+"\r"); err != nil {
 			return fmt.Errorf("write to passthrough stdin: %w", err)
@@ -0,0 +1,63 @@
+# ADR-2026-08-31-passthrough-running-publication: Defer Passthrough Running Publication Until Guard Release
+
+**Status:** accepted
+**Date:** 2026-08-31
+**Area:** backend
+
+## Context
+
+The in-memory event bus invokes subscribers synchronously to preserve event
+ordering. An ordinary queued prompt for a passthrough session is selected while
+`handleAgentReady` holds the session's cancellation and dispatch guard.
+`MarkPassthroughRunning` updates runtime status and synchronously publishes
+`agent.running`. The subscriber re-enters `handleAgentRunning` and tries to
+acquire the same non-reentrant guard. The publishing goroutine then blocks on
+its own lock indefinitely.
+
+The guard cannot simply be removed from the ready drain. It serializes queue
+selection and prompt acceptance against cancellation, manual drains, and other
+turn-start paths. ADR-0035 established the related rule that an event whose
+synchronous subscriber re-enters this guard must not be published while the
+guard is held.
+
+## Decision
+
+The lifecycle manager separates the passthrough running transition from its
+event publication for guarded callers. It updates `AgentExecution` status and
+captures an immutable `agent.running` payload under the existing runtime
+boundary, then returns a one-shot publication callback.
+
+The ordinary passthrough ready drain marks the execution and writes the prompt
+while holding the session guard. It invokes the publication callback
+synchronously only after releasing that guard. Immediate callers continue to
+use `MarkPassthroughRunning`, which performs both phases before returning.
+
+The event bus remains synchronous. Publication moves across the lock boundary.
+It does not move to an unowned goroutine.
+
+## Consequences
+
+The ready drain retains exactly-once queue selection and exclusion from
+concurrent cancellation or prompt delivery. The running-event subscriber can
+acquire the session guard, so the session does not deadlock and later lifecycle
+operations remain responsive.
+
+The lifecycle manager must atomically claim the runtime transition and snapshot
+event data under the execution-store lock before it returns the deferred
+callback. A caller must invoke the callback exactly once after it releases its
+guard. This rule also applies when a later PTY write fails after the runtime
+status transition.
+
+## Alternatives Considered
+
+- Make the entire in-memory event bus asynchronous. This would change ordering
+  for every subscriber and create broader concurrency and shutdown semantics.
+- Release and reacquire the session guard around the current synchronous
+  marker. A queued cancellation could win that gap and the original dispatch
+  could then write to a cancelled or replaced process.
+- Publish from a detached goroutine. This avoids direct re-entry but weakens
+  ordering, requires goroutine ownership, and needs stale-event protection for
+  an avoidable scheduling delay.
+- Skip `agent.running` for queued passthrough prompts. That would omit
+  `on_turn_start` processing and leave persisted task-session state out of sync
+  with the runtime.
@@ -221,3 +221,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or
 | 2026-08-30-bound-untracked-dependency-enumeration | [Bound Untracked Dependency Enumeration](2026-08-30-bound-untracked-dependency-enumeration.md) | accepted | backend | 2026-08-30 |
 | 2026-08-30-e2e-browser-cache | [Cache host-runner E2E browser provisioning](2026-08-30-e2e-browser-cache.md) | accepted | infra, workflow | 2026-08-30 |
 | 2026-08-30-environment-owned-git-status | [Project Current Git Status From the Task Environment](2026-08-30-environment-owned-git-status.md) | proposed | backend, frontend, protocol | 2026-08-30 |
+| 2026-08-31-passthrough-running-publication | [Defer Passthrough Running Publication Until Guard Release](2026-08-31-passthrough-running-publication.md) | accepted | backend | 2026-08-31 |
@@ -0,0 +1,79 @@
+---
+created: 2026-08-31
+status: done
+requirements:
+  - REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001
+system_design:
+  - ../../specs/tasks/system-design/passthrough-queued-prompt-dispatch.md
+legacy_specs: []
+---
+
+# Implementation Plan: Passthrough Running Event Deadlock
+
+## Overview
+
+Introduce a two-phase passthrough-running lifecycle operation, then use it at
+the guarded ordinary queue drain so `agent.running` is published only after the
+session guard is released. One work order keeps the lifecycle contract,
+orchestrator integration, and regression proof atomic.
+
+## Scope
+
+### In scope
+
+- Preserve the immediate marker behavior for existing unguarded callers.
+- Capture an immutable running-event payload before deferred publication.
+- Defer publication across the ready handler's session-guard boundary.
+- Add lifecycle and orchestrator regression tests for the exact re-entry chain.
+
+### Out of scope
+
+- Global event-bus scheduling changes.
+- Lock watchdogs, WebSocket deadlines, and delete-path logging.
+- Frontend reconnect behavior for terminal sessions.
+- Queue policy or passthrough submit-sequence changes.
+
+## Technical approach
+
+Add a lifecycle primitive in
+`internal/agent/runtime/lifecycle/manager_passthrough.go` that marks the
+execution running and returns a callback closing over an immutable event
+payload. Claim the transition and capture the payload under the execution-store
+lock, then persist after releasing it. Refactor `MarkPassthroughRunning` to call
+the primitive and publish immediately, preserving all existing callers.
+
+Expose the deferred form through `backendapp.lifecycleAdapter`. In the
+orchestrator, detect that narrow capability in the ordinary passthrough branch
+of `handleAgentReady`. Store the returned callback before PTY writes. Register
+its defer before the code acquires the session guard. Go's LIFO defer order then
+releases the guard first. Run the callback even when a PTY write fails after
+the status transition.
+
+Do not change `MemoryEventBus`, `handleAgentRunning`, durable lifecycle queue
+delivery, or the immediate passthrough marker used by terminal input.
+
+## Tests
+
+| Acceptance criterion | Evidence |
+| --- | --- |
+| `AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.1` | Extend the ordinary passthrough queued-message test to assert one PTY write, one queue consumption, and the successor running transition. |
+| `AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.2` | Add a synchronous re-entry regression in `internal/orchestrator` and lifecycle tests proving publication is deferred, immutable, atomic under competing mutation, at-most-once under concurrent preparation, and immediate callers still publish once. |
+
+## Work orders
+
+- [x] [Task 01: Defer passthrough running publication](task-01-defer-passthrough-running-publication.md)
+
+## Verification results
+
+- `go test -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle ./internal/backendapp -run 'TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent|TestPreparePassthroughRunning|TestMarkPassthroughRunning' -count=1` passed five tests in all three packages.
+- `go test -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle ./internal/backendapp -count=1` passed 5,086 tests.
+- `go test -race -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle -run 'TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent|TestHandleAgentReady_PassthroughQueuedMessagePublishesAfterWriteFailure|TestPreparePassthroughRunningDefersAndSnapshotsPublication|TestPreparePassthroughRunningCapturesSnapshotBeforeCompetingMutation|TestPreparePassthroughRunningClaimsTransitionOnceConcurrently|TestMarkPassthroughRunningPublishesOnceAndGuards' -count=1` passed six tests in both packages.
+
+## Risks
+
+- A deferred callback skipped on a PTY error would leave runtime status and
+  task-session state inconsistent.
+- A callback built from the mutable execution pointer could publish a later
+  execution identity or state.
+- Releasing the guard before the PTY write would reopen cancellation and
+  duplicate-dispatch races. Only event publication can cross the boundary.
@@ -0,0 +1,107 @@
+---
+id: "01-defer-passthrough-running-publication"
+title: "Defer passthrough running publication"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001
+acceptance_criteria:
+  - AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.1
+  - AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.2
+system_design:
+  - ../../specs/tasks/system-design/passthrough-queued-prompt-dispatch.md
+---
+
+# Task 01: Defer Passthrough Running Publication
+
+## Summary
+
+Prevent synchronous `agent.running` delivery from re-entering a session guard
+held by `handleAgentReady`. Preserve the current queue, PTY, runtime-status,
+and immediate-publication behavior outside that guarded path.
+
+## In scope
+
+- Add the lifecycle transition-and-deferred-publication primitive.
+- Expose it through the backend lifecycle adapter without widening unrelated
+  agent-manager interfaces.
+- Use it for ordinary queued passthrough prompt delivery from
+  `handleAgentReady`.
+- Add focused lifecycle and orchestrator tests using a synchronous re-entrant
+  running-event callback.
+
+## Out of scope
+
+- Asynchronous event-bus delivery.
+- Durable lifecycle queue dispatch, frontend terminal reconnects, lock
+  watchdogs, and WebSocket timeouts.
+- Refactors of other session-guard call sites without a proven re-entry chain.
+
+## Acceptance
+
+- The red regression test times out on the pre-fix
+  `handleAgentReady -> MarkPassthroughRunning -> handleAgentRunning` chain.
+  After the correction, the test completes with one prompt delivery and an
+  available session guard.
+- Guarded delivery marks runtime state and writes all PTY chunks before it
+  publishes the captured `agent.running` event. Publication still occurs after
+  a post-transition PTY write error.
+- The lifecycle transition and immutable payload capture are atomic under the
+  execution-store lock. Concurrent preparation calls publish at most one
+  `agent.running` event, and a competing status mutation cannot relabel it.
+- Existing immediate callers publish exactly once, and deferred publication
+  uses immutable captured event data.
+
+## Verification
+
+```bash
+go test -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle ./internal/backendapp -run 'TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent|TestPreparePassthroughRunning|TestMarkPassthroughRunning' -count=1
+go test -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle ./internal/backendapp -count=1
+```
+
+Run both commands from `apps/backend`.
+
+## Files likely touched
+
+- `apps/backend/internal/agent/runtime/lifecycle/manager_passthrough.go`
+- `apps/backend/internal/agent/runtime/lifecycle/manager_passthrough_runtime_test.go`
+- `apps/backend/internal/backendapp/adapters.go`
+- `apps/backend/internal/orchestrator/event_handlers_agent.go`
+- `apps/backend/internal/orchestrator/event_handlers_workflow.go`
+- `apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go`
+
+## Dependencies
+
+None.
+
+## Risks
+
+- The ordering of deferred callbacks relative to guard release depends on
+  explicit defer registration order and needs direct test coverage.
+- Adapter-only capability detection must fail safely in tests and alternate
+  wiring that does not publish lifecycle events.
+
+## Parallelism
+
+`sequential`
+
+## Inputs
+
+- `docs/specs/tasks/requirements/passthrough-queued-prompt-dispatch.md`
+- `docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md`
+- `docs/decisions/2026-08-31-passthrough-running-publication.md`
+- `docs/decisions/0035-version-agent-ready-events-by-prompt-generation.md`
+- GitHub issue #3177 and the diagnostic reproduction recorded in this task.
+
+## Results
+
+- Added the two-phase lifecycle capability that updates passthrough runtime state and returns a one-shot immutable `agent.running` publication callback.
+- Forwarded the capability through the production lifecycle adapter without widening `executor.AgentManagerClient`; the guarded ordinary ready drain uses it and releases its session guard before publication.
+- Preserved immediate `MarkPassthroughRunning` behavior for terminal input, prompt delivery, and test/legacy adapters without the optional capability.
+- Added lifecycle snapshot/idempotence coverage plus synchronous ready-event re-entry coverage, including publication after a PTY write failure.
+- Made the Ready-to-Running claim and event snapshot atomic under the execution-store lock; added deterministic competing-mutation and concurrent-prepare coverage.
+- `go test -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle ./internal/backendapp -run 'TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent|TestPreparePassthroughRunning|TestMarkPassthroughRunning' -count=1` passed five tests in all three packages.
+- `go test -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle ./internal/backendapp -count=1` passed 5,086 tests.
+- `go test -race -tags fts5 ./internal/orchestrator ./internal/agent/runtime/lifecycle -run 'TestHandleAgentReady_PassthroughQueuedMessageSynchronousRunningEvent|TestHandleAgentReady_PassthroughQueuedMessagePublishesAfterWriteFailure|TestPreparePassthroughRunningDefersAndSnapshotsPublication|TestPreparePassthroughRunningCapturesSnapshotBeforeCompetingMutation|TestPreparePassthroughRunningClaimsTransitionOnceConcurrently|TestMarkPassthroughRunningPublishesOnceAndGuards' -count=1` passed six tests.
@@ -59,6 +59,7 @@ signals, and task-scoped scheduling contracts.
 - [Multi-branch tasks](requirements/multi-branch.md)
 - [Parent-Child Message Interrupt](requirements/parent-child-message-interrupt.md)
 - [Parent-Child Task Stop](requirements/parent-child-task-stop.md)
+- [Passthrough Queued Prompt Dispatch](requirements/passthrough-queued-prompt-dispatch.md)
 - [Prevent Agent Auto-Start On Open](requirements/prevent-agent-autostart-on-open.md)
 - [Prompt attachments](requirements/prompt-attachments.md)
 - [Quick Chat Sessions, Persistence, and Expiration](requirements/quick-chat-expiration.md)
@@ -119,6 +120,7 @@ signals, and task-scoped scheduling contracts.
 - [External task ID idempotency](system-design/external-id-idempotency.md)
 - [Task model unification](system-design/model-unification.md)
 - [Remote Contribution Tasks](system-design/remote-contribution-tasks.md)
+- [Passthrough Queued Prompt Dispatch](system-design/passthrough-queued-prompt-dispatch.md)
 - [Task Archive Confirmation](system-design/archive-confirmation.md)
 - [Task plan write lifecycle](system-design/plan-write-lifecycle.md)
 - [Task Runtime Cleanup](system-design/runtime-cleanup.md)
@@ -135,7 +137,7 @@ signals, and task-scoped scheduling contracts.

 ## Migration record

-Migration remains in progress. The four requirements above now have
+Migration remains in progress. The five requirements above now have
 authoritative, wrapper-free requirement/design pairs. Other migrated files still
 need the same extraction before this system can return to a complete migration
 state.
@@ -0,0 +1,49 @@
+---
+status: active
+system: tasks
+created: 2026-08-31
+owners:
+  - cfl
+---
+
+# Passthrough Queued Prompt Dispatch Requirements
+
+## Overview
+
+The task system owns queued prompt selection, turn boundaries, and session
+operation serialization. CLI passthrough supplies the PTY transport, but a
+queued follow-up must start a successor turn without wedging the session that
+owns the queue.
+
+## Terminology
+
+- **Passthrough session:** A task session whose agent runs as an interactive
+  CLI under a PTY instead of through ACP prompt transport.
+- **Ready drain:** The turn-completion path that selects an eligible queued
+  prompt and starts the next turn.
+
+## Requirements
+
+### REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001: Responsive passthrough ready drain
+
+**Intent:** Deliver the next queued prompt exactly once while keeping the
+session responsive to subsequent lifecycle operations.
+
+#### Acceptance criteria
+
+- **AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.1:** When a running
+  passthrough session completes a turn with an eligible queued prompt, the
+  system shall accept that prompt exactly once. The system shall write the
+  prompt to the PTY and start the successor turn.
+- **AC-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001.2:** When starting that
+  successor turn emits the session's running-state event, the ready drain shall
+  not wait on its own lock. The event handler shall complete. Later delete,
+  cancel, and queue operations shall remain responsive without a backend
+  restart.
+
+## Out of scope
+
+- Changing the synchronous ordering contract of the in-memory event bus.
+- Adding lock watchdogs or WebSocket request deadlines.
+- Changing frontend terminal reconnect behavior for terminal-state sessions.
+- Changing passthrough prompt formatting, submit sequences, or queue policy.
@@ -0,0 +1,108 @@
+---
+status: current
+system: tasks
+requirements:
+  - REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001
+---
+
+# Passthrough Queued Prompt Dispatch System Design
+
+## Purpose and boundaries
+
+The task system owns the per-session serialization guard, queue reservation,
+turn completion, and successor-turn state transition. The agent runtime owns
+the passthrough execution status and lifecycle event payload. The in-memory
+event bus remains synchronous because existing consumers depend on ordered
+delivery.
+
+This design covers the ordinary queued-prompt drain from `agent.ready`. Durable
+lifecycle prompts already leave the ready handler's guard before their normal
+dispatch pipeline begins.
+
+## Requirement mapping
+
+| Requirement | Design section |
+| --- | --- |
+| `REQ-TASKS-PASSTHROUGH-QUEUED-PROMPT-DISPATCH-001` | [Control flow](#control-flow), [Failure and recovery](#failure-and-recovery) |
+
+## Components and responsibilities
+
+- `orchestrator.Service.handleAgentReady` owns turn completion, the
+  per-session guard, ordinary queue reservation, and PTY dispatch ordering.
+- `orchestrator.Service.deliverPassthroughPrompt` owns prompt chunk planning
+  and PTY writes.
+- `lifecycle.Manager` owns the transition of an `AgentExecution` to
+  `AgentStatusRunning` and construction of the corresponding immutable
+  `agent.running` payload.
+- `backendapp.lifecycleAdapter` exposes the lifecycle transition through the
+  orchestrator's narrow runtime boundary.
+- `MemoryEventBus` continues to invoke the watcher subscriber synchronously.
+  `handleAgentRunning` owns the task-session running transition and
+  `on_turn_start` processing.
+
+## Data and contracts
+
+The lifecycle boundary provides a two-phase passthrough-running operation for
+callers that already hold a session serialization guard:
+
+1. Under the execution-store lock, revalidate the current session and
+   passthrough execution, claim the Ready-to-Running transition, and capture an
+   immutable `agent.running` payload.
+2. Return a one-shot publication callback that emits the captured payload.
+
+The existing immediate `MarkPassthroughRunning` operation uses the same
+primitive and invokes the callback before returning. Existing terminal-input
+and unguarded prompt paths therefore retain synchronous publication.
+
+The orchestrator uses the two-phase form only for the ordinary passthrough
+ready drain. Test doubles that do not publish lifecycle events can retain the
+immediate marker contract.
+
+## Control flow
+
+1. `handleAgentReady` acquires the session guard, completes the current turn,
+   and reserves one eligible ordinary queued prompt.
+2. The orchestrator asks the lifecycle adapter to mark the execution running.
+   Lifecycle updates runtime status and snapshots the event payload without
+   publishing it.
+3. While the guard is still held, the orchestrator writes all planned prompt
+   chunks to the PTY. This preserves the existing exclusion against a
+   concurrent cancel, manual drain, or prompt dispatch.
+4. The ready handler releases the session guard.
+5. The ready handler invokes the deferred publication callback synchronously.
+   `MemoryEventBus` calls `handleAgentRunning`, which can now acquire the free
+   guard, run `on_turn_start`, and project the session as running.
+
+Publication is deferred by lock scope, not by a detached goroutine. The event
+therefore retains deterministic ordering relative to the ready handler's
+completion, and there is no unbounded publisher lifetime to manage.
+
+## Failure and recovery
+
+- In the two-phase ready-event path, a runtime-status transition failure
+  prevents the PTY write. The existing immediate `deliverPassthroughPrompt`
+  contract remains non-fatal for `MarkPassthroughRunning` failures.
+- If a PTY write fails after the execution was marked running, the deferred
+  callback still publishes after guard release. This preserves the existing
+  status/event pairing and allows ordinary recovery paths to observe the
+  runtime transition.
+- Concurrent preparation calls observe the same locked transition, so at most
+  one call claims Ready-to-Running and creates a publication.
+- The publication callback uses an immutable payload captured at the status
+  transition. Later execution mutation cannot relabel the event.
+- The event bus remains synchronous. A handler error is logged through the
+  existing bus behavior and does not reintroduce guard re-entry.
+
+## Observability
+
+The permanent regression test reproduces a synchronous running-event callback
+from the passthrough marker. The test proves that the ready handler returns and
+the PTY write occurs once. It also proves that the queue entry is consumed once
+and the session guard is available afterward. Existing lifecycle tests prove
+that immediate callers still publish once. These tests also prove that the
+deferred callback carries the captured payload.
+
+## Related decisions
+
+- [Defer passthrough running publication until guard release](../../../decisions/2026-08-31-passthrough-running-publication.md)
+- [Version AgentReady events by prompt generation](../../../decisions/0035-version-agent-ready-events-by-prompt-generation.md)