From 1c7b5afce139ae81916867d2bd3eecb8c81d16ec Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Mon, 31 Aug 2026 12:27:18 +0000 Subject: [PATCH] fix: avoid passthrough ready event deadlock (#3177) (#3181) --- .../runtime/lifecycle/manager_passthrough.go | 82 ++++++++-- .../manager_passthrough_runtime_test.go | 147 ++++++++++++++++++ apps/backend/internal/backendapp/adapters.go | 5 + .../orchestrator/event_handlers_agent.go | 21 ++- ...event_handlers_passthrough_running_test.go | 128 +++++++++++++++ .../orchestrator/event_handlers_workflow.go | 29 +++- ...6-08-31-passthrough-running-publication.md | 63 ++++++++ docs/decisions/INDEX.md | 1 + .../plan.md | 79 ++++++++++ ...1-defer-passthrough-running-publication.md | 107 +++++++++++++ docs/specs/tasks/README.md | 4 +- .../passthrough-queued-prompt-dispatch.md | 49 ++++++ .../passthrough-queued-prompt-dispatch.md | 108 +++++++++++++ 13 files changed, 800 insertions(+), 23 deletions(-) create mode 100644 apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go create mode 100644 docs/decisions/2026-08-31-passthrough-running-publication.md create mode 100644 docs/plans/passthrough-running-event-deadlock/plan.md create mode 100644 docs/plans/passthrough-running-event-deadlock/task-01-defer-passthrough-running-publication.md create mode 100644 docs/specs/tasks/requirements/passthrough-queued-prompt-dispatch.md create mode 100644 docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough.go b/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough.go index 7aab086f25..c760781894 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough.go @@ -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 } diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough_runtime_test.go b/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough_runtime_test.go index f6994fa2d2..6e42f06944 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough_runtime_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_passthrough_runtime_test.go @@ -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) diff --git a/apps/backend/internal/backendapp/adapters.go b/apps/backend/internal/backendapp/adapters.go index ebd5e769f1..e333e63815 100644 --- a/apps/backend/internal/backendapp/adapters.go +++ b/apps/backend/internal/backendapp/adapters.go @@ -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 { diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go index 9b7b2aa551..7474618a4a 100644 --- a/apps/backend/internal/orchestrator/event_handlers_agent.go +++ b/apps/backend/internal/orchestrator/event_handlers_agent.go @@ -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)) diff --git a/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go b/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go new file mode 100644 index 0000000000..ddc2082ba3 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_passthrough_running_test.go @@ -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) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 22b9460e8e..382e3d43e7 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -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) diff --git a/docs/decisions/2026-08-31-passthrough-running-publication.md b/docs/decisions/2026-08-31-passthrough-running-publication.md new file mode 100644 index 0000000000..3abd3b69d2 --- /dev/null +++ b/docs/decisions/2026-08-31-passthrough-running-publication.md @@ -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. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 88b0f8d0a0..557e13eee6 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -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 | diff --git a/docs/plans/passthrough-running-event-deadlock/plan.md b/docs/plans/passthrough-running-event-deadlock/plan.md new file mode 100644 index 0000000000..ebc7d1b35c --- /dev/null +++ b/docs/plans/passthrough-running-event-deadlock/plan.md @@ -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. diff --git a/docs/plans/passthrough-running-event-deadlock/task-01-defer-passthrough-running-publication.md b/docs/plans/passthrough-running-event-deadlock/task-01-defer-passthrough-running-publication.md new file mode 100644 index 0000000000..94bdbeb97d --- /dev/null +++ b/docs/plans/passthrough-running-event-deadlock/task-01-defer-passthrough-running-publication.md @@ -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. diff --git a/docs/specs/tasks/README.md b/docs/specs/tasks/README.md index 2a2bdb2675..6636ce5c21 100644 --- a/docs/specs/tasks/README.md +++ b/docs/specs/tasks/README.md @@ -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. diff --git a/docs/specs/tasks/requirements/passthrough-queued-prompt-dispatch.md b/docs/specs/tasks/requirements/passthrough-queued-prompt-dispatch.md new file mode 100644 index 0000000000..8c6f9434c6 --- /dev/null +++ b/docs/specs/tasks/requirements/passthrough-queued-prompt-dispatch.md @@ -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. diff --git a/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md b/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md new file mode 100644 index 0000000000..dfb028c51c --- /dev/null +++ b/docs/specs/tasks/system-design/passthrough-queued-prompt-dispatch.md @@ -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)