fix(backend): stop dynamic-route retries from double-launching the same prompt (#3360)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

apps/backend/internal/agent/runtime/dynamic/engine.go

@@ -17,6 +17,7 @@ const (
 	routeStatusStarting       = "starting"
 	routeStatusActive         = "active"
 	routeStatusActionRequired = "action_required"
+	routeStatusRetrying       = "retrying"
 )
 
 func WithClock(now func() time.Time) EngineOption {
@@ -576,6 +577,26 @@ func (e *Engine) persistInitialGeneration(
 	return true, nil
 }
 
+// persistExpectedStatus atomically updates a same-generation route status.
+// Persistence without status-fencing support fails closed.
+func (e *Engine) persistExpectedStatus(ctx context.Context, expectedGeneration int64, expectedStatus string, state RouteState) error {
+	if e.persistence == nil {
+		return nil
+	}
+	claimer, ok := e.persistence.(GenerationStatusClaimer)
+	if !ok {
+		return ErrStatusClaimUnsupported
+	}
+	claimed, err := claimer.ClaimRouteStateFrom(ctx, expectedGeneration, expectedStatus, state)
+	if err != nil {
+		return err
+	}
+	if !claimed {
+		return ErrStaleGeneration
+	}
+	return nil
+}
+
 func mustJSON(value PolicyState) []byte {
 	payload, err := json.Marshal(value)
 	if err != nil {
@@ -725,6 +746,27 @@ func (e *Engine) MarkActionRequired(
 	}, nil
 }
 
+// MarkRecoveryActionRequired returns an in-flight route to manual recovery.
+// Stale generations and states that already moved on are no-ops.
+func (e *Engine) MarkRecoveryActionRequired(ctx context.Context, sessionID string, expectedGeneration int64) error {
+	state, exists, err := e.stateForFailure(ctx, sessionID)
+	if err != nil {
+		return err
+	}
+	if !exists || state.Generation != expectedGeneration || state.Status != routeStatusRetrying {
+		return nil
+	}
+	state.Status = routeStatusActionRequired
+	state.UpdatedAt = e.now()
+	if err := e.persistExpectedStatus(ctx, expectedGeneration, routeStatusRetrying, state); err != nil {
+		return err
+	}
+	e.mu.Lock()
+	e.states[sessionID] = state
+	e.mu.Unlock()
+	return nil
+}
+
 func (e *Engine) resumePending(
 	ctx context.Context,
 	sessionID string,
@@ -741,12 +783,13 @@ func (e *Engine) resumePending(
 	if state.Generation != expectedGeneration {
 		return RouteDecision{}, ErrStaleGeneration
 	}
-	if force && state.Status == "retrying" {
+	if force && state.Status == routeStatusRetrying {
 		return RouteDecision{}, ErrRecoveryPending
 	}
 	if !force && state.Status != string(routingpolicy.DecisionRetry) && state.Status != string(routingpolicy.DecisionWaitForReset) {
 		return RouteDecision{}, ErrRecoveryPending
 	}
+	observedStatus := state.Status
 	var policyState PolicyState
 	if state.PolicyStateJSON != "" {
 		if err := json.Unmarshal([]byte(state.PolicyStateJSON), &policyState); err != nil {
@@ -757,10 +800,9 @@ func (e *Engine) resumePending(
 	if !force && policyState.Deadline != nil && now.Before(*policyState.Deadline) {
 		return RouteDecision{}, ErrRecoveryNotDue
 	}
-	expectedStatus := state.Status
-	state.Status = "retrying"
+	state.Status = routeStatusRetrying
 	state.UpdatedAt = now
-	if err := e.persistSameGeneration(ctx, expectedGeneration, expectedStatus, state); err != nil {
+	if err := e.persistExpectedStatus(ctx, expectedGeneration, observedStatus, state); err != nil {
 		return RouteDecision{}, err
 	}
 	e.mu.Lock()

apps/backend/internal/agent/runtime/dynamic/engine_test.go

@@ -404,6 +404,86 @@ func TestEngineMarkActionRequiredIsNoOpOnceRouteIsActive(t *testing.T) {
 	}
 }
 
+func TestEngineMarkRecoveryActionRequiredUnsticksManualRecoveryAfterFailedLaunch(t *testing.T) {
+	engine := NewEngine()
+	document := routingpolicy.DefaultDocument()
+	document.Transient.Retry = routingpolicy.RetryPolicy{Enabled: true, MaxRetries: 1, InitialIntervalSeconds: 60}
+	profile := Profile{ID: "launch-fail-policy", Version: 1, Candidates: []Candidate{
+		{ID: "first", Enabled: true, Policies: document},
+	}}
+	initial, err := engine.Select("launch-fail-session", profile, 0, "")
+	if err != nil {
+		t.Fatalf("initial Select: %v", err)
+	}
+	if _, err := engine.ApplyFailure(
+		"launch-fail-session", profile, initial.Generation, initial.ExecutionProfileID,
+		&routingerr.Error{Code: routingerr.CodeRateLimited, Class: routingerr.ClassTransient, FallbackAllowed: true},
+	); !errors.Is(err, ErrRecoveryPending) {
+		t.Fatalf("ApplyFailure error = %v, want recovery pending", err)
+	}
+	ctx := context.Background()
+	if _, err := engine.ResumePendingNow(ctx, "launch-fail-session", initial.Generation); err != nil {
+		t.Fatalf("ResumePendingNow (simulated timer/manual claim): %v", err)
+	}
+
+	// The durable state is now "retrying" — as if a launch was claimed but
+	// then failed before it could advance the state further. Every manual
+	// recovery path must reject it until the launch-failure handler syncs it.
+	if _, err := engine.ResumePendingNow(ctx, "launch-fail-session", initial.Generation); !errors.Is(err, ErrRecoveryPending) {
+		t.Fatalf("ResumePendingNow while stuck at retrying = %v, want recovery pending", err)
+	}
+	if _, err := engine.CancelPending(ctx, "launch-fail-session", initial.Generation, "manual_stop"); !errors.Is(err, ErrRecoveryPending) {
+		t.Fatalf("CancelPending while stuck at retrying = %v, want recovery pending", err)
+	}
+
+	if err := engine.MarkRecoveryActionRequired(ctx, "launch-fail-session", initial.Generation); err != nil {
+		t.Fatalf("MarkRecoveryActionRequired: %v", err)
+	}
+	state, exists := engine.State("launch-fail-session")
+	if !exists || state.Status != routeStatusActionRequired || state.Generation != initial.Generation {
+		t.Fatalf("state after MarkRecoveryActionRequired = %#v, exists=%v", state, exists)
+	}
+
+	// Manual retry and stop are both unblocked now that the durable state
+	// reflects action_required instead of a launch that will never resolve.
+	if stopped, err := engine.CancelPending(ctx, "launch-fail-session", initial.Generation, "manual_stop"); err != nil {
+		t.Fatalf("CancelPending after sync: %v", err)
+	} else if stopped.Status != routeStatusActionRequired {
+		t.Fatalf("stopped decision = %#v", stopped)
+	}
+}
+
+func TestEngineMarkRecoveryActionRequiredIsNoOpWhenStateMovedOn(t *testing.T) {
+	engine := NewEngine()
+	profile := Profile{ID: "no-op-policy", Version: 1, Candidates: []Candidate{
+		{ID: "first", Enabled: true},
+	}}
+	ctx := context.Background()
+	initial, err := engine.Select("no-op-session", profile, 0, "")
+	if err != nil {
+		t.Fatalf("initial Select: %v", err)
+	}
+
+	// Never entered "retrying" — no-op, not an error.
+	if err := engine.MarkRecoveryActionRequired(ctx, "no-op-session", initial.Generation); err != nil {
+		t.Fatalf("MarkRecoveryActionRequired on non-retrying state: %v", err)
+	}
+	state, _ := engine.State("no-op-session")
+	if state.Status == routeStatusActionRequired {
+		t.Fatalf("state unexpectedly advanced to action_required: %#v", state)
+	}
+
+	// Stale generation — no-op, not an error.
+	if err := engine.MarkRecoveryActionRequired(ctx, "no-op-session", initial.Generation+1); err != nil {
+		t.Fatalf("MarkRecoveryActionRequired on stale generation: %v", err)
+	}
+
+	// Unknown session — no-op, not an error.
+	if err := engine.MarkRecoveryActionRequired(ctx, "unknown-session", 1); err != nil {
+		t.Fatalf("MarkRecoveryActionRequired on unknown session: %v", err)
+	}
+}
+
 func TestBindingFingerprinterUsesOpaqueStableHMACKeys(t *testing.T) {
 	key := []byte("installation-secret")
 	fingerprinter := NewBindingFingerprinter(key)

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

@@ -27,6 +27,11 @@ var (
 	ErrRouteStateNotFound  = errors.New("dynamic route state not found")
 	ErrRecoveryPending     = errors.New("dynamic route recovery is pending")
 	ErrRecoveryNotDue      = errors.New("dynamic route recovery is not due")
+	// ErrStatusClaimUnsupported means the persistence implementation cannot
+	// fence a same-generation status transition. Status-fenced claims must
+	// fail closed rather than silently degrade to a generation-only claim,
+	// which would let two concurrent callers both win the same transition.
+	ErrStatusClaimUnsupported = errors.New("dynamic route persistence does not support status-fenced claims")
 )
 
 type Candidate struct {
@@ -134,7 +139,8 @@ type GenerationClaimer interface {
 
 // GenerationStatusClaimer updates one generation only while its status still
 // matches the caller's observation. This closes the same-generation race
-// between a launch becoming active and a concurrent recovery transition.
+// between a launch becoming active and a concurrent recovery transition, and
+// between two concurrent callers observing the same due retry/wait state.
 type GenerationStatusClaimer interface {
 	ClaimRouteStateFrom(context.Context, int64, string, RouteState) (bool, error)
 }

apps/backend/internal/agent/runtime/dynamic_resolver.go

@@ -363,7 +363,7 @@ func (r *ProfileExecutionResolver) resolveRetryRouteAction(
 		if resumeErr == nil {
 			return r.executionFromDecision(ctx, profileID, sessionID, decision)
 		}
-		if !errors.Is(resumeErr, dynamic.ErrRecoveryPending) && !errors.Is(resumeErr, dynamic.ErrRouteStateNotFound) {
+		if !errors.Is(resumeErr, dynamic.ErrRouteStateNotFound) {
 			return ProfileExecution{}, resumeErr
 		}
 	}
@@ -375,6 +375,11 @@ func (r *ProfileExecutionResolver) resolveSkipRouteAction(
 	sessionID, profileID, currentExecutionProfileID string,
 	expectedGeneration int64,
 ) (ProfileExecution, error) {
+	if state, exists, err := r.engine.LoadState(ctx, sessionID); err != nil {
+		return ProfileExecution{}, err
+	} else if exists && state.Generation == expectedGeneration && state.Status == "retrying" {
+		return ProfileExecution{}, dynamic.ErrRecoveryPending
+	}
 	profileConfig, err := r.loadDynamicProfile(ctx, profileID)
 	if err != nil {
 		return ProfileExecution{}, err
@@ -442,6 +447,19 @@ func (r *ProfileExecutionResolver) ResumePendingRoute(
 	return r.executionFromDecision(ctx, state.LogicalProfileID, sessionID, decision)
 }
 
+// MarkRouteRecoveryActionRequired returns a claimed "retrying" route to
+// manual recovery after its resumed launch failed.
+func (r *ProfileExecutionResolver) MarkRouteRecoveryActionRequired(
+	ctx context.Context,
+	sessionID string,
+	expectedGeneration int64,
+) error {
+	if r.engine == nil {
+		return nil
+	}
+	return r.engine.MarkRecoveryActionRequired(ctx, sessionID, expectedGeneration)
+}
+
 func (r *ProfileExecutionResolver) resolve(
 	ctx context.Context,
 	sessionID, profileID string,

apps/backend/internal/agent/runtime/dynamic_resolver_test.go

@@ -0,0 +1,132 @@
+package runtime
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/kandev/kandev/internal/agent/agents"
+	"github.com/kandev/kandev/internal/agent/runtime/dynamic"
+	"github.com/kandev/kandev/internal/agent/runtime/routingerr"
+	"github.com/kandev/kandev/internal/agent/runtime/routingpolicy"
+	agentsettingsmodels "github.com/kandev/kandev/internal/agent/settings/models"
+	"github.com/kandev/kandev/internal/agent/settings/store"
+)
+
+func TestResolveRouteActionRejectsManualRetryAfterRecoveryClaim(t *testing.T) {
+	ctx := context.Background()
+	resolver, engine, generation := newRetryingRouteActionResolver(t)
+	_, err := resolver.ResolveRouteAction(
+		ctx, "session-retry", "dynamic-profile", "concrete-profile", generation, "retry",
+	)
+	if !errors.Is(err, dynamic.ErrRecoveryPending) {
+		t.Fatalf("ResolveRouteAction error = %v, want recovery pending", err)
+	}
+	state, ok := engine.State("session-retry")
+	if !ok || state.Generation != generation || state.Status != "retrying" {
+		t.Fatalf("route state = %#v, ok=%v, want unchanged claimed state", state, ok)
+	}
+}
+
+func TestResolveRouteActionRejectsSkipAfterRecoveryClaim(t *testing.T) {
+	resolver, _, generation := newRetryingRouteActionResolver(t)
+	_, err := resolver.ResolveRouteAction(
+		context.Background(), "session-retry", "dynamic-profile", "concrete-profile", generation, "skip",
+	)
+	if !errors.Is(err, dynamic.ErrRecoveryPending) {
+		t.Fatalf("ResolveRouteAction error = %v, want recovery pending", err)
+	}
+}
+
+func TestMarkRouteActionRequiredUnblocksRetryAndSkipAfterFailedLaunch(t *testing.T) {
+	ctx := context.Background()
+	resolver, engine, generation := newRetryingRouteActionResolver(t)
+
+	// Simulate the launch-failure handler's sync (dynamic_policy_recovery.go
+	// and dynamic_routing.go both call this after a resumed launch fails).
+	if err := resolver.MarkRouteRecoveryActionRequired(ctx, "session-retry", generation); err != nil {
+		t.Fatalf("MarkRouteRecoveryActionRequired: %v", err)
+	}
+	state, ok := engine.State("session-retry")
+	if !ok || state.Status != "action_required" {
+		t.Fatalf("route state after sync = %#v, ok=%v, want action_required", state, ok)
+	}
+
+	if _, err := resolver.ResolveRouteAction(
+		ctx, "session-retry", "dynamic-profile", "concrete-profile", generation, "retry",
+	); err != nil {
+		t.Fatalf("ResolveRouteAction retry after sync = %v, want success", err)
+	}
+}
+
+func newRetryingRouteActionResolver(t *testing.T) (*ProfileExecutionResolver, *dynamic.Engine, int64) {
+	t.Helper()
+	ctx := context.Background()
+	engine := dynamic.NewEngine()
+	document := routingpolicy.DefaultDocument()
+	document.Transient.Retry = routingpolicy.RetryPolicy{
+		Enabled: true, MaxRetries: 1, InitialIntervalSeconds: 60,
+	}
+	profile := dynamic.Profile{ID: "dynamic-profile", Version: 1, Candidates: []dynamic.Candidate{
+		{ID: "concrete-profile", Enabled: true, Policies: document},
+	}}
+	initial, err := engine.Select("session-retry", profile, 0, "")
+	if err != nil {
+		t.Fatalf("initial Select: %v", err)
+	}
+	if _, err := engine.ApplyFailure(
+		"session-retry", profile, initial.Generation, initial.ExecutionProfileID,
+		&routingerr.Error{Code: routingerr.CodeRateLimited, Class: routingerr.ClassTransient, FallbackAllowed: true},
+	); !errors.Is(err, dynamic.ErrRecoveryPending) {
+		t.Fatalf("ApplyFailure error = %v, want recovery pending", err)
+	}
+	if _, err := engine.ResumePendingNow(ctx, "session-retry", initial.Generation); err != nil {
+		t.Fatalf("ResumePendingNow: %v", err)
+	}
+
+	profiles := &dynamicResolverTestProfiles{
+		logical:  &agentsettingsmodels.AgentProfile{ID: "dynamic-profile", AgentID: agents.DynamicAgentID, Enabled: true},
+		concrete: &agentsettingsmodels.AgentProfile{ID: "concrete-profile", AgentID: "concrete", Enabled: true},
+		dynamic:  &agentsettingsmodels.DynamicAgentProfile{ProfileID: "dynamic-profile", Version: 1},
+		routes: []agentsettingsmodels.DynamicAgentRoute{{
+			DynamicProfileID: "dynamic-profile", ExecutionProfileID: "concrete-profile", Enabled: true,
+		}},
+	}
+	return NewProfileExecutionResolver(profiles, engine, true), engine, initial.Generation
+}
+
+type dynamicResolverTestProfiles struct {
+	store.Repository
+	store.DynamicProfileRepository
+	logical  *agentsettingsmodels.AgentProfile
+	concrete *agentsettingsmodels.AgentProfile
+	dynamic  *agentsettingsmodels.DynamicAgentProfile
+	routes   []agentsettingsmodels.DynamicAgentRoute
+}
+
+func (p *dynamicResolverTestProfiles) GetAgentProfile(_ context.Context, id string) (*agentsettingsmodels.AgentProfile, error) {
+	switch id {
+	case p.logical.ID:
+		return p.logical, nil
+	case p.concrete.ID:
+		return p.concrete, nil
+	default:
+		return nil, errors.New("profile not found")
+	}
+}
+
+func (p *dynamicResolverTestProfiles) GetAgent(_ context.Context, id string) (*agentsettingsmodels.Agent, error) {
+	if id == agents.DynamicAgentID {
+		return &agentsettingsmodels.Agent{ID: id, Name: agents.DynamicAgentID}, nil
+	}
+	return &agentsettingsmodels.Agent{ID: id, Name: "concrete"}, nil
+}
+
+func (p *dynamicResolverTestProfiles) GetDynamicAgentProfile(
+	_ context.Context, profileID string,
+) (*agentsettingsmodels.DynamicAgentProfile, []agentsettingsmodels.DynamicAgentRoute, error) {
+	if profileID != p.dynamic.ProfileID {
+		return nil, nil, errors.New("dynamic profile not found")
+	}
+	return p.dynamic, p.routes, nil
+}

apps/backend/internal/backendapp/dynamic_routing.go

@@ -47,6 +47,11 @@ func applyDynamicRouteAction(
 	if err != nil {
 		return nil, err
 	}
+	if err := repairDynamicRouteAfterLaunchFailure(
+		ctx, session, request.ExpectedGeneration, resolver.MarkRouteRecoveryActionRequired,
+	); err != nil {
+		return nil, routeActionError(ctx, repo, session, err)
+	}
 	decision, err := resolveDynamicRouteAction(ctx, resolver, request, session)
 	if err != nil {
 		return handleDynamicRouteActionError(ctx, repo, session, err)
@@ -57,7 +62,7 @@ func applyDynamicRouteAction(
 	if request.Action == orchestrator.RouteActionCancelWait || request.Action == orchestrator.RouteActionStop {
 		return routeActionResult(ctx, repo, session), nil
 	}
-	return finishDynamicRouteAction(ctx, repo, session.ID, launchSuccessor)
+	return finishDynamicRouteAction(ctx, repo, resolver, session.ID, decision.Generation, launchSuccessor)
 }
 
 func loadDynamicRouteActionSession(
@@ -155,7 +160,9 @@ func persistDynamicRouteSelection(
 func finishDynamicRouteAction(
 	ctx context.Context,
 	repo *sqliterepo.Repository,
+	resolver *agentruntime.ProfileExecutionResolver,
 	sessionID string,
+	expectedGeneration int64,
 	launchSuccessor func(context.Context, string) error,
 ) (*orchestrator.RouteActionResult, error) {
 	if launchSuccessor == nil {
@@ -166,7 +173,7 @@ func finishDynamicRouteAction(
 		return routeActionResult(ctx, repo, session), nil
 	}
 	if err := launchSuccessor(ctx, sessionID); err != nil {
-		return recoverDynamicRouteAction(ctx, repo, sessionID, err)
+		return recoverDynamicRouteAction(ctx, repo, resolver, sessionID, expectedGeneration, err)
 	}
 	session, err := repo.GetTaskSession(ctx, sessionID)
 	if err != nil {
@@ -178,15 +185,23 @@ func finishDynamicRouteAction(
 func recoverDynamicRouteAction(
 	ctx context.Context,
 	repo *sqliterepo.Repository,
+	resolver *agentruntime.ProfileExecutionResolver,
 	sessionID string,
+	expectedGeneration int64,
 	launchErr error,
 ) (*orchestrator.RouteActionResult, error) {
+	if resolver != nil {
+		_ = resolver.MarkRouteRecoveryActionRequired(ctx, sessionID, expectedGeneration)
+	}
 	session, err := repo.GetTaskSession(ctx, sessionID)
 	if err != nil {
 		return nil, err
 	}
+	if session.RouteGeneration != expectedGeneration {
+		return routeActionResult(ctx, repo, session), nil
+	}
 	session.RouteState = "action_required"
-	session.RouteReason = "route_action_launch_failed"
+	session.RouteReason = orchestrator.RouteActionLaunchFailedReason
 	session.State = models.TaskSessionStateWaitingForInput
 	session.ErrorMessage = launchErr.Error()
 	session.DownstreamACPSessionID = ""
@@ -196,6 +211,23 @@ func recoverDynamicRouteAction(
 	return routeActionResult(ctx, repo, session), nil
 }
 
+func repairDynamicRouteAfterLaunchFailure(
+	ctx context.Context,
+	session *models.TaskSession,
+	expectedGeneration int64,
+	mark func(context.Context, string, int64) error,
+) error {
+	if session == nil || mark == nil ||
+		session.RouteReason != orchestrator.RouteActionLaunchFailedReason ||
+		session.RouteGeneration != expectedGeneration {
+		return nil
+	}
+	if err := mark(ctx, session.ID, expectedGeneration); err != nil {
+		return fmt.Errorf("restore dynamic route after failed launch: %w", err)
+	}
+	return nil
+}
+
 func routeActionResult(
 	ctx context.Context,
 	repo *sqliterepo.Repository,

apps/backend/internal/backendapp/dynamic_routing_test.go

@@ -0,0 +1,67 @@
+package backendapp
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/kandev/kandev/internal/orchestrator"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+func TestRepairDynamicRouteAfterLaunchFailureSurfacesMarkerError(t *testing.T) {
+	session := &models.TaskSession{
+		ID:              "session-recovery",
+		RouteGeneration: 7,
+		RouteReason:     orchestrator.RouteActionLaunchFailedReason,
+	}
+	markerErr := errors.New("database unavailable")
+	calls := 0
+	err := repairDynamicRouteAfterLaunchFailure(
+		context.Background(), session, session.RouteGeneration,
+		func(context.Context, string, int64) error {
+			calls++
+			return markerErr
+		},
+	)
+	if !errors.Is(err, markerErr) {
+		t.Fatalf("repair error = %v, want marker error", err)
+	}
+	if calls != 1 {
+		t.Fatalf("marker calls = %d, want one retry", calls)
+	}
+}
+
+func TestRepairDynamicRouteAfterLaunchFailureIgnoresUnrelatedProjection(t *testing.T) {
+	tests := []struct {
+		name       string
+		reason     string
+		generation int64
+	}{
+		{name: "different reason", reason: "manual_retry", generation: 7},
+		{name: "stale generation", reason: orchestrator.RouteActionLaunchFailedReason, generation: 8},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			calls := 0
+			session := &models.TaskSession{
+				ID:              "session-recovery",
+				RouteGeneration: 7,
+				RouteReason:     tt.reason,
+			}
+			err := repairDynamicRouteAfterLaunchFailure(
+				context.Background(), session, tt.generation,
+				func(context.Context, string, int64) error {
+					calls++
+					return nil
+				},
+			)
+			if err != nil {
+				t.Fatalf("repair error = %v, want nil", err)
+			}
+			if calls != 0 {
+				t.Fatalf("marker calls = %d, want zero", calls)
+			}
+		})
+	}
+}

apps/backend/internal/orchestrator/dynamic_policy_recovery.go

@@ -157,6 +157,18 @@ func (s *Service) runDynamicPolicyRecovery(ctx context.Context, sessionID string
 	if !ok {
 		return
 	}
+	// Serialize route recovery with manual actions through the durable
+	// projection. Release before lifecycle launch because launch takes the
+	// session lifecycle lock.
+	lock, release := s.acquireCancelInFlightGuard(sessionID)
+	lock.Lock()
+	locked := true
+	defer func() {
+		if locked {
+			lock.Unlock()
+			release()
+		}
+	}()
 	state, ok := loadDueDynamicPolicyState(ctx, loader, sessionID, generation)
 	if !ok {
 		return
@@ -168,7 +180,15 @@ func (s *Service) runDynamicPolicyRecovery(ctx context.Context, sessionID string
 	if s.handleDynamicPolicyResumeError(ctx, loader, sessionID, generation, err) {
 		return
 	}
-	s.persistDynamicPolicyRecovery(ctx, sessionID, generation, resolved)
+	if !s.persistDynamicPolicyRecovery(ctx, sessionID, generation, resolved) {
+		return
+	}
+	lock.Unlock()
+	release()
+	locked = false
+	if err := s.LaunchDynamicRouteAction(ctx, sessionID); err != nil {
+		s.markDynamicPolicyRecoveryActionRequired(ctx, sessionID, generation, err)
+	}
 }
 
 func (s *Service) removeDynamicPolicyRecoveryTimer(sessionID string) {
@@ -234,10 +254,10 @@ func (s *Service) persistDynamicPolicyRecovery(
 	sessionID string,
 	generation int64,
 	resolved agentruntime.ProfileExecution,
-) {
+) bool {
 	session, err := s.repo.GetTaskSession(ctx, sessionID)
 	if err != nil || session == nil || session.RouteGeneration != generation || session.ExecutionProfileID != resolved.ExecutionProfileID {
-		return
+		return false
 	}
 	applyDynamicRouteDecisionProjection(session, resolved.Decision)
 	session.RouteState = resolved.Decision.Status
@@ -246,11 +266,9 @@ func (s *Service) persistDynamicPolicyRecovery(
 	if err := s.repo.UpdateTaskSession(ctx, session); err != nil {
 		s.logger.Warn("failed to persist dynamic policy retry state",
 			zap.String("session_id", sessionID), zap.Error(err))
-		return
-	}
-	if err := s.LaunchDynamicRouteAction(ctx, sessionID); err != nil {
-		s.markDynamicPolicyRecoveryActionRequired(ctx, sessionID, err)
+		return false
 	}
+	return true
 }
 
 func dynamicPolicyDeadline(rawPolicyState string) *time.Time {
@@ -262,13 +280,24 @@ func dynamicPolicyDeadline(rawPolicyState string) *time.Time {
 	return &deadline
 }
 
-func (s *Service) markDynamicPolicyRecoveryActionRequired(ctx context.Context, sessionID string, launchErr error) {
+func (s *Service) markDynamicPolicyRecoveryActionRequired(
+	ctx context.Context, sessionID string, generation int64, launchErr error,
+) {
+	if s.profileExecutionResolver != nil {
+		if err := s.profileExecutionResolver.MarkRouteRecoveryActionRequired(ctx, sessionID, generation); err != nil {
+			s.logger.Warn("failed to sync durable route state to action_required after launch failure",
+				zap.String("session_id", sessionID), zap.Error(err))
+		}
+	}
 	session, err := s.repo.GetTaskSession(ctx, sessionID)
 	if err != nil || session == nil {
 		return
 	}
+	if session.RouteGeneration != generation {
+		return
+	}
 	session.RouteState = "action_required"
-	session.RouteReason = "route_action_launch_failed"
+	session.RouteReason = RouteActionLaunchFailedReason
 	session.State = models.TaskSessionStateWaitingForInput
 	session.ErrorMessage = launchErr.Error()
 	session.DownstreamACPSessionID = ""

apps/backend/internal/orchestrator/dynamic_policy_recovery_race_test.go

@@ -0,0 +1,171 @@
+package orchestrator
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/jmoiron/sqlx"
+	_ "github.com/mattn/go-sqlite3"
+
+	agentruntime "github.com/kandev/kandev/internal/agent/runtime"
+	dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic"
+	agentsettingsmodels "github.com/kandev/kandev/internal/agent/settings/models"
+	agentsettingsstore "github.com/kandev/kandev/internal/agent/settings/store"
+	"github.com/kandev/kandev/internal/orchestrator/executor"
+	sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite"
+	wfmodels "github.com/kandev/kandev/internal/workflow/models"
+	v1 "github.com/kandev/kandev/pkg/api/v1"
+)
+
+// TestConcurrentDynamicPolicyRecoveryLaunchesExactlyOnce reproduces two
+// recovery-timer fires racing on the same due generation (e.g. a duplicate
+// schedule or a backend-restart reschedule overlapping a still-live timer).
+// Before this fix, runDynamicPolicyRecovery held no per-session guard and
+// Engine.resumePending's durable claim predicate did not include the observed
+// status, so both callers could claim the same retry_wait -> retrying
+// transition and both launch a successor, delivering the same prompt twice.
+// With the exact-status CAS (ClaimRouteStatus) plus the shared
+// acquireCancelInFlightGuard held through the claim and session projection,
+// exactly one caller claims the transition and only it reaches the launch.
+func TestConcurrentDynamicPolicyRecoveryLaunchesExactlyOnce(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID            = "task-policy-race"
+		sessionID         = "session-policy-race"
+		executionID       = "execution-policy-race"
+		dynamicProfileID  = "profile-dynamic-race"
+		concreteProfileID = "profile-concrete-race"
+	)
+
+	repo := setupTestRepo(t)
+	seedSession(t, repo, taskID, sessionID, "step-race")
+
+	resolver := newDynamicPolicyRaceResolver(t, repo, dynamicProfileID, concreteProfileID)
+
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("get session: %v", err)
+	}
+	session.AgentProfileID = dynamicProfileID
+	session.ExecutionProfileID = concreteProfileID
+	session.RouteGeneration = 4
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("update session: %v", err)
+	}
+	seedExecutorRunning(t, repo, sessionID, taskID, executionID)
+
+	elapsedDeadline := time.Now().UTC().Add(-time.Minute)
+	if err := repo.SaveRouteState(ctx, dynamicruntime.RouteState{
+		SessionID: sessionID, LogicalProfileID: dynamicProfileID,
+		ExecutionProfileID: concreteProfileID, Generation: 4, ProfileVersion: 1,
+		Status:          "retry_wait",
+		PolicyStateJSON: `{"deadline":"` + elapsedDeadline.Format(time.RFC3339Nano) + `"}`,
+		UpdatedAt:       time.Now().UTC(),
+	}); err != nil {
+		t.Fatalf("SaveRouteState: %v", err)
+	}
+	if _, _, err := resolver.EngineForTest().LoadState(ctx, sessionID); err != nil {
+		t.Fatalf("warm engine cache: %v", err)
+	}
+
+	var launches int32
+	agentManager := &mockAgentManager{
+		launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) {
+			atomic.AddInt32(&launches, 1)
+			return &executor.LaunchAgentResponse{AgentExecutionID: "relaunch-" + req.SessionID}, nil
+		},
+	}
+	taskRepo := newMockTaskRepo()
+	seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress)
+	stepGetter := newMockStepGetter()
+	stepGetter.steps["step-race"] = &wfmodels.WorkflowStep{ID: "step-race", WorkflowID: "wf1", Name: "Work"}
+
+	svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentManager)
+	svc.SetProfileExecutionResolver(resolver.ProfileExecutionResolver)
+	svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"})
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	for i := 0; i < 2; i++ {
+		go func() {
+			defer wg.Done()
+			svc.runDynamicPolicyRecovery(ctx, sessionID, 4)
+		}()
+	}
+	wg.Wait()
+
+	if got := atomic.LoadInt32(&launches); got != 1 {
+		t.Fatalf("successor launches = %d, want exactly 1 (duplicate prompt delivery)", got)
+	}
+	agentManager.mu.Lock()
+	stopCalls := len(agentManager.stopAgentWithReasonArgs)
+	agentManager.mu.Unlock()
+	if stopCalls != 1 {
+		t.Fatalf("predecessor stop calls = %d, want exactly 1", stopCalls)
+	}
+}
+
+// dynamicPolicyRaceResolver exposes the underlying engine so the test can
+// warm its in-memory cache exactly as production restart-recovery does via
+// LoadState, before firing the two racing claims.
+type dynamicPolicyRaceResolver struct {
+	*agentruntime.ProfileExecutionResolver
+	engine *dynamicruntime.Engine
+}
+
+func (r *dynamicPolicyRaceResolver) EngineForTest() *dynamicruntime.Engine { return r.engine }
+
+func newDynamicPolicyRaceResolver(
+	t *testing.T, repo *sqliterepo.Repository, dynamicProfileID, concreteProfileID string,
+) *dynamicPolicyRaceResolver {
+	t.Helper()
+	dbPath := t.TempDir() + "/agent-settings.db"
+	db, err := sqlx.Open("sqlite3", dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	profileRepo, cleanup, err := agentsettingsstore.Provide(db, db, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = cleanup() })
+	ctx := context.Background()
+	for _, agent := range []*agentsettingsmodels.Agent{
+		{ID: "dynamic", Name: "dynamic"},
+		{ID: "concrete-agent", Name: "concrete-agent"},
+	} {
+		if err := profileRepo.CreateAgent(ctx, agent); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, profile := range []*agentsettingsmodels.AgentProfile{
+		{ID: dynamicProfileID, AgentID: "dynamic", Name: "Cascade", Enabled: true},
+		{ID: concreteProfileID, AgentID: "concrete-agent", Name: "Concrete", Enabled: true},
+	} {
+		if err := profileRepo.CreateAgentProfile(ctx, profile); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := profileRepo.CreateDynamicAgentProfile(ctx,
+		&agentsettingsmodels.DynamicAgentProfile{ProfileID: dynamicProfileID, Version: 1},
+		[]agentsettingsmodels.DynamicAgentRoute{{
+			DynamicProfileID:   dynamicProfileID,
+			ExecutionProfileID: concreteProfileID,
+			Enabled:            true,
+		}},
+	); err != nil {
+		t.Fatal(err)
+	}
+	engine := dynamicruntime.NewEngine(
+		dynamicruntime.WithPersistence(repo),
+		dynamicruntime.WithStateLoader(repo),
+	)
+	return &dynamicPolicyRaceResolver{
+		ProfileExecutionResolver: agentruntime.NewProfileExecutionResolver(profileRepo, engine, true),
+		engine:                   engine,
+	}
+}

apps/backend/internal/orchestrator/service.go

@@ -1176,6 +1176,10 @@ const (
 	RouteActionSkip       RouteAction = "skip"
 	RouteActionCancelWait RouteAction = "cancel_wait"
 	RouteActionStop       RouteAction = "stop"
+
+	// RouteActionLaunchFailedReason marks a task-session projection whose
+	// successor launch failed before durable recovery could be confirmed.
+	RouteActionLaunchFailedReason = "route_action_launch_failed"
 )
 
 type RouteActionRequest struct {

apps/backend/internal/task/repository/sqlite/dynamic_route_concurrency_test.go

@@ -0,0 +1,85 @@
+package sqlite
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+// TestConcurrentResumePendingClaimsExactlyOneGeneration reproduces the
+// duplicate-launch race: two callers that both observe a due retry_wait state
+// at the same generation must not both receive a decision. The durable claim
+// predicate is the authority, so exactly one caller succeeds and the other
+// gets ErrStaleGeneration.
+func TestConcurrentResumePendingClaimsExactlyOneGeneration(t *testing.T) {
+	repo := newRepoForSessionTests(t)
+	ctx := context.Background()
+	if err := repo.CreateTask(ctx, &models.Task{ID: "task-race", Title: "Race"}); err != nil {
+		t.Fatalf("CreateTask: %v", err)
+	}
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID: "session-race", TaskID: "task-race", State: models.TaskSessionStateWaitingForInput,
+	}); err != nil {
+		t.Fatalf("CreateTaskSession: %v", err)
+	}
+	elapsedDeadline := time.Now().UTC().Add(-time.Minute)
+	state := dynamicruntime.RouteState{
+		SessionID: "session-race", LogicalProfileID: "dynamic-1",
+		ExecutionProfileID: "concrete-1", Generation: 4, ProfileVersion: 1,
+		Status:          "retry_wait",
+		PolicyStateJSON: `{"deadline":"` + elapsedDeadline.Format(time.RFC3339Nano) + `"}`,
+		UpdatedAt:       time.Now().UTC(),
+	}
+	if err := repo.SaveRouteState(ctx, state); err != nil {
+		t.Fatalf("SaveRouteState: %v", err)
+	}
+
+	const callers = 2
+	engines := make([]*dynamicruntime.Engine, callers)
+	for i := range engines {
+		engines[i] = dynamicruntime.NewEngine(
+			dynamicruntime.WithPersistence(repo),
+			dynamicruntime.WithStateLoader(repo),
+		)
+		if _, _, err := engines[i].LoadState(ctx, "session-race"); err != nil {
+			t.Fatalf("warm engine cache %d: %v", i, err)
+		}
+	}
+
+	var wg sync.WaitGroup
+	decisions := make([]dynamicruntime.RouteDecision, callers)
+	errs := make([]error, callers)
+	wg.Add(callers)
+	for i := 0; i < callers; i++ {
+		go func(i int) {
+			defer wg.Done()
+			decisions[i], errs[i] = engines[i].ResumePending(ctx, "session-race", 4)
+		}(i)
+	}
+	wg.Wait()
+
+	successes := 0
+	staleCount := 0
+	for _, err := range errs {
+		if err == nil {
+			successes++
+			continue
+		}
+		if errors.Is(err, dynamicruntime.ErrStaleGeneration) || errors.Is(err, dynamicruntime.ErrRecoveryPending) {
+			staleCount++
+			continue
+		}
+		t.Fatalf("unexpected ResumePending error: %v", err)
+	}
+	if successes != 1 {
+		t.Fatalf("successful claims = %d, errors = %v (want exactly 1 success)", successes, errs)
+	}
+	if staleCount != callers-1 {
+		t.Fatalf("stale/rejected claims = %d, want %d", staleCount, callers-1)
+	}
+}

apps/backend/internal/task/repository/sqlite/dynamic_route_test.go

@@ -265,6 +265,60 @@ func TestClaimRouteStateFromRequiresExpectedStatus(t *testing.T) {
 	}
 }
 
+func TestClaimRouteStatusRejectsStateThatAlreadyAdvancedAtSameGeneration(t *testing.T) {
+	repo := newRepoForSessionTests(t)
+	ctx := context.Background()
+	if err := repo.CreateTask(ctx, &models.Task{ID: "task-claim-status", Title: "Claim status"}); err != nil {
+		t.Fatalf("CreateTask: %v", err)
+	}
+	if err := repo.CreateTaskSession(ctx, &models.TaskSession{
+		ID: "claim-status-session", TaskID: "task-claim-status", State: models.TaskSessionStateWaitingForInput,
+	}); err != nil {
+		t.Fatalf("CreateTaskSession: %v", err)
+	}
+	now := time.Now().UTC()
+	seed := dynamicruntime.RouteState{
+		SessionID: "claim-status-session", LogicalProfileID: "dynamic-1",
+		ExecutionProfileID: "concrete-1", Generation: 4, ProfileVersion: 1,
+		Status: "retry_wait", UpdatedAt: now,
+	}
+	if err := repo.SaveRouteState(ctx, seed); err != nil {
+		t.Fatalf("SaveRouteState: %v", err)
+	}
+
+	// First claim observes "retry_wait" and transitions to "retrying" at the
+	// same generation: the durable row still matches, so it succeeds.
+	winner := seed
+	winner.Status = "retrying"
+	winner.UpdatedAt = now.Add(time.Second)
+	claimed, err := repo.ClaimRouteStateFrom(ctx, 4, "retry_wait", winner)
+	if err != nil || !claimed {
+		t.Fatalf("winning ClaimRouteStateFrom = %v, %v", claimed, err)
+	}
+
+	// A second caller that also observed "retry_wait" at generation 4 loses:
+	// the durable state column has already advanced to "retrying", even
+	// though the generation itself has not changed.
+	loser := seed
+	loser.Status = "retrying"
+	loser.UpdatedAt = now.Add(2 * time.Second)
+	claimed, err = repo.ClaimRouteStateFrom(ctx, 4, "retry_wait", loser)
+	if err != nil {
+		t.Fatalf("losing ClaimRouteStateFrom: %v", err)
+	}
+	if claimed {
+		t.Fatal("second caller claimed a status transition already applied by the first")
+	}
+
+	loaded, err := repo.LoadRouteState(ctx, "claim-status-session")
+	if err != nil {
+		t.Fatalf("LoadRouteState: %v", err)
+	}
+	if loaded == nil || loaded.UpdatedAt.Unix() != winner.UpdatedAt.Unix() {
+		t.Fatalf("loaded state = %#v, want the winner's write", loaded)
+	}
+}
+
 func TestRecordRouteDecisionClaimsInitialGenerationAndWritesAttemptAtomically(t *testing.T) {
 	repo := newRepoForSessionTests(t)
 	ctx := context.Background()