diff --git a/apps/backend/internal/agent/runtime/dynamic/engine.go b/apps/backend/internal/agent/runtime/dynamic/engine.go index 08b2656699..dba53fcbfa 100644 --- a/apps/backend/internal/agent/runtime/dynamic/engine.go +++ b/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, diff --git a/apps/backend/internal/agent/runtime/dynamic/engine_test.go b/apps/backend/internal/agent/runtime/dynamic/engine_test.go index 221a50c81f..f1e89e07c4 100644 --- a/apps/backend/internal/agent/runtime/dynamic/engine_test.go +++ b/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) } diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver.go b/apps/backend/internal/agent/runtime/dynamic_resolver.go index c1accb099f..214009679f 100644 --- a/apps/backend/internal/agent/runtime/dynamic_resolver.go +++ b/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, diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver_test.go b/apps/backend/internal/agent/runtime/dynamic_resolver_test.go index 87e14bf658..1ae86aac42 100644 --- a/apps/backend/internal/agent/runtime/dynamic_resolver_test.go +++ b/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() diff --git a/apps/backend/internal/backendapp/dynamic_routing.go b/apps/backend/internal/backendapp/dynamic_routing.go index cbebee9a69..84763a5bb4 100644 --- a/apps/backend/internal/backendapp/dynamic_routing.go +++ b/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) diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index 8c8b3487de..2757262a23 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/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() { diff --git a/apps/backend/internal/orchestrator/dynamic_route_manual_retry_conflict_test.go b/apps/backend/internal/orchestrator/dynamic_route_manual_retry_conflict_test.go new file mode 100644 index 0000000000..ae98d68fe1 --- /dev/null +++ b/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) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index c6ae9d3393..82dc6fdc7d 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/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) } diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 638dd8e484..8d1e8e38ce 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/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() { diff --git a/apps/web/components/task/chat/dynamic-route-recovery.test.tsx b/apps/web/components/task/chat/dynamic-route-recovery.test.tsx index 21de8c2e48..cc8371329b 100644 --- a/apps/web/components/task/chat/dynamic-route-recovery.test.tsx +++ b/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(); + + 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", + }), + ); + }); + }); }); diff --git a/apps/web/components/task/chat/dynamic-route-recovery.tsx b/apps/web/components/task/chat/dynamic-route-recovery.tsx index a840249aaf..3564d925a8 100644 --- a/apps/web/components/task/chat/dynamic-route-recovery.tsx +++ b/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 = { 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; + 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; } }