fix(backend): recover dynamic agent routes stuck in starting with no owner (#3362)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-04 17:42:52 +00:00
committed by GitHub
parent e4eb278f21
commit f91fd4611e
18 changed files with 1730 additions and 23 deletions
@@ -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()
@@ -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)
@@ -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.
@@ -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,
@@ -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
@@ -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()
@@ -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)
}
}
@@ -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)
}
@@ -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)
}
@@ -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.
@@ -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)
}
}()
}
@@ -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{
@@ -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)
@@ -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)
}
@@ -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)
@@ -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
@@ -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()
@@ -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