fix: safely recover Cursor stream resets (#3188)

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

@@ -28,6 +28,22 @@ type AgentEventPayload struct {
 	ProviderError      *streams.ProviderError `json:"provider_error,omitempty"`
 	ExitCode           *int                   `json:"exit_code,omitempty"`
 	PromptGeneration   uint64                 `json:"prompt_generation,omitempty"`
+	// Prompt replay evidence is populated on terminal failure events. It is
+	// captured by lifecycle before the terminal event is published so consumers
+	// do not have to infer output or effects from independently subscribed
+	// stream events.
+	EvidenceKnown  bool `json:"evidence_known,omitempty"`
+	OutputObserved bool `json:"output_observed,omitempty"`
+	EffectObserved bool `json:"effect_observed,omitempty"`
+}
+
+// PromptAttemptEvidence is the immutable lifecycle snapshot attached to a
+// terminal failure event. Lifecycle conservatively treats any genuine turn
+// content as both output and effect evidence, which fails replay closed.
+type PromptAttemptEvidence struct {
+	EvidenceKnown  bool
+	OutputObserved bool
+	EffectObserved bool
 }
 
 // AgentStalledPayload describes a prompt that has stopped receiving agent events.

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

@@ -40,7 +40,17 @@ func (p *EventPublisher) PublishAgentEvent(ctx context.Context, eventType string
 func (p *EventPublisher) publishAgentEventWithTurnID(
 	ctx context.Context, eventType string, execution *AgentExecution, turnID string,
 ) {
-	p.publishAgentEventPayload(ctx, eventType, newAgentEventPayloadWithTurnID(execution, turnID))
+	p.publishAgentEventWithTurnIDAndEvidence(ctx, eventType, execution, turnID, nil)
+}
+
+func (p *EventPublisher) publishAgentEventWithTurnIDAndEvidence(
+	ctx context.Context,
+	eventType string,
+	execution *AgentExecution,
+	turnID string,
+	evidence *PromptAttemptEvidence,
+) {
+	p.publishAgentEventPayload(ctx, eventType, newAgentEventPayloadWithTurnIDAndEvidence(execution, turnID, evidence))
 }
 
 // PublishAgentStalled publishes one inactivity signal for a prompt.
@@ -108,7 +118,15 @@ func newAgentEventPayload(execution *AgentExecution) AgentEventPayload {
 }
 
 func newAgentEventPayloadWithTurnID(execution *AgentExecution, turnID string) AgentEventPayload {
-	return AgentEventPayload{
+	return newAgentEventPayloadWithTurnIDAndEvidence(execution, turnID, nil)
+}
+
+func newAgentEventPayloadWithTurnIDAndEvidence(
+	execution *AgentExecution,
+	turnID string,
+	evidence *PromptAttemptEvidence,
+) AgentEventPayload {
+	payload := AgentEventPayload{
 		AgentExecutionID:   execution.ID,
 		RunID:              execution.RunID,
 		TaskID:             execution.TaskID,
@@ -128,6 +146,12 @@ func newAgentEventPayloadWithTurnID(execution *AgentExecution, turnID string) Ag
 		ExitCode:           execution.ExitCode,
 		PromptGeneration:   execution.promptGeneration,
 	}
+	if evidence != nil {
+		payload.EvidenceKnown = evidence.EvidenceKnown
+		payload.OutputObserved = evidence.OutputObserved
+		payload.EffectObserved = evidence.EffectObserved
+	}
+	return payload
 }
 
 // PublishAgentctlEvent publishes an agentctl lifecycle event (starting, ready, error).

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

@@ -116,7 +116,12 @@ func isUninitializedStartupFailure(execution *AgentExecution, event *agentctl.Ag
 	return event != nil && event.PromptGeneration == 0 && isUninitializedStartupExecution(execution)
 }
 
-func (m *Manager) handleCompleteEventMarkState(execution *AgentExecution, event *agentctl.AgentEvent, isError bool) {
+func (m *Manager) handleCompleteEventMarkState(
+	execution *AgentExecution,
+	event *agentctl.AgentEvent,
+	isError bool,
+	failureEvidence *PromptAttemptEvidence,
+) {
 	if isError {
 		// A process can exit while the startup owner is still waiting for ACP
 		// initialization. Leave that failure non-terminal so the startup path can
@@ -144,7 +149,7 @@ func (m *Manager) handleCompleteEventMarkState(execution *AgentExecution, event
 			zap.Any("event_data", event.Data),
 			zap.String("agent_command", execution.AgentCommand),
 			zap.String("acp_session_id", execution.ACPSessionID))
-		if err := m.markCompletedWithTurnID(execution.ID, 1, errorMsg, event.TurnID); err != nil {
+		if err := m.markCompletedWithTurnID(execution.ID, 1, errorMsg, event.TurnID, failureEvidence); err != nil {
 			m.logger.Error("failed to mark execution as failed after error completion",
 				zap.String("execution_id", execution.ID),
 				zap.Error(err))
@@ -287,13 +292,14 @@ func (m *Manager) finishPromptCompletion(
 	event *agentctl.AgentEvent,
 	isError bool,
 	claim promptCompletionClaim,
+	failureEvidence *PromptAttemptEvidence,
 ) {
 	handleCompleteEventSignal(execution, event, isError)
 	if event.PromptGeneration == 0 || isError {
 		if isError {
 			setProviderError(execution, event.ProviderError)
 		}
-		m.handleCompleteEventMarkState(execution, event, isError)
+		m.handleCompleteEventMarkState(execution, event, isError, failureEvidence)
 		if claim.locked {
 			execution.promptLifecycleMu.Unlock()
 		}
@@ -340,6 +346,11 @@ func (m *Manager) handleCompleteEvent(execution *AgentExecution, event *agentctl
 	if !claimed {
 		return false
 	}
+	var failureEvidence *PromptAttemptEvidence
+	if isError {
+		evidence := execution.promptAttemptEvidenceSnapshot()
+		failureEvidence = &evidence
+	}
 	m.releaseActivity(executionActivityKey(execution.ID))
 
 	execution.markAgentActivity()
@@ -393,7 +404,7 @@ func (m *Manager) handleCompleteEvent(execution *AgentExecution, event *agentctl
 	// the drain races with the first SendPrompt's receive and can steal the signal,
 	// leaving the first SendPrompt hung and the second prompt's completion event
 	// never reaching the event bus.
-	m.finishPromptCompletion(execution, event, isError, claim)
+	m.finishPromptCompletion(execution, event, isError, claim, failureEvidence)
 	return true
 }
 

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

@@ -72,7 +72,7 @@ func TestHandleCompleteEventMarkState_ShutdownRaceMarksStopped(t *testing.T) {
 		Error: "-32603 peer disconnected before response",
 		Data:  map[string]any{"is_error": true},
 	}
-	mgr.handleCompleteEventMarkState(execution, errorEvent, true)
+	mgr.handleCompleteEventMarkState(execution, errorEvent, true, nil)
 
 	got, _ := mgr.GetExecution("exec-1")
 	if got.Status != v1.AgentStatusStopped {
@@ -103,7 +103,7 @@ func TestHandleCompleteEventMarkState_ErrorFailsWhenNotShuttingDown(t *testing.T
 		Error: "boom",
 		Data:  map[string]any{"is_error": true},
 	}
-	mgr.handleCompleteEventMarkState(execution, errorEvent, true)
+	mgr.handleCompleteEventMarkState(execution, errorEvent, true, nil)
 
 	got, _ := mgr.GetExecution("exec-1")
 	if got.Status != v1.AgentStatusFailed {

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

@@ -22,7 +22,7 @@ func TestHandleCompleteEventMarkState_DefersUninitializedStartupFailure(t *testi
 		Type:  streams.EventTypeComplete,
 		Error: "Agent process exited with code 1",
 		Data:  map[string]any{"is_error": true},
-	}, true)
+	}, true, nil)
 
 	got, _ := mgr.GetExecution(execution.ID)
 	if got.Status != v1.AgentStatusStarting {

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

@@ -2117,7 +2117,7 @@ func TestHandleCompleteEventMarkState_ErrorDoesNotRemoveExecution(t *testing.T)
 		Data:  map[string]interface{}{"is_error": true},
 	}
 
-	mgr.handleCompleteEventMarkState(execution, errorEvent, true)
+	mgr.handleCompleteEventMarkState(execution, errorEvent, true, nil)
 
 	// Execution must still be in the store so the orchestrator can clean it up
 	if _, found := mgr.executionStore.Get("exec-1"); !found {
@@ -2137,7 +2137,7 @@ func TestHandleCompleteEventMarkState_SuccessKeepsExecution(t *testing.T) {
 		Type: "complete",
 	}
 
-	mgr.handleCompleteEventMarkState(execution, successEvent, false)
+	mgr.handleCompleteEventMarkState(execution, successEvent, false, nil)
 
 	got, found := mgr.executionStore.Get("exec-1")
 	if !found {

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

@@ -1737,11 +1737,15 @@ func (m *Manager) MarkCompleted(executionID string, exitCode int, errorMessage s
 		exitCode,
 		errorMessage,
 		execution.promptTurnIDSnapshot(),
+		nil,
 	)
 }
 
 func (m *Manager) markCompletedWithTurnID(
-	executionID string, exitCode int, errorMessage, turnID string,
+	executionID string,
+	exitCode int,
+	errorMessage, turnID string,
+	failureEvidence *PromptAttemptEvidence,
 ) error {
 	execution, exists := m.executionStore.Get(executionID)
 	if !exists {
@@ -1772,6 +1776,10 @@ func (m *Manager) markCompletedWithTurnID(
 			zap.Int("exit_code", exitCode))
 		return nil
 	}
+	if (exitCode != 0 || errorMessage != "") && failureEvidence == nil {
+		evidence := execution.promptAttemptEvidenceSnapshot()
+		failureEvidence = &evidence
+	}
 
 	_ = m.executionStore.WithLock(executionID, func(exec *AgentExecution) {
 		now := time.Now()
@@ -1809,6 +1817,12 @@ func (m *Manager) markCompletedWithTurnID(
 		eventType = events.AgentFailed
 		m.classifyAndMaybeRemediate(execution, exitCode, errorMessage)
 	}
+	if eventType == events.AgentFailed {
+		m.eventPublisher.publishAgentEventWithTurnIDAndEvidence(
+			context.Background(), eventType, execution, turnID, failureEvidence,
+		)
+		return nil
+	}
 	m.eventPublisher.publishAgentEventWithTurnID(context.Background(), eventType, execution, turnID)
 
 	return nil

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

@@ -0,0 +1,82 @@
+package lifecycle
+
+import (
+	"testing"
+
+	agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	"github.com/kandev/kandev/internal/events"
+)
+
+func TestAgentFailedPayloadCarriesTerminalPromptEvidence(t *testing.T) {
+	for _, tc := range []struct {
+		name          string
+		priorActivity bool
+		useErrorEvent bool
+		wantOutput    bool
+		wantEffect    bool
+	}{
+		{
+			name: "process exit with no activity",
+		},
+		{
+			name:          "process exit after activity",
+			priorActivity: true,
+			wantOutput:    true,
+			wantEffect:    true,
+		},
+		{
+			name:          "error completion with no activity",
+			useErrorEvent: true,
+		},
+		{
+			name:          "error completion after activity",
+			priorActivity: true,
+			useErrorEvent: true,
+			wantOutput:    true,
+			wantEffect:    true,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			mgr, eventBus := createTestManagerWithTracking()
+			execution := createTestExecution("exec-1", "task-1", "session-1")
+			execution.promptGeneration = 7
+			if err := mgr.executionStore.Add(execution); err != nil {
+				t.Fatalf("add execution: %v", err)
+			}
+			if tc.priorActivity {
+				mgr.recordActivity(execution, agentctl.AgentEvent{Type: "message_chunk"})
+			}
+
+			if tc.useErrorEvent {
+				mgr.handleAgentEvent(execution, agentctl.AgentEvent{
+					Type:             "error",
+					Error:            "agent failed",
+					PromptGeneration: 7,
+				})
+			} else if err := mgr.MarkCompleted(execution.ID, 1, "agent failed"); err != nil {
+				t.Fatalf("mark completed: %v", err)
+			}
+
+			var payload AgentEventPayload
+			found := false
+			eventBus.mu.Lock()
+			for _, published := range eventBus.PublishedEvents {
+				if published.Subject != events.AgentFailed {
+					continue
+				}
+				payload, found = published.Event.Data.(AgentEventPayload)
+				break
+			}
+			eventBus.mu.Unlock()
+			if !found {
+				t.Fatal("agent.failed payload was not published")
+			}
+			if !payload.EvidenceKnown {
+				t.Fatal("agent.failed payload omitted known prompt evidence")
+			}
+			if payload.OutputObserved != tc.wantOutput || payload.EffectObserved != tc.wantEffect {
+				t.Fatalf("prompt evidence = output:%v effect:%v, want output:%v effect:%v", payload.OutputObserved, payload.EffectObserved, tc.wantOutput, tc.wantEffect)
+			}
+		})
+	}
+}

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

@@ -376,6 +376,20 @@ func (e *AgentExecution) promptActivitySnapshot() (time.Time, bool, uint64) {
 	return e.lastActivityAt, e.agentEventSincePrompt, e.promptActivityEpoch
 }
 
+// promptAttemptEvidenceSnapshot captures the replay-safety state before a
+// terminal completion handler marks that completion as activity. A genuine
+// turn-content event is deliberately conservative evidence of both output and
+// effects because lifecycle does not own the downstream persistence boundary.
+func (e *AgentExecution) promptAttemptEvidenceSnapshot() PromptAttemptEvidence {
+	e.lastActivityAtMu.Lock()
+	defer e.lastActivityAtMu.Unlock()
+	return PromptAttemptEvidence{
+		EvidenceKnown:  true,
+		OutputObserved: e.agentEventSincePrompt,
+		EffectObserved: e.agentEventSincePrompt,
+	}
+}
+
 func (e *AgentExecution) promptActivityEpochSnapshot() uint64 {
 	e.lastActivityAtMu.Lock()
 	defer e.lastActivityAtMu.Unlock()

apps/backend/internal/agent/runtime/routingerr/cursor_retriable_stream_reset_test.go

@@ -0,0 +1,145 @@
+package routingerr
+
+import "testing"
+
+const realCursorRetriableStreamReset = "Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)"
+
+const wantCursorRetriableStreamResetRuleID = "cursor.retriable_stream_reset.v1"
+
+func TestMatchRuntimeEnvironmentRules_CursorRetriable(t *testing.T) {
+	got, ok := matchRuntimeEnvironmentRules(realCursorRetriableStreamReset)
+	if !ok {
+		t.Fatal("expected Cursor stream-reset match")
+	}
+	if got.Code != CodeAgentTransportLost {
+		t.Fatalf("Code = %q, want %q", got.Code, CodeAgentTransportLost)
+	}
+	if got.ClassifierRule != wantCursorRetriableStreamResetRuleID {
+		t.Fatalf("ClassifierRule = %q, want %q", got.ClassifierRule, wantCursorRetriableStreamResetRuleID)
+	}
+	if got.Confidence != ConfHigh {
+		t.Fatalf("Confidence = %q, want %q", got.Confidence, ConfHigh)
+	}
+
+	for _, tc := range []struct {
+		name string
+		text string
+	}{
+		{
+			name: "prose before marker",
+			text: "provider said: " + realCursorRetriableStreamReset,
+		},
+		{
+			name: "partial RetriableError marker",
+			text: "Error: Retriable: HTTP/2 stream closed with error code CANCEL (0x8)",
+		},
+		{
+			name: "partial transport fingerprint",
+			text: "Error: RetriableError: HTTP/2 stream reset with error code CANCEL",
+		},
+		{
+			name: "unrelated explanation before fingerprint",
+			text: "Error: RetriableError: explanation for CANCEL (0x8)",
+		},
+		{
+			name: "missing transport fingerprint",
+			text: "Error: RetriableError: provider is busy",
+		},
+		{
+			name: "context cancellation",
+			text: "context canceled: " + realCursorRetriableStreamReset,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got, ok := matchRuntimeEnvironmentRules(tc.text); ok {
+				t.Fatalf("unexpected match: %+v", got)
+			}
+		})
+	}
+
+	withCursorCancellationToken := realCursorRetriableStreamReset + " [canceled]"
+	got, ok = matchRuntimeEnvironmentRules(withCursorCancellationToken)
+	if !ok || got.Code != CodeAgentTransportLost {
+		t.Fatalf("bracketed cancellation token classified as ok=%v err=%v, want transport loss", ok, got)
+	}
+}
+
+func TestClassifyCursorRetriable(t *testing.T) {
+	resetInjection()
+
+	e := Classify(Input{
+		Phase:      PhasePromptSend,
+		ProviderID: "cursor-acp",
+		Stderr:     realCursorRetriableStreamReset,
+	})
+	if e == nil {
+		t.Fatal("expected non-nil Error")
+	}
+	if e.Code != CodeAgentTransportLost {
+		t.Fatalf("Code = %q, want %q", e.Code, CodeAgentTransportLost)
+	}
+	if e.Class != ClassTransient {
+		t.Fatalf("Class = %q, want %q", e.Class, ClassTransient)
+	}
+	if e.CatalogueVersion != CatalogueVersion {
+		t.Fatalf("CatalogueVersion = %q, want %q", e.CatalogueVersion, CatalogueVersion)
+	}
+	if e.Confidence != ConfHigh {
+		t.Fatalf("Confidence = %q, want %q", e.Confidence, ConfHigh)
+	}
+	if !e.AutoRetryable {
+		t.Fatal("AutoRetryable = false, want true")
+	}
+	if e.FallbackAllowed {
+		t.Fatal("FallbackAllowed = true, want false")
+	}
+	if e.UserAction {
+		t.Fatal("UserAction = true, want false")
+	}
+	if e.Phase != PhasePromptSend {
+		t.Fatalf("Phase = %q, want %q", e.Phase, PhasePromptSend)
+	}
+
+	overlapped := Classify(Input{
+		Phase:  PhasePromptSend,
+		Stderr: "API Error: 529 Overloaded. " + realCursorRetriableStreamReset,
+	})
+	if overlapped.Code != CodeProviderOverloaded {
+		t.Fatalf("overlapping overload classified as %q, want %q", overlapped.Code, CodeProviderOverloaded)
+	}
+}
+
+func TestIsTransientProviderError_Cursor(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		text string
+		want bool
+	}{
+		{
+			name: "exact diagnostic",
+			text: realCursorRetriableStreamReset,
+			want: true,
+		},
+		{
+			name: "Cursor cancellation token",
+			text: realCursorRetriableStreamReset + " [canceled]",
+			want: true,
+		},
+		{
+			name: "context canceled",
+			text: "context canceled: " + realCursorRetriableStreamReset,
+			want: false,
+		},
+		{
+			name: "cancel escalated",
+			text: "cancel escalated: " + realCursorRetriableStreamReset,
+			want: false,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := IsTransientProviderError(tc.text); got != tc.want {
+				t.Errorf("IsTransientProviderError(%q) = %v, want %v", tc.text, got, tc.want)
+			}
+		})
+	}
+}

apps/backend/internal/agent/runtime/routingerr/runtime_rules.go

@@ -74,6 +74,23 @@ var runtimeEnvironmentRules = []runtimeRule{
 			}
 		},
 	},
+	{
+		// Cursor emits this exact control prefix as an assistant message chunk
+		// when its upstream HTTP/2 stream resets. Keep the fingerprint anchored
+		// to the control frame and bounded so user-authored prose cannot turn
+		// into an automatic retry.
+		id:      cursorRetriableStreamResetRuleID,
+		pattern: cursorRetriableStreamResetRe,
+		build: func(text string) *Error {
+			if cancellationOrDeadlineRe.MatchString(text) {
+				return nil
+			}
+			return &Error{
+				Code:       CodeAgentTransportLost,
+				Confidence: ConfHigh,
+			}
+		},
+	},
 	{
 		// The ACP transport pipe died mid-turn: the upstream provider service
 		// dropped the connection before a response arrived. This is
@@ -115,6 +132,8 @@ const resumeCorruptedRuleID = "anthropic.thinking_blocks.immutable.v1"
 
 const overloadedRuleID = "anthropic.overloaded.529.v1"
 
+const cursorRetriableStreamResetRuleID = "cursor.retriable_stream_reset.v1"
+
 const transportLostRuleID = "acp.transport_lost.v1"
 
 // transportLostRe matches the narrow ACP wire-level transport-death
@@ -124,6 +143,13 @@ const transportLostRuleID = "acp.transport_lost.v1"
 // error strings, which must keep falling through to manual recovery.
 var transportLostRe = regexp.MustCompile(`(?i)peer disconnected|connection closed`)
 
+// cursorRetriableStreamResetRe matches Cursor's complete control diagnostic.
+// Requiring the whole normalized message prevents ordinary provider prose from
+// combining the control prefix with an unrelated transport fragment.
+var cursorRetriableStreamResetRe = regexp.MustCompile(
+	`(?i)^\s*Error:\s*RetriableError:\s*HTTP/2 stream closed with error code CANCEL \(0x8\)(?:\s+\[canceled\])?\s*$`,
+)
+
 // overloadedRe matches the transient 529 Overloaded signature: either the
 // numeric code adjacent to "overloaded" on a single line (in either order), or
 // the Anthropic "overloaded_error" type token. Kept tight ([^\n] stays within

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go

@@ -322,19 +322,21 @@ type Adapter struct {
 
 // promptTurnState holds synchronization for one in-flight session/prompt RPC.
 type promptTurnState struct {
-	endTurn          context.CancelCauseFunc
-	rpcDone          chan struct{}
-	abortCh          chan struct{}
-	handoffCh        chan struct{}
-	providerErrorCh  chan openCodeStderrDiagnostic
-	promptGeneration uint64
-	evidenceMu       sync.Mutex
-	codexSystemError bool
-	codexCapacity    bool
-	allowHandoff     bool
-	handedOff        bool
-	gateOwned        bool
-	finishing        bool
+	endTurn           context.CancelCauseFunc
+	rpcDone           chan struct{}
+	abortCh           chan struct{}
+	handoffCh         chan struct{}
+	providerErrorCh   chan openCodeStderrDiagnostic
+	promptGeneration  uint64
+	evidenceMu        sync.Mutex
+	codexSystemError  bool
+	codexCapacity     bool
+	cursorRetriable   bool
+	cursorRetriableAt time.Time
+	allowHandoff      bool
+	handedOff         bool
+	gateOwned         bool
+	finishing         bool
 }
 
 func (t *promptTurnState) observeCodexEvidence(systemError, capacity bool) {
@@ -365,6 +367,42 @@ func (t *promptTurnState) hasCodexSystemError() bool {
 	return t.codexSystemError
 }
 
+func (t *promptTurnState) setCursorRetriable() {
+	if t == nil {
+		return
+	}
+	t.evidenceMu.Lock()
+	if !t.cursorRetriable {
+		t.cursorRetriableAt = time.Now().UTC()
+	}
+	t.cursorRetriable = true
+	t.evidenceMu.Unlock()
+}
+
+func (t *promptTurnState) clearCursorRetriable() {
+	if t == nil {
+		return
+	}
+	t.evidenceMu.Lock()
+	t.cursorRetriable = false
+	t.cursorRetriableAt = time.Time{}
+	t.evidenceMu.Unlock()
+}
+
+func (t *promptTurnState) cursorRetriableFailure() bool {
+	failure, _ := t.cursorRetriableFailureAt()
+	return failure
+}
+
+func (t *promptTurnState) cursorRetriableFailureAt() (bool, time.Time) {
+	if t == nil {
+		return false, time.Time{}
+	}
+	t.evidenceMu.Lock()
+	defer t.evidenceMu.Unlock()
+	return t.cursorRetriable, t.cursorRetriableAt
+}
+
 type asyncTurnFinalizer struct {
 	timer       *time.Timer
 	seq         uint64

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/coder/acp-go-sdk"
+	"github.com/kandev/kandev/internal/agentctl/acpcompat"
 	"github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared"
 	"github.com/kandev/kandev/internal/agentctl/types/streams"
 	v1 "github.com/kandev/kandev/pkg/api/v1"
@@ -243,6 +244,30 @@ func (a *Adapter) sendPrompt(
 	// a subagent tool_call this turn.
 	a.sweepCursorTaskMetaOnPromptEnd(sessionID)
 
+	if cursorRetriable, occurredAt := turn.cursorRetriableFailureAt(); cursorRetriable {
+		const safeMessage = cursorRetriableStreamResetMessage
+		if occurredAt.IsZero() {
+			occurredAt = time.Now().UTC()
+		}
+		a.logger.Info("cursor prompt ended with retriable stream-reset evidence",
+			zap.String("session_id", sessionID),
+			zap.Uint64("prompt_generation", promptGeneration))
+		a.cancelAsyncTurnComplete(sessionID)
+		a.sendUpdate(AgentEvent{
+			Type:             streams.EventTypeError,
+			SessionID:        sessionID,
+			PromptGeneration: promptGeneration,
+			Error:            safeMessage,
+			ProviderError: &streams.ProviderError{
+				Source:     streams.ProviderErrorSourceCursorACP,
+				ProviderID: acpcompat.CursorAgentID,
+				Message:    safeMessage,
+				OccurredAt: occurredAt,
+			},
+		})
+		return nil
+	}
+
 	if a.agentID == codexAgentID && turn.codexCapacityFailure() {
 		const safeMessage = codexModelCapacityErrorMessage
 		a.logger.Info("codex prompt ended with model-capacity evidence",

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go

@@ -11,6 +11,8 @@ import (
 	"go.uber.org/zap"
 )
 
+const acpUserRole = "user"
+
 // notifWork is the item type carried on notifQueue. notif is populated for a
 // real SDK notification (the common case); sync identifies a barrier and may
 // carry afterBarrier state finalization. The worker closes sync after the
@@ -170,14 +172,16 @@ func (a *Adapter) handleACPUpdate(
 	sessionID := string(n.SessionId)
 
 	suppressed := a.dialect.suppresses(n)
+	handled := suppressed
 	var event, leadingEvent *AgentEvent
 	if !suppressed {
 		event = a.convertNotification(n)
-		if event != nil && a.observeCodexProviderEvidence(promptGeneration, event) {
-			// Once the explicit Codex system-error marker has been observed,
-			// suppress the provider's explanatory assistant chunk. The adapter
-			// emits one normalized error after the prompt barrier instead.
+		if event != nil && (a.observeCodexProviderEvidence(promptGeneration, event) ||
+			a.observeCursorRetriableEvidence(promptGeneration, event)) {
+			// Suppress provider control/evidence chunks. The adapter emits one
+			// normalized error after the prompt barrier instead.
 			event = nil
+			handled = true
 		}
 		if n.Update.UsageUpdate != nil {
 			lifecycleEvent := usageLifecycleEvent(
@@ -220,7 +224,7 @@ func (a *Adapter) handleACPUpdate(
 			event.Type, rawData, event)
 		a.sendUpdate(*event)
 		a.maybeScheduleAsyncTurnComplete(*event)
-	} else if !suppressed {
+	} else if !handled {
 		if updateJSON, err := json.Marshal(n.Update); err == nil {
 			a.logger.Warn("unhandled ACP session notification",
 				zap.String("session_id", sessionID),
@@ -235,6 +239,38 @@ func (a *Adapter) handleACPUpdate(
 	}
 }
 
+func (a *Adapter) observeCursorRetriableEvidence(promptGeneration uint64, event *AgentEvent) bool {
+	if a.agentID != acpcompat.CursorAgentID || event == nil || promptGeneration == 0 {
+		return false
+	}
+	turn := a.currentPromptTurn()
+	if turn == nil || turn.promptGeneration != promptGeneration {
+		return false
+	}
+	if event.Type == streams.EventTypeMessageChunk && event.Role != acpUserRole &&
+		isCursorRetriableStreamReset(event.Text) {
+		turn.setCursorRetriable()
+		return true
+	}
+	if cursorProviderProgress(event) {
+		turn.clearCursorRetriable()
+	}
+	return false
+}
+
+func cursorProviderProgress(event *AgentEvent) bool {
+	switch event.Type {
+	case streams.EventTypeMessageChunk:
+		return event.Role != acpUserRole && strings.TrimSpace(event.Text) != ""
+	case streams.EventTypeReasoning:
+		return strings.TrimSpace(event.ReasoningText) != ""
+	case streams.EventTypeToolCall:
+		return event.ToolCallID != ""
+	default:
+		return false
+	}
+}
+
 func (a *Adapter) observeCodexProviderEvidence(promptGeneration uint64, event *AgentEvent) bool {
 	if a.agentID != codexAgentID || event == nil || promptGeneration == 0 {
 		return false
@@ -307,7 +343,7 @@ func (a *Adapter) convertNotification(n acp.SessionNotification) *AgentEvent {
 		return a.convertMessageChunkWithProtocolID(
 			sessionID,
 			u.UserMessageChunk.Content,
-			"user",
+			acpUserRole,
 			derefStr(u.UserMessageChunk.MessageId),
 		)
 
@@ -572,7 +608,7 @@ func (a *Adapter) convertMessageChunkWithProtocolID(
 	}
 
 	// Only set Role for user messages (assistant is the default)
-	if role == "user" {
+	if role == acpUserRole {
 		event.Role = role
 	}
 

apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_cursor.go

@@ -2,10 +2,34 @@ package acp
 
 import (
 	"encoding/json"
+	"strings"
 
 	"github.com/kandev/kandev/internal/agentctl/types/streams"
 )
 
+const (
+	cursorRetriableStreamResetPrefix  = "Error: RetriableError:"
+	cursorRetriableStreamResetMessage = "Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)"
+	cursorRetriableStreamResetMaxTail = 256
+)
+
+// isCursorRetriableStreamReset recognizes Cursor's complete transport control
+// chunk. The prefix check keeps the common per-token path allocation-free;
+// requiring the complete diagnostic prevents ordinary provider prose from
+// combining the control prefix with an unrelated transport fragment.
+func isCursorRetriableStreamReset(text string) bool {
+	trimmed := strings.TrimSpace(text)
+	if len(trimmed) < len(cursorRetriableStreamResetPrefix) ||
+		!strings.EqualFold(trimmed[:len(cursorRetriableStreamResetPrefix)], cursorRetriableStreamResetPrefix) {
+		return false
+	}
+	if len(trimmed)-len(cursorRetriableStreamResetPrefix) > cursorRetriableStreamResetMaxTail {
+		return false
+	}
+	return strings.EqualFold(trimmed, cursorRetriableStreamResetMessage) ||
+		strings.EqualFold(trimmed, cursorRetriableStreamResetMessage+" [canceled]")
+}
+
 type cursorTaskMeta struct {
 	ToolCallID  string
 	AgentID     string

apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_cursor_retriable_test.go

@@ -0,0 +1,258 @@
+package acp
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	acpsdk "github.com/coder/acp-go-sdk"
+	"github.com/kandev/kandev/internal/agentctl/acpcompat"
+	"github.com/kandev/kandev/internal/agentctl/types/streams"
+)
+
+const cursorRetriableStreamResetChunk = "Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)"
+
+func cursorMessageNotification(sessionID, text string) acpsdk.SessionNotification {
+	return makeNotification(sessionID, acpsdk.SessionUpdate{
+		AgentMessageChunk: &acpsdk.SessionUpdateAgentMessageChunk{
+			Content: acpsdk.TextBlock(text),
+		},
+	})
+}
+
+func newCursorPromptTurn(t *testing.T, generation uint64) (*Adapter, *promptTurnState) {
+	t.Helper()
+	a := newTestAdapter()
+	a.agentID = acpcompat.CursorAgentID
+	a.normalizer = NewNormalizer(acpcompat.CursorAgentID)
+	a.dialect = newACPDialect(acpcompat.CursorAgentID)
+	t.Cleanup(func() { _ = a.Close() })
+	_, turn := a.registerPromptTurn(context.Background(), generation)
+	t.Cleanup(func() { a.clearPromptTurn(turn) })
+	return a, turn
+}
+
+func TestCursorRetriableStreamResetSuppressesExactCurrentChunk(t *testing.T) {
+	a, _ := newCursorPromptTurn(t, 7)
+
+	a.handleACPUpdate(cursorMessageNotification("session-1", cursorRetriableStreamResetChunk), 7)
+
+	if events := drainEvents(a); len(events) != 0 {
+		t.Fatalf("exact Cursor control chunk emitted %d events: %+v", len(events), events)
+	}
+}
+
+func TestCursorRetriableStreamResetRejectsUnrelatedEvidence(t *testing.T) {
+	tests := []struct {
+		name       string
+		agentID    string
+		generation uint64
+		text       string
+	}{
+		{
+			name:       "prose before marker",
+			generation: 7,
+			text:       "The provider returned Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)",
+		},
+		{
+			name:       "unrelated explanation before fingerprint",
+			generation: 7,
+			text:       "Error: RetriableError: explanation for CANCEL (0x8)",
+		},
+		{
+			name:       "partial marker",
+			generation: 7,
+			text:       "Error: RetriableError:",
+		},
+		{
+			name:       "missing transport fingerprint",
+			generation: 7,
+			text:       "Error: RetriableError: provider is busy",
+		},
+		{
+			name:       "stale generation",
+			generation: 6,
+			text:       cursorRetriableStreamResetChunk,
+		},
+		{
+			name:       "zero generation",
+			generation: 0,
+			text:       cursorRetriableStreamResetChunk,
+		},
+		{
+			name:       "other adapter",
+			agentID:    "claude-acp",
+			generation: 7,
+			text:       cursorRetriableStreamResetChunk,
+		},
+		{
+			name:       "user message",
+			generation: 7,
+			text:       cursorRetriableStreamResetChunk,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			a, _ := newCursorPromptTurn(t, 7)
+			if test.agentID != "" {
+				a.agentID = test.agentID
+			}
+
+			notification := cursorMessageNotification("session-1", test.text)
+			if test.name == "user message" {
+				notification.Update.UserMessageChunk = &acpsdk.SessionUpdateUserMessageChunk{
+					Content: acpsdk.TextBlock(test.text),
+				}
+				notification.Update.AgentMessageChunk = nil
+			}
+			a.handleACPUpdate(notification, test.generation)
+
+			events := drainEvents(a)
+			if len(events) != 1 || events[0].Type != streams.EventTypeMessageChunk || events[0].Text != test.text {
+				t.Fatalf("events = %+v, want one ordinary message chunk", events)
+			}
+		})
+	}
+}
+
+func TestCursorRetriableStreamResetClearsOnProgressAndRearmsOnLaterMarker(t *testing.T) {
+	a, _ := newCursorPromptTurn(t, 7)
+
+	a.handleACPUpdate(cursorMessageNotification("session-1", cursorRetriableStreamResetChunk), 7)
+	a.handleACPUpdate(cursorMessageNotification("session-1", "Cursor resumed generation"), 7)
+	a.handleACPUpdate(cursorMessageNotification("session-1", cursorRetriableStreamResetChunk), 7)
+
+	events := drainEvents(a)
+	if len(events) != 1 || events[0].Type != streams.EventTypeMessageChunk || events[0].Text != "Cursor resumed generation" {
+		t.Fatalf("events = %+v, want only resumed provider progress", events)
+	}
+}
+
+func TestCursorRetriableStreamResetNotClearedByToolUpdate(t *testing.T) {
+	a, _ := newCursorPromptTurn(t, 7)
+	a.handleACPUpdate(cursorMessageNotification("session-1", cursorRetriableStreamResetChunk), 7)
+
+	completed := acpsdk.ToolCallStatus("completed")
+	a.handleACPUpdate(makeNotification("session-1", acpsdk.SessionUpdate{
+		ToolCallUpdate: &acpsdk.SessionToolCallUpdate{
+			ToolCallId: "tool-1",
+			Status:     &completed,
+		},
+	}), 7)
+
+	if !a.currentPromptTurn().cursorRetriableFailure() {
+		t.Fatal("tool update cleared pending Cursor retriable evidence")
+	}
+}
+
+func TestSendPromptCursorRetriableStreamResetSettlesAfterNotificationBarrier(t *testing.T) {
+	a, fake, conn := setupHandoffFakeAgent(t)
+	a.agentID = acpcompat.CursorAgentID
+	a.normalizer = NewNormalizer(acpcompat.CursorAgentID)
+	a.dialect = newACPDialect(acpcompat.CursorAgentID)
+
+	ctx := context.Background()
+	if err := a.Initialize(ctx); err != nil {
+		t.Fatalf("initialize: %v", err)
+	}
+	if _, err := a.NewSession(ctx, nil); err != nil {
+		t.Fatalf("new session: %v", err)
+	}
+	_ = drainEvents(a)
+
+	done := make(chan error, 1)
+	go func() { done <- a.Prompt(ctx, "try this", nil, 7) }()
+	select {
+	case <-fake.entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("prompt did not reach the fake Cursor agent")
+	}
+
+	sendCapturedUpdate(t, conn, `{"sessionId":"session-handoff","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"`+cursorRetriableStreamResetChunk+`"}}}`)
+	fake.releasePrompts()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Prompt returned error: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("prompt did not settle")
+	}
+
+	events := drainEvents(a)
+	var providerErrors []*streams.ProviderError
+	var errors, completes, messageChunks int
+	for _, event := range events {
+		switch event.Type {
+		case streams.EventTypeError:
+			errors++
+			providerErrors = append(providerErrors, event.ProviderError)
+		case streams.EventTypeComplete:
+			completes++
+		case streams.EventTypeMessageChunk:
+			messageChunks++
+		}
+	}
+	if errors != 1 || completes != 0 || messageChunks != 0 || len(providerErrors) != 1 {
+		t.Fatalf("events = %+v, want one structured error, no raw chunk, and no complete", events)
+	}
+	providerError := providerErrors[0]
+	if !providerError.Valid() || providerError.Source != streams.ProviderErrorSourceCursorACP ||
+		providerError.ProviderID != acpcompat.CursorAgentID ||
+		providerError.Message != cursorRetriableStreamResetChunk {
+		t.Fatalf("provider error = %+v, want valid sanitized Cursor diagnostic", providerError)
+	}
+}
+
+func TestSendPromptCursorProgressSupersedesRetriableMarker(t *testing.T) {
+	a, fake, conn := setupHandoffFakeAgent(t)
+	a.agentID = acpcompat.CursorAgentID
+	a.normalizer = NewNormalizer(acpcompat.CursorAgentID)
+	a.dialect = newACPDialect(acpcompat.CursorAgentID)
+
+	ctx := context.Background()
+	if err := a.Initialize(ctx); err != nil {
+		t.Fatalf("initialize: %v", err)
+	}
+	if _, err := a.NewSession(ctx, nil); err != nil {
+		t.Fatalf("new session: %v", err)
+	}
+	_ = drainEvents(a)
+
+	done := make(chan error, 1)
+	go func() { done <- a.Prompt(ctx, "try this", nil, 7) }()
+	select {
+	case <-fake.entered:
+	case <-time.After(2 * time.Second):
+		t.Fatal("prompt did not reach the fake Cursor agent")
+	}
+
+	sendCapturedUpdate(t, conn, `{"sessionId":"session-handoff","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"`+cursorRetriableStreamResetChunk+`"}}}`)
+	sendCapturedUpdate(t, conn, `{"sessionId":"session-handoff","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"Cursor resumed generation"}}}`)
+	fake.releasePrompts()
+
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatalf("Prompt returned error: %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("prompt did not settle")
+	}
+
+	events := drainEvents(a)
+	var errors, completes int
+	for _, event := range events {
+		switch event.Type {
+		case streams.EventTypeError:
+			errors++
+		case streams.EventTypeComplete:
+			completes++
+		}
+	}
+	if errors != 0 || completes != 1 {
+		t.Fatalf("events = %+v, want no error and one complete", events)
+	}
+}

apps/backend/internal/agentctl/types/streams/provider_error.go

@@ -10,6 +10,9 @@ const (
 	// ProviderErrorSourceCodexACP marks a safe diagnostic reconstructed from
 	// Codex ACP metadata and its matching capacity message.
 	ProviderErrorSourceCodexACP = "codex_acp"
+	// ProviderErrorSourceCursorACP marks Cursor's bounded HTTP/2 stream-reset
+	// diagnostic reconstructed from its terminal ACP control chunk.
+	ProviderErrorSourceCursorACP = "cursor_acp"
 )
 
 // ProviderError is the bounded, sanitized provider diagnostic that may cross

apps/backend/internal/orchestrator/background_work_accounting_test.go

@@ -1221,6 +1221,7 @@ func countActivityClears(recorded *recordingEventBus) int {
 func TestTransientFailurePreservesBackgroundRegistration(t *testing.T) {
 	svc, _ := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)
+	svc.beginPromptAttempt("s1", "exec-transient", 7, false)
 	recorded := &recordingEventBus{}
 	svc.eventBus = recorded
 	taskEvents := &recordingTaskEvents{}
@@ -1229,7 +1230,11 @@ func TestTransientFailurePreservesBackgroundRegistration(t *testing.T) {
 	svc.markForegroundIdle("s1")
 
 	svc.handleAgentFailed(t.Context(), watcher.AgentEventData{
-		TaskID: "t1", SessionID: "s1", AgentExecutionID: "exec-transient", ErrorMessage: overloaded529,
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "exec-transient",
+		PromptGeneration: 7,
+		ErrorMessage:     overloaded529,
 	})
 
 	if !svc.hasBackgroundTask("s1", "tool-transient") {

apps/backend/internal/orchestrator/dynamic_evidence.go

@@ -1,127 +1,275 @@
 package orchestrator
 
 import (
+	"context"
 	"sync"
 
 	"github.com/kandev/kandev/internal/orchestrator/watcher"
+	"github.com/kandev/kandev/internal/task/models"
 )
 
-// dynamicAttemptEvidence is deliberately process-local. It is only an
-// observation fence for the current downstream execution; the durable
-// continuation package is stored with the route generation separately.
-type dynamicAttemptEvidence struct {
-	mu            sync.Mutex
-	executionID   string
-	evidenceKnown bool
-	output        bool
-	effect        bool
+// promptAttemptEvidence is deliberately process-local. It fences recovery to
+// the concrete execution and prompt generation that produced the failure; the
+// durable dynamic continuation package is stored with the route generation
+// separately.
+type promptAttemptEvidence struct {
+	mu               sync.Mutex
+	executionID      string
+	promptGeneration uint64
+	evidenceKnown    bool
+	output           bool
+	effect           bool
+	dynamic          bool
 }
 
-func (s *Service) beginDynamicAttempt(sessionID string) {
+func (s *Service) beginPromptAttempt(
+	sessionID, executionID string,
+	promptGeneration uint64,
+	dynamic bool,
+) {
 	if sessionID == "" {
 		return
 	}
-	s.dynamicAttemptEvidence.Store(sessionID, &dynamicAttemptEvidence{evidenceKnown: true})
+	s.dynamicAttemptEvidence.Store(sessionID, &promptAttemptEvidence{
+		executionID:      executionID,
+		promptGeneration: promptGeneration,
+		evidenceKnown:    true,
+		dynamic:          dynamic,
+	})
 }
 
-func (s *Service) bindDynamicAttemptExecution(sessionID, executionID string) {
-	if sessionID == "" || executionID == "" {
-		return
-	}
-	v, ok := s.dynamicAttemptEvidence.Load(sessionID)
-	if !ok {
-		return
-	}
-	evidence, ok := v.(*dynamicAttemptEvidence)
-	if !ok {
-		return
-	}
-	evidence.mu.Lock()
-	evidence.executionID = executionID
-	evidence.mu.Unlock()
+func (s *Service) beginDynamicAttempt(sessionID string) {
+	s.beginPromptAttempt(sessionID, "", 1, true)
 }
 
-func (s *Service) observeDynamicAttempt(sessionID, executionID string, output, effect bool) {
-	if sessionID == "" || (!output && !effect) {
-		return
+func (s *Service) beginInteractivePromptAttempt(
+	ctx context.Context,
+	sessionID, executionID string,
+	dynamic bool,
+) {
+	s.beginPromptAttempt(sessionID, executionID, s.nextPromptGeneration(ctx, sessionID), dynamic)
+}
+
+func (s *Service) beginInitialPromptAttempt(sessionID string, dynamic bool) {
+	s.beginPromptAttempt(sessionID, "", 1, dynamic)
+}
+
+func (s *Service) bindPromptAttemptToExecution(ctx context.Context, sessionID, executionID string) {
+	s.bindPromptAttempt(sessionID, executionID, s.promptGenerationForSession(ctx, sessionID))
+}
+
+func (s *Service) nextPromptGeneration(ctx context.Context, sessionID string) uint64 {
+	generation := s.promptGenerationForSession(ctx, sessionID)
+	if generation == ^uint64(0) {
+		return 0
 	}
-	v, ok := s.dynamicAttemptEvidence.Load(sessionID)
+	return generation + 1
+}
+
+func (s *Service) promptGenerationForSession(ctx context.Context, sessionID string) uint64 {
+	if s.agentManager == nil || sessionID == "" {
+		return 0
+	}
+	reader, ok := s.agentManager.(interface {
+		GetPromptGenerationForSession(context.Context, string) (uint64, error)
+	})
 	if !ok {
+		return 0
+	}
+	generation, err := reader.GetPromptGenerationForSession(ctx, sessionID)
+	if err != nil {
+		return 0
+	}
+	return generation
+}
+
+func (s *Service) isDynamicPromptSession(session *models.TaskSession) bool {
+	return s.profileExecutionResolver != nil && session != nil &&
+		session.RouteGeneration > 0 && session.ExecutionProfileID != ""
+}
+
+func (s *Service) bindPromptAttempt(sessionID, executionID string, promptGeneration uint64) {
+	if sessionID == "" || (executionID == "" && promptGeneration == 0) {
 		return
 	}
-	evidence, ok := v.(*dynamicAttemptEvidence)
+	evidence, ok := s.promptAttemptForSession(sessionID)
 	if !ok {
 		return
 	}
 	evidence.mu.Lock()
 	defer evidence.mu.Unlock()
-	if evidence.executionID != "" && executionID != "" && evidence.executionID != executionID {
-		return
-	}
-	if evidence.executionID != "" && executionID == "" {
-		// Once a concrete execution is bound, an observation without its
-		// execution identity is ambiguous. Keep the route gate fail-closed.
-		evidence.evidenceKnown = false
-		return
-	}
-	if evidence.executionID == "" && executionID != "" {
+	if executionID != "" {
+		if evidence.executionID != "" && evidence.executionID != executionID {
+			evidence.evidenceKnown = false
+			return
+		}
 		evidence.executionID = executionID
 	}
+	if promptGeneration != 0 {
+		if evidence.promptGeneration != 0 && evidence.promptGeneration != promptGeneration {
+			evidence.evidenceKnown = false
+			return
+		}
+		evidence.promptGeneration = promptGeneration
+	}
+}
+
+func (s *Service) bindDynamicAttemptExecution(sessionID, executionID string) {
+	s.bindPromptAttempt(sessionID, executionID, 0)
+}
+
+func (s *Service) observePromptAttempt(
+	sessionID, executionID string,
+	promptGeneration uint64,
+	output, effect bool,
+) {
+	if sessionID == "" || (!output && !effect) {
+		return
+	}
+	evidence, ok := s.promptAttemptForSession(sessionID)
+	if !ok {
+		return
+	}
+	evidence.mu.Lock()
+	defer evidence.mu.Unlock()
+	if !evidence.promptIdentityMatchesLocked(executionID, promptGeneration) {
+		return
+	}
 	evidence.output = evidence.output || output
 	evidence.effect = evidence.effect || effect
 }
 
-func (s *Service) withDynamicAttemptEvidence(data watcher.AgentEventData) watcher.AgentEventData {
+func (s *Service) observeDynamicAttempt(sessionID, executionID string, output, effect bool) {
+	s.observePromptAttempt(sessionID, executionID, 0, output, effect)
+}
+
+func (s *Service) withPromptAttemptEvidence(data watcher.AgentEventData) watcher.AgentEventData {
 	if data.SessionID == "" {
 		return data
 	}
-	v, ok := s.dynamicAttemptEvidence.Load(data.SessionID)
+	// Lifecycle captures terminal failure evidence before publishing the
+	// failure event. That snapshot is authoritative when stream and lifecycle
+	// events travel through separate bus subscriptions; retain it while still
+	// using the process-local record to fence the session, execution, and
+	// generation identity.
+	lifecycleEvidenceKnown := data.EvidenceKnown
+	lifecycleOutputObserved := data.OutputObserved
+	lifecycleEffectObserved := data.EffectObserved
+	evidence, ok := s.promptAttemptForSession(data.SessionID)
 	if !ok {
-		return data
-	}
-	evidence, ok := v.(*dynamicAttemptEvidence)
-	if !ok {
-		return data
-	}
-	evidence.mu.Lock()
-	defer evidence.mu.Unlock()
-	data.DynamicRouteAttempt = true
-	if evidence.executionID != "" &&
-		(data.AgentExecutionID == "" || evidence.executionID != data.AgentExecutionID) {
-		// The event belongs to a predecessor or successor that is not the
-		// attempt represented by this evidence record. Keep the route gate
-		// fail-closed and let the durable execution fence reject it too.
+		// Lifecycle evidence is only authoritative after the process-local
+		// attempt record fences the session, execution, and generation. Without
+		// that record, a terminal snapshot could authorize a replacement for an
+		// unrelated or already-cleared attempt.
 		data.EvidenceKnown = false
 		data.OutputObserved = false
 		data.EffectObserved = false
 		return data
 	}
-	data.EvidenceKnown = evidence.evidenceKnown
-	data.OutputObserved = evidence.output
-	data.EffectObserved = evidence.effect
+	evidence.mu.Lock()
+	defer evidence.mu.Unlock()
+	if !evidence.promptIdentityMatchesLocked(data.AgentExecutionID, data.PromptGeneration) {
+		data.EvidenceKnown = false
+		data.OutputObserved = false
+		data.EffectObserved = false
+		if evidence.dynamic {
+			data.DynamicRouteAttempt = true
+		}
+		return data
+	}
+	if evidence.dynamic {
+		data.DynamicRouteAttempt = true
+	}
+	if lifecycleEvidenceKnown {
+		data.EvidenceKnown = true
+		data.OutputObserved = lifecycleOutputObserved || evidence.output
+		data.EffectObserved = lifecycleEffectObserved || evidence.effect
+	} else {
+		data.EvidenceKnown = evidence.evidenceKnown
+		data.OutputObserved = evidence.output
+		data.EffectObserved = evidence.effect
+	}
 	return data
 }
 
-func (s *Service) clearDynamicAttemptEvidence(sessionID, executionID string) {
-	if sessionID == "" || executionID == "" {
-		return
+func (s *Service) withDynamicAttemptEvidence(data watcher.AgentEventData) watcher.AgentEventData {
+	data.DynamicRouteAttempt = true
+	return s.withPromptAttemptEvidence(data)
+}
+
+func (s *Service) promptAttemptPreResultSafe(data watcher.AgentEventData) bool {
+	if data.SessionID == "" || data.DynamicRouteAttempt ||
+		data.AgentExecutionID == "" || data.PromptGeneration == 0 ||
+		!data.EvidenceKnown || data.OutputObserved || data.EffectObserved {
+		return false
 	}
-	v, ok := s.dynamicAttemptEvidence.Load(sessionID)
+	evidence, ok := s.promptAttemptForSession(data.SessionID)
 	if !ok {
+		return false
+	}
+	evidence.mu.Lock()
+	defer evidence.mu.Unlock()
+	return !evidence.dynamic && evidence.evidenceKnown &&
+		evidence.executionID == data.AgentExecutionID &&
+		evidence.promptGeneration == data.PromptGeneration &&
+		!evidence.output && !evidence.effect
+}
+
+func (s *Service) clearPromptAttemptEvidence(sessionID, executionID string, promptGeneration uint64) {
+	if sessionID == "" {
 		return
 	}
-	evidence, ok := v.(*dynamicAttemptEvidence)
+	evidence, ok := s.promptAttemptForSession(sessionID)
 	if !ok {
 		return
 	}
 	evidence.mu.Lock()
-	currentExecutionID := evidence.executionID
+	matches := evidence.promptIdentityMatchesForClearLocked(executionID, promptGeneration)
 	evidence.mu.Unlock()
-	if currentExecutionID == executionID {
+	if matches {
 		s.dynamicAttemptEvidence.CompareAndDelete(sessionID, evidence)
 	}
 }
 
+func (s *Service) promptAttemptForSession(sessionID string) (*promptAttemptEvidence, bool) {
+	v, ok := s.dynamicAttemptEvidence.Load(sessionID)
+	if !ok {
+		return nil, false
+	}
+	evidence, ok := v.(*promptAttemptEvidence)
+	return evidence, ok
+}
+
+func (e *promptAttemptEvidence) promptIdentityMatchesLocked(executionID string, promptGeneration uint64) bool {
+	if e.executionID != "" {
+		if executionID == "" || e.executionID != executionID {
+			e.evidenceKnown = false
+			return false
+		}
+	} else if executionID != "" {
+		e.executionID = executionID
+	}
+	if e.promptGeneration != 0 {
+		if promptGeneration == 0 || e.promptGeneration != promptGeneration {
+			e.evidenceKnown = false
+			return false
+		}
+	} else if promptGeneration != 0 {
+		e.promptGeneration = promptGeneration
+	}
+	return true
+}
+
+func (e *promptAttemptEvidence) promptIdentityMatchesForClearLocked(executionID string, promptGeneration uint64) bool {
+	if e.executionID != "" && (executionID == "" || e.executionID != executionID) {
+		return false
+	}
+	if e.promptGeneration != 0 && (promptGeneration == 0 || e.promptGeneration != promptGeneration) {
+		return false
+	}
+	return true
+}
+
 func dynamicPreResultSafe(data watcher.AgentEventData) bool {
 	return data.DynamicRouteAttempt && data.EvidenceKnown && !data.OutputObserved && !data.EffectObserved
 }

apps/backend/internal/orchestrator/dynamic_evidence_test.go

@@ -44,6 +44,22 @@ func TestDynamicPreResultRequiresExplicitKnownEvidence(t *testing.T) {
 	}
 }
 
+func TestDynamicAttemptEvidenceRequiresLocalIdentityFence(t *testing.T) {
+	var service Service
+	got := service.withDynamicAttemptEvidence(watcher.AgentEventData{
+		SessionID:        "session-1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		EvidenceKnown:    true,
+	})
+	if got.EvidenceKnown {
+		t.Fatal("lifecycle evidence without a local attempt record was accepted")
+	}
+	if dynamicPreResultSafe(got) {
+		t.Fatal("lifecycle evidence without a local attempt record was treated as pre-result safe")
+	}
+}
+
 func TestDynamicAttemptEvidenceRejectsAmbiguousExecutionEvents(t *testing.T) {
 	var service Service
 	service.beginDynamicAttempt("session-1")

apps/backend/internal/orchestrator/dynamic_launch.go

@@ -62,6 +62,7 @@ func (d *dynamicTaskDownstream) Launch(
 		return dynamicruntime.DownstreamExecution{}, fmt.Errorf("%w: %v", classified, err)
 	}
 	d.service.bindDynamicAttemptExecution(d.sessionID, execution.AgentExecutionID)
+	d.service.bindPromptAttemptToExecution(ctx, d.sessionID, execution.AgentExecutionID)
 	d.execution = execution
 	acpSessionID := ""
 	if session, sessionErr := d.service.repo.GetTaskSession(ctx, d.sessionID); sessionErr == nil && session != nil {
@@ -204,7 +205,7 @@ func (s *Service) launchPreparedSessionWithDynamicFallbackWithContinuation(
 	continuationInput *dynamicruntime.ContinuationInput,
 ) (*executor.TaskExecution, error) {
 	if s.profileExecutionResolver == nil {
-		return s.executor.LaunchPreparedSession(ctx, task, sessionID, options)
+		return s.launchConcretePreparedSession(ctx, task, sessionID, options)
 	}
 	session, err := s.repo.GetTaskSession(ctx, sessionID)
 	if err != nil {
@@ -215,7 +216,7 @@ func (s *Service) launchPreparedSessionWithDynamicFallbackWithContinuation(
 		return nil, err
 	}
 	if !dynamic {
-		return s.executor.LaunchPreparedSession(ctx, task, sessionID, options)
+		return s.launchConcretePreparedSession(ctx, task, sessionID, options)
 	}
 	downstream := &dynamicTaskDownstream{
 		service: s, task: task, sessionID: sessionID, options: options,
@@ -250,6 +251,22 @@ func (s *Service) launchPreparedSessionWithDynamicFallbackWithContinuation(
 	return downstream.execution, nil
 }
 
+func (s *Service) launchConcretePreparedSession(
+	ctx context.Context,
+	task *v1.Task,
+	sessionID string,
+	options executor.LaunchOptions,
+) (*executor.TaskExecution, error) {
+	if options.StartAgent && (options.Prompt != "" || len(options.Attachments) > 0) {
+		s.beginInitialPromptAttempt(sessionID, false)
+	}
+	execution, err := s.executor.LaunchPreparedSession(ctx, task, sessionID, options)
+	if execution != nil {
+		s.bindPromptAttemptToExecution(ctx, sessionID, execution.AgentExecutionID)
+	}
+	return execution, err
+}
+
 func (s *Service) dynamicContinuationForLaunch(
 	ctx context.Context,
 	task *v1.Task,

apps/backend/internal/orchestrator/event_handlers_agent.go

@@ -1356,7 +1356,8 @@ func (s *Service) handleAgentFailed(ctx context.Context, data watcher.AgentEvent
 }
 
 func (s *Service) handleAgentFailedLocked(ctx context.Context, data watcher.AgentEventData) {
-	data = s.withDynamicAttemptEvidence(data)
+	data = s.withPromptAttemptEvidence(data)
+	defer s.clearPromptAttemptEvidence(data.SessionID, data.AgentExecutionID, data.PromptGeneration)
 	s.logger.Warn("handling agent failed",
 		zap.String("task_id", data.TaskID),
 		zap.String("session_id", data.SessionID),

apps/backend/internal/orchestrator/event_handlers_streaming.go

@@ -94,18 +94,37 @@ func (s *Service) handleAgentStreamEvent(ctx context.Context, payload *lifecycle
 		return
 	}
 	switch eventType {
-	case "message_streaming", "thinking_streaming", agentEventComplete:
-		s.observeDynamicAttempt(
+	case "message_streaming":
+		s.observePromptAttempt(
 			payload.SessionID,
-			payload.ExecutionID,
+			eventExecutionID,
+			payload.Data.PromptGeneration,
+			strings.TrimSpace(payload.Data.Text) != "",
+			false,
+		)
+	case "thinking_streaming":
+		s.observePromptAttempt(
+			payload.SessionID,
+			eventExecutionID,
+			payload.Data.PromptGeneration,
 			strings.TrimSpace(payload.Data.Text) != "",
 			false,
 		)
 	case agentEventToolCall, agentEventToolUpdate:
-		s.observeDynamicAttempt(payload.SessionID, payload.ExecutionID, false, true)
+		s.observePromptAttempt(
+			payload.SessionID,
+			eventExecutionID,
+			payload.Data.PromptGeneration,
+			false,
+			true,
+		)
 	}
 	if eventType == agentEventComplete {
-		defer s.clearDynamicAttemptEvidence(payload.SessionID, payload.ExecutionID)
+		defer s.clearPromptAttemptEvidence(
+			payload.SessionID,
+			eventExecutionID,
+			payload.Data.PromptGeneration,
+		)
 	}
 
 	if !terminalCompleteStream {
@@ -303,19 +322,24 @@ func (s *Service) completeTurnForStreamEvent(
 func (s *Service) handleAgentErrorEvent(ctx context.Context, payload *lifecycle.AgentStreamEventPayload) {
 	taskID := payload.TaskID
 	sessionID := payload.SessionID
+	executionID := payload.ExecutionID
+	if executionID == "" {
+		executionID = payload.AgentID
+	}
 	if sessionID != "" {
 		failure := watcher.AgentEventData{
 			TaskID:           taskID,
 			SessionID:        sessionID,
-			AgentExecutionID: payload.ExecutionID,
+			AgentExecutionID: executionID,
 			AgentID:          payload.AgentID,
+			PromptGeneration: payload.Data.PromptGeneration,
 			ErrorMessage:     payload.Data.Error,
 			ProviderError:    payload.Data.ProviderError,
 		}
 		if failure.ErrorMessage == "" {
 			failure.ErrorMessage = payload.Data.Text
 		}
-		failure = s.withDynamicAttemptEvidence(failure)
+		failure = s.withPromptAttemptEvidence(failure)
 		if s.routeDynamicAgentFailure(ctx, failure, classifyKanbanFailure(failure)) {
 			return
 		}

apps/backend/internal/orchestrator/event_handlers_transient.go

@@ -125,13 +125,21 @@ func (s *Service) rememberTurnPrompt(sessionID, text, model string, planMode boo
 // handleRecoverableFailure); false for non-transient errors, office tasks,
 // or an exhausted retry budget.
 func (s *Service) handleTransientFailure(ctx context.Context, data watcher.AgentEventData) bool {
-	data = s.withDynamicAttemptEvidence(data)
+	data = s.withPromptAttemptEvidence(data)
 	// Dynamic profiles own both error classes and their retry/reset policy. The
 	// legacy Kanban retry ladder must not consume a configured dynamic retry
 	// budget before the shared evaluator sees the failure.
 	if data.DynamicRouteAttempt {
 		return false
 	}
+	if !s.promptAttemptPreResultSafe(data) {
+		s.logger.Debug("refusing automatic transient retry without safe prompt-attempt evidence",
+			zap.String("task_id", data.TaskID),
+			zap.String("session_id", data.SessionID),
+			zap.String("agent_execution_id", data.AgentExecutionID),
+			zap.Uint64("prompt_generation", data.PromptGeneration))
+		return false
+	}
 	classified := classifyKanbanFailure(data)
 	if data.SessionID == "" || routingerr.Decide(routingerr.ContextKanban, classified, time.Now().UTC()) != routingerr.DecisionShortRetry {
 		return false

apps/backend/internal/orchestrator/event_handlers_transient_replay_safety_test.go

@@ -0,0 +1,183 @@
+package orchestrator
+
+import (
+	"context"
+	"testing"
+
+	"github.com/kandev/kandev/internal/agent/runtime/lifecycle"
+	"github.com/kandev/kandev/internal/orchestrator/watcher"
+)
+
+const cursorTransportLostDiagnostic = "Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)"
+
+func TestHandleTransientFailure_ReplayRequiresPromptAttemptEvidence(t *testing.T) {
+	svc, _ := newTransientTestService(t)
+	t.Cleanup(svc.cancelAllTransientRetries)
+
+	data := watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		EvidenceKnown:    true,
+		ErrorMessage:     cursorTransportLostDiagnostic,
+	}
+	if svc.handleTransientFailure(context.Background(), data) {
+		t.Fatal("handleTransientFailure scheduled replay without a process-local prompt-attempt record")
+	}
+	if _, ok := svc.transientRetries.Load("s1"); ok {
+		t.Fatal("missing prompt-attempt evidence armed a retry entry")
+	}
+}
+
+func TestHandleTransientFailure_ReplayRejectsOutputOrToolEvidence(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		data watcher.AgentEventData
+	}{
+		{
+			name: "assistant output",
+			data: watcher.AgentEventData{OutputObserved: true},
+		},
+		{
+			name: "tool activity",
+			data: watcher.AgentEventData{EffectObserved: true},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			svc, _ := newTransientTestService(t)
+			t.Cleanup(svc.cancelAllTransientRetries)
+			svc.beginPromptAttempt("s1", "execution-1", 7, false)
+			svc.observePromptAttempt("s1", "execution-1", 7, tc.data.OutputObserved, tc.data.EffectObserved)
+			data := tc.data
+			data.TaskID = "t1"
+			data.SessionID = "s1"
+			data.AgentExecutionID = "execution-1"
+			data.PromptGeneration = 7
+			data.EvidenceKnown = true
+			data.ErrorMessage = cursorTransportLostDiagnostic
+
+			if svc.handleTransientFailure(context.Background(), data) {
+				t.Fatalf("handleTransientFailure scheduled replay after %s", tc.name)
+			}
+		})
+	}
+}
+
+func TestHandleTransientFailure_ReplayUsesCurrentPromptAttempt(t *testing.T) {
+	svc, mc := newTransientTestService(t)
+	t.Cleanup(svc.cancelAllTransientRetries)
+	svc.beginPromptAttempt("s1", "execution-1", 7, false)
+
+	data := watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     cursorTransportLostDiagnostic,
+	}
+	if !svc.handleTransientFailure(context.Background(), data) {
+		t.Fatal("current output-free prompt attempt did not schedule the existing retry owner")
+	}
+	v, ok := svc.transientRetries.Load("s1")
+	if !ok {
+		t.Fatal("expected one transient retry entry")
+	}
+	entry, ok := v.(*transientRetryEntry)
+	if !ok || entry.attempt != 1 {
+		t.Fatalf("retry entry = %#v, want first attempt", v)
+	}
+	if len(mc.sessionMessages) != 1 || mc.sessionMessages[0].metadata["failure_code"] != "agent_transport_lost" {
+		t.Fatalf("retry status messages = %+v, want one transport-loss warning", mc.sessionMessages)
+	}
+}
+
+func TestPromptAttemptEvidence_ObservesThoughtAndToolActivity(t *testing.T) {
+	svc, _ := newTransientTestService(t)
+	t.Cleanup(svc.cancelAllTransientRetries)
+	svc.beginPromptAttempt("s1", "execution-1", 7, false)
+
+	svc.handleAgentStreamEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "execution-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:             "thinking_streaming",
+			Text:             "thinking about the request",
+			PromptGeneration: 7,
+		},
+	})
+	svc.handleAgentStreamEvent(context.Background(), &lifecycle.AgentStreamEventPayload{
+		TaskID:      "t1",
+		SessionID:   "s1",
+		ExecutionID: "execution-1",
+		Data: &lifecycle.AgentStreamEventData{
+			Type:             agentEventToolCall,
+			ToolCallID:       "read-file-1",
+			ToolTitle:        "Read File",
+			ToolStatus:       "running",
+			PromptGeneration: 7,
+		},
+	})
+
+	data := watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     cursorTransportLostDiagnostic,
+	}
+	if svc.handleTransientFailure(context.Background(), data) {
+		t.Fatal("thought and tool activity incorrectly authorized automatic replay")
+	}
+	if _, ok := svc.transientRetries.Load("s1"); ok {
+		t.Fatal("unsafe thought/tool attempt armed a retry entry")
+	}
+}
+
+func TestPromptAttemptEvidence_RejectsReplacedAttempt(t *testing.T) {
+	svc, _ := newTransientTestService(t)
+	t.Cleanup(svc.cancelAllTransientRetries)
+	svc.beginPromptAttempt("s1", "execution-1", 7, false)
+	svc.beginPromptAttempt("s1", "execution-2", 8, false)
+
+	data := watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     cursorTransportLostDiagnostic,
+	}
+	if svc.handleTransientFailure(context.Background(), data) {
+		t.Fatal("stale execution/generation authorized automatic replay")
+	}
+}
+
+func TestCursorTransportLost_UsesOneSameProviderRetryOwner(t *testing.T) {
+	svc, _ := newTransientTestService(t)
+	t.Cleanup(svc.cancelAllTransientRetries)
+	svc.beginPromptAttempt("s1", "execution-1", 7, false)
+
+	data := watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		AgentID:          "cursor-acp",
+		PromptGeneration: 7,
+		ErrorMessage:     cursorTransportLostDiagnostic,
+	}
+	if !svc.handleTransientFailure(context.Background(), data) {
+		t.Fatal("Cursor transport-loss failure did not use the existing retry owner")
+	}
+	if !svc.handleTransientFailure(context.Background(), data) {
+		t.Fatal("second eligible failure did not reuse the existing retry owner")
+	}
+	v, ok := svc.transientRetries.Load("s1")
+	if !ok {
+		t.Fatal("retry owner disappeared after second eligible failure")
+	}
+	entry, ok := v.(*transientRetryEntry)
+	if !ok || entry.attempt != 2 {
+		t.Fatalf("retry entry = %#v, want the same owner at attempt 2", v)
+	}
+}

apps/backend/internal/orchestrator/event_handlers_transient_test.go

@@ -32,14 +32,21 @@ func newTransientTestService(t *testing.T) (*Service, *mockMessageCreator) {
 	return svc, mc
 }
 
+func armTransientPromptEvidence(svc *Service) {
+	svc.beginPromptAttempt("s1", "execution-1", 7, false)
+}
+
 func TestHandleTransientFailure_SchedulesRetryAndEmitsWarning(t *testing.T) {
 	svc, mc := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)
+	armTransientPromptEvidence(svc)
 
 	took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
-		TaskID:       "t1",
-		SessionID:    "s1",
-		ErrorMessage: overloaded529,
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     overloaded529,
 	})
 	if !took {
 		t.Fatal("handleTransientFailure = false, want true (should own the transient failure)")
@@ -86,12 +93,15 @@ func TestHandleTransientFailure_SchedulesRetryAndEmitsWarning(t *testing.T) {
 func TestHandleTransientFailure_ModelCapacityUsesProviderNeutralPolicy(t *testing.T) {
 	svc, mc := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)
+	armTransientPromptEvidence(svc)
 
 	took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
-		TaskID:       "t1",
-		SessionID:    "s1",
-		AgentID:      "codex-acp",
-		ErrorMessage: "Selected model is at capacity. Please try a different model.",
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		AgentID:          "codex-acp",
+		PromptGeneration: 7,
+		ErrorMessage:     "Selected model is at capacity. Please try a different model.",
 	})
 	if !took {
 		t.Fatal("model-capacity failure was not owned by the Kanban short-retry policy")
@@ -217,14 +227,17 @@ func TestHandleTransientFailure_NonTransientReturnsFalse(t *testing.T) {
 func TestHandleTransientFailure_ExhaustedFallsThrough(t *testing.T) {
 	svc, mc := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)
+	armTransientPromptEvidence(svc)
 
 	// Pre-seed an entry that has already used the full budget.
 	svc.transientRetries.Store("s1", &transientRetryEntry{attempt: transientMaxAttempts, cancel: func() {}})
 
 	took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
-		TaskID:       "t1",
-		SessionID:    "s1",
-		ErrorMessage: overloaded529,
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     overloaded529,
 	})
 	if took {
 		t.Fatal("handleTransientFailure = true after budget exhausted, want false (fall through to recovery)")
@@ -551,8 +564,15 @@ func TestTransientRetryDelayHonorsShortProviderReset(t *testing.T) {
 func TestHandleTransientFailure_AttemptCounterIncrements(t *testing.T) {
 	svc, _ := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)
+	armTransientPromptEvidence(svc)
 
-	data := watcher.AgentEventData{TaskID: "t1", SessionID: "s1", ErrorMessage: overloaded529}
+	data := watcher.AgentEventData{
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     overloaded529,
+	}
 
 	svc.handleTransientFailure(context.Background(), data)
 	svc.handleTransientFailure(context.Background(), data)

apps/backend/internal/orchestrator/event_handlers_transient_transport_lost_test.go

@@ -15,11 +15,14 @@ const realPeerDisconnected = `{"code":-32603,"message":"Internal error","data":{
 func TestHandleTransientFailure_TransportLostSchedulesRetry(t *testing.T) {
 	svc, mc := newTransientTestService(t)
 	t.Cleanup(svc.cancelAllTransientRetries)
+	armTransientPromptEvidence(svc)
 
 	took := svc.handleTransientFailure(context.Background(), watcher.AgentEventData{
-		TaskID:       "t1",
-		SessionID:    "s1",
-		ErrorMessage: realPeerDisconnected,
+		TaskID:           "t1",
+		SessionID:        "s1",
+		AgentExecutionID: "execution-1",
+		PromptGeneration: 7,
+		ErrorMessage:     realPeerDisconnected,
 	})
 	if !took {
 		t.Fatal("handleTransientFailure = false, want true (should own the transport-lost failure)")

apps/backend/internal/orchestrator/service.go

@@ -1032,10 +1032,11 @@ type Service struct {
 	// context. key: sessionID, value: capturedPrompt. Replaced every turn.
 	lastTurnPrompt sync.Map
 
-	// dynamicAttemptEvidence is keyed by logical session. A dynamic attempt is
-	// replaced at every concrete launch, and its execution ID fences late
-	// stream/lifecycle events from a predecessor. Fallback requires an explicit
-	// no-output/no-effect result from this map.
+	// dynamicAttemptEvidence is keyed by logical session and stores the current
+	// concrete or dynamic prompt attempt. It is replaced for every attempt, and
+	// its execution ID and prompt generation fence late stream/lifecycle events
+	// from a predecessor. Automatic recovery requires an explicit no-output,
+	// no-effect result from this map.
 	dynamicAttemptEvidence sync.Map
 
 	// Service state

apps/backend/internal/orchestrator/task_operations.go

@@ -4026,17 +4026,23 @@ func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prom
 		return nil, fmt.Errorf("%w, please wait for completion", ErrAgentPromptInProgress)
 	}
 
+	// Cache and reserve the replay identity before model switching. A restart
+	// based switch dispatches its prompt from StartAgentProcess, before the
+	// ordinary PromptTask admission path can initialize these records.
+	s.rememberTurnPrompt(sessionID, prompt, model, planMode, attachments)
+	if modelSwitchRequired(session, model) {
+		s.beginInitialPromptAttempt(sessionID, s.isDynamicPromptSession(session))
+	}
+
 	if result, handled, switchErr := s.trySwitchModelForPrompt(
 		ctx, taskID, sessionID, model, effectivePrompt, session, foregroundDispatch,
 	); handled {
+		if switchErr != nil {
+			s.clearPromptAttemptEvidence(sessionID, "", 0)
+		}
 		return result, switchErr
 	}
 
-	// Cache the raw prompt so a transient-provider-error (529) retry can
-	// re-drive this turn after backoff without the caller's context. Stores
-	// the pre-injection prompt; PromptTask re-applies config/plan transforms.
-	s.rememberTurnPrompt(sessionID, prompt, model, planMode, attachments)
-
 	session, rollback, err := s.claimPromptDispatch(
 		ctx, taskID, sessionID, options.claimEntryID, options.lifecyclePrompt,
 		options.reserveTurnUntilDispatch, options.promptDispatchRecovery,
@@ -4099,6 +4105,12 @@ func (s *Service) promptTask(ctx context.Context, taskID, sessionID string, prom
 	// Only bounded-ack callers preserve request context; ordinary prompts can take minutes.
 	promptCtx := options.executorContext(ctx)
 	s.bindPromptTurnID(promptCtx, session, rollback.turnID)
+	s.beginInteractivePromptAttempt(
+		promptCtx,
+		sessionID,
+		session.AgentExecutionID,
+		s.isDynamicPromptSession(session),
+	)
 	dispatchOutcome := &promptDispatchOutcome{}
 	onDispatched := s.promptDispatchCallback(
 		promptCtx, taskID, sessionID, rollback.reservedTurn, foregroundDispatch, dispatchOutcome,
@@ -4287,6 +4299,8 @@ func (s *Service) promptDispatchCallback(
 	outcome *promptDispatchOutcome,
 ) func() {
 	return func() {
+		executionID, _ := s.agentManager.GetExecutionIDForSession(ctx, sessionID)
+		s.bindPromptAttemptToExecution(ctx, sessionID, executionID)
 		var publicationErr error
 		if reservedTurn != nil && s.turnService != nil {
 			// Agentctl has already accepted the prompt. Publication must not inherit
@@ -4329,6 +4343,9 @@ func (s *Service) trySwitchModelForPrompt(ctx context.Context, taskID, sessionID
 		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)
+	}
 	revealedBackground := s.acceptForegroundDispatch(dispatch)
 	if revealedBackground || dispatch.yieldedBeforeBegin || s.acceptedForegroundDispatchClaim(dispatch) {
 		s.publishForegroundActivityChanged(ctx, taskID, sessionID)
@@ -5026,9 +5043,9 @@ func (s *Service) handlePromptError(ctx context.Context, taskID, sessionID strin
 
 // trySwitchModel handles model switching for a prompt. Returns (result, true, nil) if a switch was
 // performed, (nil, false, err) on error, or (nil, false, nil) if no switch was needed.
-func (s *Service) trySwitchModel(ctx context.Context, taskID, sessionID, model, effectivePrompt string, session *models.TaskSession) (*PromptResult, bool, error) {
-	if model == "" {
-		return nil, false, nil
+func modelSwitchRequired(session *models.TaskSession, model string) bool {
+	if model == "" || session == nil {
+		return false
 	}
 	var currentModel string
 	if session.AgentProfileSnapshot != nil {
@@ -5036,9 +5053,19 @@ func (s *Service) trySwitchModel(ctx context.Context, taskID, sessionID, model,
 			currentModel = m
 		}
 	}
-	if currentModel == model {
+	return currentModel != model
+}
+
+func (s *Service) trySwitchModel(ctx context.Context, taskID, sessionID, model, effectivePrompt string, session *models.TaskSession) (*PromptResult, bool, error) {
+	if !modelSwitchRequired(session, model) {
 		return nil, false, nil
 	}
+	var currentModel string
+	if session.AgentProfileSnapshot != nil {
+		if m, ok := session.AgentProfileSnapshot["model"].(string); ok {
+			currentModel = m
+		}
+	}
 	s.logger.Info("switching model",
 		zap.String("task_id", taskID),
 		zap.String("session_id", sessionID),

docs/plans/cursor-retriable-stream-reset/plan.md

@@ -0,0 +1,148 @@
+---
+created: 2026-08-31
+status: done
+requirements:
+  - REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001
+system_design:
+  - ../../specs/platform/system-design/provider-error-recovery.md
+legacy_specs: []
+---
+
+# Implementation Plan: Cursor Retriable Stream Reset
+
+## Overview
+
+Project Cursor's terminal HTTP/2 `RetriableError` control frame into one
+structured provider failure. Classify its safe diagnostic as the existing
+transient transport-loss cause. Authorize the existing same-provider retry only
+for prompt attempts that have no output or tool activity. Adapter detection and
+classifier work can proceed independently. Replay-safety wiring depends on both
+contracts and completes the end-to-end behavior.
+
+## Scope
+
+### In scope
+
+- Cursor-only, prompt-generation-scoped detection of the complete terminal
+  stream reset control frame.
+- Suppression of the control frame and one post-barrier structured error.
+- Suppression of a Kandev retry when Cursor resumes its own attempt after the
+  marker.
+- A dedicated deterministic classifier rule for the safe Cursor diagnostic.
+- Provider-neutral replay evidence for concrete interactive retry decisions.
+- Existing same-provider retry ownership, attempt limit, delay schedule, and
+  manual recovery fallback.
+
+### Out of scope
+
+- Generic scanning of arbitrary assistant output.
+- Widening the existing generic `transportLostRe` expression.
+- Cursor-specific retry timers, counters, or backoff values.
+- Automatic provider or model switching for `agent_transport_lost`.
+- New UI, user-facing copy, persistence, or public API changes.
+
+## Technical approach
+
+### Cursor ACP evidence projection
+
+- Add the allocation-conscious matcher and bounded safe diagnostic to
+  `apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_cursor.go`.
+- Extend `promptTurnState` in `adapter.go` with mutex-guarded pending Cursor
+  evidence. Matching control frames set the pending marker. Later
+  same-generation provider progress clears it. A later matching frame re-arms
+  it.
+- Add `observeCursorRetriableEvidence` beside the Codex observer in
+  `adapter_updates.go`. Check agent identity, event type, prompt generation, and
+  the anchored prefix before the full fingerprint. Suppress only a matching
+  control frame.
+- In `adapter_prompt.go`, inspect pending evidence only after `syncNotifQueue()`.
+  Emit one `EventTypeError` with a valid `ProviderError` and return instead of
+  emitting the ordinary complete event.
+- Add `ProviderErrorSourceCursorACP` to
+  `apps/backend/internal/agentctl/types/streams/provider_error.go`.
+
+### Deterministic classification
+
+- Add `cursor.retriable_stream_reset.v1` to
+  `apps/backend/internal/agent/runtime/routingerr/runtime_rules.go` before the
+  generic transport-loss rule.
+- Require the complete normalized Cursor diagnostic, with the existing
+  bracketed `canceled` decoration allowed.
+  Apply the existing context-cancellation veto without changing
+  `transportLostRe`.
+- Assert the full `Classify` result is `CodeAgentTransportLost`,
+  `ClassTransient`, high confidence, and auto-retryable. Pin that `[canceled]`
+  does not trigger the cancellation veto.
+
+### Replay authorization and existing retry owner
+
+- Generalize the process-local attempt evidence in
+  `apps/backend/internal/orchestrator/dynamic_evidence.go`. Concrete interactive
+  prompts also retain output and tool observations. The evidence has session,
+  execution, and prompt-generation scope. Preserve `DynamicRouteAttempt` as
+  routing identity. Do not turn concrete prompts into dynamic attempts.
+- Carry a conservative terminal evidence snapshot from lifecycle on
+  `agent.failed`, captured before completion bookkeeping marks the terminal
+  event as activity. The orchestrator must prefer this immutable snapshot over
+  stream-subscription timing while retaining local identity fencing.
+- Begin or replace concrete prompt evidence at prompt dispatch. Bind it to the
+  execution and generation. Update it from `event_handlers_streaming.go`. Clear
+  only the matching completed or replaced attempt.
+- Cache and reserve replay identity before a model-switch restart can dispatch
+  its replacement prompt, then bind it to the replacement execution.
+- Make `handleTransientFailure` require known evidence with no output or tool
+  activity before it schedules any automatic prompt replay. Missing, stale,
+  output-bearing, or tool-bearing evidence falls through to manual recovery.
+- Keep `CodeAgentTransportLost.FallbackAllowed` false. Dynamic routing retains
+  its current fail-closed gate and cannot reinterpret the Cursor signature as
+  permission to switch providers.
+- Reuse `transientMaxAttempts`, `transientRetryDelayFor`, the existing single
+  retry entry, and the existing resume-before-reprompt path without adding a
+  Cursor-specific retry loop.
+
+## Tests
+
+- `AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.12`: a new ACP test file covers the
+  exact captured payload, anchored negatives, partial negatives, agent scope,
+  and generation scope. It also covers chunk suppression, queue-barrier
+  ordering, and one structured error instead of complete.
+- `AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.13`: ACP tests cover progress that
+  clears the pending marker. They also cover a later matching marker that re-arms
+  it.
+- `AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.14`: a new `routingerr` test file
+  covers the semantic code, class, rule ID, invariants, and negative cases. It
+  also covers the `[canceled]` non-veto case.
+- `AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.15`: a new orchestrator test file
+  covers safe retry scheduling. It covers fail-closed behavior for missing,
+  stale, output, and tool evidence. It also covers prompt-generation
+  replacement.
+
+## Work orders
+
+- [x] [Task 01: Project Cursor terminal evidence](task-01-project-cursor-terminal-evidence.md)
+- [x] [Task 02: Classify Cursor stream resets](task-02-classify-cursor-stream-resets.md)
+- [x] [Task 03: Gate automatic prompt replay](task-03-gate-automatic-prompt-replay.md)
+
+## Verification results
+
+- Task 01: `cd apps/backend && go test -race ./internal/agentctl/server/adapter/transport/acp -run 'Test(CursorRetriable|ObserveCursorRetriable|SendPrompt.*Cursor)'` passed.
+- Task 02: `cd apps/backend && go test -race ./internal/agent/runtime/routingerr -run 'Test(ClassifyCursorRetriable|MatchRuntimeEnvironmentRules_CursorRetriable|IsTransientProviderError_Cursor)'` passed.
+- Task 03 targeted suite: `cd apps/backend && go test -race -tags fts5 ./internal/orchestrator -run 'Test(HandleTransientFailure.*Replay|PromptAttemptEvidence|CursorTransportLost)'` passed.
+- Remediation identity-fence suite: `cd apps/backend && go test -race -tags fts5 ./internal/orchestrator -run 'TestDynamicAttemptEvidence|TestDynamicPreResultRequiresExplicitKnownEvidence|TestHandleTransientFailure.*Replay|TestCursorTransportLost'` passed.
+- Full lifecycle race suite: `go test -race ./internal/agent/runtime/lifecycle` passed, 2088 tests.
+- Full orchestrator race suite: `go test -race -tags fts5 ./internal/orchestrator` passed, 2261 tests.
+- `make -C apps/backend lint` and changed-lines `golangci-lint` passed with 0 issues.
+- Changed Go files reported no `gofmt` differences; `git diff --check` passed.
+- `python3 scripts/lint-spec-files.py --all` passed.
+- `make -C apps/backend test` was attempted. The changed packages passed, while unrelated baseline tests failed because the host config discovery selected `/root/.kandev/config.yaml` and the GitHub fixture lacked `github_task_prs`.
+
+## Risks
+
+- ACP notification ordering can misclassify Cursor-owned retry progress if the
+  pending marker is examined before the queue barrier.
+- Clearing on an old tool update can hide a true terminal failure. Only renewed
+  provider generation or a newly-started tool can clear the marker.
+- Generalizing replay evidence can expose existing transient tests that assumed
+  unknown evidence was safe. Those tests must provide explicit safe evidence or
+  assert manual recovery.
+- A stale prompt generation must not clear or authorize a successor turn.

docs/plans/cursor-retriable-stream-reset/task-01-project-cursor-terminal-evidence.md

@@ -0,0 +1,93 @@
+---
+id: "01-project-cursor-terminal-evidence"
+title: "Project Cursor terminal evidence"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001
+acceptance_criteria:
+  - AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.12
+  - AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.13
+system_design:
+  - ../../specs/platform/system-design/provider-error-recovery.md
+---
+
+# Task 01: Project Cursor Terminal Evidence
+
+## Summary
+
+Recognize Cursor's exact HTTP/2 `RetriableError` control frame on the active
+prompt. Suppress it and settle one structured provider error after the prompt
+notification barrier. Treat later provider progress as Cursor-owned retry
+activity. Thus, a superseded marker does not become a Kandev failure.
+
+## In scope
+
+- Cursor dialect matcher and stable safe diagnostic.
+- Turn-scoped pending evidence with generation fencing.
+- Ordered notification observation and control-chunk suppression.
+- Post-barrier `EventTypeError` projection and Cursor provider-error source.
+- Exact, negative, stale-generation, internal-progress, and settlement tests.
+
+## Out of scope
+
+- Error classification or orchestrator recovery policy.
+- Generic assistant-output scanning.
+- Provider, model, or UI changes.
+
+## Acceptance
+
+- The exact captured chunk is matched only for `cursor-acp`, suppressed, and
+  converted into one valid structured error after all preceding notifications.
+- Embedded prose, partial signatures, zero or stale generations, and non-Cursor
+  events are not matched or suppressed.
+- Same-generation provider progress clears a pending marker, while a later
+  matching marker can become terminal evidence again.
+
+## Verification
+
+```bash
+cd apps/backend && go test -race ./internal/agentctl/server/adapter/transport/acp -run 'Test(CursorRetriable|ObserveCursorRetriable|SendPrompt.*Cursor)'
+```
+
+## Files likely touched
+
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_cursor.go`
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go`
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go`
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go`
+- `apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_cursor_retriable_test.go`
+- `apps/backend/internal/agentctl/types/streams/provider_error.go`
+
+## Dependencies
+
+None.
+
+## Risks
+
+- The hot observer runs for every chunk, so cheap identity and prefix checks
+  must precede the full case-insensitive fingerprint.
+- Notification and prompt-response races can emit complete before the error if
+  the barrier is bypassed.
+
+## Parallelism
+
+`parallel-safe` with Task 02. The files are disjoint.
+
+## Inputs
+
+- Provider Error Recovery requirement criteria `.12` and `.13`.
+- Cursor projection and retry-ownership sections in the system design.
+- The Codex capacity evidence observer and prompt-settlement precedent.
+- The embedded raw ACP evidence in the task request.
+
+## Results
+
+- Implemented Cursor-specific, generation-scoped ACP evidence projection and
+  post-barrier structured error settlement.
+- Added exact, anchored, stale-generation, adapter-scope, progress-reset, and
+  re-arm tests.
+- Verification passed:
+  `cd apps/backend && go test -race ./internal/agentctl/server/adapter/transport/acp -run 'Test(CursorRetriable|ObserveCursorRetriable|SendPrompt.*Cursor)'`.

docs/plans/cursor-retriable-stream-reset/task-02-classify-cursor-stream-resets.md

@@ -0,0 +1,83 @@
+---
+id: "02-classify-cursor-stream-resets"
+title: "Classify Cursor stream resets"
+status: done
+wave: 1
+depends_on: []
+plan: "plan.md"
+requirements:
+  - REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001
+acceptance_criteria:
+  - AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.14
+system_design:
+  - ../../specs/platform/system-design/provider-error-recovery.md
+---
+
+# Task 02: Classify Cursor Stream Resets
+
+## Summary
+
+Add a dedicated deterministic fingerprint for Cursor's safe stream-reset
+diagnostic and map it to the existing transient transport-loss cause. Preserve
+the generic transport matcher and the context-cancellation safety veto.
+
+## In scope
+
+- Dedicated runtime rule and stable classifier rule ID.
+- Exact positive, prose and partial negatives, and collision tests.
+- Full classification invariants and transient helper coverage.
+- Explicit `[canceled]` non-veto coverage.
+
+## Out of scope
+
+- ACP event detection or suppression.
+- Retry scheduling or effect-safety changes.
+- Widening `transportLostRe`.
+
+## Acceptance
+
+- The stable emitted diagnostic classifies as high-confidence
+  `CodeAgentTransportLost` with class `ClassTransient` and `AutoRetryable=true`.
+- The rule does not match prose or incomplete transport tokens and does not
+  change the priority of more specific existing rules.
+- `[canceled]` remains distinct from context cancellation, while existing
+  context-cancelled composite envelopes remain non-retryable.
+
+## Verification
+
+```bash
+cd apps/backend && go test -race ./internal/agent/runtime/routingerr -run 'Test(ClassifyCursorRetriable|MatchRuntimeEnvironmentRules_CursorRetriable|IsTransientProviderError_Cursor)'
+```
+
+## Files likely touched
+
+- `apps/backend/internal/agent/runtime/routingerr/runtime_rules.go`
+- `apps/backend/internal/agent/runtime/routingerr/cursor_retriable_stream_reset_test.go`
+
+## Dependencies
+
+None.
+
+## Risks
+
+- An overly broad expression can classify user-authored prose or unrelated
+  cancellation output as transport loss.
+- Rule ordering can accidentally supersede overload or resume-corruption
+  fingerprints.
+
+## Parallelism
+
+`parallel-safe` with Task 01. The files are disjoint.
+
+## Inputs
+
+- Provider Error Recovery criterion `.14`.
+- Cursor classification section in the system design.
+- Existing transport-loss rule and tests.
+
+## Results
+
+- Added `cursor.retriable_stream_reset.v1` with complete normalized-diagnostic
+  matching and the existing cancellation veto.
+- Verification passed:
+  `cd apps/backend && go test -race ./internal/agent/runtime/routingerr -run 'Test(ClassifyCursorRetriable|MatchRuntimeEnvironmentRules_CursorRetriable|IsTransientProviderError_Cursor)'`.

docs/plans/cursor-retriable-stream-reset/task-03-gate-automatic-prompt-replay.md

@@ -0,0 +1,113 @@
+---
+id: "03-gate-automatic-prompt-replay"
+title: "Gate automatic prompt replay"
+status: done
+wave: 2
+depends_on:
+  - "01-project-cursor-terminal-evidence"
+  - "02-classify-cursor-stream-resets"
+plan: "plan.md"
+requirements:
+  - REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001
+acceptance_criteria:
+  - AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.15
+system_design:
+  - ../../specs/platform/system-design/provider-error-recovery.md
+---
+
+# Task 03: Gate Automatic Prompt Replay
+
+## Summary
+
+Make the concrete interactive retry owner consume current prompt replay
+evidence before it schedules the existing retry loop. Preserve same-provider
+resume, current retry bounds, and dynamic routing's no-fallback invariant.
+
+## In scope
+
+- Provider-neutral prompt attempt evidence for concrete and dynamic sessions.
+- Session, execution, and prompt-generation fencing.
+- Output and tool-activity observation through the real stream path.
+- Safe scheduling through the existing transient retry owner.
+- Manual recovery for missing, stale, output-bearing, or tool-bearing evidence.
+- End-to-end coverage from the Cursor structured diagnostic through classifier
+  and retry decision.
+
+## Out of scope
+
+- A new retry loop, delay schedule, attempt cap, or persistent retry state.
+- Dynamic candidate switching for `agent_transport_lost`.
+- Cursor-specific orchestration branches.
+- UI or localization changes.
+
+## Acceptance
+
+- A current, known, output-free and tool-free transport reset schedules the
+  existing same-provider retry and no second owner.
+- Missing, stale, output, or tool evidence cannot schedule automatic replay and
+  falls through to the existing manual recovery surface.
+- The observed sequence with thoughts and a `Read File` call is unsafe, while
+  Cursor-owned progress before prompt settlement never creates a Kandev failure.
+
+## Verification
+
+```bash
+(cd apps/backend && go test -race -tags fts5 ./internal/orchestrator -run 'Test(HandleTransientFailure.*Replay|PromptAttemptEvidence|CursorTransportLost)')
+make -C apps/backend test
+make -C apps/backend lint
+(cd apps/backend && golangci-lint run ./... --new-from-rev="$(git merge-base HEAD origin/main)" --timeout=5m)
+git diff --name-only -- '*.go' | xargs -r gofmt -l
+```
+
+## Files likely touched
+
+- `apps/backend/internal/orchestrator/dynamic_evidence.go`
+- `apps/backend/internal/orchestrator/service.go`
+- `apps/backend/internal/orchestrator/task_operations.go`
+- `apps/backend/internal/orchestrator/event_handlers_streaming.go`
+- `apps/backend/internal/orchestrator/event_handlers_transient.go`
+- `apps/backend/internal/orchestrator/event_handlers_transient_replay_safety_test.go`
+
+## Dependencies
+
+- Task 01 supplies the structured Cursor provider error.
+- Task 02 supplies the transient transport-loss classification.
+
+## Risks
+
+- Evidence initialized too late can turn a safe no-output failure into unknown
+  and manual recovery.
+- Evidence retained too long can authorize or block the wrong prompt generation.
+- Existing transient tests must explicitly establish safe attempt evidence after
+  the fail-closed gate is applied.
+
+## Parallelism
+
+`sequential` after Tasks 01 and 02.
+
+## Inputs
+
+- Provider Error Recovery criterion `.15`.
+- Cursor retry-safety section in the system design.
+- Existing dynamic attempt evidence and transient retry owner.
+- Completed Task 01 and Task 02 contracts.
+
+## Results
+
+- Generalized process-local prompt evidence to cover concrete and dynamic
+  attempts with execution and prompt-generation fencing.
+- Added lifecycle-owned terminal evidence so NATS delivery order between
+  stream and failure subscriptions cannot authorize replay after prior activity.
+- Registered the cached prompt and replay identity before model-switch startup,
+  then bound it to the replacement execution.
+- Automatic concrete replay now requires known, output-free, tool-free evidence
+  and continues to use the existing same-provider retry owner and budget.
+- Tightened Cursor matching to the complete normalized diagnostic and covered
+  tool-update preservation, consumed-event handling, and timestamp capture.
+- Verification passed:
+  `cd apps/backend && go test -race -tags fts5 ./internal/orchestrator -run 'Test(HandleTransientFailure.*Replay|PromptAttemptEvidence|CursorTransportLost)'`.
+- The remediation identity-fence suite passed, including the dynamic missing-record
+  regression. The full lifecycle race suite passed with 2088 tests, and the full
+  orchestrator race suite passed with 2261 tests.
+- Backend lint and changed-lines lint passed with 0 issues. The full backend test target was
+  attempted; only unrelated host-config and missing-fixture failures remained.

docs/specs/platform/requirements/provider-error-recovery.md

@@ -2,7 +2,7 @@
 status: draft
 system: platform
 created: 2026-08-08
-updated: 2026-08-27
+updated: 2026-08-31
 owners:
   - Kandev
 ---
@@ -31,6 +31,10 @@ Agent CLIs and providers report equivalent failures through different ACP frames
 - **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.9:** While an eligible interactive transient retry is pending, desktop and mobile task chat shall show the retry reason, provider, attempt count, countdown, and Cancel action only while the backend owns that retry.
 - **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.10:** When an interactive transient retry succeeds, is exhausted, reaches a terminal state, is stopped, or is cancelled, the system shall attempt to durably retire every outstanding retry notice for the session so that a successful cleanup prevents the notice from reappearing after the session becomes idle, the task changes, the page reloads, or another viewer observes the session. If listing or deletion fails, the system shall log and swallow the failure so a later authorized cleanup can retry the remaining notices.
 - **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.11:** After authorization, cancelling an interactive transient retry shall attempt to retire outstanding retry notices even when the retry loop has already ended. Cancelling an active loop shall also expose manual Resume and Start fresh recovery actions. Listing or deletion failures follow AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.10. A denied or foreign task-session pair shall not expose whether retry state or retry notices exist.
+- **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.12:** When Cursor reports a current-prompt `RetriableError` control frame and then settles without later provider progress, Kandev shall suppress the control frame. After Kandev processes all earlier prompt notifications, it shall publish exactly one structured provider failure. Similar prose, partial signatures, stale prompt evidence, and non-Cursor output shall remain ordinary agent output.
+- **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.13:** When Cursor resumes generation before the prompt settles, Kandev shall treat the earlier matching frame as internal retry telemetry. Kandev shall not start a retry from the superseded frame. A later matching terminal frame for the same prompt can become new failure evidence.
+- **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.14:** A terminal Cursor `RetriableError` HTTP/2 stream reset shall classify as the existing transient agent transport-loss cause. The bracketed word `canceled` alone shall not classify the failure as an operator or context cancellation.
+- **AC-PLATFORM-PROVIDER-ERROR-RECOVERY-001.15:** Kandev shall automatically recover a terminal Cursor stream reset only when current-turn evidence proves that the attempt had no output or tool activity. Eligible recovery shall retain the selected provider. It shall use the existing same-provider retry owner, attempt limit, and backoff. Missing evidence, output, or tool activity shall stop automatic replay and expose manual recovery.
 
 ## System design
 

docs/specs/platform/system-design/provider-error-recovery.md

@@ -4,7 +4,7 @@ system: platform
 requirements:
   - REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001
 created: 2026-08-08
-updated: 2026-08-27
+updated: 2026-08-31
 owners:
   - Kandev
 ---
@@ -18,7 +18,7 @@ This design preserves the technical source detail for `REQ-PLATFORM-PROVIDER-ERR
 
 | Requirement | Design section |
 | --- | --- |
-| `REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001` | [Migrated source detail](#migrated-source-detail), [Interactive transient retry notice lifecycle](#interactive-transient-retry-notice-lifecycle) |
+| `REQ-PLATFORM-PROVIDER-ERROR-RECOVERY-001` | [Migrated source detail](#migrated-source-detail), [Cursor normal-completion failure projection](#cursor-normal-completion-failure-projection), [Cursor retry-safety semantics](#cursor-retry-safety-semantics), [Interactive transient retry notice lifecycle](#interactive-transient-retry-notice-lifecycle) |
 
 ## Migrated source detail
 
@@ -72,6 +72,60 @@ workspace modes.
 - Classification is deterministic in this version. Calling a model to classify
   an error or extract timing data is deferred.
 
+#### Cursor normal-completion failure projection
+
+`cursor-agent` can report an upstream HTTP/2 stream reset as an ordinary
+`agent_message_chunk`. It can later return a successful `session/prompt`
+response. The ACP transport therefore owns a Cursor-specific evidence
+projection that mirrors the existing Codex capacity projection. It does not add
+generic content scanning to orchestration.
+
+The observer runs in the ordered ACP notification worker and applies these
+checks in order:
+
+1. The adapter identity is `cursor-acp`.
+2. The normalized event is a non-empty assistant message chunk for a non-zero
+   prompt generation that matches the active turn.
+3. After trimming leading and trailing whitespace, the chunk begins with
+   `Error: RetriableError:`.
+4. A case-insensitive bounded match finds `RetriableError` and either
+   `http/2 stream closed` or `CANCEL (0x8)`.
+
+The identity, event-type, and prefix checks precede the full fingerprint. The
+prefix check uses `strings.HasPrefix(strings.TrimSpace(text), ...)`, so ordinary
+per-token traffic does not allocate a normalized copy. Prose that mentions the
+error after other text, a partial `RetriableError`, a stale generation, and the
+same text from another adapter do not match.
+
+A match sets pending evidence on the active `promptTurnState` under its existing
+evidence mutex. The observer suppresses the control chunk. A later non-empty
+assistant or thought chunk clears the marker for the same generation. A new
+tool call also clears it. This activity proves that Cursor resumed its own
+attempt. A later matching control chunk sets the marker again. Updates for an
+in-flight tool do not clear it. Such updates can arrive during error settlement
+without proving renewed provider generation.
+
+After `session/prompt` returns, `sendPrompt` drains the notification queue. Then
+it examines the turn. If the Cursor marker is pending, the adapter cancels the
+async completion owner. It emits exactly one `EventTypeError` instead of the
+ordinary complete event. The error carries this bounded constant diagnostic:
+`Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)`.
+The valid `ProviderError` uses source `cursor_acp`, provider ID `cursor-acp`, and
+a UTC occurrence time. The adapter does not copy raw assistant text into the
+structured diagnostic.
+
+The deterministic catalogue adds the dedicated rule
+`cursor.retriable_stream_reset.v1` before the generic transport-loss rule. The
+rule requires the complete normalized Cursor diagnostic
+`Error: RetriableError: HTTP/2 stream closed with error code CANCEL (0x8)`
+(with the existing bracketed `canceled` decoration allowed). It maps to
+`agent_transport_lost` with high confidence and class `transient`.
+It retains the existing `AutoRetryable` invariant. The rule applies the shared
+context-cancellation veto. Cursor's `[canceled]` token does not satisfy that
+veto. It lacks `context canceled`, `context deadline exceeded`, or
+`cancel escalated`. The deliberately narrow `transportLostRe` remains
+unchanged.
+
 ### Error classes
 
 The policy layer has two configurable classes:
@@ -111,6 +165,52 @@ Classification does not by itself authorize retry or switching.
   failure stops for manual recovery even when its class policy requests retry
   or skip.
 
+#### Cursor retry-safety semantics
+
+The Cursor label `RetriableError` is evidence about the upstream transport. It
+is not permission for Kandev to repeat a turn. The following rules define the
+Cursor recovery choices. They preserve the provider-neutral safety boundary in
+[ADR-2026-08-08-provider-neutral-agent-error-recovery](../../../decisions/2026-08-08-provider-neutral-agent-error-recovery.md):
+
+1. **Safe point.** A terminal marker is automatically replayable only when its
+   prompt-generation-correlated evidence is known and records neither assistant
+   output nor tool activity. Thoughts, message output, a pending or completed
+   tool call, and missing evidence all fail closed. The observed incident had
+   thoughts and a `Read File` call in flight, so it enters manual recovery even
+   though its classification is transient.
+2. **Continuation mode.** An eligible concrete-profile retry retains the
+   selected execution profile. It uses the existing provider-native resume
+   identity before it sends the cached original prompt again. It does not create
+   a fresh provider route or switch providers. This replay is allowed only at
+   the safe point above. An unsafe turn exposes the existing manual Resume and
+   Start fresh choices. The user, not the classifier, chooses continuation.
+3. **Cursor-owned retry.** Kandev does not schedule while the original
+   `session/prompt` RPC remains open. Provider progress after a marker clears
+   the pending marker. Only a later terminal marker can re-arm it. The prompt
+   barrier then proves that Cursor's internal retry has either resumed or
+   finished before Kandev chooses recovery. This prevents overlapping Cursor
+   and Kandev retry loops.
+4. **Budget and delay.** Eligible concrete-profile recovery reuses the single
+   orchestrator-owned retry entry, `transientMaxAttempts`, and
+   `transientRetryDelayFor`. There is no Cursor-specific nested counter, timer,
+   or backoff. Exhaustion uses the existing manual recovery path.
+
+The orchestrator records replay evidence for every interactive prompt, not only
+dynamic route attempts. The evidence remains scoped by session, execution, and
+prompt generation. Lifecycle snapshots the current prompt's evidence before it
+marks a terminal completion as activity and carries that immutable snapshot on
+`agent.failed`. This prevents separate NATS subscriptions for `agent.stream.*`
+and `agent.failed` from changing the replay decision based on delivery order.
+The concrete-profile `handleTransientFailure` path requires the same known,
+no-output, no-tool condition before scheduling. A missing record is unsafe.
+Dynamic profiles retain `dynamicPreResultSafe` and their configured policy
+owner. A model-switch restart reserves the cached prompt and replay identity
+before `StartAgentProcess` can dispatch the replacement prompt, then binds the
+identity to the replacement execution. `agent_transport_lost` remains
+same-provider recovery evidence and does not gain permission to switch
+candidates merely because Cursor supplied the new fingerprint. No orchestration
+branch inspects `cursor-acp`, `cursor_acp`, or the raw diagnostic.
+
 ### Per-class policy
 
 Each dynamic candidate stores one policy for `transient` and one for `hard`.