diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go index 3a3f6e2f62..8a19da5f5b 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_interaction.go @@ -99,6 +99,19 @@ func (m *Manager) PromptAgent(ctx context.Context, executionID string, prompt st return m.PromptAgentWithDispatchCallback(ctx, executionID, prompt, attachments, dispatchOnly, nil) } +// RegisterInitialPromptDispatchCallbacks installs one-shot callbacks for the +// initial prompt sent during StartAgentProcess. Model-switch startup launches +// that prompt asynchronously, so callers that own startup cancellation must +// wait for either provider acceptance or a pre-acceptance delivery failure. +func (m *Manager) RegisterInitialPromptDispatchCallbacks(executionID string, onDispatched, onFailure func()) error { + execution, exists := m.executionStore.Get(executionID) + if !exists { + return fmt.Errorf("execution %q not found: %w", executionID, ErrExecutionNotFound) + } + execution.setInitialPromptDispatchCallbacks(onDispatched, onFailure) + return nil +} + // PromptAgentWithDispatchCallback exposes agentctl acceptance to callers that // must keep admission serialized until the queued prompt is actually dispatched. func (m *Manager) PromptAgentWithDispatchCallback(ctx context.Context, executionID string, prompt string, attachments []v1.MessageAttachment, dispatchOnly bool, onDispatched func()) (*PromptResult, error) { diff --git a/apps/backend/internal/agent/runtime/lifecycle/session.go b/apps/backend/internal/agent/runtime/lifecycle/session.go index c75146f4dd..0657c8556d 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session.go @@ -952,6 +952,19 @@ func (sm *SessionManager) dispatchInitialPrompt(ctx context.Context, execution * zap.Int("effective_length", len(effectivePrompt))) } acpAttachments := convertAttachments(attachments) + onDispatched, onInitialPromptFailure := execution.takeInitialPromptDispatchCallbacks() + var failureHandler func(InitialPromptFailure) + if onInitialPromptFailure != nil { + initialPromptFailure := sm.initialPromptFailure + failureHandler = func(failure InitialPromptFailure) { + onInitialPromptFailure() + if initialPromptFailure != nil { + initialPromptFailure(failure) + } + } + } else { + failureHandler = sm.initialPromptFailure + } go func() { promptCtx, cancel := appctx.Detached(ctx, sm.stopCh, 0) defer cancel() @@ -962,7 +975,7 @@ func (sm *SessionManager) dispatchInitialPrompt(ctx context.Context, execution * false, acpAttachments, false, - sendPromptCallbacks{onFailure: sm.initialPromptFailure}, + sendPromptCallbacks{onDispatched: onDispatched, onFailure: failureHandler}, false, ) if err != nil { diff --git a/apps/backend/internal/agent/runtime/lifecycle/session_test.go b/apps/backend/internal/agent/runtime/lifecycle/session_test.go index a54b3c7651..01a6ab7091 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/session_test.go +++ b/apps/backend/internal/agent/runtime/lifecycle/session_test.go @@ -628,7 +628,6 @@ func TestInitializeAndPrompt_StreamBeforeInitialize(t *testing.T) { agentctl: client, promptDoneCh: make(chan PromptCompletionSignal, 1), } - agentConfig := &testAgent{ id: "test-agent", enabled: true, @@ -1235,6 +1234,8 @@ func TestInitializeAndPrompt_WithTaskDescription(t *testing.T) { agentctl: client, promptDoneCh: make(chan PromptCompletionSignal, 1), } + dispatched := make(chan struct{}, 1) + execution.setInitialPromptDispatchCallbacks(func() { dispatched <- struct{}{} }, nil) agentConfig := &testAgent{ id: "test-agent", @@ -1260,7 +1261,11 @@ func TestInitializeAndPrompt_WithTaskDescription(t *testing.T) { } // Wait for the prompt to be sent asynchronously - time.Sleep(500 * time.Millisecond) + select { + case <-dispatched: + case <-time.After(2 * time.Second): + t.Fatal("timed out waiting for initial prompt dispatch callback") + } actions := mock.getActionLog() diff --git a/apps/backend/internal/agent/runtime/lifecycle/types.go b/apps/backend/internal/agent/runtime/lifecycle/types.go index 0213f690d5..e77d6259b7 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/types.go +++ b/apps/backend/internal/agent/runtime/lifecycle/types.go @@ -274,6 +274,13 @@ type AgentExecution struct { // asynchronously, so its transport-level gate alone cannot provide this. promptMu sync.Mutex dispatchedPromptPending atomic.Bool + // Initial-prompt callbacks are installed before StartAgentProcess for + // model-switch launches. Lifecycle sends the initial prompt asynchronously, + // so they must be captured before startup begins and consumed once that + // prompt is accepted or fails before acceptance. + initialPromptDispatchCallback func() + initialPromptFailureCallback func() + initialPromptDispatchCallbackMu sync.Mutex // Closed when the current SendPrompt returns, so CancelAgent can wait // for the in-flight prompt to finish before the caller retries. @@ -346,6 +353,23 @@ func (e *AgentExecution) setSessionInitialized(value bool) { e.sessionInitializedMu.Unlock() } +func (e *AgentExecution) setInitialPromptDispatchCallbacks(onDispatched, onFailure func()) { + e.initialPromptDispatchCallbackMu.Lock() + e.initialPromptDispatchCallback = onDispatched + e.initialPromptFailureCallback = onFailure + e.initialPromptDispatchCallbackMu.Unlock() +} + +func (e *AgentExecution) takeInitialPromptDispatchCallbacks() (func(), func()) { + e.initialPromptDispatchCallbackMu.Lock() + onDispatched := e.initialPromptDispatchCallback + onFailure := e.initialPromptFailureCallback + e.initialPromptDispatchCallback = nil + e.initialPromptFailureCallback = nil + e.initialPromptDispatchCallbackMu.Unlock() + return onDispatched, onFailure +} + type activeTopLevelTool struct { ToolCallID string Name string diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go index 042dffe549..8f376a9146 100644 --- a/apps/backend/internal/orchestrator/event_handlers_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_test.go @@ -301,6 +301,8 @@ type mockAgentManager struct { promptAcceptedOnError bool promptAgentFunc func(context.Context, string, string, []v1.MessageAttachment, bool) (*executor.PromptResult, error) launchAgentFunc func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) + initialPromptDispatchCallback func() + initialPromptFailureCallback func() startAgentProcessCalls []string startAgentProcessErr error startAgentProcessFunc func(context.Context, string) error @@ -469,6 +471,15 @@ func (m *mockAgentManager) StartAgentProcess(ctx context.Context, sessionID stri } return err } + +func (m *mockAgentManager) RegisterInitialPromptDispatchCallbacks(_ string, onDispatched, onFailure func()) error { + m.mu.Lock() + m.initialPromptDispatchCallback = onDispatched + m.initialPromptFailureCallback = onFailure + m.mu.Unlock() + return nil +} + func (m *mockAgentManager) IsAgentCommandConfigured(_ string) bool { return true } func (m *mockAgentManager) StopAgent(ctx context.Context, agentExecutionID string, force bool) error { m.mu.Lock() diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction.go b/apps/backend/internal/orchestrator/executor/executor_interaction.go index 971de5d12d..ed8c302130 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction.go @@ -889,6 +889,28 @@ func workspaceFromTaskEnvironment(env *models.TaskEnvironment) string { // the agent doesn't support in-place switching, it falls back to stopping and // restarting the agent with the new model. func (e *Executor) SwitchModel(ctx context.Context, taskID, sessionID, newModel, prompt string) (*PromptResult, error) { + return e.switchModel(ctx, taskID, sessionID, newModel, prompt, nil, nil) +} + +// SwitchModelWithDispatchCallbacks is the model-switch variant used by a +// resume-owned prompt. The fallback startup sends its initial prompt from a +// lifecycle goroutine, so the caller must receive the actual acceptance or +// pre-acceptance failure instead of treating StartAgentProcess as acceptance. +func (e *Executor) SwitchModelWithDispatchCallbacks( + ctx context.Context, + taskID, sessionID, newModel, prompt string, + onDispatched func(executionID string), + onFailure func(), +) (*PromptResult, error) { + return e.switchModel(ctx, taskID, sessionID, newModel, prompt, onDispatched, onFailure) +} + +func (e *Executor) switchModel( + ctx context.Context, + taskID, sessionID, newModel, prompt string, + onDispatched func(executionID string), + onFailure func(), +) (*PromptResult, error) { e.logger.Info("switching model for session", zap.String("task_id", taskID), zap.String("session_id", sessionID), @@ -954,7 +976,7 @@ func (e *Executor) SwitchModel(ctx context.Context, taskID, sessionID, newModel, zap.Bool("use_worktree", req.UseWorktree), zap.String("repository_path", req.RepositoryPath)) - if err := e.launchModelSwitchAgent(launchCtx, task.ID, sessionID, newModel, session, req, existingRunning); err != nil { + if err := e.launchModelSwitchAgent(launchCtx, task.ID, sessionID, newModel, session, req, existingRunning, onDispatched, onFailure); err != nil { return nil, err } @@ -1123,7 +1145,15 @@ func (e *Executor) stopPreparedModelSwitchAgent( } // launchModelSwitchAgent launches the new agent, persists state, and starts the process. -func (e *Executor) launchModelSwitchAgent(ctx context.Context, taskID, sessionID, newModel string, session *models.TaskSession, req *LaunchAgentRequest, existingRunning *models.ExecutorRunning) error { +func (e *Executor) launchModelSwitchAgent( + ctx context.Context, + taskID, sessionID, newModel string, + session *models.TaskSession, + req *LaunchAgentRequest, + existingRunning *models.ExecutorRunning, + onDispatched func(executionID string), + onFailure func(), +) error { resp, err := e.agentManager.LaunchAgent(ctx, req) if err != nil { e.logger.Error("failed to launch agent with new model", @@ -1132,6 +1162,28 @@ func (e *Executor) launchModelSwitchAgent(ctx context.Context, taskID, sessionID zap.Error(err)) return fmt.Errorf("failed to launch agent with new model: %w", err) } + if onDispatched != nil || onFailure != nil { + registrar, ok := e.agentManager.(interface { + RegisterInitialPromptDispatchCallbacks(string, func(), func()) error + }) + if !ok { + registrationErr := errors.New("agent manager cannot register initial prompt dispatch callbacks") + e.cleanupUnstartedExecutionAfterPersistError(ctx, sessionID, resp.AgentExecutionID, registrationErr) + return registrationErr + } + if err := registrar.RegisterInitialPromptDispatchCallbacks( + resp.AgentExecutionID, + func() { + if onDispatched != nil { + onDispatched(resp.AgentExecutionID) + } + }, + onFailure, + ); err != nil { + e.cleanupUnstartedExecutionAfterPersistError(ctx, sessionID, resp.AgentExecutionID, err) + return fmt.Errorf("failed to register initial prompt dispatch callbacks: %w", err) + } + } if err := e.persistModelSwitchState(ctx, taskID, sessionID, session, newModel); err != nil { e.cleanupUnstartedExecutionAfterPersistError(ctx, sessionID, resp.AgentExecutionID, err) diff --git a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go index 88cf27e535..ed76b20283 100644 --- a/apps/backend/internal/orchestrator/executor/executor_interaction_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_interaction_test.go @@ -831,6 +831,8 @@ func TestLaunchModelSwitchAgent_CleansStartedExecutionAfterTerminalRace(t *testi &callerSession, &LaunchAgentRequest{}, nil, + nil, + nil, ) if !errors.Is(err, ErrSessionStateSuperseded) { diff --git a/apps/backend/internal/orchestrator/resume_attempt.go b/apps/backend/internal/orchestrator/resume_attempt.go index ed78c87722..2448f686f3 100644 --- a/apps/backend/internal/orchestrator/resume_attempt.go +++ b/apps/backend/internal/orchestrator/resume_attempt.go @@ -37,6 +37,19 @@ type resumeAttempt struct { // can be retained when a replacement is admitted and retained again when // its owner eventually returns; both paths must describe one tombstone. retained bool + // accepted is protected by resumeAttemptRegistry.mu. Once provider + // acceptance is recorded, explicit session cancellation no longer owns + // startup teardown for this attempt. + accepted bool + // awaitingInitialPrompt keeps startup ownership active after the model + // switch path returns. Its initial prompt is dispatched asynchronously by + // lifecycle, so promptTask's deferred finish must wait for the real provider + // acceptance callback. + awaitingInitialPrompt bool + // finishRequested is set when the owning operation returns while the + // initial prompt is still pending. The acceptance callback completes the + // attempt after it transfers ownership. + finishRequested bool } type resumeAttemptRegistry struct { @@ -56,6 +69,7 @@ type resumeAttemptTombstone struct { id uint64 executionID string cancelled bool + accepted bool } func newResumeAttemptRegistry() *resumeAttemptRegistry { @@ -109,12 +123,20 @@ func (r *resumeAttemptRegistry) begin(parent context.Context, taskID, sessionID // against prompt admission and lifecycle event handling. func (r *resumeAttemptRegistry) invalidate(sessionID string) bool { r.mu.Lock() - defer r.mu.Unlock() attempt := r.attempts[sessionID] - if attempt == nil || attempt.ctx.Err() != nil { + if attempt == nil || attempt.ctx.Err() != nil || attempt.accepted { + r.mu.Unlock() return false } attempt.cancel() + finishPending := attempt.awaitingInitialPrompt + r.mu.Unlock() + if finishPending { + // The operation that owns this attempt may already have returned while + // lifecycle is still dispatching its initial prompt. Finish the pending + // ownership record now so a cancelled callback cannot leave it active. + attempt.finish(r) + } return true } @@ -161,6 +183,9 @@ func (r *resumeAttemptRegistry) canCleanup(attempt *resumeAttempt) bool { if current := r.attempts[attempt.sessionID]; current != nil && current != attempt { return false } + if attempt.accepted { + return false + } for _, tombstone := range r.tombstones[attempt.sessionID] { if tombstone.id > attempt.id && tombstone.executionID == executionID { return false @@ -203,6 +228,9 @@ func (r *resumeAttemptRegistry) retainLocked(attempt *resumeAttempt) { for index := range entries { if entries[index].id == attempt.id && entries[index].executionID == "" && executionID != "" { entries[index].executionID = executionID + } + if entries[index].id == attempt.id { + entries[index].accepted = attempt.accepted break } } @@ -215,6 +243,7 @@ func (r *resumeAttemptRegistry) retainLocked(attempt *resumeAttempt) { id: attempt.id, executionID: executionID, cancelled: attempt.ctx.Err() != nil, + accepted: attempt.accepted, }) if len(entries) > maxResumeAttemptTombstones { entries = entries[len(entries)-maxResumeAttemptTombstones:] @@ -253,12 +282,103 @@ func (r *resumeAttemptRegistry) canCleanupIdentity(sessionID, executionID, origi if !found || tombstone.executionID != executionID { return false } + if tombstone.accepted { + return false + } if current := r.attempts[sessionID]; current != nil && current.id > id { return false } return r.latestExecution[sessionID][executionID] <= id } +// accept transfers startup authority to the provider turn while retaining the +// attempt identity for later execution events. The registry lock orders this +// transition with explicit cancellation, so an invalidated attempt cannot be +// revived by a late provider callback. +func (r *resumeAttemptRegistry) accept(attempt *resumeAttempt, executionID string) bool { + if attempt == nil || executionID == "" { + return false + } + r.mu.Lock() + defer r.mu.Unlock() + if r.attempts[attempt.sessionID] != attempt || attempt.ctx.Err() != nil { + return false + } + attempt.executionMu.Lock() + knownExecutionID := attempt.executionID + if knownExecutionID == "" { + attempt.executionID = executionID + knownExecutionID = executionID + } + attempt.executionMu.Unlock() + if knownExecutionID != executionID { + return false + } + attempt.accepted = true + return true +} + +// holdForInitialPrompt keeps an admitted model-switch attempt active until +// lifecycle reports acceptance of the replacement's asynchronously dispatched +// initial prompt. The caller must set this before starting the replacement. +func (r *resumeAttemptRegistry) holdForInitialPrompt(attempt *resumeAttempt) bool { + if attempt == nil { + return false + } + r.mu.Lock() + defer r.mu.Unlock() + if r.attempts[attempt.sessionID] != attempt || attempt.ctx.Err() != nil || attempt.accepted { + return false + } + attempt.awaitingInitialPrompt = true + return true +} + +// releaseInitialPromptHold releases the model-switch-specific deferred finish +// when the model switch stayed in place and ordinary prompt dispatch will own +// the acceptance boundary instead. +func (r *resumeAttemptRegistry) releaseInitialPromptHold(attempt *resumeAttempt) { + if attempt == nil { + return + } + r.mu.Lock() + if r.attempts[attempt.sessionID] == attempt { + attempt.awaitingInitialPrompt = false + } + r.mu.Unlock() +} + +// abortInitialPromptHold closes a model-switch attempt when lifecycle reports +// a delivery failure before provider acceptance. There is no later owner that +// can safely finish the attempt after the asynchronous failure callback. +func (r *resumeAttemptRegistry) abortInitialPromptHold(attempt *resumeAttempt) { + if attempt == nil { + return + } + r.mu.Lock() + if r.attempts[attempt.sessionID] == attempt { + attempt.awaitingInitialPrompt = false + } + r.mu.Unlock() + attempt.finish(r) +} + +// finishAfterInitialPromptAcceptance completes a deferred owner only after +// the callback has transferred startup authority. If the outer operation is +// still active, its normal finish call remains responsible for completion. +func (r *resumeAttemptRegistry) finishAfterInitialPromptAcceptance(attempt *resumeAttempt) { + if attempt == nil { + return + } + r.mu.Lock() + shouldFinish := r.attempts[attempt.sessionID] == attempt && + attempt.accepted && attempt.finishRequested + r.mu.Unlock() + if shouldFinish { + attempt.finish(r) + } +} + func (r *resumeAttemptRegistry) cancelledExecutionLocked(sessionID, executionID string) bool { if executionID == "" { return false @@ -317,17 +437,21 @@ func (attempt *resumeAttempt) finish(registry *resumeAttemptRegistry) { if attempt == nil { return } - attempt.finishOnce.Do(func() { - if registry != nil { - registry.mu.Lock() - if registry.attempts[attempt.sessionID] == attempt { - delete(registry.attempts, attempt.sessionID) - } - registry.retainLocked(attempt) + if registry != nil { + registry.mu.Lock() + if registry.attempts[attempt.sessionID] == attempt && + attempt.awaitingInitialPrompt && !attempt.accepted && attempt.ctx.Err() == nil { + attempt.finishRequested = true registry.mu.Unlock() + return } - close(attempt.done) - }) + if registry.attempts[attempt.sessionID] == attempt { + delete(registry.attempts, attempt.sessionID) + } + registry.retainLocked(attempt) + registry.mu.Unlock() + } + attempt.finishOnce.Do(func() { close(attempt.done) }) } func (attempt *resumeAttempt) wait(ctx context.Context) error { diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index de616f3b83..07f8319e7f 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -5922,6 +5922,7 @@ func (s *Service) runPromptTurn( promptCtx = dispatchCtx onDispatched, dispatchOutcome := s.preparePromptDispatchCallback( promptCtx, taskID, sessionID, session, rollback, options, foregroundDispatch, releaseDispatchGuard, + resumeAttempt, ) result, execErr := s.executor.PromptWithDispatchCallback( promptCtx, taskID, sessionID, effectivePrompt, attachments, dispatchOnly, @@ -6015,7 +6016,7 @@ func (s *Service) resolveModelSwitchAttempt( ) (switchResult *PromptResult, handled bool, remainingGuard *lockedCancelInFlightGuard, switchErr error) { result, handledSwitch, attemptErr := s.attemptModelSwitchForPrompt( ctx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch, runBeforeDispatch, - runAfterDispatchAdmission, + runAfterDispatchAdmission, resumeAttempt, ) if !handledSwitch { if modelSwitchGuard != nil { @@ -6296,13 +6297,14 @@ func (s *Service) validateAndRunDispatchBoundary( // preparePromptDispatchCallback binds the turn, arms the interactive-prompt // attempt evidence, and builds the onDispatched callback executor. -// PromptWithDispatchCallback invokes once agentctl accepts the prompt — -// wrapping in the caller's afterDispatch hook and the nonterminal dispatch -// guard release, innermost first, when either applies. +// PromptWithDispatchCallback invokes once agentctl accepts the prompt. For a +// resumed turn, ownership transfer wraps the durable afterDispatch hook and +// identity publication; the nonterminal dispatch guard releases last. func (s *Service) preparePromptDispatchCallback( promptCtx context.Context, taskID, sessionID string, session *models.TaskSession, rollback promptClaimRollback, options promptTaskOptions, foregroundDispatch *foregroundDispatch, releaseDispatchGuard func(), + resumeAttempt *resumeAttempt, ) (onDispatched func(), dispatchOutcome *promptDispatchOutcome) { s.bindPromptTurnID(promptCtx, session, rollback.turnID) s.beginInteractivePromptAttempt( @@ -6323,9 +6325,10 @@ func (s *Service) preparePromptDispatchCallback( } } } + acceptedExecutionID := session.AgentExecutionID onDispatched = s.promptDispatchCallbackForIdentity( promptCtx, taskID, sessionID, rollback.sessionIdentity, - session.AgentExecutionID, rollback.reservedTurn, foregroundDispatch, dispatchOutcome, + acceptedExecutionID, rollback.reservedTurn, foregroundDispatch, dispatchOutcome, ) if options.afterDispatch != nil { originalOnDispatched := onDispatched @@ -6336,6 +6339,15 @@ func (s *Service) preparePromptDispatchCallback( dispatchOutcome.recordAccepted(errors.Join(durableOutcomeErr, publicationErr)) } } + if resumeAttempt != nil { + originalOnDispatched := onDispatched + onDispatched = func() { + // Transfer startup ownership before any durable post-acceptance hook + // or identity publication can fail and before the dispatch guard releases. + s.acceptResumeAttemptAtPromptAcceptance(acceptedExecutionID, resumeAttempt) + originalOnDispatched() + } + } if releaseDispatchGuard != nil { originalOnDispatched := onDispatched onDispatched = func() { @@ -6350,6 +6362,25 @@ func (s *Service) preparePromptDispatchCallback( return onDispatched, dispatchOutcome } +func (s *Service) acceptResumeAttemptAtPromptAcceptance( + expectedExecutionID string, + attempt *resumeAttempt, +) bool { + if attempt == nil { + return false + } + if expectedExecutionID == "" { + expectedExecutionID = attempt.execution() + } + if expectedExecutionID == "" { + return false + } + // Provider acceptance is the irreversible boundary for startup teardown. + // Use the execution admitted before the provider call; stale-incarnation + // publication remains separately fenced by promptDispatchCallbackForIdentity. + return s.resumeAttemptStore().accept(attempt, expectedExecutionID) +} + func (s *Service) finishPromptDispatchFailure( ctx context.Context, taskID, sessionID, prompt string, @@ -6782,8 +6813,12 @@ func (s *Service) promptDispatchCallbackForIdentity( func (s *Service) attemptModelSwitchForPrompt( ctx context.Context, taskID, sessionID, model, effectivePrompt string, session *models.TaskSession, foregroundDispatch *foregroundDispatch, runBeforeDispatch func() error, - runAfterDispatchAdmission func() error, + runAfterDispatchAdmission func() error, resumeAttempts ...*resumeAttempt, ) (result *PromptResult, handled bool, err error) { + var resumeAttempt *resumeAttempt + if len(resumeAttempts) > 0 { + resumeAttempt = resumeAttempts[0] + } if modelSwitchRequired(session, model) { s.beginInitialPromptAttempt(sessionID, s.isDynamicPromptSession(session)) admissionErr := runBeforeDispatch() @@ -6799,7 +6834,9 @@ func (s *Service) attemptModelSwitchForPrompt( return nil, true, admissionErr } } - result, handled, switchErr := s.trySwitchModelForPrompt(ctx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch) + result, handled, switchErr := s.trySwitchModelForPrompt( + ctx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch, resumeAttempt, + ) if handled && switchErr != nil { s.clearPromptAttemptEvidence(sessionID, "", 0) } @@ -6808,17 +6845,37 @@ func (s *Service) attemptModelSwitchForPrompt( // trySwitchModelForPrompt keeps foreground admission consistent when a model // switch either dispatches the prompt itself or fails before reaching the agent. -func (s *Service) trySwitchModelForPrompt(ctx context.Context, taskID, sessionID, model, prompt string, session *models.TaskSession, dispatch *foregroundDispatch) (*PromptResult, bool, error) { - result, switched, err := s.trySwitchModel(ctx, taskID, sessionID, model, prompt, session) +func (s *Service) trySwitchModelForPrompt( + ctx context.Context, + taskID, sessionID, model, prompt string, + session *models.TaskSession, + dispatch *foregroundDispatch, + resumeAttempt *resumeAttempt, +) (*PromptResult, bool, error) { + releaseHold, onDispatched, onFailure, holdErr := s.modelSwitchResumeCallbacks(session, model, resumeAttempt) + if holdErr != nil { + return nil, true, holdErr + } + result, switched, err := s.trySwitchModelWithDispatchCallbacks( + ctx, taskID, sessionID, model, prompt, session, onDispatched, onFailure, + ) if !switched && err == nil { + releaseHold() return nil, false, nil } if err != nil { + releaseHold() s.rollbackForegroundDispatchOnFailure(ctx, taskID, sessionID, dispatch) return result, true, err } if executionID, executionErr := s.agentManager.GetExecutionIDForSession(ctx, sessionID); executionErr == nil && executionID != "" { s.bindPromptAttemptToExecution(ctx, sessionID, executionID) + if resumeAttempt != nil { + // The replacement's initial prompt is dispatched asynchronously by + // lifecycle. Bind its immutable execution identity now; startup + // ownership transfers only from the callback installed for that prompt. + resumeAttempt.setExecutionID(executionID) + } } revealedBackground := s.acceptForegroundDispatch(dispatch) if revealedBackground || dispatch.yieldedBeforeBegin || s.acceptedForegroundDispatchClaim(dispatch) { @@ -6827,6 +6884,30 @@ func (s *Service) trySwitchModelForPrompt(ctx context.Context, taskID, sessionID return result, true, nil } +func (s *Service) modelSwitchResumeCallbacks( + session *models.TaskSession, + model string, + resumeAttempt *resumeAttempt, +) (release func(), onDispatched func(executionID string), onFailure func(), err error) { + release = func() {} + if resumeAttempt == nil || !modelSwitchRequired(session, model) { + return release, nil, nil, nil + } + registry := s.resumeAttemptStore() + if !registry.holdForInitialPrompt(resumeAttempt) { + s.cleanupCancelledResumeAttempt(resumeAttempt) + return nil, nil, nil, ErrResumeAttemptCancelled + } + release = func() { registry.releaseInitialPromptHold(resumeAttempt) } + onDispatched = func(executionID string) { + if s.acceptResumeAttemptAtPromptAcceptance(executionID, resumeAttempt) { + registry.finishAfterInitialPromptAcceptance(resumeAttempt) + } + } + onFailure = func() { registry.abortInitialPromptHold(resumeAttempt) } + return release, onDispatched, onFailure, nil +} + type promptClaimRollback struct { previousSessionState models.TaskSessionState previousTaskState v1.TaskState @@ -7888,6 +7969,16 @@ func modelSwitchRequired(session *models.TaskSession, model string) bool { } func (s *Service) trySwitchModel(ctx context.Context, taskID, sessionID, model, effectivePrompt string, session *models.TaskSession) (*PromptResult, bool, error) { + return s.trySwitchModelWithDispatchCallbacks(ctx, taskID, sessionID, model, effectivePrompt, session, nil, nil) +} + +func (s *Service) trySwitchModelWithDispatchCallbacks( + ctx context.Context, + taskID, sessionID, model, effectivePrompt string, + session *models.TaskSession, + onDispatched func(executionID string), + onFailure func(), +) (*PromptResult, bool, error) { if !modelSwitchRequired(session, model) { return nil, false, nil } @@ -7903,7 +7994,9 @@ func (s *Service) trySwitchModel(ctx context.Context, taskID, sessionID, model, zap.String("from", currentModel), zap.String("to", model)) switchCtx := context.WithoutCancel(ctx) - switchResult, err := s.executor.SwitchModel(switchCtx, taskID, sessionID, model, effectivePrompt) + switchResult, err := s.executor.SwitchModelWithDispatchCallbacks( + switchCtx, taskID, sessionID, model, effectivePrompt, onDispatched, onFailure, + ) if err != nil { return nil, true, fmt.Errorf("model switch failed: %w", err) } diff --git a/apps/backend/internal/orchestrator/task_operations_model_switch_cancellation_test.go b/apps/backend/internal/orchestrator/task_operations_model_switch_cancellation_test.go new file mode 100644 index 0000000000..663aa9dc1f --- /dev/null +++ b/apps/backend/internal/orchestrator/task_operations_model_switch_cancellation_test.go @@ -0,0 +1,113 @@ +package orchestrator + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.7 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 +func TestResumeAttempt_ModelSwitchFallbackCancellationBeforeInitialPromptAcceptance(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-model-switch-before-acceptance" + sessionID = "session-model-switch-before-acceptance" + oldExecution = "execution-model-switch-before-old" + newExecution = "execution-model-switch-before-new" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + session.AgentProfileSnapshot = map[string]interface{}{"model": "old-model"} + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session model: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, oldExecution) + + manager := &mockAgentManager{ + repoForExecutionLookup: repo, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + if err := repo.UpsertExecutorRunning(context.Background(), &models.ExecutorRunning{ + ID: sessionID, SessionID: sessionID, TaskID: taskID, + AgentExecutionID: newExecution, Status: "starting", + }); err != nil { + return nil, err + } + return &executor.LaunchAgentResponse{AgentExecutionID: newExecution}, nil + }, + } + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, State: v1.TaskStateInProgress} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, manager) + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + attempt, owner, err := svc.beginResumeAttempt(ctx, taskID, sessionID) + if err != nil || !owner { + t.Fatalf("begin model-switch attempt: attempt=%v owner=%v err=%v", attempt, owner, err) + } + attempt.setExecutionID(oldExecution) + t.Cleanup(func() { attempt.finish(svc.resumeAttemptStore()) }) + + _, handled, err := svc.trySwitchModelForPrompt( + ctx, taskID, sessionID, "new-model", "model-switch prompt", session, + &foregroundDispatch{}, attempt, + ) + if err != nil { + t.Fatalf("model-switch fallback: %v", err) + } + if !handled { + t.Fatal("model-switch fallback was not handled") + } + manager.mu.Lock() + onDispatched := manager.initialPromptDispatchCallback + started := append([]string(nil), manager.startAgentProcessCalls...) + manager.mu.Unlock() + if len(started) != 1 || started[0] != newExecution { + t.Fatalf("StartAgentProcess calls = %#v, want [%q]", started, newExecution) + } + if onDispatched == nil { + t.Fatal("model-switch fallback did not retain the initial-prompt callback") + } + // promptTask returns before lifecycle's asynchronous initial prompt callback. + // Its normal deferred finish must leave the attempt cancellable in this gap. + attempt.finish(svc.resumeAttemptStore()) + if _, active := svc.resumeAttemptStore().current(sessionID); !active { + t.Fatal("prompt-task completion removed the pending model-switch attempt before acceptance") + } + + registry := svc.resumeAttemptStore() + registry.mu.Lock() + acceptedBeforeCancel := attempt.accepted + registry.mu.Unlock() + if acceptedBeforeCancel { + t.Fatal("startup ownership transferred before the provider accepted the initial prompt") + } + svc.invalidateResumeAttempt(sessionID) + svc.cleanupCancelledResumeAttempt(attempt) + + manager.mu.Lock() + stops := append([]stopAgentCall(nil), manager.stopAgentWithReasonArgs...) + manager.mu.Unlock() + if len(stops) != 1 { + t.Fatalf("forced cleanup calls = %#v, want one call", stops) + } + if stops[0].ExecutionID != newExecution || !stops[0].Force { + t.Fatalf("forced cleanup call = %#v, want execution %q with force", stops[0], newExecution) + } + + // A delayed lifecycle callback from the cancelled startup cannot revive + // provider ownership after exact-execution cleanup has won. + onDispatched() + registry.mu.Lock() + acceptedAfterLateCallback := attempt.accepted + registry.mu.Unlock() + if acceptedAfterLateCallback { + t.Fatal("late initial-prompt callback revived a cancelled resume attempt") + } +} diff --git a/apps/backend/internal/orchestrator/task_operations_resumed_turn_cancellation_test.go b/apps/backend/internal/orchestrator/task_operations_resumed_turn_cancellation_test.go new file mode 100644 index 0000000000..40df044a81 --- /dev/null +++ b/apps/backend/internal/orchestrator/task_operations_resumed_turn_cancellation_test.go @@ -0,0 +1,833 @@ +package orchestrator + +import ( + "context" + "errors" + "fmt" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/messagequeue" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// acceptedTurnPromptManager lets the first provider turn report acceptance +// before it waits for the cancellation owner to release the turn. +type acceptedTurnPromptManager struct { + *mockAgentManager + accepted chan struct{} + release chan struct{} + releaseOnce sync.Once + firstPrompt atomic.Bool + firstPromptErr error + beforeDispatch func() +} + +func (m *acceptedTurnPromptManager) PromptAgentWithDispatchCallback( + ctx context.Context, + executionID, prompt string, + attachments []v1.MessageAttachment, + dispatchOnly bool, + onDispatched func(), +) (*executor.PromptResult, error) { + _, err := m.PromptAgent(ctx, executionID, prompt, attachments, dispatchOnly) + if err != nil { + return nil, err + } + first := m.firstPrompt.CompareAndSwap(false, true) + if first && m.beforeDispatch != nil { + m.beforeDispatch() + } + if onDispatched != nil { + onDispatched() + } + if !first { + return &executor.PromptResult{ + StopReason: "end_turn", + AgentMessage: "follow-up response", + }, nil + } + close(m.accepted) + <-m.release + return &executor.PromptResult{StopReason: "cancelled"}, m.firstPromptErr +} + +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.7 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.9 +func TestPromptTask_ResumedTurnCancelPreservesExecution(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-resumed-turn-cancel" + sessionID = "session-resumed-turn-cancel" + executionID = "execution-resumed-turn-cancel" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + session.AgentProfileID = "profile1" + session.AgentExecutionID = executionID + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session profile: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, "execution-before-resume") + + var launches atomic.Int32 + var currentExecution atomic.Value + currentExecution.Store(executionID) + agentRunning := &atomic.Bool{} + baseManager := &mockAgentManager{ + repoForExecutionLookup: repo, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return currentExecution.Load().(string), nil + }, + isAgentRunningFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + isAgentReadyFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launches.Add(1) + currentExecution.Store(executionID) + agentRunning.Store(true) + if err := repo.UpsertExecutorRunning(context.Background(), &models.ExecutorRunning{ + ID: sessionID, SessionID: sessionID, TaskID: taskID, + AgentExecutionID: executionID, Status: "ready", + }); err != nil { + return nil, err + } + if err := repo.UpdateTaskSessionState( + context.Background(), req.SessionID, models.TaskSessionStateWaitingForInput, "", + ); err != nil { + return nil, err + } + return &executor.LaunchAgentResponse{AgentExecutionID: executionID}, nil + }, + } + manager := &acceptedTurnPromptManager{ + mockAgentManager: baseManager, + accepted: make(chan struct{}), + release: make(chan struct{}), + } + baseManager.cancelAgentFunc = func(context.Context, string) error { + manager.releaseOnce.Do(func() { close(manager.release) }) + return nil + } + baseManager.stopAgentWithReasonFunc = func(_ context.Context, _ string, _ string, _ bool) error { + agentRunning.Store(false) + return nil + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, State: v1.TaskStateInProgress} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, manager) + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + + firstPromptDone := make(chan error, 1) + go func() { + _, promptErr := svc.PromptTask(ctx, taskID, sessionID, "first resumed prompt", "", false, nil, false) + firstPromptDone <- promptErr + }() + select { + case <-manager.accepted: + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("resumed prompt did not reach provider acceptance") + } + acceptedAttempt, active := svc.resumeAttemptStore().current(sessionID) + if !active || acceptedAttempt == nil { + t.Fatal("accepted resume attempt was removed before the provider turn returned") + } + registry := svc.resumeAttemptStore() + registry.mu.Lock() + isAccepted := acceptedAttempt.accepted + registry.mu.Unlock() + if !isAccepted { + t.Fatal("provider acceptance did not transfer startup ownership") + } + + cancelDone := make(chan error, 1) + go func() { cancelDone <- svc.CancelAgent(ctx, sessionID) }() + select { + case cancelErr := <-cancelDone: + if cancelErr != nil { + t.Fatalf("CancelAgent: %v", cancelErr) + } + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("CancelAgent did not settle accepted turn") + } + select { + case promptErr := <-firstPromptDone: + if promptErr != nil { + t.Fatalf("accepted cancelled prompt: %v", promptErr) + } + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("accepted prompt did not return after cancellation") + } + + baseManager.mu.Lock() + forcedStops := append([]stopAgentCall(nil), baseManager.stopAgentWithReasonArgs...) + baseManager.mu.Unlock() + if len(forcedStops) != 0 { + t.Fatalf("forced execution stops after accepted cancellation = %d, want 0: %#v", len(forcedStops), forcedStops) + } + if got := launches.Load(); got != 1 { + t.Fatalf("resume launches after accepted cancellation = %d, want 1", got) + } + assertResumeSessionState(t, repo, sessionID, models.TaskSessionStateWaitingForInput) + running, err := repo.GetExecutorRunningBySessionID(ctx, sessionID) + if err != nil { + t.Fatalf("load preserved execution: %v", err) + } + if running == nil || running.AgentExecutionID != executionID { + t.Fatalf("preserved execution = %#v, want %q", running, executionID) + } + + if _, err := svc.PromptTask(ctx, taskID, sessionID, "follow-up prompt", "", false, nil, false); err != nil { + t.Fatalf("follow-up prompt: %v", err) + } + if got := launches.Load(); got != 1 { + t.Fatalf("follow-up caused %d resume launches, want 1", got) + } + baseManager.mu.Lock() + promptCalls := append([]promptCall(nil), baseManager.capturedPromptCalls...) + baseManager.mu.Unlock() + if len(promptCalls) != 2 { + t.Fatalf("prompt calls = %d, want 2: %#v", len(promptCalls), promptCalls) + } + for index, call := range promptCalls { + if call.ExecutionID != executionID { + t.Errorf("prompt %d execution ID = %q, want %q", index, call.ExecutionID, executionID) + } + } +} + +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.7 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.9 +func TestPromptTask_QueuedAcceptedTurnIdentityReadFailurePreservesExecution(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-queued-accepted-identity-failure" + sessionID = "session-queued-accepted-identity-failure" + executionID = "execution-queued-accepted-identity-failure" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + session.AgentProfileID = "profile1" + session.AgentExecutionID = executionID + if session.QueueIncarnationID == "" { + session.QueueIncarnationID = "queue-incarnation-accepted-identity-failure" + } + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session profile: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, "execution-before-queued-resume") + + var launches atomic.Int32 + var currentExecution atomic.Value + currentExecution.Store(executionID) + agentRunning := &atomic.Bool{} + identityReadFailure := &atomic.Bool{} + var identityReadFailures atomic.Int32 + queueRepo := messagequeue.NewMemoryRepositoryWithAuthority( + func(ctx context.Context, requestedTaskID, requestedSessionID string) (messagequeue.QueueSessionIdentity, error) { + if identityReadFailure.Load() { + identityReadFailures.Add(1) + return messagequeue.QueueSessionIdentity{}, errors.New("injected session identity read failure") + } + current, resolveErr := repo.GetTaskSession(ctx, requestedSessionID) + if resolveErr != nil { + return messagequeue.QueueSessionIdentity{}, resolveErr + } + if current == nil || current.TaskID != requestedTaskID || current.QueueIncarnationID == "" { + return messagequeue.QueueSessionIdentity{}, messagequeue.ErrSessionIdentityMismatch + } + return messagequeue.QueueSessionIdentity{ + TaskID: requestedTaskID, + SessionID: requestedSessionID, + SessionIncarnationID: current.QueueIncarnationID, + }, nil + }, + ) + baseManager := &mockAgentManager{ + repoForExecutionLookup: repo, + getExecutionIDForSessionFunc: func(context.Context, string) (string, error) { + return currentExecution.Load().(string), nil + }, + isAgentRunningFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + isAgentReadyFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launches.Add(1) + currentExecution.Store(executionID) + agentRunning.Store(true) + if err := repo.UpsertExecutorRunning(context.Background(), &models.ExecutorRunning{ + ID: sessionID, SessionID: sessionID, TaskID: taskID, + AgentExecutionID: executionID, Status: "ready", + }); err != nil { + return nil, err + } + if err := repo.UpdateTaskSessionState( + context.Background(), req.SessionID, models.TaskSessionStateWaitingForInput, "", + ); err != nil { + return nil, err + } + return &executor.LaunchAgentResponse{AgentExecutionID: executionID}, nil + }, + } + manager := &acceptedTurnPromptManager{ + mockAgentManager: baseManager, + accepted: make(chan struct{}), + release: make(chan struct{}), + beforeDispatch: func() { identityReadFailure.Store(true) }, + } + baseManager.cancelAgentFunc = func(context.Context, string) error { + manager.releaseOnce.Do(func() { close(manager.release) }) + return nil + } + baseManager.stopAgentWithReasonFunc = func(_ context.Context, _ string, _ string, _ bool) error { + agentRunning.Store(false) + return nil + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, State: v1.TaskStateInProgress} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, manager) + svc.messageQueue = messagequeue.NewService(queueRepo, messagequeue.DefaultMaxPerSession, testLogger()) + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + t.Cleanup(func() { + svc.clearAcceptedQueuedDispatch(sessionID) + svc.stopSendNowWorkers() + }) + + identity, err := svc.messageQueue.ResolveSessionIdentity(ctx, taskID, sessionID) + if err != nil { + t.Fatalf("resolve queued session identity: %v", err) + } + if identity.SessionIncarnationID == "" { + t.Fatal("queued session identity has no incarnation") + } + if err := svc.messageQueue.SetAutoRun(ctx, sessionID, true); err != nil { + t.Fatalf("enable auto-run: %v", err) + } + if _, err := svc.messageQueue.QueueMessageWithMetadata( + ctx, sessionID, taskID, "queued accepted identity failure prompt", "", + messagequeue.QueuedByUser, false, nil, nil, + ); err != nil { + t.Fatalf("queue prompt: %v", err) + } + queued, ok, _, err := svc.messageQueue.ReserveQueuedWithAutoRunForSession(ctx, identity) + if err != nil || !ok || queued == nil { + t.Fatalf("reserve queued prompt: queued=%+v ok=%v err=%v", queued, ok, err) + } + if reservation := svc.markQueuedDispatchInFlightWithIdentityLocked(identity, queued.ID, queued); reservation == nil { + t.Fatal("mark queued dispatch in flight returned nil") + } + + var afterDispatchSawAccepted atomic.Bool + promptDone := make(chan error, 1) + go func() { + _, promptErr := svc.promptTask( + ctx, taskID, sessionID, "queued accepted identity failure prompt", "", false, nil, true, + launchOriginAutomatic, + promptTaskOptions{ + claimEntryID: queued.ID, + expectedSessionIdentity: &identity, + afterDispatch: func() error { + afterDispatchSawAccepted.Store(activeResumeAttemptAccepted(svc, sessionID)) + return nil + }, + }, + ) + promptDone <- promptErr + }() + select { + case <-manager.accepted: + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("queued resumed prompt did not reach provider acceptance") + } + if identityReadFailures.Load() == 0 { + t.Fatal("acceptance callback did not exercise the injected identity read failure") + } + if !afterDispatchSawAccepted.Load() { + t.Fatal("afterDispatch ran before startup ownership transferred") + } + assertActiveResumeAttemptAccepted(t, svc, sessionID) + + cancelDone := make(chan error, 1) + go func() { cancelDone <- svc.CancelAgent(ctx, sessionID) }() + select { + case cancelErr := <-cancelDone: + if cancelErr != nil { + t.Fatalf("CancelAgent: %v", cancelErr) + } + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("CancelAgent did not settle queued accepted turn") + } + select { + case promptErr := <-promptDone: + if promptErr != nil { + t.Fatalf("queued accepted prompt: %v", promptErr) + } + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("queued accepted prompt did not return after cancellation") + } + + baseManager.mu.Lock() + forcedStops := append([]stopAgentCall(nil), baseManager.stopAgentWithReasonArgs...) + baseManager.mu.Unlock() + if len(forcedStops) != 0 { + t.Fatalf("forced execution stops after queued accepted cancellation = %d, want 0: %#v", len(forcedStops), forcedStops) + } + if got := launches.Load(); got != 1 { + t.Fatalf("queued resume launches after accepted cancellation = %d, want 1", got) + } + assertResumeSessionState(t, repo, sessionID, models.TaskSessionStateWaitingForInput) + running, err := repo.GetExecutorRunningBySessionID(ctx, sessionID) + if err != nil { + t.Fatalf("load preserved queued execution: %v", err) + } + if running == nil || running.AgentExecutionID != executionID { + t.Fatalf("preserved queued execution = %#v, want %q", running, executionID) + } + + if _, err := svc.PromptTask(ctx, taskID, sessionID, "queued identity follow-up", "", false, nil, false); err != nil { + t.Fatalf("queued identity follow-up prompt: %v", err) + } + if got := launches.Load(); got != 1 { + t.Fatalf("queued identity follow-up caused %d resume launches, want 1", got) + } + baseManager.mu.Lock() + promptCalls := append([]promptCall(nil), baseManager.capturedPromptCalls...) + baseManager.mu.Unlock() + if len(promptCalls) != 2 { + t.Fatalf("queued identity prompt calls = %d, want 2: %#v", len(promptCalls), promptCalls) + } + for index, call := range promptCalls { + if call.ExecutionID != executionID { + t.Errorf("queued identity prompt %d execution ID = %q, want %q", index, call.ExecutionID, executionID) + } + } +} + +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.7 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.9 +func TestResumeTaskSessionAndPrompt_AcceptedTurnCancelPreservesExecution(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-compound-resumed-turn-cancel" + sessionID = "session-compound-resumed-turn-cancel" + executionID = "execution-compound-resumed-turn-cancel" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + session.AgentProfileID = "profile1" + session.AgentExecutionID = executionID + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session profile: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + + var launches atomic.Int32 + agentRunning := &atomic.Bool{} + baseManager := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunningFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + isAgentReadyFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launches.Add(1) + agentRunning.Store(true) + if err := repo.UpdateTaskSessionState( + context.Background(), req.SessionID, models.TaskSessionStateWaitingForInput, "", + ); err != nil { + return nil, err + } + return &executor.LaunchAgentResponse{AgentExecutionID: executionID}, nil + }, + } + manager := &acceptedTurnPromptManager{ + mockAgentManager: baseManager, + accepted: make(chan struct{}), + release: make(chan struct{}), + firstPromptErr: fmt.Errorf("prompt abandoned after cancel: %w", lifecycle.ErrCancelEscalated), + } + baseManager.cancelAgentFunc = func(context.Context, string) error { + manager.releaseOnce.Do(func() { close(manager.release) }) + return nil + } + baseManager.stopAgentWithReasonFunc = func(_ context.Context, _ string, _ string, _ bool) error { + agentRunning.Store(false) + return nil + } + + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, State: v1.TaskStateInProgress} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, manager) + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + + resultDone := make(chan struct { + result *PromptResult + err error + }, 1) + go func() { + result, promptErr := svc.ResumeTaskSessionAndPrompt( + ctx, taskID, sessionID, "first compound prompt", "", false, nil, + ) + resultDone <- struct { + result *PromptResult + err error + }{result: result, err: promptErr} + }() + select { + case <-manager.accepted: + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("compound resume did not reach provider acceptance") + } + assertActiveResumeAttemptAccepted(t, svc, sessionID) + + cancelDone := make(chan error, 1) + go func() { cancelDone <- svc.CancelAgent(ctx, sessionID) }() + select { + case cancelErr := <-cancelDone: + if cancelErr != nil { + t.Fatalf("CancelAgent: %v", cancelErr) + } + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("CancelAgent did not settle the accepted compound turn") + } + select { + case outcome := <-resultDone: + if outcome.err == nil || !errors.Is(outcome.err, lifecycle.ErrCancelEscalated) { + t.Fatalf("accepted compound error = %v, want wrapped ErrCancelEscalated", outcome.err) + } + case <-time.After(resumeCancellationTestTimeout(t)): + t.Fatal("accepted compound prompt did not return after cancellation") + } + + baseManager.mu.Lock() + forcedStops := append([]stopAgentCall(nil), baseManager.stopAgentWithReasonArgs...) + baseManager.mu.Unlock() + if len(forcedStops) != 0 { + t.Fatalf("forced execution stops after accepted compound cancellation = %d, want 0: %#v", len(forcedStops), forcedStops) + } + if got := launches.Load(); got != 1 { + t.Fatalf("compound resume launches = %d, want 1", got) + } + assertResumeSessionState(t, repo, sessionID, models.TaskSessionStateWaitingForInput) + + if _, err := svc.PromptTask(ctx, taskID, sessionID, "compound follow-up", "", false, nil, false); err != nil { + t.Fatalf("compound follow-up prompt: %v", err) + } + if got := launches.Load(); got != 1 { + t.Fatalf("compound follow-up caused %d resume launches, want 1", got) + } +} + +func TestResumeAttempt_AcceptanceCancellationOrdering(t *testing.T) { + t.Run("provider acceptance wins", func(t *testing.T) { + registry := newResumeAttemptRegistry() + attempt, owner := registry.begin(context.Background(), "task-acceptance-order", "session-acceptance-order") + if !owner { + t.Fatal("acceptance-order attempt was not admitted") + } + attempt.setExecutionID("execution-acceptance-order") + if !registry.accept(attempt, "execution-acceptance-order") { + t.Fatal("provider acceptance did not win before cancellation") + } + if registry.invalidate(attempt.sessionID) { + t.Fatal("accepted attempt remained startup-cancellable") + } + if err := attempt.context().Err(); err != nil { + t.Fatalf("accepted attempt context was cancelled: %v", err) + } + if !registry.accept(attempt, "execution-acceptance-order") { + t.Fatal("repeated acceptance was not idempotent") + } + if registry.canCleanup(attempt) { + t.Fatal("accepted attempt retained startup cleanup authority") + } + attempt.finish(registry) + if registry.canCleanupIdentity( + attempt.sessionID, "execution-acceptance-order", attempt.identity(), + ) { + t.Fatal("accepted tombstone retained startup cleanup authority") + } + }) + + t.Run("startup cancellation wins", func(t *testing.T) { + registry := newResumeAttemptRegistry() + attempt, owner := registry.begin(context.Background(), "task-cancellation-order", "session-cancellation-order") + if !owner { + t.Fatal("cancellation-order attempt was not admitted") + } + attempt.setExecutionID("execution-cancellation-order") + if !registry.invalidate(attempt.sessionID) { + t.Fatal("startup attempt was not invalidated") + } + if registry.accept(attempt, "execution-cancellation-order") { + t.Fatal("late provider acceptance revived cancelled startup") + } + if attempt.context().Err() == nil { + t.Fatal("startup cancellation did not cancel the attempt context") + } + if !registry.canCleanup(attempt) { + t.Fatal("cancelled startup lost cleanup authority") + } + attempt.finish(registry) + }) +} + +func TestResumeAttempt_AcceptedExecutionEventsRemainValid(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-accepted-events" + sessionID = "session-accepted-events" + executionID = "execution-accepted-events" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo()) + attempt, owner, err := svc.beginResumeAttempt(ctx, taskID, sessionID) + if err != nil || !owner { + t.Fatalf("begin accepted event attempt: attempt=%v owner=%v err=%v", attempt, owner, err) + } + attempt.setExecutionID(executionID) + if !svc.resumeAttemptStore().accept(attempt, executionID) { + t.Fatal("accepted event attempt was not accepted") + } + attemptID := attempt.identity() + attempt.finish(svc.resumeAttemptStore()) + + if !svc.resumeAttemptAllowsExecution(sessionID, executionID, attemptID) { + t.Fatal("accepted execution event was rejected after attempt completion") + } + if svc.resumeAttemptAllowsExecution(sessionID, executionID, "resume-999999") { + t.Fatal("unknown accepted event identity was allowed") + } + svc.handleACPSessionCreated(ctx, watcher.ACPSessionEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + AttemptID: attemptID, + ACPSessionID: "accepted-provider-session", + }) + assertResumeToken(t, repo, sessionID, "accepted-provider-session") + svc.handleACPSessionCreated(ctx, watcher.ACPSessionEventData{ + TaskID: taskID, + SessionID: sessionID, + AgentExecutionID: executionID, + AttemptID: "resume-999999", + ACPSessionID: "stale-provider-session", + }) + assertResumeToken(t, repo, sessionID, "accepted-provider-session") +} + +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 +func TestResumeAttempt_AcceptedPublicationFailureDoesNotCleanup(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-accepted-publication-failure" + sessionID = "session-accepted-publication-failure" + executionID = "execution-accepted-publication-failure" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + session.AgentProfileID = "profile1" + session.AgentExecutionID = executionID + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session profile: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + + var launches atomic.Int32 + agentRunning := &atomic.Bool{} + publicationErr := errors.New("accepted turn publication failed") + agentManager := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunningFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + isAgentReadyFn: func(context.Context, string) bool { + return agentRunning.Load() + }, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + launches.Add(1) + agentRunning.Store(true) + if err := repo.UpdateTaskSessionState( + context.Background(), req.SessionID, models.TaskSessionStateWaitingForInput, "", + ); err != nil { + return nil, err + } + return &executor.LaunchAgentResponse{AgentExecutionID: executionID}, nil + }, + promptResult: &executor.PromptResult{StopReason: "end_turn", AgentMessage: "accepted"}, + } + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, State: v1.TaskStateInProgress} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentManager) + svc.executor = executor.NewExecutor(agentManager, repo, testLogger(), executor.ExecutorConfig{}) + svc.turnService = failingReservedTurnPublisher{ + TurnService: &repoBackedTurnService{repo: repo}, + err: publicationErr, + } + + _, err = svc.promptTask( + ctx, taskID, sessionID, "accepted publication prompt", "", false, nil, false, + launchOriginManual, + promptTaskOptions{reserveTurnUntilDispatch: true}, + ) + if !errors.Is(err, publicationErr) { + t.Fatalf("promptTask error = %v, want publication error %v", err, publicationErr) + } + var accepted interface{ DetachedResumeAccepted() bool } + if !errors.As(err, &accepted) || !accepted.DetachedResumeAccepted() { + t.Fatalf("promptTask error = %v, want accepted-dispatch marker", err) + } + if got := launches.Load(); got != 1 { + t.Fatalf("resume launches = %d, want 1", got) + } + agentManager.mu.Lock() + forcedStops := append([]stopAgentCall(nil), agentManager.stopAgentWithReasonArgs...) + agentManager.mu.Unlock() + if len(forcedStops) != 0 { + t.Fatalf("accepted publication failure forced %d startup stops: %#v", len(forcedStops), forcedStops) + } + if pending := svc.reservedPromptTurnID(sessionID); pending != "" { + t.Fatalf("accepted publication failure left turn rollback-eligible: %q", pending) + } +} + +// @covers AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 +func TestResumeAttempt_ModelSwitchFallbackTransfersAcceptance(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-model-switch-acceptance" + sessionID = "session-model-switch-acceptance" + oldExecution = "execution-model-switch-old" + newExecution = "execution-model-switch-new" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("load session: %v", err) + } + session.AgentProfileSnapshot = map[string]interface{}{"model": "old-model"} + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session model: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, oldExecution) + + manager := &mockAgentManager{ + repoForExecutionLookup: repo, + isAgentRunning: true, + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + if err := repo.UpsertExecutorRunning(context.Background(), &models.ExecutorRunning{ + ID: sessionID, SessionID: sessionID, TaskID: taskID, + AgentExecutionID: newExecution, Status: "ready", + }); err != nil { + return nil, err + } + return &executor.LaunchAgentResponse{AgentExecutionID: newExecution}, nil + }, + } + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, State: v1.TaskStateInProgress} + svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, manager) + svc.executor = executor.NewExecutor(manager, repo, testLogger(), executor.ExecutorConfig{}) + attempt, owner, err := svc.beginResumeAttempt(ctx, taskID, sessionID) + if err != nil || !owner { + t.Fatalf("begin model-switch attempt: attempt=%v owner=%v err=%v", attempt, owner, err) + } + attempt.setExecutionID(oldExecution) + t.Cleanup(func() { attempt.finish(svc.resumeAttemptStore()) }) + + result, handled, err := svc.trySwitchModelForPrompt( + ctx, taskID, sessionID, "new-model", "model-switch prompt", session, + &foregroundDispatch{}, attempt, + ) + if err != nil { + t.Fatalf("model-switch fallback: %v", err) + } + if !handled { + t.Fatal("model-switch fallback was not handled") + } + if result == nil || result.StopReason != "model_switched" { + t.Fatalf("model-switch result = %#v, want model_switched", result) + } + manager.mu.Lock() + onDispatched := manager.initialPromptDispatchCallback + manager.mu.Unlock() + if onDispatched == nil { + t.Fatal("model-switch fallback did not register its initial-prompt acceptance callback") + } + registry := svc.resumeAttemptStore() + registry.mu.Lock() + acceptedBeforeDispatch := attempt.accepted + registry.mu.Unlock() + if acceptedBeforeDispatch { + t.Fatal("model-switch fallback transferred startup ownership before provider acceptance") + } + onDispatched() + if got := attempt.execution(); got != newExecution { + t.Fatalf("resume attempt execution = %q, want %q", got, newExecution) + } + registry.mu.Lock() + accepted := attempt.accepted + registry.mu.Unlock() + if !accepted { + t.Fatal("model-switch fallback did not transfer startup ownership") + } + if err := attempt.validate(registry); err != nil { + t.Fatalf("accepted model-switch attempt validation: %v", err) + } +} + +func assertActiveResumeAttemptAccepted(t *testing.T, svc *Service, sessionID string) { + t.Helper() + if !activeResumeAttemptAccepted(svc, sessionID) { + t.Fatalf("provider acceptance did not transfer startup ownership for %s", sessionID) + } +} + +func activeResumeAttemptAccepted(svc *Service, sessionID string) bool { + attempt, active := svc.resumeAttemptStore().current(sessionID) + if !active || attempt == nil { + return false + } + registry := svc.resumeAttemptStore() + registry.mu.Lock() + defer registry.mu.Unlock() + return attempt.accepted +} diff --git a/apps/web/e2e/helpers/api-client.ts b/apps/web/e2e/helpers/api-client.ts index 08e0335274..b9d9471269 100644 --- a/apps/web/e2e/helpers/api-client.ts +++ b/apps/web/e2e/helpers/api-client.ts @@ -2522,6 +2522,7 @@ export class ApiClient { id: string; task_id: string; queue_incarnation_id: string; + agent_execution_id?: string; agent_profile_id?: string; executor_id?: string; executor_profile_id?: string; diff --git a/apps/web/e2e/helpers/session-resume-prompt-queue.ts b/apps/web/e2e/helpers/session-resume-prompt-queue.ts index 4ea13f9d49..6b67319222 100644 --- a/apps/web/e2e/helpers/session-resume-prompt-queue.ts +++ b/apps/web/e2e/helpers/session-resume-prompt-queue.ts @@ -182,3 +182,80 @@ export async function waitForQueuedCount( }) .toBe(count); } + +export type SessionRuntimeIdentity = { + executionId: string; + acpSessionId: string; +}; + +/** Read the durable execution and provider conversation identities for a session. */ +export async function readSessionRuntimeIdentity( + apiClient: ApiClient, + taskId: string, + sessionId: string, +): Promise { + const { sessions } = await apiClient.listTaskSessions(taskId); + const session = sessions.find((candidate) => candidate.id === sessionId); + const status = await apiClient.wsRequest<{ acp_session_id?: string }>("task.session.status", { + task_id: taskId, + session_id: sessionId, + }); + if (!session?.agent_execution_id) { + throw new Error(`Session ${sessionId} has no durable agent execution identity`); + } + if (!status.acp_session_id) { + throw new Error(`Session ${sessionId} has no durable ACP conversation identity`); + } + return { + executionId: session.agent_execution_id, + acpSessionId: status.acp_session_id, + }; +} + +/** Read message IDs whose persisted content contains a turn-specific marker. */ +export async function readSessionMessageIdsContaining( + apiClient: ApiClient, + sessionId: string, + marker: string, +): Promise> { + const { messages } = await apiClient.listSessionMessages(sessionId); + return new Set( + messages.filter((message) => message.content.includes(marker)).map((message) => message.id), + ); +} + +/** Wait until a new persisted message contains the marker. */ +export async function waitForNewSessionMessage( + apiClient: ApiClient, + sessionId: string, + previousMessageIds: ReadonlySet, + marker: string, + timeout = 30_000, +): Promise { + await expect + .poll( + async () => { + const { messages } = await apiClient.listSessionMessages(sessionId); + return messages.some( + (message) => !previousMessageIds.has(message.id) && message.content.includes(marker), + ); + }, + { + timeout, + message: `Waiting for a new session message containing ${marker}`, + }, + ) + .toBe(true); +} + +/** Count persisted resume boot messages without relying on the UI deduplication. */ +export async function countResumeBootMessages( + apiClient: ApiClient, + sessionId: string, +): Promise { + const { messages } = await apiClient.listSessionMessages(sessionId); + return messages.filter( + (message) => + message.metadata?.script_type === "agent_boot" && message.metadata?.is_resuming === true, + ).length; +} diff --git a/apps/web/e2e/tests/session/mobile-session-resume-recovery.spec.ts b/apps/web/e2e/tests/session/mobile-session-resume-recovery.spec.ts index c2da7a3cbc..677206873a 100644 --- a/apps/web/e2e/tests/session/mobile-session-resume-recovery.spec.ts +++ b/apps/web/e2e/tests/session/mobile-session-resume-recovery.spec.ts @@ -6,8 +6,12 @@ import { waitForSessionState } from "../../helpers/session"; import { SessionPage } from "../../pages/session-page"; import { cleanupDelayedResumeFixture, + countResumeBootMessages, createFailOnResumeProfile, + readSessionMessageIdsContaining, + readSessionRuntimeIdentity, seedDelayedResumeFixture, + waitForNewSessionMessage, waitForSessionReady, } from "../../helpers/session-resume-prompt-queue"; import { @@ -101,6 +105,116 @@ test.describe("mobile: delayed resume cancellation", () => { await cleanupDelayedResumeFixture(apiClient, fixture); } }); + + test("pausing an accepted lazy resume preserves the runtime for later turns", async ({ + testPage, + apiClient, + seedData, + backend, + }) => { + test.setTimeout(150_000); + await apiClient.saveUserSettings({ prevent_auto_start_agent_on_open: true }); + + try { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Mobile accepted lazy resume pause recovery", + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + if (!task.session_id) throw new Error("mobile accepted lazy resume task has no session_id"); + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ + timeout: 30_000, + }); + await session.waitForChatIdle({ timeout: 30_000 }); + + await backend.restart(); + await testPage.reload(); + await session.waitForLoad(); + await expect(testPage.getByTestId("composer-agent-start-hint")).toBeVisible({ + timeout: 60_000, + }); + + const resumeBootsBeforeMessage = await countResumeBootMessages(apiClient, task.session_id); + + // Provider output proves that the resumed prompt crossed acceptance + // before the touch cancellation is sent. + await session.sendMessageViaButton("/slow 8s"); + await expect(session.chat.getByText("Running slow response", { exact: false })).toBeVisible({ + timeout: 30_000, + }); + const initialRuntimeIdentity = await readSessionRuntimeIdentity( + apiClient, + task.id, + task.session_id, + ); + await session.cancelAgentButton().tap(); + await waitForSessionState(apiClient, { + taskId: task.id, + sessionId: task.session_id, + expectedState: "WAITING_FOR_INPUT", + message: "Waiting for mobile accepted lazy resume cancellation", + timeout: 30_000, + }); + await expect(session.idleInput()).toBeVisible({ timeout: 30_000 }); + + expect(await readSessionRuntimeIdentity(apiClient, task.id, task.session_id)).toEqual( + initialRuntimeIdentity, + ); + const resumeBootsAfterFirstPause = await countResumeBootMessages(apiClient, task.session_id); + expect(resumeBootsAfterFirstPause).toBe(resumeBootsBeforeMessage + 1); + + const slowResponseMessageIdsBeforeSecond = await readSessionMessageIdsContaining( + apiClient, + task.session_id, + "Running slow response", + ); + expect(slowResponseMessageIdsBeforeSecond.size).toBeGreaterThan(0); + await session.sendMessageViaButton("/slow 8s"); + await waitForNewSessionMessage( + apiClient, + task.session_id, + slowResponseMessageIdsBeforeSecond, + "Running slow response", + ); + await session.cancelAgentButton().tap(); + await waitForSessionState(apiClient, { + taskId: task.id, + sessionId: task.session_id, + expectedState: "WAITING_FOR_INPUT", + message: "Waiting for the later mobile accepted pause", + timeout: 30_000, + }); + + expect(await readSessionRuntimeIdentity(apiClient, task.id, task.session_id)).toEqual( + initialRuntimeIdentity, + ); + expect(await countResumeBootMessages(apiClient, task.session_id)).toBe( + resumeBootsAfterFirstPause, + ); + + await session.sendMessageViaButton("/e2e:simple-message"); + await session.expectChatResponseVisible("simple mock response", 1, { timeout: 30_000 }); + expect(await readSessionRuntimeIdentity(apiClient, task.id, task.session_id)).toEqual( + initialRuntimeIdentity, + ); + expect(await countResumeBootMessages(apiClient, task.session_id)).toBe( + resumeBootsAfterFirstPause, + ); + await assertNoDocumentHorizontalOverflow(testPage, "mobile accepted lazy resume pause"); + } finally { + await apiClient.saveUserSettings({ prevent_auto_start_agent_on_open: false }); + } + }); }); test.describe("mobile: failed resume recovery", () => { diff --git a/apps/web/e2e/tests/session/session-recovery.spec.ts b/apps/web/e2e/tests/session/session-recovery.spec.ts index e98429b471..8591c459a0 100644 --- a/apps/web/e2e/tests/session/session-recovery.spec.ts +++ b/apps/web/e2e/tests/session/session-recovery.spec.ts @@ -6,8 +6,12 @@ import { waitForSessionState } from "../../helpers/session"; import { SessionPage } from "../../pages/session-page"; import { cleanupDelayedResumeFixture, + countResumeBootMessages, createFailOnResumeProfile, + readSessionMessageIdsContaining, + readSessionRuntimeIdentity, seedDelayedResumeFixture, + waitForNewSessionMessage, waitForSessionReady, waitForQueuedCount, } from "../../helpers/session-resume-prompt-queue"; @@ -153,6 +157,116 @@ test.describe("Session recovery", () => { } }); + test("pausing an accepted lazy resume preserves the runtime for later turns", async ({ + testPage, + apiClient, + seedData, + backend, + }) => { + test.setTimeout(150_000); + await apiClient.saveUserSettings({ prevent_auto_start_agent_on_open: true }); + + try { + const task = await apiClient.createTaskWithAgent( + seedData.workspaceId, + "Accepted lazy resume pause recovery", + seedData.agentProfileId, + { + description: "/e2e:simple-message", + workflow_id: seedData.workflowId, + workflow_step_id: seedData.startStepId, + repository_ids: [seedData.repositoryId], + }, + ); + if (!task.session_id) throw new Error("accepted lazy resume task has no session_id"); + + await testPage.goto(`/t/${task.id}`); + const session = new SessionPage(testPage); + await session.waitForLoad(); + await expect(session.chat.getByText("simple mock response", { exact: false })).toBeVisible({ + timeout: 30_000, + }); + await session.waitForChatIdle({ timeout: 30_000 }); + + await backend.restart(); + await testPage.reload(); + await session.waitForLoad(); + await expect(testPage.getByTestId("composer-agent-start-hint")).toBeVisible({ + timeout: 60_000, + }); + + const resumeBootsBeforeMessage = await countResumeBootMessages(apiClient, task.session_id); + + // The slow response is the acceptance witness: provider output can only + // arrive after the resumed prompt crossed the dispatch callback. + await session.sendMessage("/slow 8s"); + await expect(session.chat.getByText("Running slow response", { exact: false })).toBeVisible({ + timeout: 30_000, + }); + const initialRuntimeIdentity = await readSessionRuntimeIdentity( + apiClient, + task.id, + task.session_id, + ); + await session.cancelAgentButton().click(); + await waitForSessionState(apiClient, { + taskId: task.id, + sessionId: task.session_id, + expectedState: "WAITING_FOR_INPUT", + message: "Waiting for accepted lazy resume cancellation", + timeout: 30_000, + }); + await expect(session.idleInput()).toBeVisible({ timeout: 30_000 }); + + expect(await readSessionRuntimeIdentity(apiClient, task.id, task.session_id)).toEqual( + initialRuntimeIdentity, + ); + const resumeBootsAfterFirstPause = await countResumeBootMessages(apiClient, task.session_id); + expect(resumeBootsAfterFirstPause).toBe(resumeBootsBeforeMessage + 1); + + // A second accepted pause must use the same process and conversation. + const slowResponseMessageIdsBeforeSecond = await readSessionMessageIdsContaining( + apiClient, + task.session_id, + "Running slow response", + ); + expect(slowResponseMessageIdsBeforeSecond.size).toBeGreaterThan(0); + await session.sendMessage("/slow 8s"); + await waitForNewSessionMessage( + apiClient, + task.session_id, + slowResponseMessageIdsBeforeSecond, + "Running slow response", + ); + await session.cancelAgentButton().click(); + await waitForSessionState(apiClient, { + taskId: task.id, + sessionId: task.session_id, + expectedState: "WAITING_FOR_INPUT", + message: "Waiting for the later accepted pause", + timeout: 30_000, + }); + + expect(await readSessionRuntimeIdentity(apiClient, task.id, task.session_id)).toEqual( + initialRuntimeIdentity, + ); + expect(await countResumeBootMessages(apiClient, task.session_id)).toBe( + resumeBootsAfterFirstPause, + ); + + await session.sendMessage("/e2e:simple-message"); + await session.expectChatResponseVisible("simple mock response", 1, { timeout: 30_000 }); + expect(await readSessionRuntimeIdentity(apiClient, task.id, task.session_id)).toEqual( + initialRuntimeIdentity, + ); + expect(await countResumeBootMessages(apiClient, task.session_id)).toBe( + resumeBootsAfterFirstPause, + ); + } finally { + await apiClient.saveUserSettings({ prevent_auto_start_agent_on_open: false }); + } + }); + test("session startup keeps the composer editable and queues a submitted prompt", async ({ testPage, apiClient, diff --git a/docs/plans/resume-cancellation/plan.md b/docs/plans/resume-cancellation/plan.md index 954863376f..0afc5aaf09 100644 --- a/docs/plans/resume-cancellation/plan.md +++ b/docs/plans/resume-cancellation/plan.md @@ -213,3 +213,10 @@ pass. The [error scope package](../error-scope-and-history/plan.md) supersedes the session card placement and removal behavior. Completed results here remain historical evidence. Provider recovery, timeout budgets, cancellation, and authorization remain unchanged. The successor owns chronological error retention, ordinary scroll behavior, and shared task alerts. + +## Accepted-turn cancellation successor + +The [resumed turn cancellation package](../resumed-turn-cancellation/plan.md) +corrects startup authority that remains active after prompt acceptance. +It adds process-survival and follow-up assertions to the existing cancellation coverage. +This completed record and its verification results remain historical evidence. diff --git a/docs/plans/resume-cancellation/task-02-startup-cancellation.md b/docs/plans/resume-cancellation/task-02-startup-cancellation.md index 17aebcd607..980f8dfbfc 100644 --- a/docs/plans/resume-cancellation/task-02-startup-cancellation.md +++ b/docs/plans/resume-cancellation/task-02-startup-cancellation.md @@ -98,3 +98,10 @@ identities use an explicit prefix, bind the first callback execution atomically, and fail closed for untagged or compacted callbacks. Dynamic launch callbacks preserve their originating resume context instead of borrowing the current replacement identity. + +## Accepted-turn cancellation successor + +The [resumed turn cancellation package](../resumed-turn-cancellation/plan.md) +corrects startup authority that remains active after prompt acceptance. +It adds process-survival and follow-up assertions to the existing cancellation coverage. +This completed record and its verification results remain historical evidence. diff --git a/docs/plans/resumed-turn-cancellation/plan.md b/docs/plans/resumed-turn-cancellation/plan.md new file mode 100644 index 0000000000..eeb14c1596 --- /dev/null +++ b/docs/plans/resumed-turn-cancellation/plan.md @@ -0,0 +1,187 @@ +--- +created: 2026-09-18 +status: complete +requirements: + - REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007 +system_design: + - ../../specs/agents/system-design/session-recovery-failures.md +legacy_specs: [] +--- + +# Implementation Plan: Preserve Resumed Agents After Pause + +## Overview + +Preserve the healthy agent process when the user pauses an accepted resumed turn. +One sequential work order implements the ownership correction and its service and browser regressions. +Implementation is complete. + +## Evidence and root cause + +The reported session is `7c6a9ba2-e2be-4ff8-bf56-d0e339fee3df` in task +`742b1dca-39bc-4035-856b-39240f0b9340`. The running build was `81b59416ab`. +The provider session was `01a0b19b-c373-7853-9c0e-7c68e9c2b196`. +Backend diagnostics establish this September 18 timeline in Europe/Lisbon: + +| Time | Event | +| --- | --- | +| 00:32:01 | Resumed execution `5e11de73-89a6-4fb6-8f0c-a53a892859b0` accepts work. | +| 00:37:26 | The user requests turn cancellation. | +| 00:37:30.341 | The provider completes cancellation normally. | +| 00:37:30.450 | Kandev force-stops the execution as `cancelled resume startup`. | +| 00:37:33 | Kandev removes the instance and execution tracking. | +| 00:37:35 | The next user message finds no agent and requests another resume. | +| 00:37:46 | Replacement execution `94b839f4-0c9e-4a91-ba87-74c557295d16` becomes ready. | + +The source confirms that `promptTask` defers attempt completion until its blocking +prompt call returns. `runExplicitCancellationOwned` invalidates that attempt. +`finishPromptExecutorDispatch` then invokes startup cleanup after normal turn cancellation. +The same defect can recur during the first turn of each replacement execution. + +Evidence came from diagnostic bundle `b581ffed368b708481e536590e5d5847`. +The archive byte limit truncated older history, but the incident window is present. +This package uses the existing trace and source inspection plus deterministic service and browser regressions. + +## Requirement conformance + +The agents system owns provider continuity and startup attempt outcomes. +The [existing requirement](../../specs/agents/requirements/session-recovery-failures.md) +adds criteria 007.7 through 007.9 for accepted-turn cancellation and process reuse. +Criteria 007.2 through 007.4 retain startup fencing, retry, and disconnect protection. +The [design amendment](../../specs/agents/system-design/session-recovery-failures.md#accepted-turn-ownership-requirement-007) +separates startup authority from execution identity. + +The completed [resume cancellation package](../resume-cancellation/plan.md) +remains the baseline. Its recorded results remain historical evidence. +Its acceptance-barrier test permits `ErrResumeAttemptCancelled` after acceptance +and never asserts process survival. This package closes that coverage gap. +Other contribution recovery, error-history, and resume-queue packages retain their scope and results. + +## Scope + +### In scope + +- Transfer startup cancellation authority at provider acceptance. +- Preserve accepted execution callbacks, runtime identity, and normal cancellation outcomes. +- Keep model-switch fallback startup cancellable until lifecycle reports initial-prompt acceptance or failure. +- Cover lazy resume, compound resume-and-prompt, handler retry, and relevant model-switch paths. +- Preserve cancellation before readiness, queued work, and exact replacement cleanup safety. +- Prove pause and follow-up behavior on desktop and phone. + +### Out of scope + +- Changes to workflow parking, explicit runtime stops, archive, or restart recovery policy. +- Removal of cancellation timeout escalation when a provider does not settle. +- New UI layout, labels, APIs, schema, flags, provider fallback, or automatic replay. +- Mutation of the reported live session or implementation during this design turn. + +## Technical approach + +Use the accepted phase described in the owning design. Integrate it into +`resume_attempt.go`, the prompt acceptance callback, and cancellation cleanup. +The existing cancellation guard serializes transfer and cancellation. +Registry completion and tombstones retain event provenance after the transfer. +Publication errors after provider acceptance remain accepted outcomes. +Audit the compound continuation and model-switch return paths for stale startup checks. +Do not fix only the final cleanup call while leaving valid events classified as stale. + +## Tests + +Add `task_operations_resumed_turn_cancellation_test.go` with these named regressions: + +| Test | Criteria and evidence | +| --- | --- | +| `TestPromptTask_ResumedTurnCancelPreservesExecution` | 007.7-.9: first resumed turn accepts, pauses, preserves execution/token, and accepts the next message once. | +| `TestResumeTaskSessionAndPrompt_AcceptedTurnCancelPreservesExecution` | 007.7-.8: outer compound continuation cannot reclaim an accepted runtime. | +| `TestResumeAttempt_AcceptanceCancellationOrdering` | 007.2-.4, .8: barriers prove cancellation before acceptance and after acceptance, including duplicate acceptance callbacks. | +| `TestResumeAttempt_AcceptedExecutionEventsRemainValid` | 007.2, .9: token, stream, and completion events remain valid after pause; stale/replaced events remain rejected. | +| `TestResumeAttempt_AcceptedPublicationFailureDoesNotCleanup` | 007.8: accepted publication failure causes no startup teardown, rollback, or replay. | +| `TestResumeAttempt_ModelSwitchFallbackTransfersAcceptance` | 007.8: a replacement execution launched by model-switch fallback inherits accepted-turn ownership. | +| `TestResumeAttempt_ModelSwitchFallbackCancellationBeforeInitialPromptAcceptance` | 007.7-.8: cancellation after process startup but before asynchronous initial-prompt acceptance force-cleans the replacement and fences a delayed callback. | +| `TestPromptTask_QueuedAcceptedTurnIdentityReadFailurePreservesExecution` | 007.7-.9: a queued dispatch with a nonempty incarnation transfers ownership before an injected identity-read failure and preserves the same execution for pause and follow-up. | + +Use real service paths with controlled provider barriers. Assert zero forced-stop +calls after successful turn cancellation. Include both nil-error cancelled results +and the existing provider cancellation error form. Preserve unrelated queued work. +The compound regression uses the wrapped `ErrCancelEscalated` provider cancellation form. +Add handler retry cases in their existing test files if those paths differ. +A mere successful return or unchanged boot-row count does not prove process survival. +The queued-dispatch regression also observes the ordinary `afterDispatch` hook and +proves that acceptance has already revoked startup cleanup authority before that +hook or any fallible identity publication read can run. + +## E2E tests + +Extend `session-recovery.spec.ts` in `chromium` and +`mobile-session-resume-recovery.spec.ts` in `mobile-chrome`. +Use the existing isolated backend and mock provider. +Make the first message perform lazy resume while passive resume is disabled. +Wait for actual accepted work before clicking or tapping Pause. +After cancellation settles, submit a follow-up through the composer. +Assert a response, unchanged execution/provider identity, and no added durable boot event. +Repeat pause and follow-up on that same execution to cover later turns. +Capture the initial runtime identity after the first message is accepted and before +the first pause. For the second pause, snapshot the existing persisted response +message IDs and wait for a newly persisted response ID before pausing, so the witness +cannot match the first turn's output. Compare the runtime identity after cancellation +and after each follow-up. +The phone case retains reachable touch controls and no horizontal page overflow. +Keep existing delayed-startup cancel/retry scenarios as the negative-boundary coverage. + +No rendered structure changes are planned, so an ASCII UI preview is unnecessary. +The existing task chat and mobile session layout remain the interaction exemplars. + +## Work orders + +- [x] [Task 01: Transfer startup ownership at prompt acceptance](task-01-transfer-startup-ownership.md) + +## Verification results + +Implementation is complete. The first RED run of +`TestPromptTask_ResumedTurnCancelPreservesExecution` failed with +`resume attempt cancelled: resume-1` after provider acceptance, reproducing +startup cleanup of the accepted execution. The final verification block passed: + +- `pnpm install --frozen-lockfile` from `apps`: passed. +- Focused orchestrator race suite for accepted cancellation, ordering, events, + publication failure, and model-switch acceptance: passed. +- Queued accepted-turn regression with injected identity-read failure and + same-execution follow-up: passed. +- Full affected-package race suite for orchestrator, executor, lifecycle, and + task handlers: passed. +- Chromium `session-recovery.spec.ts`: 7 tests passed. +- Mobile Chromium `mobile-session-resume-recovery.spec.ts`: 4 tests passed. +- Web typecheck: passed. +- `python3 scripts/list-docs.py validate`: passed. +- `python3 scripts/lint-spec-files.py --all`: passed. +- `git diff --check`: passed. +- Model-switch lifecycle callback propagation and pre-acceptance cancellation barrier: passed. + +The browser helper counts persisted `agent_boot` metadata because the boot row's +content is empty and the rendered resume label is deduplicated. The package is +uncommitted, as required by the work order. +Design validation on 2026-09-18: + +- `python3 scripts/list-docs.py validate`: passed (288 decisions, 1004 specifications). +- `python3 scripts/lint-spec-files.test.py`: passed (36 tests). +- `python3 scripts/lint-spec-files.py --all`: passed. +- Package links and acceptance references: passed. +- `git diff --check`: passed. +- `git status --short -- docs/plans/resumed-turn-cancellation`: both new package files are present and untracked. + +Implementation verification is complete. The package remains uncommitted for review. + +## Documentation impact + +Internal requirements, design, and delivery records change. Public documentation +has no new command, configuration, control, or recovery procedure to describe. +The existing cancellation ADR is sufficient for this local boundary correction. +No new ADR or system index inventory is necessary. + +## Risks + +- Early transfer can let cancelled startup dispatch. Late transfer can kill accepted work. +- Deleting identity evidence can reject valid events or admit stale callbacks. +- Callback publication failure can incorrectly restore startup cleanup authority. +- A guard held during provider settlement can deadlock cancellation. +- Passive browser recovery can hide the defect by launching a replacement before assertions. diff --git a/docs/plans/resumed-turn-cancellation/task-01-transfer-startup-ownership.md b/docs/plans/resumed-turn-cancellation/task-01-transfer-startup-ownership.md new file mode 100644 index 0000000000..f50df8fffe --- /dev/null +++ b/docs/plans/resumed-turn-cancellation/task-01-transfer-startup-ownership.md @@ -0,0 +1,168 @@ +--- +id: "01-transfer-startup-ownership" +title: "Transfer startup ownership at prompt acceptance" +status: done +wave: 1 +depends_on: [] +plan: "plan.md" +requirements: + - REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007 +acceptance_criteria: + - AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.2 + - AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.3 + - AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.4 + - AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.7 + - AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8 + - AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.9 +system_design: + - ../../specs/agents/system-design/session-recovery-failures.md +--- + +# Task 01: Transfer Startup Ownership at Prompt Acceptance + +## Summary + +End startup cancellation authority when the provider accepts the resumed prompt. +Keep the execution identity valid so pause and follow-up use the same healthy process. + +## In scope + +- Implement the accepted phase and guarded authority transfer in the existing attempt registry. +- Integrate lazy and compound resume, prompt acceptance, final result handling, and exact startup cleanup. +- Audit handler retries and model switches for the same accepted boundary. +- Propagate model-switch startup acceptance and pre-acceptance failure callbacks through lifecycle. +- Add the named service regressions from the plan and desktop/phone process-reuse scenarios. +- Preserve pre-acceptance cancellation, stale-callback rejection, and unrelated queued work. + +## Out of scope + +New public contracts, UI components, provider policies, schema, workflow parking changes, +commits or PR delivery, and mutation of the reported live instance. + +## Acceptance + +1. Pausing an accepted resumed turn preserves the healthy execution and provider identity. Follow-up dispatches exactly once on that execution. +2. Cancellation before acceptance still fences startup. Accepted events remain valid while stale callbacks cannot affect a replacement execution. +3. Desktop and phone regressions pass through real cancellation and follow-up controls. Accepted publication failure causes no replay or startup cleanup. + +## TDD and reproduction + +First add `TestPromptTask_ResumedTurnCancelPreservesExecution` in the new test file. +Hold the resumed provider turn after its acceptance callback, then call `CancelAgent`. +Release normal provider cancellation and assert zero forced stops and unchanged execution identity. +Send another prompt and assert one delivery on that execution. +Run it before production edits. The expected failure is startup teardown of the accepted execution. +Do not accept unrelated setup failures as the red result. + +Add the remaining named barrier regressions from the plan before each corresponding correction. +Use channels and bounded waits, not arbitrary sleeps. +Read the backend TDD reference before creating fixtures or subprocesses. + +Write the desktop and phone scenarios before the correction and record the failing process-reuse assertion. +Use existing mock scenarios and isolated fixtures. Disable passive resume for the reproduction. +The first new message must trigger lazy resume and remain active after provider acceptance. +Use existing authorized test API reads for execution/provider identity and durable message IDs. +If the current helper lacks those fields, extend its typed read helper without adding a production endpoint. +Do not rely on the deduplicated visible resume label as the only evidence. + +## Verification + +Run the complete block from the repository root after implementation. +Install workspace dependencies once before the first frontend command in a fresh worktree. +The managed E2E runner rebuilds the binaries and serves the production web build. +Run the two browser suites sequentially without worker overrides. + +```bash +(cd apps && pnpm install --frozen-lockfile) +(cd apps/backend && go test -race ./internal/orchestrator -run 'Test(PromptTask_(ResumedTurnCancel|QueuedAcceptedTurnIdentityReadFailure)|ResumeTaskSessionAndPrompt_AcceptedTurnCancel|ResumeAttempt_)' -count=1 -timeout=10m) +(cd apps/backend && go test -race ./internal/orchestrator ./internal/orchestrator/executor ./internal/agent/runtime/lifecycle ./internal/task/handlers -count=1 -timeout=20m) +(cd apps/web && pnpm e2e:run --project chromium tests/session/session-recovery.spec.ts) +(cd apps/web && pnpm e2e:run --project mobile-chrome tests/session/mobile-session-resume-recovery.spec.ts) +(cd apps/web && pnpm run typecheck) +python3 scripts/list-docs.py validate +python3 scripts/lint-spec-files.py --all +git diff --check +``` + +## Files likely touched + +- `apps/backend/internal/orchestrator/resume_attempt.go` +- `apps/backend/internal/orchestrator/task_operations.go` +- `apps/backend/internal/orchestrator/task_operations_resumed_turn_cancellation_test.go` (new) +- `apps/backend/internal/orchestrator/task_operations_resume_cancellation_test.go` +- `apps/backend/internal/orchestrator/executor/executor_interaction.go` (only if acceptance propagation needs correction) +- `apps/backend/internal/agent/runtime/lifecycle/{types,manager_interaction,session}.go` (model-switch initial-prompt callbacks) +- `apps/backend/internal/task/handlers/message_handlers_resume_readiness_test.go` +- `apps/web/e2e/tests/session/session-recovery.spec.ts` +- `apps/web/e2e/tests/session/mobile-session-resume-recovery.spec.ts` +- `apps/web/e2e/helpers/session-resume-prompt-queue.ts` +- `apps/web/e2e/helpers/api-client.ts` (only if an existing identity read needs typing) +- This plan, work order, and the linked requirement/design amendment for result updates. + +## Dependencies + +None. Existing resume cancellation and provider acceptance callbacks are in this checkout. + +## Risks + +Keep transfer inside the cancellation guard, but never hold that guard during turn settlement. +Retain event provenance without keeping startup teardown authority alive. +Preserve accepted outcomes when publication fails. Do not replay ambiguous provider work. +Resume without a prompt, dispatch-only calls, and model-switch branches must retain their lifecycle rules. + +## Parallelism + +sequential + +## Inputs + +- [Requirements](../../specs/agents/requirements/session-recovery-failures.md#recovery-attempt-isolation). +- [Accepted-turn design](../../specs/agents/system-design/session-recovery-failures.md#accepted-turn-ownership-requirement-007). +- [Evidence and test matrix](plan.md). +- [Cancellation ownership ADR](../../decisions/2026-08-03-backend-owned-cancellation-progress.md). +- Existing `TestPromptTask_ResumeAttemptKeepsCancellationOutOfAcceptanceCallback` and same-execution replacement tests. +- Existing desktop and phone delayed-startup cancellation fixtures. + +## Results + +The first RED run of `TestPromptTask_ResumedTurnCancelPreservesExecution` +failed with `resume attempt cancelled: resume-1` after provider acceptance. +The fix adds an accepted phase to the resume-attempt registry, transfers startup +ownership at the ordinary dispatch callback and model-switch replacement boundary, +and retains the attempt identity for valid late provider events. Accepted cleanup +and invalidation are fenced while pre-acceptance cancellation remains cancellable. +The review correction makes the transfer use the captured execution identity and +occur before `afterDispatch`. It does not depend on a second session identity read, +so a queued dispatch with a nonempty incarnation cannot leave accepted work startup- +cancellable when that read fails. The new queued regression injects that failure, +observes the hook ordering, pauses the accepted turn, and verifies follow-up reuse +of the same execution. + +The final verification block passed: + +- `pnpm install --frozen-lockfile` from `apps`. +- Focused and full affected-package Go race suites. +- Chromium `session-recovery.spec.ts`: 7 passed. +- Mobile Chromium `mobile-session-resume-recovery.spec.ts`: 4 passed. +- Web typecheck. +- Specification catalog validation and full specification lint. +- `git diff --check`. + +The accepted direct regression returns the provider's nil-error cancelled result. +The compound regression returns the wrapped `lifecycle.ErrCancelEscalated` form. +Publication failure and model-switch replacement regressions confirm that accepted +work does not trigger startup teardown or replay. The desktop and phone accepted-turn +scenarios capture runtime identity after the first accepted response, wait for a new +persisted response ID before the second pause, and compare that identity after every +pause and follow-up. The package is uncommitted and no live runtime state was changed. + +The PR fixup review found that the model-switch fallback starts its initial prompt +from a lifecycle goroutine after `StartAgentProcess` returns. The implementation now +registers one-shot dispatch and pre-acceptance failure callbacks before startup, +keeps the resume attempt active until one of them settles, and transfers ownership +using the replacement execution ID captured by the launch response. The new +`TestResumeAttempt_ModelSwitchFallbackCancellationBeforeInitialPromptAcceptance` +barrier proves cancellation still force-cleans the replacement before provider +acceptance and that a delayed callback cannot revive the cancelled attempt. +Lifecycle coverage also verifies that the callback reaches the asynchronous initial +prompt acceptance point. diff --git a/docs/specs/agents/requirements/session-recovery-failures.md b/docs/specs/agents/requirements/session-recovery-failures.md index 977faf3bcd..657dc2da23 100644 --- a/docs/specs/agents/requirements/session-recovery-failures.md +++ b/docs/specs/agents/requirements/session-recovery-failures.md @@ -2,7 +2,7 @@ status: active system: agents created: 2026-09-11 -updated: 2026-09-14 +updated: 2026-09-18 owners: - Kandev --- @@ -60,10 +60,13 @@ This supersedes the reveal-at-top behavior from the completed startup recovery s The task system owns durable history and shared error scope through [task error ownership](../../tasks/requirements/task-launch-failure-recovery.md). -## Proposed recovery attempt amendment +## Recovery attempt isolation -The following requirement is draft. The existing requirements remain active. -Implementation belongs to the [resume cancellation package](../../../plans/resume-cancellation/plan.md). +Criteria 007.1 through 007.6 are implemented in the +[resume cancellation package](../../../plans/resume-cancellation/plan.md). +The accepted-turn amendment adds criteria 007.7 through 007.9. Its +implementation and verification are recorded in the +[resumed turn cancellation package](../../../plans/resumed-turn-cancellation/plan.md). ### REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007: Isolated recovery attempts @@ -78,6 +81,10 @@ Implementation belongs to the [resume cancellation package](../../../plans/resum - **AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.5:** A failed resume before prompt dispatch shall show the existing recovery card with the resume cause. It shall not claim that the agent is busy with a complex task. Reload shall preserve the applicable recovery action and failure details. - **AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.6:** On desktop and phone, users shall be able to cancel startup and retry after cancellation settles. Existing recovery actions, touch targets, keyboard access, and transcript scrolling shall remain available. +- **AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.7:** After the provider accepts a resumed prompt, successful turn cancellation shall preserve the healthy agent process and saved conversation identity. The next message shall use that process without another startup, resume, or recovery error. +- **AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.8:** Cancellation concurrent with prompt acceptance shall have one outcome. Before acceptance, cancelled startup shall not dispatch. After acceptance, normal turn cancellation shall own the outcome. An accepted prompt shall not be replayed or reported as cancelled startup because its turn was paused. +- **AC-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007.9:** Desktop and phone shall support pause followed by another message during the first resumed turn and later turns. Valid responses and session updates shall continue after each pause. A further resume shall require an independently established loss or stop of the runtime. + ## Out of scope Automatic history replacement, new credential permissions, and changes to diff --git a/docs/specs/agents/system-design/session-recovery-failures.md b/docs/specs/agents/system-design/session-recovery-failures.md index 4da987c04d..ec7494893f 100644 --- a/docs/specs/agents/system-design/session-recovery-failures.md +++ b/docs/specs/agents/system-design/session-recovery-failures.md @@ -2,7 +2,7 @@ status: draft system: agents created: 2026-09-11 -updated: 2026-09-14 +updated: 2026-09-18 requirements: - REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-005 - REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-006 @@ -23,7 +23,7 @@ to own contribution admission and durable bootstrap failure projection. | --- | --- | | REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-005 | Workspace-only registration | | REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-006 | Recovery presentation ownership; post-start recoverable failure detail; responsive amendment | -| REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007 | Proposed attempt isolation | +| REQ-AGENTS-AGENT-RESUME-RUNTIME-RECOVERY-007 | Attempt isolation; accepted-turn ownership | The following amendments are implemented in the [contribution resume recovery package](../../../plans/contribution-resume-recovery/plan.md). @@ -270,3 +270,75 @@ Desktop and phone tests exercise the actual backend resume path and reload. This applies the accepted [backend cancellation ownership decision](../../../decisions/2026-08-03-backend-owned-cancellation-progress.md). It extends that implementation to startup attempts without a new durable state or an alternative cancellation owner. No new ADR is required. + +## Accepted-turn ownership (requirement 007) + +This amendment clarifies the startup boundary in the preceding attempt +isolation design. The [repair package](../../../plans/resumed-turn-cancellation/plan.md) +records implementation and validation of criteria 007.7 through 007.9. + +### Startup authority and execution identity + +`resumeAttemptRegistry` owns startup cancellation authority. An attempt identity +also identifies valid runtime events after startup. These are separate responsibilities. +The registry must retain event provenance when startup cancellation authority ends. + +Add an explicit accepted phase to `resumeAttempt`, protected by the registry +mutex. Transfer authority under the existing session cancellation guard. +The transition checks the current attempt and captured execution before acceptance. +It is idempotent and cannot revive an invalidated attempt. + +`preparePromptDispatchCallback` receives provider acceptance through +`PromptWithDispatchCallback`. Record the accepted phase before releasing its +dispatch guard and before publication errors can select startup cleanup. +The callback is evidence of provider acceptance even when a subsequent durable +publication or `afterDispatch` hook fails. Preserve the existing accepted-error path. +Do not wait for the blocking prompt call to return before transferring authority. + +`runExplicitCancellationOwned` continues to call `invalidateResumeAttempt`. +Invalidation cancels only attempts that still own startup. An accepted attempt +keeps its event identity while normal runtime cancellation settles the turn. +A cancelled turn does not cancel the accepted attempt context or confer startup +teardown authority. Shutdown and explicit runtime teardown retain their existing owners. + +`finishPromptExecutorDispatch` and the compound resume continuation distinguish +accepted work from cancelled startup. Normal cancellation returns the turn outcome. +It does not roll back an accepted claim, decorate it as recovery failure, or +call `cleanupCancelledResumeAttempt`. Apply the same rule to model-switch paths +that bypass the ordinary prompt callback. + +Retain normal deferred attempt completion so outer compound calls can finish +without treating the successful transfer as a missing registry owner. +Successful tombstones continue to authorize matching execution callbacks. +Both direct and identity-based startup cleanup reject accepted attempts. +Old cancelled attempts remain fenced, including reused execution IDs and retries. +Do not weaken execution generation checks or accept unknown callback identities. + +### Entry points and failure boundaries + +Apply the transfer to lazy message resume, `ResumeTaskSessionAndPrompt`, and +the handler retry path. Preserve the original identity across internal retry. +Resume without a prompt relinquishes startup ownership once readiness succeeds. +A later ordinary prompt must not inherit startup cancellation authority. + +Before acceptance, cancellation still interrupts startup and prevents dispatch. +After acceptance, provider errors use normal prompt failure handling. +Publication failure does not authorize prompt replay or startup cleanup. +A provider that cannot settle cancellation retains the existing bounded escalation +policy. Archive, workflow parking, explicit stop, crash, and backend restart +remain independent reasons for runtime removal or recovery. + +### Compatibility, presentation, and evidence + +The change is process-local. It adds no durable state, public API, runtime flag, +locale string, or new interface control. Desktop and phone retain their existing +composer, pause control, cancellation progress, and transcript scroll owner. +Tests use those surfaces and require a successful follow-up on the same execution. +Visible boot-row count alone is insufficient because the UI deduplicates resume entries. + +Barrier-controlled service tests prove both acceptance orderings and process +survival. They also prove valid events after cancellation and stale-event rejection. +The existing backend cancellation ADR remains authoritative. This local lifecycle +correction needs no separate ADR. Removing attempt tracking at initial readiness +is insufficient for compound resume because it leaves a pre-dispatch cancellation gap. +Removing all attempt checks would lose stale-callback protection.