fix(backend): stop manual Retry from starting a second agent when a route already resumed (#3399)

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

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

@@ -53,14 +53,20 @@ type Engine struct {
 	persistence Persistence
 	loader      StateLoader
 	probes      map[string]ProbeLease
+	// retryClaims identifies retry launches owned by this process. A durable
+	// "retrying" state can survive a restart without its in-memory owner, so
+	// manual recovery may reclaim it only when this map does not contain the
+	// same generation.
+	retryClaims map[string]int64
 }
 
 func NewEngine(options ...EngineOption) *Engine {
 	engine := &Engine{
-		now:      time.Now,
-		circuits: NewCircuitRegistry(),
-		states:   make(map[string]RouteState),
-		probes:   make(map[string]ProbeLease),
+		now:         time.Now,
+		circuits:    NewCircuitRegistry(),
+		states:      make(map[string]RouteState),
+		probes:      make(map[string]ProbeLease),
+		retryClaims: make(map[string]int64),
 	}
 	for _, option := range options {
 		option(engine)
@@ -313,6 +319,18 @@ func (e *Engine) LoadState(ctx context.Context, sessionID string) (RouteState, b
 	return e.loadStateLocked(ctx, sessionID)
 }
 
+// OwnsRetryClaim reports whether this process still owns a retry launch for
+// the session generation. The durable route row alone cannot answer this
+// after a restart because the previous process owner is gone.
+func (e *Engine) OwnsRetryClaim(sessionID string, generation int64) bool {
+	if e == nil || sessionID == "" || generation <= 0 {
+		return false
+	}
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	return e.retryClaims[sessionID] == generation
+}
+
 func (e *Engine) ActionFor(profile Profile, candidateID string, code routingerr.Code) Action {
 	for _, candidate := range profile.Candidates {
 		if candidate.ID != candidateID {
@@ -669,13 +687,8 @@ func (e *Engine) CancelPending(
 	}, nil
 }
 
-// MarkActive completes the claimed route's starting phase once a concrete
-// launch has actually succeeded. It is the only producer of the durable
-// "active" status, so a startup sweep can tell a healthy idling route apart
-// from one still holding "starting" with no launch in flight. A no-op when
-// the route has already left "starting" (a later generation, a failure
-// transition, or a duplicate call), so it is safe to call unconditionally
-// after a successful launch.
+// MarkActive records a successful launch. It accepts both a fresh starting
+// route and a resumed retry. Other states are unchanged.
 func (e *Engine) MarkActive(ctx context.Context, sessionID string, expectedGeneration int64) error {
 	e.mu.Lock()
 	defer e.mu.Unlock()
@@ -689,7 +702,7 @@ func (e *Engine) MarkActive(ctx context.Context, sessionID string, expectedGener
 	if state.Generation != expectedGeneration {
 		return ErrStaleGeneration
 	}
-	if state.Status != routeStatusStarting {
+	if state.Status != routeStatusStarting && state.Status != routeStatusRetrying {
 		return nil
 	}
 	expectedStatus := state.Status
@@ -699,19 +712,13 @@ func (e *Engine) MarkActive(ctx context.Context, sessionID string, expectedGener
 		return err
 	}
 	e.states[sessionID] = state
+	delete(e.retryClaims, sessionID)
 	return nil
 }
 
-// MarkActionRequired transitions a claimed route to durable action_required,
-// fenced to the caller's known generation. It is the catch-all recovery
-// marker for a launch that claimed a generation but failed before reaching a
-// terminal status of its own, so the recovery UI always has something to act
-// on instead of a route silently stuck at "starting". Like MarkActive, it
-// only transitions a route that is still "starting": a route a launch has
-// already carried past that phase (active) is left untouched, so a later,
-// unrelated failure on a healthy route cannot demote it and offer a fallback
-// the failure classifier explicitly declined. Calling it on a route that is
-// already action_required is a no-op.
+// MarkActionRequired moves an unconfirmed launch to durable action_required.
+// The transition is generation and status fenced. Active and already resolved
+// routes are left unchanged.
 func (e *Engine) MarkActionRequired(
 	ctx context.Context,
 	sessionID string,
@@ -730,7 +737,7 @@ func (e *Engine) MarkActionRequired(
 	if state.Generation != expectedGeneration {
 		return RouteDecision{}, ErrStaleGeneration
 	}
-	if state.Status == routeStatusStarting {
+	if state.Status == routeStatusStarting || state.Status == routeStatusRetrying {
 		expectedStatus := state.Status
 		state.Status = routeStatusActionRequired
 		state.UpdatedAt = e.now()
@@ -738,6 +745,7 @@ func (e *Engine) MarkActionRequired(
 			return RouteDecision{}, err
 		}
 		e.states[sessionID] = state
+		delete(e.retryClaims, sessionID)
 	}
 	return RouteDecision{
 		SessionID: sessionID, LogicalProfileID: state.LogicalProfileID,
@@ -746,25 +754,47 @@ 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)
+// ReclaimRetrying returns an orphaned durable retry claim to manual recovery.
+// A retry claim owned by this process is not reclaimable because its launch
+// may still cross the provider boundary.
+func (e *Engine) ReclaimRetrying(ctx context.Context, sessionID string, expectedGeneration int64) (bool, error) {
+	return e.reclaimRetrying(ctx, sessionID, expectedGeneration, false)
+}
+
+func (e *Engine) reclaimRetrying(
+	ctx context.Context,
+	sessionID string,
+	expectedGeneration int64,
+	allowOwned bool,
+) (bool, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	state, exists, err := e.loadStateLocked(ctx, sessionID)
 	if err != nil {
-		return err
+		return false, err
 	}
 	if !exists || state.Generation != expectedGeneration || state.Status != routeStatusRetrying {
-		return nil
+		return false, nil
+	}
+	if !allowOwned && e.retryClaims[sessionID] == expectedGeneration {
+		return false, ErrRecoveryPending
 	}
 	state.Status = routeStatusActionRequired
 	state.UpdatedAt = e.now()
 	if err := e.persistExpectedStatus(ctx, expectedGeneration, routeStatusRetrying, state); err != nil {
-		return err
+		return false, err
 	}
-	e.mu.Lock()
 	e.states[sessionID] = state
-	e.mu.Unlock()
-	return nil
+	delete(e.retryClaims, sessionID)
+	return true, nil
+}
+
+// MarkRecoveryActionRequired returns an in-flight route to manual recovery.
+// The launch owner calls this after a concrete launch failure, so it may
+// reclaim a retry claim owned by this process.
+func (e *Engine) MarkRecoveryActionRequired(ctx context.Context, sessionID string, expectedGeneration int64) error {
+	_, err := e.reclaimRetrying(ctx, sessionID, expectedGeneration, true)
+	return err
 }
 
 func (e *Engine) resumePending(
@@ -773,7 +803,9 @@ func (e *Engine) resumePending(
 	expectedGeneration int64,
 	force bool,
 ) (RouteDecision, error) {
-	state, exists, err := e.stateForFailure(ctx, sessionID)
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	state, exists, err := e.loadStateLocked(ctx, sessionID)
 	if err != nil {
 		return RouteDecision{}, err
 	}
@@ -783,8 +815,15 @@ func (e *Engine) resumePending(
 	if state.Generation != expectedGeneration {
 		return RouteDecision{}, ErrStaleGeneration
 	}
-	if force && state.Status == routeStatusRetrying {
-		return RouteDecision{}, ErrRecoveryPending
+	if force {
+		if state.Status == routeStatusRetrying {
+			return RouteDecision{}, ErrRecoveryPending
+		}
+		if state.Status != string(routingpolicy.DecisionRetry) &&
+			state.Status != string(routingpolicy.DecisionWaitForReset) &&
+			state.Status != routeStatusActionRequired {
+			return RouteDecision{}, ErrRecoveryPending
+		}
 	}
 	if !force && state.Status != string(routingpolicy.DecisionRetry) && state.Status != string(routingpolicy.DecisionWaitForReset) {
 		return RouteDecision{}, ErrRecoveryPending
@@ -805,9 +844,8 @@ func (e *Engine) resumePending(
 	if err := e.persistExpectedStatus(ctx, expectedGeneration, observedStatus, state); err != nil {
 		return RouteDecision{}, err
 	}
-	e.mu.Lock()
 	e.states[sessionID] = state
-	e.mu.Unlock()
+	e.retryClaims[sessionID] = state.Generation
 	return RouteDecision{
 		SessionID: sessionID, LogicalProfileID: state.LogicalProfileID,
 		ExecutionProfileID: state.ExecutionProfileID, Generation: state.Generation,

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

@@ -404,38 +404,137 @@ func TestEngineMarkActionRequiredIsNoOpOnceRouteIsActive(t *testing.T) {
 	}
 }
 
+// TestEngineMarkActionRequiredFromRetryingTransitionsAtSameGeneration is the
+// regression test restoring the deferred recovery guard in
+// LaunchDynamicRouteAction for the timer-recovery path: persistDynamicPolicyRecovery
+// leaves a resumed route at "retrying" (resume never advances the generation),
+// so a launch that fails after that resume must still reach action_required
+// instead of sticking at "retrying" with a dead Retry/Stop banner.
+func TestEngineMarkActionRequiredFromRetryingTransitionsAtSameGeneration(t *testing.T) {
+	engine := NewEngine()
+	document := routingpolicy.DefaultDocument()
+	document.Transient.Retry = routingpolicy.RetryPolicy{Enabled: true, MaxRetries: 1, InitialIntervalSeconds: 60}
+	profile := Profile{ID: "dynamic-retrying-action-required", Candidates: []Candidate{
+		{ID: "first", Enabled: true, Policies: document},
+	}}
+	initial, err := engine.Select("retrying-action-required-session", profile, 0, "")
+	if err != nil {
+		t.Fatalf("initial Select: %v", err)
+	}
+	if _, err := engine.ApplyFailure(
+		"retrying-action-required-session", profile, initial.Generation, initial.ExecutionProfileID,
+		&routingerr.Error{Code: routingerr.CodeRateLimited, Class: routingerr.ClassTransient, FallbackAllowed: true},
+	); !errors.Is(err, ErrRecoveryPending) {
+		t.Fatalf("ApplyFailure: %v", err)
+	}
+	resumed, err := engine.ResumePendingNow(context.Background(), "retrying-action-required-session", initial.Generation)
+	if err != nil {
+		t.Fatalf("ResumePendingNow: %v", err)
+	}
+	if resumed.Status != routeStatusRetrying {
+		t.Fatalf("precondition: resumed status = %q, want %q", resumed.Status, routeStatusRetrying)
+	}
+
+	decision, err := engine.MarkActionRequired(
+		context.Background(), "retrying-action-required-session", initial.Generation, "route_action_launch_failed",
+	)
+	if err != nil {
+		t.Fatalf("MarkActionRequired from retrying: %v", err)
+	}
+	if decision.Status != routeStatusActionRequired || decision.Generation != initial.Generation {
+		t.Fatalf("decision = %#v", decision)
+	}
+	state, ok := engine.State("retrying-action-required-session")
+	if !ok || state.Status != routeStatusActionRequired || state.Generation != initial.Generation {
+		t.Fatalf("state after MarkActionRequired = %#v, ok=%v", state, ok)
+	}
+}
+
+// TestEngineMarkActiveFromRetryingThenActionRequiredIsNoOp is the regression
+// test for review finding F1: a resumed route (status "retrying") that then
+// launches successfully must reach "active" via MarkActive, exactly as a
+// freshly claimed "starting" route does. Without this, a resumed route stays
+// at "retrying" forever, and MarkActionRequired's starting-or-retrying guard
+// (widened for the resumed-then-FAILED case) would wrongly demote a healthy,
+// successfully launched route on any later unrelated failure.
+func TestEngineMarkActiveFromRetryingThenActionRequiredIsNoOp(t *testing.T) {
+	engine := NewEngine()
+	document := routingpolicy.DefaultDocument()
+	document.Transient.Retry = routingpolicy.RetryPolicy{Enabled: true, MaxRetries: 1, InitialIntervalSeconds: 60}
+	profile := Profile{ID: "dynamic-retrying-active", Candidates: []Candidate{
+		{ID: "first", Enabled: true, Policies: document},
+	}}
+	initial, err := engine.Select("retrying-active-session", profile, 0, "")
+	if err != nil {
+		t.Fatalf("initial Select: %v", err)
+	}
+	if _, err := engine.ApplyFailure(
+		"retrying-active-session", profile, initial.Generation, initial.ExecutionProfileID,
+		&routingerr.Error{Code: routingerr.CodeRateLimited, Class: routingerr.ClassTransient, FallbackAllowed: true},
+	); !errors.Is(err, ErrRecoveryPending) {
+		t.Fatalf("ApplyFailure: %v", err)
+	}
+	resumed, err := engine.ResumePendingNow(context.Background(), "retrying-active-session", initial.Generation)
+	if err != nil {
+		t.Fatalf("ResumePendingNow: %v", err)
+	}
+	if resumed.Status != routeStatusRetrying {
+		t.Fatalf("precondition: resumed status = %q, want %q", resumed.Status, routeStatusRetrying)
+	}
+
+	if err := engine.MarkActive(context.Background(), "retrying-active-session", initial.Generation); err != nil {
+		t.Fatalf("MarkActive from retrying: %v", err)
+	}
+	state, ok := engine.State("retrying-active-session")
+	if !ok || state.Status != routeStatusActive || state.Generation != initial.Generation {
+		t.Fatalf("state after MarkActive = %#v, ok=%v, want active", state, ok)
+	}
+
+	// A later, unrelated failure that reaches the catch-all guard must not
+	// demote this healthy, successfully launched route.
+	decision, err := engine.MarkActionRequired(
+		context.Background(), "retrying-active-session", initial.Generation, "unrelated_declined_failure",
+	)
+	if err != nil {
+		t.Fatalf("MarkActionRequired on active route: %v", err)
+	}
+	if decision.Status != routeStatusActive {
+		t.Fatalf("decision.Status = %q, want unchanged %q", decision.Status, routeStatusActive)
+	}
+	state, ok = engine.State("retrying-active-session")
+	if !ok || state.Status != routeStatusActive {
+		t.Fatalf("state after no-op MarkActionRequired = %#v, ok=%v, want unchanged active", state, ok)
+	}
+}
+
 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},
-	}}
+	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)
 	}
+	ctx := context.Background()
 	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)
+		t.Fatalf("ResumePendingNow: %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 reclaimed, err := engine.ReclaimRetrying(ctx, "launch-fail-session", initial.Generation); !errors.Is(err, ErrRecoveryPending) || reclaimed {
+		t.Fatalf("ReclaimRetrying for owned claim = reclaimed %v, error %v, want recovery pending", reclaimed, err)
+	}
 	if err := engine.MarkRecoveryActionRequired(ctx, "launch-fail-session", initial.Generation); err != nil {
 		t.Fatalf("MarkRecoveryActionRequired: %v", err)
 	}
@@ -443,9 +542,6 @@ func TestEngineMarkRecoveryActionRequiredUnsticksManualRecoveryAfterFailedLaunch
 	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 {
@@ -455,16 +551,12 @@ func TestEngineMarkRecoveryActionRequiredUnsticksManualRecoveryAfterFailedLaunch
 
 func TestEngineMarkRecoveryActionRequiredIsNoOpWhenStateMovedOn(t *testing.T) {
 	engine := NewEngine()
-	profile := Profile{ID: "no-op-policy", Version: 1, Candidates: []Candidate{
-		{ID: "first", Enabled: true},
-	}}
+	profile := Profile{ID: "no-op-policy", 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)
 	}
@@ -472,13 +564,9 @@ func TestEngineMarkRecoveryActionRequiredIsNoOpWhenStateMovedOn(t *testing.T) {
 	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)
 	}

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

@@ -21,6 +21,8 @@ import (
 
 var ErrDynamicRoutingDisabled = errors.New("dynamic agent routing is disabled")
 
+const dynamicRouteStatusRetrying = "retrying"
+
 // ProfileExecution is the caller-facing result of resolving a logical
 // profile. Concrete callers receive the same ID for both fields. Dynamic
 // callers retain their logical ID while the resolver records the concrete
@@ -359,9 +361,24 @@ func (r *ProfileExecutionResolver) resolveRetryRouteAction(
 		return ProfileExecution{}, dynamic.ErrStaleGeneration
 	}
 	if exists {
+		if state.Status == dynamicRouteStatusRetrying {
+			// A retry can survive a process restart without its in-memory owner.
+			// Reclaim it only when this process does not own the launch.
+			if r.engine.OwnsRetryClaim(sessionID, expectedGeneration) {
+				return ProfileExecution{}, dynamic.ErrRecoveryPending
+			}
+			reclaimed, reclaimErr := r.engine.ReclaimRetrying(ctx, sessionID, expectedGeneration)
+			if reclaimErr != nil {
+				return ProfileExecution{}, reclaimErr
+			}
+			if reclaimed {
+				return r.resolve(ctx, sessionID, profileID, expectedGeneration, "", currentExecutionProfileID)
+			}
+			return ProfileExecution{}, dynamic.ErrRecoveryPending
+		}
 		decision, resumeErr := r.engine.ResumePendingNow(ctx, sessionID, expectedGeneration)
 		if resumeErr == nil {
-			return r.executionFromDecision(ctx, profileID, sessionID, decision)
+			return r.executionFromDecisionWithRecovery(ctx, profileID, sessionID, decision)
 		}
 		if !errors.Is(resumeErr, dynamic.ErrRouteStateNotFound) {
 			return ProfileExecution{}, resumeErr
@@ -377,8 +394,17 @@ func (r *ProfileExecutionResolver) resolveSkipRouteAction(
 ) (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
+	} else if exists && state.Generation == expectedGeneration && state.Status == dynamicRouteStatusRetrying {
+		if r.engine.OwnsRetryClaim(sessionID, expectedGeneration) {
+			return ProfileExecution{}, dynamic.ErrRecoveryPending
+		}
+		reclaimed, reclaimErr := r.engine.ReclaimRetrying(ctx, sessionID, expectedGeneration)
+		if reclaimErr != nil {
+			return ProfileExecution{}, reclaimErr
+		}
+		if !reclaimed {
+			return ProfileExecution{}, dynamic.ErrRecoveryPending
+		}
 	}
 	profileConfig, err := r.loadDynamicProfile(ctx, profileID)
 	if err != nil {
@@ -412,6 +438,22 @@ func (r *ProfileExecutionResolver) resolveCancelRouteAction(
 	if action == "stop" {
 		reason = "manual_stop"
 	}
+	state, exists, err := r.engine.LoadState(ctx, sessionID)
+	if err != nil {
+		return ProfileExecution{}, err
+	}
+	if exists && state.Generation == expectedGeneration && state.Status == dynamicRouteStatusRetrying {
+		if r.engine.OwnsRetryClaim(sessionID, expectedGeneration) {
+			return ProfileExecution{}, dynamic.ErrRecoveryPending
+		}
+		reclaimed, reclaimErr := r.engine.ReclaimRetrying(ctx, sessionID, expectedGeneration)
+		if reclaimErr != nil {
+			return ProfileExecution{}, reclaimErr
+		}
+		if !reclaimed {
+			return ProfileExecution{}, dynamic.ErrRecoveryPending
+		}
+	}
 	decision, err := r.engine.CancelPending(ctx, sessionID, expectedGeneration, reason)
 	if err != nil {
 		return ProfileExecution{}, err
@@ -444,7 +486,7 @@ func (r *ProfileExecutionResolver) ResumePendingRoute(
 	if err != nil {
 		return ProfileExecution{}, err
 	}
-	return r.executionFromDecision(ctx, state.LogicalProfileID, sessionID, decision)
+	return r.executionFromDecisionWithRecovery(ctx, state.LogicalProfileID, sessionID, decision)
 }
 
 // MarkRouteRecoveryActionRequired returns a claimed "retrying" route to
@@ -460,6 +502,21 @@ func (r *ProfileExecutionResolver) MarkRouteRecoveryActionRequired(
 	return r.engine.MarkRecoveryActionRequired(ctx, sessionID, expectedGeneration)
 }
 
+func (r *ProfileExecutionResolver) executionFromDecisionWithRecovery(
+	ctx context.Context,
+	profileID, sessionID string,
+	decision dynamic.RouteDecision,
+) (ProfileExecution, error) {
+	execution, err := r.executionFromDecision(ctx, profileID, sessionID, decision)
+	if err == nil || decision.Status != dynamicRouteStatusRetrying {
+		return execution, err
+	}
+	if recoveryErr := r.MarkRouteRecoveryActionRequired(ctx, sessionID, decision.Generation); recoveryErr != nil {
+		return ProfileExecution{}, fmt.Errorf("%w; restore route recovery: %v", err, recoveryErr)
+	}
+	return ProfileExecution{}, fmt.Errorf("%w: %v", dynamic.ErrRecoveryPending, err)
+}
+
 func (r *ProfileExecutionResolver) resolve(
 	ctx context.Context,
 	sessionID, profileID string,

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

@@ -59,6 +59,52 @@ func TestMarkRouteActionRequiredUnblocksRetryAndSkipAfterFailedLaunch(t *testing
 	}
 }
 
+func TestResolveRouteActionReclaimsRetryingRouteAfterRestart(t *testing.T) {
+	ctx := context.Background()
+	_, firstEngine, generation := newRetryingRouteActionResolver(t)
+	persisted, ok := firstEngine.State("session-retry")
+	if !ok || persisted.Status != "retrying" {
+		t.Fatalf("precondition route state = %#v, ok=%v", persisted, ok)
+	}
+
+	restartedEngine := dynamic.NewEngine(dynamic.WithStateLoader(&singleRouteStateLoader{state: persisted}))
+	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,
+		}},
+	}
+	restartedResolver := NewProfileExecutionResolver(profiles, restartedEngine, true)
+
+	result, err := restartedResolver.ResolveRouteAction(
+		ctx, "session-retry", "dynamic-profile", "concrete-profile", generation, "retry",
+	)
+	if err != nil {
+		t.Fatalf("ResolveRouteAction after restart: %v", err)
+	}
+	if result.Generation != generation+1 || result.ExecutionProfileID != "concrete-profile" || result.Decision.Status != "starting" {
+		t.Fatalf("result = %+v, want new generation %d on same candidate", result, generation+1)
+	}
+	state, ok := restartedEngine.State("session-retry")
+	if !ok || state.Generation != generation+1 || state.Status != "starting" {
+		t.Fatalf("restarted route state = %#v, ok=%v, want fenced successor", state, ok)
+	}
+}
+
+type singleRouteStateLoader struct {
+	state dynamic.RouteState
+}
+
+func (l *singleRouteStateLoader) LoadRouteState(_ context.Context, sessionID string) (*dynamic.RouteState, error) {
+	if sessionID != l.state.SessionID {
+		return nil, nil
+	}
+	state := l.state
+	return &state, nil
+}
+
 func newRetryingRouteActionResolver(t *testing.T) (*ProfileExecutionResolver, *dynamic.Engine, int64) {
 	t.Helper()
 	ctx := context.Background()

apps/backend/internal/backendapp/dynamic_routing.go

@@ -270,7 +270,8 @@ func routeActionError(
 	session *models.TaskSession,
 	err error,
 ) error {
-	if !errors.Is(err, dynamicruntime.ErrStaleGeneration) {
+	if !errors.Is(err, dynamicruntime.ErrStaleGeneration) &&
+		!errors.Is(err, dynamicruntime.ErrRecoveryPending) {
 		return err
 	}
 	return routeActionConflict(ctx, repo, session, err)

apps/backend/internal/orchestrator/dynamic_launch.go

@@ -209,11 +209,12 @@ func (s *Service) markDynamicRouteActive(ctx context.Context, sessionID string,
 // durable action_required and mirrors the status onto the task-session
 // projection. It is the catch-all recovery marker every declined or failed
 // dynamic launch path falls back to, so a claimed route is never left
-// silently stuck at "starting" with no recovery affordance. The underlying
-// engine call only transitions a route that is still "starting" (see
-// Engine.MarkActionRequired), so calling this on a route a launch already
-// carried to "active" is a safe no-op: neither store is touched, and an
-// unrelated later failure cannot overwrite a healthy route's reason.
+// silently stuck at "starting" or "retrying" with no recovery affordance.
+// The underlying engine call only transitions a route that is still
+// "starting" or "retrying" (see Engine.MarkActionRequired), so calling this
+// on a route a launch already carried to "active" is a safe no-op: neither
+// store is touched, and an unrelated later failure cannot overwrite a
+// healthy route's reason.
 func (s *Service) markDynamicRouteActionRequired(ctx context.Context, sessionID string, generation int64, reason string) {
 	if s.profileExecutionResolver == nil || sessionID == "" || generation <= 0 {
 		return
@@ -663,9 +664,10 @@ func (s *Service) routeDynamicAgentFailure(
 	// failed attempt, not the one this call started from. The guard is safe
 	// to arm this early even for a failure the classifier forbids fallback
 	// for: markDynamicRouteActionRequired only transitions a route that is
-	// still "starting", so it is a no-op once this generation's launch has
-	// already reached "active" (an ordinary runtime failure on a healthy
-	// session), and only fires for a route genuinely stranded mid-launch.
+	// still "starting" or "retrying", so it is a no-op once this generation's
+	// launch has already reached "active" (an ordinary runtime failure on a
+	// healthy session), and only fires for a route genuinely stranded
+	// mid-launch.
 	generation := session.RouteGeneration
 	handled := false
 	defer func() {

apps/backend/internal/orchestrator/dynamic_route_manual_retry_conflict_test.go

@@ -0,0 +1,275 @@
+package orchestrator
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic"
+	"github.com/kandev/kandev/internal/agent/runtime/routingpolicy"
+	"github.com/kandev/kandev/internal/task/models"
+)
+
+// seedDueDynamicRetryWaitRoute claims generation 1 for a two-candidate dynamic
+// profile through the real resolver, then persists a durable retry_wait state
+// with an already-elapsed deadline - the same shape a policy-classified
+// transient failure leaves behind, and the precondition both
+// ProfileExecutionResolver.ResumePendingRoute (the recovery-timer path) and
+// ProfileExecutionResolver.ResolveRouteAction "retry" (the manual path) act on.
+func seedDueDynamicRetryWaitRoute(
+	t *testing.T,
+	ctx context.Context,
+	repo interface {
+		GetTaskSession(context.Context, string) (*models.TaskSession, error)
+		UpdateTaskSession(context.Context, *models.TaskSession) error
+		SaveRouteState(context.Context, dynamicruntime.RouteState) error
+	},
+	dynamicProfileID, sessionID string,
+) {
+	t.Helper()
+	deadline := time.Now().Add(-time.Minute)
+	policyState := dynamicruntime.PolicyState{Deadline: &deadline}
+	raw, err := json.Marshal(policyState)
+	if err != nil {
+		t.Fatalf("marshal policy state: %v", err)
+	}
+	if err := repo.SaveRouteState(ctx, dynamicruntime.RouteState{
+		SessionID:          sessionID,
+		LogicalProfileID:   dynamicProfileID,
+		ExecutionProfileID: "candidate-1",
+		Generation:         1,
+		ProfileVersion:     1,
+		Status:             string(routingpolicy.DecisionRetry),
+		PolicyStateJSON:    string(raw),
+		UpdatedAt:          time.Now().Add(-time.Minute),
+	}); err != nil {
+		t.Fatalf("seed retry_wait route state: %v", err)
+	}
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("GetTaskSession: %v", err)
+	}
+	session.AgentProfileID = dynamicProfileID
+	session.ExecutionProfileID = "candidate-1"
+	session.RouteGeneration = 1
+	session.RouteState = string(routingpolicy.DecisionRetry)
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("UpdateTaskSession: %v", err)
+	}
+}
+
+// TestResolveRetryRouteAction_TimerResumeThenManualRetrySameGeneration is the
+// regression test for the manual-retry fall-through: once the recovery timer
+// (ProfileExecutionResolver.ResumePendingRoute, as called from
+// runDynamicPolicyRecovery) has resumed a due retry_wait route to "retrying",
+// a manual Retry click observing the same expected generation must be told
+// its view is stale instead of claiming a second, competing successor
+// generation via the unfenced-by-status resolve() fallback.
+func TestResolveRetryRouteAction_TimerResumeThenManualRetrySameGeneration(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-manual-retry-after-timer"
+		sessionID = "session-manual-retry-after-timer"
+		dynamicID = "dynamic-manual-retry-after-timer"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	resolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{
+		{executionProfileID: "candidate-1", enabled: true},
+		{executionProfileID: "candidate-2", enabled: true},
+	}, dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo))
+	seedDueDynamicRetryWaitRoute(t, ctx, repo, dynamicID, sessionID)
+
+	// The recovery timer resumes generation 1 first.
+	timerResult, err := resolver.ResumePendingRoute(ctx, sessionID, 1)
+	if err != nil {
+		t.Fatalf("timer ResumePendingRoute: %v", err)
+	}
+	if timerResult.Generation != 1 || timerResult.Decision.Status != "retrying" {
+		t.Fatalf("timer result = %+v, want generation 1 status retrying", timerResult)
+	}
+
+	// A manual retry click still observing generation 1 must not claim a
+	// second, independent generation.
+	_, err = resolver.ResolveRouteAction(ctx, sessionID, dynamicID, "candidate-1", 1, "retry")
+	if !errors.Is(err, dynamicruntime.ErrRecoveryPending) {
+		t.Fatalf("manual retry after timer resume error = %v, want %v", err, dynamicruntime.ErrRecoveryPending)
+	}
+
+	state, err := repo.LoadRouteState(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("LoadRouteState: %v", err)
+	}
+	if state == nil || state.Generation != 1 || state.Status != "retrying" {
+		t.Fatalf("durable route state = %#v, want unchanged generation 1 status retrying (exactly one decision)", state)
+	}
+}
+
+// TestResolveRetryRouteAction_DoubleManualRetrySameGeneration is the
+// regression test for the more likely real-world trigger: two Retry clicks
+// racing (or simply double-clicked) with no recovery timer involved at all.
+// The first claims the resume; the second, still observing the pre-resume
+// generation, must be rejected rather than launching a second successor.
+func TestResolveRetryRouteAction_DoubleManualRetrySameGeneration(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-double-manual-retry"
+		sessionID = "session-double-manual-retry"
+		dynamicID = "dynamic-double-manual-retry"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	resolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{
+		{executionProfileID: "candidate-1", enabled: true},
+		{executionProfileID: "candidate-2", enabled: true},
+	}, dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo))
+	seedDueDynamicRetryWaitRoute(t, ctx, repo, dynamicID, sessionID)
+
+	first, err := resolver.ResolveRouteAction(ctx, sessionID, dynamicID, "candidate-1", 1, "retry")
+	if err != nil {
+		t.Fatalf("first manual retry: %v", err)
+	}
+	if first.Generation != 1 || first.ExecutionProfileID != "candidate-1" {
+		t.Fatalf("first manual retry result = %+v, want generation 1 candidate-1", first)
+	}
+
+	_, err = resolver.ResolveRouteAction(ctx, sessionID, dynamicID, "candidate-1", 1, "retry")
+	if !errors.Is(err, dynamicruntime.ErrRecoveryPending) {
+		t.Fatalf("second manual retry error = %v, want %v", err, dynamicruntime.ErrRecoveryPending)
+	}
+
+	state, err := repo.LoadRouteState(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("LoadRouteState: %v", err)
+	}
+	if state == nil || state.Generation != 1 || state.Status != "retrying" {
+		t.Fatalf("durable route state = %#v, want unchanged generation 1 status retrying (exactly one decision)", state)
+	}
+}
+
+func TestResolveRetryRouteAction_ReclaimsPersistedRetryingAfterRestart(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-retry-after-restart"
+		sessionID = "session-retry-after-restart"
+		dynamicID = "dynamic-retry-after-restart"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	seedDueDynamicRetryWaitRoute(t, ctx, repo, dynamicID, sessionID)
+
+	firstResolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{
+		{executionProfileID: "candidate-1", enabled: true},
+		{executionProfileID: "candidate-2", enabled: true},
+	}, dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo))
+	if _, err := firstResolver.ResumePendingRoute(ctx, sessionID, 1); err != nil {
+		t.Fatalf("timer resume: %v", err)
+	}
+
+	restartedResolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{
+		{executionProfileID: "candidate-1", enabled: true},
+		{executionProfileID: "candidate-2", enabled: true},
+	}, dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo))
+	result, err := restartedResolver.ResolveRouteAction(ctx, sessionID, dynamicID, "candidate-1", 1, "retry")
+	if err != nil {
+		t.Fatalf("manual retry after restart: %v", err)
+	}
+	if result.Generation != 2 || result.ExecutionProfileID != "candidate-1" || result.Decision.Status != "starting" {
+		t.Fatalf("result = %+v, want generation 2 candidate-1 starting", result)
+	}
+	state, err := repo.LoadRouteState(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("LoadRouteState: %v", err)
+	}
+	if state == nil || state.Generation != 2 || state.Status != "starting" {
+		t.Fatalf("durable route state = %#v, want fenced successor", state)
+	}
+}
+
+// TestResolveRetryRouteAction_NoExistingRouteStateStillResolvesFresh proves
+// the fall-through is preserved for its legitimate case: a retry action on a
+// session with no durable route state yet (first launch via a retry action).
+func TestResolveRetryRouteAction_NoExistingRouteStateStillResolvesFresh(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-manual-retry-fresh"
+		sessionID = "session-manual-retry-fresh"
+		dynamicID = "dynamic-manual-retry-fresh"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	resolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{
+		{executionProfileID: "candidate-1", enabled: true},
+	}, dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo))
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("GetTaskSession: %v", err)
+	}
+	session.AgentProfileID = dynamicID
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("UpdateTaskSession: %v", err)
+	}
+
+	result, err := resolver.ResolveRouteAction(ctx, sessionID, dynamicID, "", 0, "retry")
+	if err != nil {
+		t.Fatalf("manual retry with no existing route state: %v", err)
+	}
+	if result.ExecutionProfileID != "candidate-1" || result.Generation != 1 {
+		t.Fatalf("result = %+v, want fresh generation 1 candidate-1", result)
+	}
+}
+
+// TestResolveRetryRouteAction_FromActionRequiredStillResumes is the
+// not-regressed case: a route already at durable action_required must still
+// resume at the same generation and execution profile through the manual
+// retry path, since force=true never gates on that status.
+func TestResolveRetryRouteAction_FromActionRequiredStillResumes(t *testing.T) {
+	ctx := context.Background()
+	const (
+		taskID    = "task-manual-retry-action-required"
+		sessionID = "session-manual-retry-action-required"
+		dynamicID = "dynamic-manual-retry-action-required"
+	)
+	repo := setupTestRepo(t)
+	seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput)
+	resolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{
+		{executionProfileID: "candidate-1", enabled: true},
+		{executionProfileID: "candidate-2", enabled: true},
+	}, dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo))
+	if err := repo.SaveRouteState(ctx, dynamicruntime.RouteState{
+		SessionID: sessionID, LogicalProfileID: dynamicID,
+		ExecutionProfileID: "candidate-1", Generation: 1, ProfileVersion: 1,
+		Status: "action_required", UpdatedAt: time.Now().UTC(),
+	}); err != nil {
+		t.Fatalf("seed action_required route state: %v", err)
+	}
+	session, err := repo.GetTaskSession(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("GetTaskSession: %v", err)
+	}
+	session.AgentProfileID = dynamicID
+	session.ExecutionProfileID = "candidate-1"
+	session.RouteGeneration = 1
+	session.RouteState = "action_required"
+	if err := repo.UpdateTaskSession(ctx, session); err != nil {
+		t.Fatalf("UpdateTaskSession: %v", err)
+	}
+
+	result, err := resolver.ResolveRouteAction(ctx, sessionID, dynamicID, "candidate-1", 1, "retry")
+	if err != nil {
+		t.Fatalf("manual retry from action_required: %v", err)
+	}
+	if result.Generation != 1 || result.ExecutionProfileID != "candidate-1" {
+		t.Fatalf("result = %+v, want unchanged generation 1 candidate-1", result)
+	}
+
+	state, err := repo.LoadRouteState(ctx, sessionID)
+	if err != nil {
+		t.Fatalf("LoadRouteState: %v", err)
+	}
+	if state == nil || state.Status != "retrying" || state.Generation != 1 {
+		t.Fatalf("durable route state = %#v, want generation 1 status retrying", state)
+	}
+}

apps/backend/internal/orchestrator/event_handlers_workflow.go

@@ -2527,7 +2527,7 @@ func (s *Service) prepareWorkflowReplacementSession(
 		// resolveDynamicLaunchExecution above may have claimed newSession's
 		// route generation as "starting"; nothing else transitions it if this
 		// workspace-only attach then fails. Safe to call unconditionally: it
-		// only fires while that generation is still "starting".
+		// only fires while that generation is still "starting" or "retrying".
 		s.markDynamicRouteActionRequired(ctx, sessionID, newSession.RouteGeneration, "workflow_replacement_launch_failed")
 		return nil, fmt.Errorf("failed to attach workflow replacement workspace: %w", err)
 	}

apps/backend/internal/orchestrator/task_operations.go

@@ -352,10 +352,11 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr
 			// agent, so it never reaches "active" itself — the eventual
 			// StartCreatedSession call does that. But if a route was claimed
 			// below and this launch then fails, nothing else will ever move
-			// it off "starting"; the deferred guard covers both failure
-			// points uniformly. It only fires while the route is still
-			// "starting" (see Engine.MarkActionRequired), so it is a no-op
-			// if a concurrent real launch already marked it active.
+			// it off "starting" or "retrying"; the deferred guard covers both
+			// failure points uniformly. It only fires while the route is
+			// still "starting" or "retrying" (see Engine.MarkActionRequired),
+			// so it is a no-op if a concurrent real launch already marked it
+			// active.
 			var claimedRouteGeneration int64
 			launchOwned := false
 			defer func() {

apps/web/components/task/chat/dynamic-route-recovery.test.tsx

@@ -16,6 +16,7 @@ vi.mock("@/components/state-provider", () => ({
 }));
 
 import type { TaskSession } from "@/lib/types/http";
+import { WebSocketRequestError } from "@/lib/ws/client";
 import { DynamicRouteRecovery } from "./dynamic-route-recovery";
 
 const session = {
@@ -65,4 +66,31 @@ describe("DynamicRouteRecovery", () => {
       expect.objectContaining({ route_generation: 4, route_state: "starting" }),
     );
   });
+
+  it("applies the authoritative route from a conflict response", async () => {
+    mocks.request.mockRejectedValueOnce(
+      new WebSocketRequestError("route changed", "CONFLICT", {
+        route: {
+          execution_profile_id: "profile-3",
+          route_generation: 5,
+          state: "action_required",
+          reason: "route_action_launch_failed",
+        },
+      }),
+    );
+    render(<DynamicRouteRecovery session={session} />);
+
+    fireEvent.click(screen.getByTestId("dynamic-route-retry"));
+
+    await waitFor(() => {
+      expect(mocks.setTaskSession).toHaveBeenCalledWith(
+        expect.objectContaining({
+          execution_profile_id: "profile-3",
+          route_generation: 5,
+          route_state: "action_required",
+          route_reason: "route_action_launch_failed",
+        }),
+      );
+    });
+  });
 });

apps/web/components/task/chat/dynamic-route-recovery.tsx

@@ -8,6 +8,7 @@ import { useAppStore } from "@/components/state-provider";
 import { agentProfileId as toAgentProfileId } from "@/lib/types/ids";
 import type { TaskSession } from "@/lib/types/http";
 import { getWebSocketClient } from "@/lib/ws/connection";
+import { WebSocketRequestError } from "@/lib/ws/client";
 
 const ROUTE_REASON_KEYS: Record<string, string> = {
   candidate_order: "dynamicRouteReasonCandidateOrder",
@@ -30,6 +31,12 @@ type RouteActionResult = {
   pending_outcome?: string;
 };
 
+function isRouteActionResult(value: unknown): value is RouteActionResult {
+  if (typeof value !== "object" || value === null) return false;
+  const result = value as Partial<RouteActionResult>;
+  return typeof result.route_generation === "number" && typeof result.state === "string";
+}
+
 async function applyRouteAction(
   session: TaskSession,
   action: RouteAction,
@@ -47,7 +54,11 @@ async function applyRouteAction(
       30000,
     );
     return result;
-  } catch {
+  } catch (error) {
+    if (error instanceof WebSocketRequestError && error.code === "CONFLICT") {
+      const route = error.details?.route;
+      return isRouteActionResult(route) ? route : null;
+    }
     return null;
   }
 }