fix: preserve resumed agents after accepted turn cancellation (#3791)

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

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 {

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

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

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

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)

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

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 {

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

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

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
+}

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;

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<SessionRuntimeIdentity> {
+  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<Set<string>> {
+  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<string>,
+  marker: string,
+  timeout = 30_000,
+): Promise<void> {
+  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<number> {
+  const { messages } = await apiClient.listSessionMessages(sessionId);
+  return messages.filter(
+    (message) =>
+      message.metadata?.script_type === "agent_boot" && message.metadata?.is_resuming === true,
+  ).length;
+}

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", () => {

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,