diff --git a/apps/backend/internal/agent/runtime/dynamic/engine.go b/apps/backend/internal/agent/runtime/dynamic/engine.go index dd4d1c622c..63c021c72a 100644 --- a/apps/backend/internal/agent/runtime/dynamic/engine.go +++ b/apps/backend/internal/agent/runtime/dynamic/engine.go @@ -13,7 +13,11 @@ import ( type EngineOption func(*Engine) -const routeStatusActionRequired = "action_required" +const ( + routeStatusStarting = "starting" + routeStatusActive = "active" + routeStatusActionRequired = "action_required" +) func WithClock(now func() time.Time) EngineOption { return func(engine *Engine) { @@ -154,7 +158,7 @@ func (e *Engine) selectContext( Generation: generation, ProfileVersion: profile.Version, Reason: reason, - Status: "starting", + Status: routeStatusStarting, } nextState := RouteState{ SessionID: sessionID, @@ -162,7 +166,7 @@ func (e *Engine) selectContext( ExecutionProfileID: candidate.ID, Generation: generation, ProfileVersion: profile.Version, - Status: "starting", + Status: routeStatusStarting, UpdatedAt: now, } if err := e.claimAndPersist(ctx, expectedGeneration, decision, nextState); err != nil { @@ -421,7 +425,7 @@ func (e *Engine) applyPolicyFailure( if evaluation.Kind == routingpolicy.DecisionStop { nextState.Status = routeStatusActionRequired } - if err := e.persistSameGeneration(ctx, expectedGeneration, nextState); err != nil { + if err := e.persistSameGeneration(ctx, expectedGeneration, state.Status, nextState); err != nil { return RouteDecision{}, err } e.mu.Lock() @@ -512,10 +516,28 @@ func (e *Engine) stateForFailure(ctx context.Context, sessionID string) (RouteSt return e.loadStateLocked(ctx, sessionID) } -func (e *Engine) persistSameGeneration(ctx context.Context, expectedGeneration int64, state RouteState) error { +func (e *Engine) persistSameGeneration( + ctx context.Context, + expectedGeneration int64, + expectedStatus string, + state RouteState, +) error { if e.persistence == nil { return nil } + if handled, err := e.persistInitialGeneration(ctx, expectedGeneration, expectedStatus, state); handled { + return err + } + if claimer, ok := e.persistence.(GenerationStatusClaimer); ok { + claimed, err := claimer.ClaimRouteStateFrom(ctx, expectedGeneration, expectedStatus, state) + if err != nil { + return err + } + if !claimed { + return ErrStaleGeneration + } + return nil + } if claimer, ok := e.persistence.(GenerationClaimer); ok { claimed, err := claimer.ClaimRouteState(ctx, expectedGeneration, state) if err != nil { @@ -529,6 +551,31 @@ func (e *Engine) persistSameGeneration(ctx context.Context, expectedGeneration i return e.persistence.SaveRouteState(ctx, state) } +// persistInitialGeneration preserves the insert path for an adapter that +// observes a policy failure before the initial route row exists. +func (e *Engine) persistInitialGeneration( + ctx context.Context, + expectedGeneration int64, + expectedStatus string, + state RouteState, +) (bool, error) { + if expectedGeneration != 0 || expectedStatus != "" { + return false, nil + } + claimer, ok := e.persistence.(GenerationClaimer) + if !ok { + return false, nil + } + claimed, err := claimer.ClaimRouteState(ctx, expectedGeneration, state) + if err != nil { + return true, err + } + if !claimed { + return true, ErrStaleGeneration + } + return true, nil +} + func mustJSON(value PolicyState) []byte { payload, err := json.Marshal(value) if err != nil { @@ -581,10 +628,11 @@ func (e *Engine) CancelPending( } } policyState.PendingOutcome = routingpolicy.OutcomeStop + expectedStatus := state.Status state.PolicyStateJSON = string(mustJSON(policyState)) state.Status = routeStatusActionRequired state.UpdatedAt = e.now() - if err := e.persistSameGeneration(ctx, expectedGeneration, state); err != nil { + if err := e.persistSameGeneration(ctx, expectedGeneration, expectedStatus, state); err != nil { return RouteDecision{}, err } e.mu.Lock() @@ -600,6 +648,83 @@ 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. +func (e *Engine) MarkActive(ctx context.Context, sessionID string, expectedGeneration int64) error { + e.mu.Lock() + defer e.mu.Unlock() + state, exists, err := e.loadStateLocked(ctx, sessionID) + if err != nil { + return err + } + if !exists { + return ErrRouteStateNotFound + } + if state.Generation != expectedGeneration { + return ErrStaleGeneration + } + if state.Status != routeStatusStarting { + return nil + } + expectedStatus := state.Status + state.Status = routeStatusActive + state.UpdatedAt = e.now() + if err := e.persistSameGeneration(ctx, expectedGeneration, expectedStatus, state); err != nil { + return err + } + e.states[sessionID] = state + 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. +func (e *Engine) MarkActionRequired( + ctx context.Context, + sessionID string, + expectedGeneration int64, + reason string, +) (RouteDecision, error) { + e.mu.Lock() + defer e.mu.Unlock() + state, exists, err := e.loadStateLocked(ctx, sessionID) + if err != nil { + return RouteDecision{}, err + } + if !exists { + return RouteDecision{}, ErrRouteStateNotFound + } + if state.Generation != expectedGeneration { + return RouteDecision{}, ErrStaleGeneration + } + if state.Status == routeStatusStarting { + expectedStatus := state.Status + state.Status = routeStatusActionRequired + state.UpdatedAt = e.now() + if err := e.persistSameGeneration(ctx, expectedGeneration, expectedStatus, state); err != nil { + return RouteDecision{}, err + } + e.states[sessionID] = state + } + return RouteDecision{ + SessionID: sessionID, LogicalProfileID: state.LogicalProfileID, + ExecutionProfileID: state.ExecutionProfileID, Generation: state.Generation, + ProfileVersion: state.ProfileVersion, Reason: reason, Status: state.Status, + }, nil +} + func (e *Engine) resumePending( ctx context.Context, sessionID string, @@ -632,9 +757,10 @@ func (e *Engine) resumePending( if !force && policyState.Deadline != nil && now.Before(*policyState.Deadline) { return RouteDecision{}, ErrRecoveryNotDue } + expectedStatus := state.Status state.Status = "retrying" state.UpdatedAt = now - if err := e.persistSameGeneration(ctx, expectedGeneration, state); err != nil { + if err := e.persistSameGeneration(ctx, expectedGeneration, expectedStatus, state); err != nil { return RouteDecision{}, err } e.mu.Lock() diff --git a/apps/backend/internal/agent/runtime/dynamic/engine_test.go b/apps/backend/internal/agent/runtime/dynamic/engine_test.go index 99c2377c6c..cce722554e 100644 --- a/apps/backend/internal/agent/runtime/dynamic/engine_test.go +++ b/apps/backend/internal/agent/runtime/dynamic/engine_test.go @@ -281,6 +281,129 @@ func TestEngineCancelPendingStopsWithoutAdvancingGeneration(t *testing.T) { } } +func TestEngineMarkActiveCompletesStartingPhase(t *testing.T) { + engine := NewEngine() + profile := Profile{ID: "dynamic-active", Candidates: []Candidate{{ID: "first", Enabled: true}}} + initial, err := engine.Select("active-session", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + state, ok := engine.State("active-session") + if !ok || state.Status != routeStatusStarting { + t.Fatalf("precondition: state = %#v, ok=%v, want starting", state, ok) + } + if err := engine.MarkActive(context.Background(), "active-session", initial.Generation); err != nil { + t.Fatalf("MarkActive: %v", err) + } + state, ok = engine.State("active-session") + if !ok || state.Status != routeStatusActive || state.Generation != initial.Generation { + t.Fatalf("state after MarkActive = %#v, ok=%v", state, ok) + } + // A repeat call (e.g. from a duplicate callback) is a no-op, not an error. + if err := engine.MarkActive(context.Background(), "active-session", initial.Generation); err != nil { + t.Fatalf("repeat MarkActive: %v", err) + } + if err := engine.MarkActive(context.Background(), "active-session", initial.Generation-1); !errors.Is(err, ErrStaleGeneration) { + t.Fatalf("stale MarkActive error = %v, want %v", err, ErrStaleGeneration) + } + if err := engine.MarkActive(context.Background(), "missing-session", 1); !errors.Is(err, ErrRouteStateNotFound) { + t.Fatalf("missing session MarkActive error = %v, want %v", err, ErrRouteStateNotFound) + } +} + +func TestEngineMarkActiveIsNoOpWhenRouteAlreadyLeftStarting(t *testing.T) { + engine := NewEngine() + profile := Profile{ID: "dynamic-active-skip", Candidates: []Candidate{ + {ID: "first", Enabled: true}, + {ID: "second", Enabled: true}, + }} + initial, err := engine.Select("active-skip-session", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + next, err := engine.SelectContextWithReason( + context.Background(), "active-skip-session", profile, initial.Generation, initial.ExecutionProfileID, "manual_skip", + ) + if err != nil { + t.Fatalf("manual skip: %v", err) + } + // A late MarkActive call for the superseded generation must not resurrect + // it as active over the newer claim. + if err := engine.MarkActive(context.Background(), "active-skip-session", initial.Generation); !errors.Is(err, ErrStaleGeneration) { + t.Fatalf("MarkActive for superseded generation error = %v, want %v", err, ErrStaleGeneration) + } + state, ok := engine.State("active-skip-session") + if !ok || state.Generation != next.Generation || state.Status != routeStatusStarting { + t.Fatalf("state after stale MarkActive = %#v, ok=%v", state, ok) + } +} + +func TestEngineMarkActionRequiredFromStartingAndFencesGeneration(t *testing.T) { + engine := NewEngine() + profile := Profile{ID: "dynamic-action-required", Candidates: []Candidate{{ID: "first", Enabled: true}}} + initial, err := engine.Select("action-required-session", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + // A route sitting in "starting" (the exact stranded case this exists to + // fix) must be acceptable input. + decision, err := engine.MarkActionRequired(context.Background(), "action-required-session", initial.Generation, "launch_failed") + if err != nil { + t.Fatalf("MarkActionRequired: %v", err) + } + if decision.Status != routeStatusActionRequired || decision.Reason != "launch_failed" || decision.Generation != initial.Generation { + t.Fatalf("decision = %#v", decision) + } + state, ok := engine.State("action-required-session") + if !ok || state.Status != routeStatusActionRequired { + t.Fatalf("state after MarkActionRequired = %#v, ok=%v", state, ok) + } + // A second call (a different failure path re-declaring the same route) is + // idempotent, not an error, and keeps its own reason. + decision, err = engine.MarkActionRequired(context.Background(), "action-required-session", initial.Generation, "second_reason") + if err != nil { + t.Fatalf("repeat MarkActionRequired: %v", err) + } + if decision.Reason != "second_reason" || decision.Status != routeStatusActionRequired { + t.Fatalf("repeat decision = %#v", decision) + } + if _, err := engine.MarkActionRequired(context.Background(), "action-required-session", initial.Generation-1, "stale"); !errors.Is(err, ErrStaleGeneration) { + t.Fatalf("stale MarkActionRequired error = %v, want %v", err, ErrStaleGeneration) + } + if _, err := engine.MarkActionRequired(context.Background(), "missing-session", 1, "missing"); !errors.Is(err, ErrRouteStateNotFound) { + t.Fatalf("missing session MarkActionRequired error = %v, want %v", err, ErrRouteStateNotFound) + } +} + +// TestEngineMarkActionRequiredIsNoOpOnceRouteIsActive is the regression test +// for review finding F1: a route a launch already carried to "active" must +// not be demoted to action_required by an unrelated later failure (for +// example a permission-denied or resume-corrupted error the classifier +// forbids fallback for), which would surface a "Try next provider" recovery +// banner on a healthy session. +func TestEngineMarkActionRequiredIsNoOpOnceRouteIsActive(t *testing.T) { + engine := NewEngine() + profile := Profile{ID: "dynamic-action-required-active", Candidates: []Candidate{{ID: "first", Enabled: true}}} + initial, err := engine.Select("active-route-session", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + if err := engine.MarkActive(context.Background(), "active-route-session", initial.Generation); err != nil { + t.Fatalf("MarkActive: %v", err) + } + decision, err := engine.MarkActionRequired(context.Background(), "active-route-session", initial.Generation, "permission_denied") + 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("active-route-session") + if !ok || state.Status != routeStatusActive { + t.Fatalf("state after no-op MarkActionRequired = %#v, ok=%v, want unchanged active", state, ok) + } +} + func TestBindingFingerprinterUsesOpaqueStableHMACKeys(t *testing.T) { key := []byte("installation-secret") fingerprinter := NewBindingFingerprinter(key) diff --git a/apps/backend/internal/agent/runtime/dynamic/types.go b/apps/backend/internal/agent/runtime/dynamic/types.go index 347f62852f..6112eab65d 100644 --- a/apps/backend/internal/agent/runtime/dynamic/types.go +++ b/apps/backend/internal/agent/runtime/dynamic/types.go @@ -132,6 +132,13 @@ type GenerationClaimer interface { ClaimRouteState(context.Context, int64, RouteState) (bool, error) } +// 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. +type GenerationStatusClaimer interface { + ClaimRouteStateFrom(context.Context, int64, string, RouteState) (bool, error) +} + // DecisionRecorder lets a repository commit the state row and immutable // attempt row in one transaction. Persistence remains backwards compatible // for callers that only need the narrow Save/Append contract. diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver.go b/apps/backend/internal/agent/runtime/dynamic_resolver.go index 03d785fe1b..f5f947852c 100644 --- a/apps/backend/internal/agent/runtime/dynamic_resolver.go +++ b/apps/backend/internal/agent/runtime/dynamic_resolver.go @@ -321,6 +321,31 @@ func (r *ProfileExecutionResolver) ResolveRouteAction( } } +// MarkRouteActive completes a claimed route's starting phase after the +// asynchronous agent process-start callback confirms success. +func (r *ProfileExecutionResolver) MarkRouteActive(ctx context.Context, sessionID string, expectedGeneration int64) error { + if r.engine == nil { + return errors.New("dynamic profile execution is not configured") + } + return r.engine.MarkActive(ctx, sessionID, expectedGeneration) +} + +// MarkRouteActionRequired transitions a claimed route to durable +// action_required regardless of its current status. Callers use this so a +// launch failure after a generation claim always exposes a recovery action +// instead of leaving the route stuck at "starting". +func (r *ProfileExecutionResolver) MarkRouteActionRequired( + ctx context.Context, + sessionID string, + expectedGeneration int64, + reason string, +) (dynamic.RouteDecision, error) { + if r.engine == nil { + return dynamic.RouteDecision{}, errors.New("dynamic profile execution is not configured") + } + return r.engine.MarkActionRequired(ctx, sessionID, expectedGeneration, reason) +} + func (r *ProfileExecutionResolver) resolveRetryRouteAction( ctx context.Context, sessionID, profileID, currentExecutionProfileID string, diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index f079f911e3..8c8b3487de 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/apps/backend/internal/orchestrator/dynamic_launch.go @@ -21,6 +21,8 @@ import ( ) const dynamicRouteStatusWaiting = "waiting" +const dynamicRouteStatusActive = "active" +const dynamicRouteStatusActionRequired = "action_required" // dynamicSuccessorDetachedTimeout bounds one detached fallback launch: the // predecessor stop plus the successor launch. Without it the repository calls @@ -174,6 +176,167 @@ func (s *Service) persistDynamicACPSession( return s.repo.UpdateTaskSession(ctx, session) } +// markDynamicRouteActive completes the durable "starting" -> "active" +// transition after a concrete launch has actually succeeded, and mirrors the +// status onto the task-session projection so both durable stores agree. +// Best-effort: a launch that already succeeded is not undone by a failure to +// record this side transition. +func (s *Service) markDynamicRouteActive(ctx context.Context, sessionID string, generation int64) { + if s.profileExecutionResolver == nil || sessionID == "" || generation <= 0 { + return + } + if err := s.profileExecutionResolver.MarkRouteActive(ctx, sessionID, generation); err != nil { + if !errors.Is(err, dynamicruntime.ErrStaleGeneration) && !errors.Is(err, dynamicruntime.ErrRouteStateNotFound) { + s.logger.Warn("failed to mark dynamic route active", + zap.String("session_id", sessionID), zap.Error(err)) + } + return + } + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil || session == nil || session.RouteGeneration != generation { + return + } + if loader, ok := s.repo.(dynamicRouteStateLoader); ok { + state, loadErr := loader.LoadRouteState(ctx, sessionID) + if loadErr != nil || state == nil || state.Generation != generation || state.Status != dynamicRouteStatusActive { + return + } + } + s.mirrorDynamicRouteProjection(ctx, session, generation, dynamicRouteStatusActive, session.RouteReason) +} + +// markDynamicRouteActionRequired transitions a claimed dynamic route to +// 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. +func (s *Service) markDynamicRouteActionRequired(ctx context.Context, sessionID string, generation int64, reason string) { + if s.profileExecutionResolver == nil || sessionID == "" || generation <= 0 { + return + } + decision, err := s.profileExecutionResolver.MarkRouteActionRequired(ctx, sessionID, generation, reason) + if err != nil { + if !errors.Is(err, dynamicruntime.ErrStaleGeneration) && !errors.Is(err, dynamicruntime.ErrRouteStateNotFound) { + s.logger.Warn("failed to mark dynamic route action_required", + zap.String("session_id", sessionID), zap.Error(err)) + } + return + } + if decision.Status != dynamicRouteStatusActionRequired { + return + } + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil || session == nil || session.RouteGeneration != generation { + return + } + s.mirrorDynamicRouteProjection(ctx, session, generation, decision.Status, reason) +} + +type dynamicRouteSessionProjector interface { + UpdateTaskSessionDynamicRouteIfCurrent( + context.Context, + string, + int64, + string, + string, + string, + ) (bool, time.Time, error) +} + +func (s *Service) mirrorDynamicRouteProjection( + ctx context.Context, + session *models.TaskSession, + generation int64, + status, reason string, +) { + if session == nil || session.RouteGeneration != generation || status == "" { + return + } + if session.RouteState == status && session.RouteReason == reason { + return + } + oldState := session.State + if projector, ok := s.repo.(dynamicRouteSessionProjector); ok { + changed, updatedAt, err := projector.UpdateTaskSessionDynamicRouteIfCurrent( + ctx, session.ID, generation, session.RouteState, status, reason, + ) + if err != nil { + s.logger.Warn("failed to mirror dynamic route state to task session", + zap.String("session_id", session.ID), zap.Error(err)) + return + } + if !changed { + return + } + session.RouteState = status + session.RouteReason = reason + session.UpdatedAt = updatedAt + s.publishTaskSessionStateChanged(ctx, session.TaskID, session.ID, oldState, session.State, session.ErrorMessage, &updatedAt, session) + return + } + // Legacy repositories do not expose the narrow route projection update. Keep + // the fallback for tests and older adapters, while production SQLite uses the + // generation/status-guarded method above. + session.RouteState = status + session.RouteReason = reason + if err := s.repo.UpdateTaskSession(ctx, session); err != nil { + s.logger.Warn("failed to mirror dynamic route state to task session", + zap.String("session_id", session.ID), zap.Error(err)) + return + } + updatedAt := session.UpdatedAt + s.publishTaskSessionStateChanged(ctx, session.TaskID, session.ID, oldState, session.State, session.ErrorMessage, &updatedAt, session) +} + +// handleAgentProcessStarted settles a dynamic route only after the asynchronous +// process start succeeds. LaunchPreparedSession returns before this point. +func (s *Service) handleAgentProcessStarted( + ctx context.Context, + _, sessionID, agentExecutionID string, +) { + if s.profileExecutionResolver == nil || sessionID == "" { + return + } + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil || session == nil || session.RouteGeneration <= 0 || session.ExecutionProfileID == "" { + return + } + if session.AgentExecutionID != "" && agentExecutionID != "" && session.AgentExecutionID != agentExecutionID { + return + } + if session.State != models.TaskSessionStateStarting && session.State != models.TaskSessionStateRunning { + return + } + s.markDynamicRouteActive(ctx, sessionID, session.RouteGeneration) +} + +// handleAgentProcessStartFailed settles a claimed dynamic route when process +// startup fails after LaunchPreparedSession already returned successfully. +func (s *Service) handleAgentProcessStartFailed( + ctx context.Context, + _, sessionID, agentExecutionID string, + _ error, +) { + if s.profileExecutionResolver == nil || sessionID == "" { + return + } + session, err := s.repo.GetTaskSession(ctx, sessionID) + if err != nil || session == nil || session.RouteGeneration <= 0 || session.ExecutionProfileID == "" { + return + } + if session.AgentExecutionID != "" && agentExecutionID != "" && session.AgentExecutionID != agentExecutionID { + return + } + if session.State == models.TaskSessionStateCancelled || session.State == models.TaskSessionStateCompleted { + return + } + s.markDynamicRouteActionRequired(ctx, sessionID, session.RouteGeneration, "agent_process_start_failed") +} + func applyDynamicRouteDecisionProjection( session *models.TaskSession, decision dynamicruntime.RouteDecision, @@ -482,14 +645,37 @@ func (s *Service) routeDynamicAgentFailure( data watcher.AgentEventData, classified *routingerr.Error, ) bool { - if classified == nil || !classified.FallbackAllowed { - return false - } data = s.withDynamicAttemptEvidence(data) session, ok := s.dynamicFailureSession(ctx, data) if !ok { return false } + reason := "dynamic_recovery_declined" + if classified != nil { + reason = classified.Error() + } + // The route already holds a claimed generation the moment + // dynamicFailureSession succeeds. Every return below is a decline, so the + // deferred guard covers every early return uniformly (including any added + // later) instead of relying on each branch to remember its own + // transition. `generation` is updated in place once RouteAfterFailure + // claims a new one so the guard marks the generation actually holding the + // 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. + generation := session.RouteGeneration + handled := false + defer func() { + if !handled { + s.markDynamicRouteActionRequired(ctx, session.ID, generation, reason) + } + }() + if classified == nil || !classified.FallbackAllowed { + return false + } if !dynamicPreResultSafe(data) { // A dynamic route is never allowed to guess that a failed turn was // pre-result. Missing evidence is as unsafe as observed output or a @@ -512,9 +698,12 @@ func (s *Service) routeDynamicAgentFailure( session.RouteGeneration, classified, ) if err != nil { - return s.persistPendingDynamicRecovery(ctx, session, decision, err) + handled = s.persistPendingDynamicRecovery(ctx, session, decision, err) + return handled } - return s.launchDynamicSuccessorAfterFailure(ctx, data, session, conductor, decision, continuationInput) + generation = decision.Generation + handled = s.launchDynamicSuccessorAfterFailure(ctx, data, session, conductor, decision, continuationInput) + return handled } func (s *Service) persistPendingDynamicRecovery( @@ -752,6 +941,15 @@ func (s *Service) LaunchDynamicRouteAction(ctx context.Context, sessionID string } return errors.New("dynamic route action session not found") } + // The backend route-action handler has already claimed session.RouteGeneration + // as "starting" before calling this. Every error return below leaves that + // claim without an owner unless the deferred guard marks it action_required. + succeeded := false + defer func() { + if !succeeded { + s.markDynamicRouteActionRequired(ctx, sessionID, session.RouteGeneration, "manual dynamic route action failed") + } + }() task, err := s.scheduler.GetTask(ctx, session.TaskID) if err != nil { return err @@ -786,6 +984,7 @@ func (s *Service) LaunchDynamicRouteAction(ctx context.Context, sessionID string if !s.relaunchDynamicTaskAfterFailure(ctx, data, session.ExecutionProfileID) { return errors.New("dynamic route action successor launch failed") } + succeeded = true return nil } @@ -811,7 +1010,29 @@ func (s *Service) relaunchDynamicTaskAfterFailure( ctx context.Context, data watcher.AgentEventData, executionProfileID string, -) bool { +) (succeeded bool) { + defer func() { + if succeeded || s.profileExecutionResolver == nil || data.SessionID == "" { + return + } + // This helper is used by both the synchronous route-failure path and + // the detached successor worker. The caller's deferred guard covers + // the former, but the detached path reports handled=true before the + // worker completes, so this helper must settle the claimed generation + // when the relaunch fails. + markCtx := context.WithoutCancel(ctx) + session, err := s.repo.GetTaskSession(markCtx, data.SessionID) + if err != nil || session == nil || session.RouteGeneration <= 0 { + return + } + s.markDynamicRouteActionRequired( + markCtx, + data.SessionID, + session.RouteGeneration, + "dynamic_successor_launch_failed", + ) + }() + prompt, ok := s.dynamicRelaunchPrompt(ctx, data.SessionID) if !ok { return false diff --git a/apps/backend/internal/orchestrator/dynamic_policy_recovery.go b/apps/backend/internal/orchestrator/dynamic_policy_recovery.go index c0193732ac..e3c872815b 100644 --- a/apps/backend/internal/orchestrator/dynamic_policy_recovery.go +++ b/apps/backend/internal/orchestrator/dynamic_policy_recovery.go @@ -17,6 +17,10 @@ type dynamicRouteStateLister interface { ListPendingRouteStates(context.Context) ([]dynamicruntime.RouteState, error) } +type dynamicStartingRouteLister interface { + ListStartingRouteStates(context.Context) ([]dynamicruntime.RouteState, error) +} + // startDynamicPolicyRecovery restores only durable waits whose dispatch has // not started. A persisted retrying state is intentionally left for manual // recovery because a process restart cannot prove whether its launch crossed @@ -48,6 +52,60 @@ func (s *Service) startDynamicPolicyRecovery(ctx context.Context) { } } +// reconcileOrphanedDynamicStartingRoutes recovers routes left at "starting" +// with no launch owner: a launch that failed after claiming a generation but +// before reaching a terminal route status (see routeDynamicAgentFailure and +// LaunchDynamicRouteAction) can strand a route this way, and no timer or +// event fires for a durable status that is not a pending wait. A route that +// reached MarkActive is no longer "starting", so this sweep cannot demote a +// healthy idling dynamic session. +func (s *Service) reconcileOrphanedDynamicStartingRoutes(ctx context.Context) { + if s.profileExecutionResolver == nil { + return + } + lister, ok := s.repo.(dynamicStartingRouteLister) + if !ok { + return + } + states, err := lister.ListStartingRouteStates(ctx) + if err != nil { + s.logger.Warn("failed to list starting dynamic route states", zap.Error(err)) + return + } + for _, state := range states { + s.reconcileOrphanedDynamicStartingRoute(ctx, state) + } +} + +func (s *Service) reconcileOrphanedDynamicStartingRoute(ctx context.Context, state dynamicruntime.RouteState) { + session, err := s.repo.GetTaskSession(ctx, state.SessionID) + if err != nil || session == nil || session.RouteGeneration != state.Generation { + return + } + if !isOrphanableDynamicSessionState(session.State) { + return + } + s.markDynamicRouteActionRequired(ctx, state.SessionID, state.Generation, "orphaned_starting_route") +} + +// isOrphanableDynamicSessionState reports whether a session's current state +// is consistent with "a route claimed as starting has no in-flight launch +// owner". STARTING means a launch was under way when the process that owned +// it stopped; IDLE (office runs only) means the executor was already torn +// down between turns with no successor launched. Every other state either +// already has a live owner (RUNNING), has not attempted a launch for this +// claim yet (CREATED — PrepareSession's background workspace launch claims a +// route ahead of the user's first prompt, which can sit unstarted for a +// long time by design), or is a terminal/parked outcome the ordinary +// session UI already explains (WAITING_FOR_INPUT, COMPLETED, FAILED, +// CANCELLED). Flagging those would put a stale "Retry / Try next / Stop" +// banner on a session with nothing left to recover — including every +// dynamic session that predates MarkActive, which is "starting" for exactly +// this reason and not because anything is actually stuck. +func isOrphanableDynamicSessionState(state models.TaskSessionState) bool { + return state == models.TaskSessionStateStarting || state == models.TaskSessionStateIdle +} + func (s *Service) stopDynamicPolicyRecovery() { s.dynamicRecoveryMu.Lock() defer s.dynamicRecoveryMu.Unlock() diff --git a/apps/backend/internal/orchestrator/dynamic_route_recovery_test.go b/apps/backend/internal/orchestrator/dynamic_route_recovery_test.go new file mode 100644 index 0000000000..8b12b4985b --- /dev/null +++ b/apps/backend/internal/orchestrator/dynamic_route_recovery_test.go @@ -0,0 +1,706 @@ +package orchestrator + +import ( + "context" + "errors" + "testing" + "time" + + agentruntime "github.com/kandev/kandev/internal/agent/runtime" + dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic" + "github.com/kandev/kandev/internal/agent/runtime/routingerr" + agentruntimekind "github.com/kandev/kandev/internal/agentruntime" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/orchestrator/executor" + "github.com/kandev/kandev/internal/orchestrator/watcher" + "github.com/kandev/kandev/internal/task/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// seedClaimedDynamicRoute claims generation 1 for a fixed single-candidate +// dynamic profile directly through the engine (bypassing the profile store, +// which these tests do not need) and projects the claim onto the +// task_sessions row the way persistDynamicLaunchDecision would. +func seedClaimedDynamicRoute( + t *testing.T, + ctx context.Context, + repo interface { + GetTaskSession(context.Context, string) (*models.TaskSession, error) + UpdateTaskSession(context.Context, *models.TaskSession) error + }, + engine *dynamicruntime.Engine, + sessionID, executionID string, +) dynamicruntime.RouteDecision { + t.Helper() + profile := dynamicruntime.Profile{ + ID: "dynamic-logical", Version: 1, + Candidates: []dynamicruntime.Candidate{{ID: "candidate-1", Enabled: true}}, + } + decision, err := engine.Select(sessionID, profile, 0, "") + if err != nil { + t.Fatalf("claim initial route generation: %v", err) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + session.AgentProfileID = profile.ID + session.ExecutionProfileID = decision.ExecutionProfileID + session.RouteGeneration = decision.Generation + session.RouteState = decision.Status + session.AgentExecutionID = executionID + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("UpdateTaskSession: %v", err) + } + return decision +} + +func TestRouteDynamicAgentFailure_MarksActionRequiredWhenPreResultUnsafe(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-pre-result-unsafe" + sessionID = "session-dynamic-pre-result-unsafe" + executionID = "execution-dynamic-pre-result-unsafe" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + // Reproduce the observed production failure verbatim: the agent already + // streamed output for this attempt (OutputObserved=true), so + // dynamicPreResultSafe must refuse to auto-replace the turn. + svc.beginDynamicAttempt(sessionID) + svc.bindDynamicAttemptExecution(sessionID, executionID) + svc.observeDynamicAttempt(sessionID, executionID, true, false) + + classified := &routingerr.Error{ + Code: routingerr.CodeProviderOverloaded, Class: routingerr.ClassTransient, + FallbackAllowed: true, AutoRetryable: true, + } + handled := svc.routeDynamicAgentFailure(ctx, watcher.AgentEventData{ + TaskID: taskID, SessionID: sessionID, AgentExecutionID: executionID, + }, classified) + if handled { + t.Fatal("routeDynamicAgentFailure claimed to handle a pre-result-unsafe failure") + } + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != "action_required" { + t.Fatalf("durable route state = %#v, want action_required", routeState) + } + if routeState.Generation != 1 { + t.Fatalf("route generation = %d, want 1 (no successor was claimed)", routeState.Generation) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.RouteState != "action_required" { + t.Fatalf("task session route_state = %q, want action_required", session.RouteState) + } +} + +func TestRouteDynamicAgentFailure_MarksActionRequiredWhenFallbackNotAllowed(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-fallback-not-allowed" + sessionID = "session-dynamic-fallback-not-allowed" + executionID = "execution-dynamic-fallback-not-allowed" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + classified := &routingerr.Error{Code: routingerr.CodeTask, FallbackAllowed: false} + handled := svc.routeDynamicAgentFailure(ctx, watcher.AgentEventData{ + TaskID: taskID, SessionID: sessionID, AgentExecutionID: executionID, + }, classified) + if handled { + t.Fatal("routeDynamicAgentFailure claimed to handle a non-fallback failure") + } + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != "action_required" { + t.Fatalf("durable route state = %#v, want action_required", routeState) + } +} + +func TestLaunchDynamicRouteAction_MarksRouteActionRequiredWhenRelaunchFails(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-route-action-failure" + sessionID = "session-dynamic-route-action-failure" + executionID = "execution-dynamic-route-action-failure" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + stopErr := errors.New("runtime teardown failed") + agentManager := &mockAgentManager{stopAgentWithReasonErr: stopErr} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + // Simulates the backend route-action handler having already claimed the + // successor generation as "starting" before calling LaunchDynamicRouteAction. + if err := svc.LaunchDynamicRouteAction(ctx, sessionID); err == nil { + t.Fatal("LaunchDynamicRouteAction succeeded despite predecessor stop failing") + } + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != "action_required" { + t.Fatalf("durable route state = %#v, want action_required", routeState) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.RouteState != "action_required" { + t.Fatalf("task session route_state = %q, want action_required", session.RouteState) + } +} + +func TestReconcileOrphanedDynamicStartingRoutes_SweepsInFlightLaunchStates(t *testing.T) { + // STARTING means a launch was under way when the owning process stopped; + // IDLE (office runs) means the executor was already torn down between + // turns. Both have no possible owner left after a restart. + for _, state := range []models.TaskSessionState{ + models.TaskSessionStateStarting, + models.TaskSessionStateIdle, + } { + t.Run(string(state), func(t *testing.T) { + ctx := context.Background() + taskID := "task-dynamic-orphan-sweep-" + string(state) + sessionID := "session-dynamic-orphan-sweep-" + string(state) + executionID := "execution-dynamic-orphan-sweep-" + string(state) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, state) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + + seedEngine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, seedEngine, true)) + seedClaimedDynamicRoute(t, ctx, repo, seedEngine, sessionID, executionID) + // Reconcile through a fresh engine. This proves the startup path loads + // the durable route row instead of relying on the old process cache. + recoveryEngine := dynamicruntime.NewEngine( + dynamicruntime.WithPersistence(repo), + dynamicruntime.WithStateLoader(repo), + ) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, recoveryEngine, true)) + + svc.reconcileOrphanedDynamicStartingRoutes(ctx) + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != "action_required" { + t.Fatalf("durable route state = %#v, want action_required", routeState) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.RouteState != "action_required" { + t.Fatalf("task session route_state = %q, want action_required", session.RouteState) + } + }) + } +} + +// TestReconcileOrphanedDynamicStartingRoutes_PrecedesGeneralStartupReconciliation +// is the regression test for the CodeRabbit finding on Create PR review +// (service.go#L2580): Service.Start runs reconcileExecutorSessionsOnStartup +// before reconcileOrphanedDynamicStartingRoutes, and the former flips every +// STARTING/RUNNING session with an executors_running row to +// WAITING_FOR_INPUT — a state the orphan sweep treats as an already-explained +// terminal outcome and skips. Almost every claimed dynamic route has an +// executors_running row by the time a crash strands it (PrepareSession's +// background workspace launch creates the row early), so running the two in +// Start's original order would make the sweep a no-op for the exact scenario +// it exists to recover. This test drives the two reconcilers in Start's +// actual order and requires the route to still reach action_required. +func TestReconcileOrphanedDynamicStartingRoutes_PrecedesGeneralStartupReconciliation(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-orphan-ordering" + sessionID = "session-dynamic-orphan-ordering" + executionID = "execution-dynamic-orphan-ordering" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + if err := repo.UpsertExecutorRunning(ctx, &models.ExecutorRunning{ + ID: sessionID, + SessionID: sessionID, + TaskID: taskID, + Runtime: agentruntimekind.RuntimeStandalone, + Status: models.ExecutorRunningStatusRunning, + CreatedAt: time.Now().UTC(), + UpdatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("upsert executor row: %v", err) + } + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + + seedEngine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, seedEngine, true)) + seedClaimedDynamicRoute(t, ctx, repo, seedEngine, sessionID, executionID) + recoveryEngine := dynamicruntime.NewEngine( + dynamicruntime.WithPersistence(repo), + dynamicruntime.WithStateLoader(repo), + ) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, recoveryEngine, true)) + + // Mirror Service.Start's exact ordering: the orphan sweep must run before + // the general startup reconciler. + svc.reconcileOrphanedDynamicStartingRoutes(ctx) + svc.reconcileExecutorSessionsOnStartup(ctx) + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != "action_required" { + t.Fatalf("durable route state = %#v, want action_required", routeState) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.RouteState != "action_required" { + t.Fatalf("task session route_state = %q, want action_required", session.RouteState) + } + // The general reconciler still ran (and ran second): the session itself + // settles to WAITING_FOR_INPUT the ordinary way. + if session.State != models.TaskSessionStateWaitingForInput { + t.Fatalf("task session state = %q, want WAITING_FOR_INPUT", session.State) + } +} + +// TestReconcileOrphanedDynamicStartingRoutes_SkipsNonOrphanStates is the +// regression test for review finding F3: RUNNING already has a live owner; +// CREATED is PrepareSession's ordinary pre-first-prompt claim, which can sit +// unstarted for a long time by design; WAITING_FOR_INPUT/COMPLETED/FAILED/ +// CANCELLED are terminal or parked outcomes the normal session UI already +// explains — including every dynamic session that predates MarkActive and is +// "starting" only because that transition did not exist yet, not because +// anything is stuck. None of these should grow a stale recovery banner. +func TestReconcileOrphanedDynamicStartingRoutes_SkipsNonOrphanStates(t *testing.T) { + for _, state := range []models.TaskSessionState{ + models.TaskSessionStateRunning, + models.TaskSessionStateCreated, + models.TaskSessionStateWaitingForInput, + models.TaskSessionStateCompleted, + models.TaskSessionStateFailed, + models.TaskSessionStateCancelled, + } { + t.Run(string(state), func(t *testing.T) { + ctx := context.Background() + taskID := "task-dynamic-orphan-skip-" + string(state) + sessionID := "session-dynamic-orphan-skip-" + string(state) + executionID := "execution-dynamic-orphan-skip-" + string(state) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, state) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + svc.reconcileOrphanedDynamicStartingRoutes(ctx) + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != "starting" { + t.Fatalf("durable route state = %#v, want unchanged starting for a %s session", routeState, state) + } + }) + } +} + +func TestMarkDynamicRouteActive_MirrorsBothStores(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-mark-active" + sessionID = "session-dynamic-mark-active" + executionID = "execution-dynamic-mark-active" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + decision := seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + svc.markDynamicRouteActive(ctx, sessionID, decision.Generation) + + routeState, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if routeState == nil || routeState.Status != dynamicRouteStatusActive { + t.Fatalf("durable route state = %#v, want active", routeState) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.RouteState != dynamicRouteStatusActive { + t.Fatalf("task session route_state = %q, want active", session.RouteState) + } + + // A healthy active route must not be swept by startup reconciliation: it + // is no longer "starting", so ListStartingRouteStates never returns it. + svc.reconcileOrphanedDynamicStartingRoutes(ctx) + routeState, err = repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState after sweep: %v", err) + } + if routeState.Status != dynamicRouteStatusActive { + t.Fatalf("route state after sweep = %#v, want unchanged active", routeState) + } +} + +func TestDynamicTaskDownstreamLaunchMarksRouteActiveAfterProcessStart(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-downstream-active" + sessionID = "session-dynamic-downstream-active" + executionID = "execution-dynamic-downstream-active" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + agentManager := &mockAgentManager{ + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return &executor.LaunchAgentResponse{AgentExecutionID: executionID}, nil + }, + startAgentProcessFunc: func(context.Context, string) error { return nil }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + decision := seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + started := make(chan struct{}) + svc.executor.SetOnAgentProcessStarted(func(callbackCtx context.Context, callbackTaskID, callbackSessionID, callbackExecutionID string) { + svc.handleAgentProcessStarted(callbackCtx, callbackTaskID, callbackSessionID, callbackExecutionID) + close(started) + }) + + downstream := &dynamicTaskDownstream{ + service: svc, + task: &v1.Task{ID: taskID, WorkspaceID: "ws1", Description: "test"}, + sessionID: sessionID, + options: executor.LaunchOptions{ + AgentProfileID: "candidate-1", + StartAgent: true, + }, + } + if _, err := downstream.Launch(ctx, dynamicruntime.DownstreamLaunch{ + ExecutionProfileID: "candidate-1", + Decision: decision, + }); err != nil { + t.Fatalf("dynamic downstream launch: %v", err) + } + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for process-start settlement") + } + + state, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if state == nil || state.Status != dynamicRouteStatusActive || state.Generation != decision.Generation { + t.Fatalf("route state after downstream launch = %#v, want active generation %d", state, decision.Generation) + } +} + +func TestMarkDynamicRouteActionRequiredPublishesRouteProjection(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-action-event" + sessionID = "session-dynamic-action-event" + executionID = "execution-dynamic-action-event" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + eventBus := &recordingEventBus{} + svc.eventBus = eventBus + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + decision := seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + svc.markDynamicRouteActionRequired(ctx, sessionID, decision.Generation, "launch_failed") + + if len(eventBus.events) != 1 || eventBus.events[0].subject != events.TaskSessionStateChanged { + t.Fatalf("published events = %#v, want one session state event", eventBus.events) + } + data, ok := eventBus.events[0].event.Data.(map[string]interface{}) + if !ok { + t.Fatalf("event data = %T, want map[string]interface{}", eventBus.events[0].event.Data) + } + if data["route_state"] != dynamicRouteStatusActionRequired { + t.Fatalf("event route_state = %#v, want action_required", data["route_state"]) + } + if data["route_reason"] != "launch_failed" { + t.Fatalf("event route_reason = %#v, want launch_failed", data["route_reason"]) + } +} + +func TestMarkDynamicRouteActiveDoesNotOverwriteActionRequiredRoute(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-active-noop" + sessionID = "session-dynamic-active-noop" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateWaitingForInput) + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + session.AgentProfileID = "dynamic-logical" + session.ExecutionProfileID = "candidate-1" + session.RouteGeneration = 1 + session.RouteState = dynamicRouteStatusActionRequired + session.RouteReason = "launch_failed" + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("UpdateTaskSession: %v", err) + } + if err := repo.SaveRouteState(ctx, dynamicruntime.RouteState{ + SessionID: sessionID, LogicalProfileID: session.AgentProfileID, + ExecutionProfileID: session.ExecutionProfileID, Generation: 1, + Status: dynamicRouteStatusActionRequired, UpdatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("SaveRouteState: %v", err) + } + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + engine := dynamicruntime.NewEngine( + dynamicruntime.WithPersistence(repo), + dynamicruntime.WithStateLoader(repo), + ) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + + svc.markDynamicRouteActive(ctx, sessionID, 1) + + loaded, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if loaded == nil || loaded.Status != dynamicRouteStatusActionRequired { + t.Fatalf("durable route state = %#v, want action_required", loaded) + } + updated, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession after MarkActive: %v", err) + } + if updated.RouteState != dynamicRouteStatusActionRequired || updated.RouteReason != "launch_failed" { + t.Fatalf("session route projection = (%q, %q), want action_required/launch_failed", updated.RouteState, updated.RouteReason) + } +} + +func TestHandleAgentProcessStartedMarksDynamicRouteActive(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-process-started" + sessionID = "session-dynamic-process-started" + executionID = "execution-dynamic-process-started" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + decision := seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + svc.handleAgentProcessStarted(ctx, taskID, sessionID, executionID) + + state, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if state == nil || state.Generation != decision.Generation || state.Status != dynamicRouteStatusActive { + t.Fatalf("route state after process start = %#v, want active generation %d", state, decision.Generation) + } +} + +func TestHandleAgentProcessStartedSkipsCancelledSession(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-process-cancelled" + sessionID = "session-dynamic-process-cancelled" + executionID = "execution-dynamic-process-cancelled" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateCancelled) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + decision := seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + svc.handleAgentProcessStarted(ctx, taskID, sessionID, executionID) + + state, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if state == nil || state.Generation != decision.Generation || state.Status != "starting" { + t.Fatalf("route state after cancelled process start = %#v, want unchanged starting generation %d", state, decision.Generation) + } +} + +func TestHandleAgentProcessStartFailedMarksDynamicRouteActionRequired(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-process-failed" + sessionID = "session-dynamic-process-failed" + executionID = "execution-dynamic-process-failed" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateStarting) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, &mockAgentManager{}) + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo)) + svc.SetProfileExecutionResolver(agentruntime.NewProfileExecutionResolver(nil, engine, true)) + decision := seedClaimedDynamicRoute(t, ctx, repo, engine, sessionID, executionID) + + svc.handleAgentProcessStartFailed(ctx, taskID, sessionID, executionID, errors.New("provider process failed")) + + state, err := repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if state == nil || state.Generation != decision.Generation || state.Status != dynamicRouteStatusActionRequired { + t.Fatalf("route state after process failure = %#v, want action_required generation %d", state, decision.Generation) + } +} + +func TestRouteDynamicAgentFailureMarksSuccessorGenerationActionRequiredWhenLaunchFails(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-dynamic-successor-failure" + sessionID = "session-dynamic-successor-failure" + executionID = "execution-dynamic-successor-failure" + dynamicID = "dynamic-successor-failure" + ) + repo := setupTestRepo(t) + seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateRunning) + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + agentManager := &mockAgentManager{stopAgentWithReasonErr: errors.New("predecessor stop failed")} + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentManager) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + engineOptions := []dynamicruntime.EngineOption{dynamicruntime.WithPersistence(repo)} + resolver := newWorkflowDynamicProfileResolverWithCandidates(t, dynamicID, []workflowDynamicCandidate{ + { + executionProfileID: "candidate-1", + enabled: true, + rulesJSON: `{"on_provider_error":"try_next"}`, + }, + {executionProfileID: "candidate-2", enabled: true}, + }, engineOptions...) + svc.SetProfileExecutionResolver(resolver) + selected, err := resolver.Resolve(ctx, sessionID, dynamicID, 0, "") + if err != nil { + t.Fatalf("initial dynamic resolve: %v", err) + } + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + session.AgentProfileID = dynamicID + session.ExecutionProfileID = selected.ExecutionProfileID + session.RouteGeneration = selected.Generation + session.RouteState = selected.Decision.Status + session.AgentExecutionID = executionID + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("project initial route: %v", err) + } + svc.beginDynamicAttempt(sessionID) + svc.bindDynamicAttemptExecution(sessionID, executionID) + + handled := svc.routeDynamicAgentFailure(ctx, watcher.AgentEventData{ + TaskID: taskID, SessionID: sessionID, AgentExecutionID: executionID, + PromptGeneration: 1, + }, &routingerr.Error{ + Code: routingerr.CodeProviderOverloaded, Class: routingerr.ClassTransient, + FallbackAllowed: true, + }) + // The successor launch can complete in this call or on a detached worker. + // Both paths must settle the claimed generation before this recovery can + // be presented to the user. + deadline := time.Now().Add(2 * time.Second) + var state *dynamicruntime.RouteState + for { + state, err = repo.LoadRouteState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadRouteState while waiting for successor failure: %v", err) + } + if state != nil && state.Generation == 2 && state.Status == dynamicRouteStatusActionRequired { + break + } + if time.Now().After(deadline) { + t.Fatalf("successor launch result not settled (handled=%v): %#v", handled, state) + } + time.Sleep(10 * time.Millisecond) + } + + updated, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("GetTaskSession after successor failure: %v", err) + } + if updated.RouteGeneration != 2 || updated.RouteState != dynamicRouteStatusActionRequired { + t.Fatalf("session route projection = generation %d state %q, want generation 2/action_required", updated.RouteGeneration, updated.RouteState) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_workflow.go b/apps/backend/internal/orchestrator/event_handlers_workflow.go index 50e1ab0b24..c6ae9d3393 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow.go @@ -2524,6 +2524,11 @@ func (s *Service) prepareWorkflowReplacementSession( WorkflowStepID: dbTask.WorkflowStepID, StartAgent: false, }); err != nil { + // 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". + 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/event_handlers_workflow_dynamic_profile_test.go b/apps/backend/internal/orchestrator/event_handlers_workflow_dynamic_profile_test.go index 57f487463b..886c3fb5af 100644 --- a/apps/backend/internal/orchestrator/event_handlers_workflow_dynamic_profile_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_workflow_dynamic_profile_test.go @@ -71,6 +71,59 @@ func TestCreateNewSessionForStep_ResolvesDynamicProfileBeforeWorkspaceAttach(t * } } +// TestCreateNewSessionForStep_MarksRouteActionRequiredWhenWorkspaceAttachFails +// is the regression test for review finding F2: a dynamic route resolved and +// claimed as "starting" by prepareWorkflowReplacementSession must not stay +// there forever when the workspace attach that follows fails, since nothing +// else transitions it — the replacement session is never actually started. +// +// @covers AC-AGENTS-DYNAMIC-AGENT-ROUTING-001.3 +func TestCreateNewSessionForStep_MarksRouteActionRequiredWhenWorkspaceAttachFails(t *testing.T) { + ctx := context.Background() + taskRepo, current := seedDynamicWorkflowSwitch(t) + const dynamicProfileID = "profile-dynamic" + const concreteProfileID = "profile-concrete" + + resolver := newWorkflowDynamicProfileResolver(t, dynamicProfileID, concreteProfileID) + attachErr := fmt.Errorf("workspace attach failed") + agentManager := &mockAgentManager{ + repoForExecutionLookup: taskRepo, + launchAgentFunc: func(_ context.Context, _ *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, attachErr + }, + } + schedulerRepo := newMockTaskRepo() + schedulerRepo.tasks[current.TaskID] = &v1.Task{ID: current.TaskID, WorkspaceID: "ws1", Title: "Test Task"} + stepGetter := newMockStepGetter() + stepGetter.steps["step-work"] = &wfmodels.WorkflowStep{ID: "step-work", WorkflowID: "wf1", Name: "Work"} + svc := createTestServiceWithScheduler(taskRepo, stepGetter, schedulerRepo, agentManager) + svc.SetProfileExecutionResolver(resolver) + + if _, err := svc.createNewSessionForStep(ctx, current.TaskID, current, dynamicProfileID); err == nil { + t.Fatal("createNewSessionForStep succeeded despite a failed workspace attach") + } + + sessions, err := taskRepo.ListTaskSessions(ctx, current.TaskID) + if err != nil { + t.Fatalf("list task sessions: %v", err) + } + var replacement *models.TaskSession + for _, session := range sessions { + if session.ID != current.ID { + replacement = session + } + } + if replacement == nil { + t.Fatalf("expected the replacement session to remain for inspection, sessions = %+v", sessions) + } + if replacement.RouteGeneration == 0 { + t.Fatal("expected the replacement session to have claimed a route generation") + } + if replacement.RouteState != "action_required" { + t.Fatalf("replacement session route_state = %q, want action_required", replacement.RouteState) + } +} + // @covers AC-AGENTS-DYNAMIC-AGENT-ROUTING-001.1 func TestCreateNewSessionForStep_RemovesPreparedSessionWhenDynamicResolutionFails(t *testing.T) { ctx := context.Background() @@ -162,6 +215,30 @@ func newWorkflowDynamicProfileResolverWithCandidate( t *testing.T, dynamicProfileID, concreteProfileID string, candidateEnabled bool, + engineOptions ...dynamicruntime.EngineOption, +) *agentruntime.ProfileExecutionResolver { + return newWorkflowDynamicProfileResolverWithCandidates( + t, + dynamicProfileID, + []workflowDynamicCandidate{{ + executionProfileID: concreteProfileID, + enabled: candidateEnabled, + }}, + engineOptions..., + ) +} + +type workflowDynamicCandidate struct { + executionProfileID string + enabled bool + rulesJSON string +} + +func newWorkflowDynamicProfileResolverWithCandidates( + t *testing.T, + dynamicProfileID string, + candidates []workflowDynamicCandidate, + engineOptions ...dynamicruntime.EngineOption, ) *agentruntime.ProfileExecutionResolver { t.Helper() db, err := sqlx.Open("sqlite3", ":memory:") @@ -183,23 +260,35 @@ func newWorkflowDynamicProfileResolverWithCandidate( 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}, - } { + profiles := []*agentsettingsmodels.AgentProfile{{ + ID: dynamicProfileID, AgentID: "dynamic", Name: "Cascade", Enabled: true, + }} + for _, candidate := range candidates { + profiles = append(profiles, &agentsettingsmodels.AgentProfile{ + ID: candidate.executionProfileID, AgentID: "concrete-agent", + Name: candidate.executionProfileID, Enabled: true, + }) + } + for _, profile := range profiles { if err := repo.CreateAgentProfile(ctx, profile); err != nil { t.Fatal(err) } } + routes := make([]agentsettingsmodels.DynamicAgentRoute, 0, len(candidates)) + for position, candidate := range candidates { + routes = append(routes, agentsettingsmodels.DynamicAgentRoute{ + DynamicProfileID: dynamicProfileID, + Position: position, + ExecutionProfileID: candidate.executionProfileID, + Enabled: candidate.enabled, + RulesJSON: candidate.rulesJSON, + }) + } if err := repo.CreateDynamicAgentProfile(ctx, &agentsettingsmodels.DynamicAgentProfile{ProfileID: dynamicProfileID, Version: 1}, - []agentsettingsmodels.DynamicAgentRoute{{ - DynamicProfileID: dynamicProfileID, - ExecutionProfileID: concreteProfileID, - Enabled: candidateEnabled, - }}, + routes, ); err != nil { t.Fatal(err) } - return agentruntime.NewProfileExecutionResolver(repo, dynamicruntime.NewEngine(), true) + return agentruntime.NewProfileExecutionResolver(repo, dynamicruntime.NewEngine(engineOptions...), true) } diff --git a/apps/backend/internal/orchestrator/executor/executor.go b/apps/backend/internal/orchestrator/executor/executor.go index d90d4dbdaf..7ed75a3ee7 100644 --- a/apps/backend/internal/orchestrator/executor/executor.go +++ b/apps/backend/internal/orchestrator/executor/executor.go @@ -742,6 +742,15 @@ type TaskReviewStateReconcileFunc func(ctx context.Context, taskID, completedSes // its default FAILED state updates. type AgentStartFailedFunc func(ctx context.Context, taskID, sessionID, agentExecutionID string, err error, fromResume bool) (handled bool) +// AgentProcessStartedFunc is called after the agent process starts +// successfully and the executor has run its normal success callback. +type AgentProcessStartedFunc func(ctx context.Context, taskID, sessionID, agentExecutionID string) + +// AgentProcessStartFailedFunc is called after the executor has handled a +// failed process start. It lets an owning subsystem settle any durable claim +// that was made before the asynchronous start began. +type AgentProcessStartFailedFunc func(ctx context.Context, taskID, sessionID, agentExecutionID string, err error) + // LaunchFailedFunc is called when session launch fails before the agent starts. // repositoryID identifies the repository whose launch failed. Useful for // creating repository-scoped user-facing status messages tied to launch errors. @@ -842,6 +851,10 @@ type Executor struct { // delegates failure handling to this callback, allowing the orchestrator // to detect auth errors and treat them as recoverable. onAgentStartFailed AgentStartFailedFunc + // Callback after a process start succeeds or fails. These callbacks run + // after the executor's normal lifecycle bookkeeping. + onAgentProcessStarted AgentProcessStartedFunc + onAgentProcessStartFailed AgentProcessStartFailedFunc // Callback for session launch failures (pre-start). Allows orchestrator // to emit user-friendly guidance for known failure patterns. @@ -1147,6 +1160,18 @@ func (e *Executor) SetOnAgentStartFailed(fn AgentStartFailedFunc) { e.onAgentStartFailed = fn } +// SetOnAgentProcessStarted sets a callback invoked after asynchronous process +// startup succeeds and the session has been reconciled to RUNNING. +func (e *Executor) SetOnAgentProcessStarted(fn AgentProcessStartedFunc) { + e.onAgentProcessStarted = fn +} + +// SetOnAgentProcessStartFailed sets a callback invoked after asynchronous +// process startup fails and the normal failure handler has run. +func (e *Executor) SetOnAgentProcessStartFailed(fn AgentProcessStartFailedFunc) { + e.onAgentProcessStartFailed = fn +} + // SetOnPrimarySessionSet sets a callback for when the first session for a task // is marked primary. This publishes a task.updated event so the frontend // receives primary_session_id. diff --git a/apps/backend/internal/orchestrator/executor/executor_execute.go b/apps/backend/internal/orchestrator/executor/executor_execute.go index 41ea5b7fcf..8773abeef8 100644 --- a/apps/backend/internal/orchestrator/executor/executor_execute.go +++ b/apps/backend/internal/orchestrator/executor/executor_execute.go @@ -138,6 +138,9 @@ func (e *Executor) runAgentProcessAsync(ctx context.Context, taskID, sessionID, updateCtx, taskID, sessionID, agentExecutionID, err, escalateTaskOnFailure, fromResume, ) + if e.onAgentProcessStartFailed != nil { + e.onAgentProcessStartFailed(updateCtx, taskID, sessionID, agentExecutionID, err) + } return } if _, terminal := e.stopStartedExecutionIfSessionTerminal( @@ -150,6 +153,9 @@ func (e *Executor) runAgentProcessAsync(ctx context.Context, taskID, sessionID, } onSuccess(updateCtx) + if e.onAgentProcessStarted != nil { + e.onAgentProcessStarted(updateCtx, taskID, sessionID, agentExecutionID) + } }() } diff --git a/apps/backend/internal/orchestrator/executor/executor_test.go b/apps/backend/internal/orchestrator/executor/executor_test.go index 5927af7a61..bbb1925a38 100644 --- a/apps/backend/internal/orchestrator/executor/executor_test.go +++ b/apps/backend/internal/orchestrator/executor/executor_test.go @@ -2415,6 +2415,67 @@ func TestStartAgentProcessAsync_MarksStartingSessionRunning(t *testing.T) { } } +func TestStartAgentProcessAsyncNotifiesAfterProcessStart(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-123"] = &models.TaskSession{ + ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting, + } + repo.tasks["task-123"] = &models.Task{ID: "task-123", State: v1.TaskStateScheduling} + exec := newTestExecutor(t, &mockAgentManager{ + startAgentProcessFunc: func(context.Context, string) error { return nil }, + }, repo) + started := make(chan struct{}) + exec.SetOnAgentProcessStarted(func(_ context.Context, taskID, sessionID, executionID string) { + if taskID != "task-123" || sessionID != "session-123" || executionID != "exec-456" { + t.Errorf("process-start callback args = (%q, %q, %q)", taskID, sessionID, executionID) + } + close(started) + }) + + exec.startAgentProcessAsync(context.Background(), "task-123", "session-123", "exec-456") + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for process start") + } +} + +func TestStartAgentProcessAsyncNotifiesAfterProcessStartFailure(t *testing.T) { + repo := newMockRepository() + repo.sessions["session-123"] = &models.TaskSession{ + ID: "session-123", TaskID: "task-123", State: models.TaskSessionStateStarting, + } + repo.tasks["task-123"] = &models.Task{ID: "task-123", State: v1.TaskStateScheduling} + startErr := errors.New("process bootstrap failed") + stopped := make(chan struct{}) + exec := newTestExecutor(t, &mockAgentManager{ + startAgentProcessFunc: func(context.Context, string) error { return startErr }, + stopAgentFunc: func(context.Context, string, bool) error { close(stopped); return nil }, + }, repo) + failed := make(chan error, 1) + exec.SetOnAgentProcessStartFailed(func(_ context.Context, taskID, sessionID, executionID string, err error) { + if taskID != "task-123" || sessionID != "session-123" || executionID != "exec-456" { + t.Errorf("process-start failure callback args = (%q, %q, %q)", taskID, sessionID, executionID) + } + failed <- err + }) + + exec.startAgentProcessAsync(context.Background(), "task-123", "session-123", "exec-456") + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for failed process cleanup") + } + select { + case err := <-failed: + if !errors.Is(err, startErr) { + t.Fatalf("process-start callback error = %v, want %v", err, startErr) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for process-start failure callback") + } +} + func TestStartAgentProcessAsync_StopWinningStartRacePreservesReview(t *testing.T) { repo := newMockRepository() repo.sessions["session-123"] = &models.TaskSession{ diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index c72bf8fe91..c3aea6b485 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -1422,6 +1422,8 @@ func NewService( }) exec.SetLaunchFailureReviewEligibility(s.resolveLaunchFailureReviewEligibility) exec.SetOnAgentStartFailed(s.handleAgentStartFailed) + exec.SetOnAgentProcessStarted(s.handleAgentProcessStarted) + exec.SetOnAgentProcessStartFailed(s.handleAgentProcessStartFailed) if caps, ok := agentManager.(executor.ExecutorTypeCapabilities); ok { exec.SetCapabilities(caps) } @@ -2531,6 +2533,15 @@ func (s *Service) Start(ctx context.Context) error { s.mu.Unlock() return err } + // Recover routes orphaned at "starting" by a launch failure that never + // reached a terminal route status. Must run before + // reconcileExecutorSessionsOnStartup and scheduler.Start: the former + // normalizes every STARTING/RUNNING session to WAITING_FOR_INPUT, which + // would make every orphaned route look like a terminal, already-explained + // session and hide it from this sweep; the latter can begin dispatching + // new launches, which would let a genuinely in-flight claim race this + // snapshot of "starting" routes. + s.reconcileOrphanedDynamicStartingRoutes(ctx) s.reconcileExecutorSessionsOnStartup(ctx) if s.workflowStore != nil { s.workflowStore.ReconcileQueuedTasks(ctx) diff --git a/apps/backend/internal/orchestrator/task_operations.go b/apps/backend/internal/orchestrator/task_operations.go index 4002d4072d..4333973797 100644 --- a/apps/backend/internal/orchestrator/task_operations.go +++ b/apps/backend/internal/orchestrator/task_operations.go @@ -348,6 +348,21 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr go func() { bgCtx := context.Background() launchProfileID := agentProfileID + // This workspace-only launch (StartAgent unset) never starts the + // 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. + var claimedRouteGeneration int64 + launchOwned := false + defer func() { + if claimedRouteGeneration > 0 && !launchOwned { + s.markDynamicRouteActionRequired(bgCtx, sessionID, claimedRouteGeneration, "prepared_session_launch_failed") + } + }() if s.profileExecutionResolver != nil { launchSession, loadErr := s.repo.GetTaskSession(bgCtx, sessionID) if loadErr != nil { @@ -367,6 +382,7 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr if resolved.ExecutionProfileID != "" { launchProfileID = resolved.ExecutionProfileID if routeClaimed { + claimedRouteGeneration = resolved.Generation applyResolvedExecution(launchSession, resolved) if updateErr := s.repo.UpdateTaskSession(bgCtx, launchSession); updateErr != nil { s.logger.Warn("failed to persist dynamic route for prepared session", @@ -389,6 +405,7 @@ func (s *Service) PrepareTaskSession(ctx context.Context, taskID string, agentPr zap.Error(launchErr)) return } + launchOwned = true if prepExec != nil { s.ensureSessionPRWatch(bgCtx, taskID, prepExec.SessionID, prepExec.WorktreeBranch) } diff --git a/apps/backend/internal/orchestrator/task_operations_test.go b/apps/backend/internal/orchestrator/task_operations_test.go index 67951eceb0..1b15c67c78 100644 --- a/apps/backend/internal/orchestrator/task_operations_test.go +++ b/apps/backend/internal/orchestrator/task_operations_test.go @@ -3869,6 +3869,64 @@ func TestPrepareTaskSession_WorkspaceLaunchFailureRecovery(t *testing.T) { }) } +// TestPrepareTaskSession_MarksRouteActionRequiredWhenWorkspaceLaunchFails is +// the regression test for review finding F2: PrepareTaskSession's own +// background workspace launch resolves and claims a dynamic route ahead of +// the actual agent start (see resolveExecutionForLaunchSession), but this +// launch never starts the agent — it only reaches "active" later, via +// StartCreatedSession. If it fails here, nothing else revisits the claim, so +// it must not stay "starting" forever. +func TestPrepareTaskSession_MarksRouteActionRequiredWhenWorkspaceLaunchFails(t *testing.T) { + const taskID = "task-dynamic-prepare-launch-fail" + const dynamicProfileID = "profile-dynamic" + const concreteProfileID = "profile-concrete" + + repo := setupTestRepo(t) + ctx := context.Background() + now := time.Now().UTC() + if err := repo.CreateWorkspace(ctx, &models.Workspace{ID: "ws1", Name: "Test", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("CreateWorkspace: %v", err) + } + if err := repo.CreateWorkflow(ctx, &models.Workflow{ID: "wf1", WorkspaceID: "ws1", Name: "Test Workflow", CreatedAt: now, UpdatedAt: now}); err != nil { + t.Fatalf("CreateWorkflow: %v", err) + } + if err := repo.CreateTask(ctx, &models.Task{ + ID: taskID, WorkflowID: "wf1", Title: "Test Task", State: v1.TaskStateInProgress, + CreatedAt: now, UpdatedAt: now, + }); err != nil { + t.Fatalf("CreateTask: %v", err) + } + taskRepo := newMockTaskRepo() + taskRepo.tasks[taskID] = &v1.Task{ID: taskID, WorkspaceID: "ws1", Title: "Test Task", State: v1.TaskStateInProgress} + launchErr := errors.New("workspace launch failed") + agentMgr := &mockAgentManager{ + launchAgentFunc: func(context.Context, *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + return nil, launchErr + }, + } + svc := createTestServiceWithScheduler(repo, newMockStepGetter(), taskRepo, agentMgr) + svc.eventBus = bus.NewMemoryEventBus(testLogger()) + svc.SetProfileExecutionResolver(newWorkflowDynamicProfileResolver(t, dynamicProfileID, concreteProfileID)) + + sessionID, err := svc.PrepareTaskSession(context.Background(), taskID, dynamicProfileID, "", "", "", true) + if err != nil { + t.Fatalf("PrepareTaskSession: %v", err) + } + + require.Eventually(t, func() bool { + session, getErr := repo.GetTaskSession(context.Background(), sessionID) + return getErr == nil && session.RouteState == "action_required" + }, time.Second, 10*time.Millisecond, "expected the claimed dynamic route to reach action_required after the workspace launch failed") + + session, err := repo.GetTaskSession(context.Background(), sessionID) + if err != nil { + t.Fatalf("GetTaskSession: %v", err) + } + if session.RouteGeneration == 0 { + t.Fatal("expected the session to have claimed a route generation before the launch failed") + } +} + func TestHandleSessionLaunchFailure_SkipsTaskFailureWhenSessionTransitionLoses(t *testing.T) { repo := setupTestRepo(t) seedTaskAndSession(t, repo, "task1", "session1", models.TaskSessionStateCancelled) diff --git a/apps/backend/internal/task/repository/sqlite/dynamic_route.go b/apps/backend/internal/task/repository/sqlite/dynamic_route.go index 6876c3636b..a73c3eb4e5 100644 --- a/apps/backend/internal/task/repository/sqlite/dynamic_route.go +++ b/apps/backend/internal/task/repository/sqlite/dynamic_route.go @@ -17,6 +17,7 @@ import ( var _ dynamicruntime.Persistence = (*Repository)(nil) var _ dynamicruntime.ContinuationPersistence = (*Repository)(nil) +var _ dynamicruntime.GenerationStatusClaimer = (*Repository)(nil) func (r *Repository) SaveRouteState(ctx context.Context, state dynamicruntime.RouteState) error { if isTransientRouteSession(state.SessionID) { @@ -83,6 +84,34 @@ func (r *Repository) ClaimRouteState(ctx context.Context, expectedGeneration int return rows == 1, err } +// ClaimRouteStateFrom updates a route generation only while both its +// generation and status still match the caller's observation. Same-generation +// transitions use this narrower fence so a late recovery callback cannot +// overwrite a route that became active (or vice versa). +func (r *Repository) ClaimRouteStateFrom( + ctx context.Context, + expectedGeneration int64, + expectedStatus string, + state dynamicruntime.RouteState, +) (bool, error) { + if isTransientRouteSession(state.SessionID) { + return true, nil + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE dynamic_route_states + SET logical_profile_id = ?, execution_profile_id = ?, + route_generation = ?, profile_version = ?, state = ?, continuation_json = ?, policy_state_json = ?, updated_at = ? + WHERE session_id = ? AND route_generation = ? AND state = ? + `), state.LogicalProfileID, state.ExecutionProfileID, + state.Generation, state.ProfileVersion, state.Status, state.ContinuationJSON, state.PolicyStateJSON, state.UpdatedAt, + state.SessionID, expectedGeneration, expectedStatus) + if err != nil { + return false, err + } + rows, err := result.RowsAffected() + return rows == 1, err +} + // RecordRouteDecision commits the generation claim and immutable attempt row // together. A stale claim returns dynamicruntime.ErrStaleGeneration. func (r *Repository) RecordRouteDecision(ctx context.Context, decision dynamicruntime.RouteDecision, state dynamicruntime.RouteState) error { @@ -220,6 +249,40 @@ func (r *Repository) ListPendingRouteStates(ctx context.Context) ([]dynamicrunti return states, nil } +// ListStartingRouteStates returns every route whose durable status is still +// "starting". Startup reconciliation is the only caller: a healthy route also +// passes through "starting" en route to "active", so listing this status is +// only a safe orphan signal once the caller has confirmed there is no live +// session backing the row (see Service.reconcileOrphanedDynamicStartingRoutes). +func (r *Repository) ListStartingRouteStates(ctx context.Context) ([]dynamicruntime.RouteState, error) { + rows, err := r.ro.QueryContext(ctx, r.ro.Rebind(` + SELECT session_id, logical_profile_id, execution_profile_id, + route_generation, profile_version, state, continuation_json, policy_state_json, updated_at + FROM dynamic_route_states + WHERE state = ? ORDER BY updated_at ASC + `), "starting") + if err != nil { + return nil, err + } + defer func() { _ = rows.Close() }() + states := make([]dynamicruntime.RouteState, 0) + for rows.Next() { + var state dynamicruntime.RouteState + if err := rows.Scan( + &state.SessionID, &state.LogicalProfileID, &state.ExecutionProfileID, + &state.Generation, &state.ProfileVersion, &state.Status, + &state.ContinuationJSON, &state.PolicyStateJSON, &state.UpdatedAt, + ); err != nil { + return nil, err + } + states = append(states, state) + } + if err := rows.Err(); err != nil { + return nil, err + } + return states, nil +} + func (r *Repository) SaveRouteContinuation(ctx context.Context, record dynamicruntime.ContinuationRecord) error { if isTransientRouteSession(record.SessionID) { return nil diff --git a/apps/backend/internal/task/repository/sqlite/dynamic_route_test.go b/apps/backend/internal/task/repository/sqlite/dynamic_route_test.go index fdb804f979..30911ae72a 100644 --- a/apps/backend/internal/task/repository/sqlite/dynamic_route_test.go +++ b/apps/backend/internal/task/repository/sqlite/dynamic_route_test.go @@ -139,6 +139,37 @@ func TestListPendingRouteStatesExcludesAmbiguousRetryingStates(t *testing.T) { } } +func TestListStartingRouteStatesReturnsOnlyStartingRows(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + now := time.Now().UTC() + for _, sessionID := range []string{"starting-orphan", "active-session", "retrying-session"} { + taskID := "task-" + sessionID + if err := repo.CreateTask(ctx, &models.Task{ID: taskID, Title: sessionID}); err != nil { + t.Fatalf("CreateTask(%s): %v", taskID, err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ID: sessionID, TaskID: taskID, State: models.TaskSessionStateWaitingForInput}); err != nil { + t.Fatalf("CreateTaskSession(%s): %v", sessionID, err) + } + } + for _, state := range []dynamicruntime.RouteState{ + {SessionID: "starting-orphan", LogicalProfileID: "dynamic", Generation: 1, Status: "starting", UpdatedAt: now}, + {SessionID: "active-session", LogicalProfileID: "dynamic", Generation: 1, Status: "active", UpdatedAt: now.Add(time.Second)}, + {SessionID: "retrying-session", LogicalProfileID: "dynamic", Generation: 1, Status: "retrying", UpdatedAt: now.Add(2 * time.Second)}, + } { + if err := repo.SaveRouteState(ctx, state); err != nil { + t.Fatalf("SaveRouteState(%s): %v", state.SessionID, err) + } + } + states, err := repo.ListStartingRouteStates(ctx) + if err != nil { + t.Fatalf("ListStartingRouteStates: %v", err) + } + if len(states) != 1 || states[0].SessionID != "starting-orphan" { + t.Fatalf("starting states = %#v", states) + } +} + func TestClaimRouteStateAdvancesGenerationWithFence(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() @@ -185,6 +216,55 @@ func TestClaimRouteStateAdvancesGenerationWithFence(t *testing.T) { } } +func TestClaimRouteStateFromRequiresExpectedStatus(t *testing.T) { + ctx := context.Background() + repo := newRepoForSessionTests(t) + if err := repo.CreateTask(ctx, &models.Task{ID: "task-conditional-status"}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "conditional-status-session", TaskID: "task-conditional-status", + State: models.TaskSessionStateStarting, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + initial := dynamicruntime.RouteState{ + SessionID: "conditional-status-session", LogicalProfileID: "logical", + ExecutionProfileID: "candidate", Generation: 1, Status: "starting", + UpdatedAt: time.Now().UTC(), + } + claimed, err := repo.ClaimRouteState(ctx, 0, initial) + if err != nil || !claimed { + t.Fatalf("initial ClaimRouteState = %v, %v", claimed, err) + } + + active := initial + active.Status = "active" + active.UpdatedAt = time.Now().UTC() + claimed, err = repo.ClaimRouteStateFrom(ctx, 1, "starting", active) + if err != nil || !claimed { + t.Fatalf("conditional active transition = %v, %v", claimed, err) + } + + stale := active + stale.Status = "action_required" + stale.UpdatedAt = time.Now().UTC() + claimed, err = repo.ClaimRouteStateFrom(ctx, 1, "starting", stale) + if err != nil { + t.Fatalf("stale conditional transition: %v", err) + } + if claimed { + t.Fatal("conditional transition succeeded after status changed") + } + loaded, err := repo.LoadRouteState(ctx, initial.SessionID) + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if loaded == nil || loaded.Status != "active" { + t.Fatalf("route state after stale transition = %#v, want active", loaded) + } +} + func TestRecordRouteDecisionClaimsInitialGenerationAndWritesAttemptAtomically(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index e5ebc88e40..b440e94788 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -1568,6 +1568,32 @@ func (r *Repository) UpdateTaskSessionStateIfCurrent( return rows > 0, now, nil } +// UpdateTaskSessionDynamicRouteIfCurrent changes only the route projection +// while the session generation and projected route state still match the +// caller's observation. It avoids writing a stale full session row after an +// asynchronous launch or recovery callback. +func (r *Repository) UpdateTaskSessionDynamicRouteIfCurrent( + ctx context.Context, + id string, + expectedGeneration int64, + expectedRouteState, routeState, routeReason string, +) (bool, time.Time, error) { + now := time.Now().UTC() + result, err := r.db.ExecContext(ctx, r.db.Rebind(` + UPDATE task_sessions + SET route_state = ?, route_reason = ?, updated_at = ? + WHERE id = ? AND route_generation = ? AND route_state = ? + `), routeState, routeReason, now, id, expectedGeneration, expectedRouteState) + if err != nil { + return false, time.Time{}, err + } + rows, err := result.RowsAffected() + if err != nil { + return false, time.Time{}, err + } + return rows > 0, now, nil +} + // CancelActiveTaskSession atomically transitions one active session to // CANCELLED. A false result means the row exists in a non-active state or was // concurrently changed before this conditional write; callers re-read to