diff --git a/apps/backend/internal/agent/runtime/dynamic/engine.go b/apps/backend/internal/agent/runtime/dynamic/engine.go index 63c021c72a..08b2656699 100644 --- a/apps/backend/internal/agent/runtime/dynamic/engine.go +++ b/apps/backend/internal/agent/runtime/dynamic/engine.go @@ -17,6 +17,7 @@ const ( routeStatusStarting = "starting" routeStatusActive = "active" routeStatusActionRequired = "action_required" + routeStatusRetrying = "retrying" ) func WithClock(now func() time.Time) EngineOption { @@ -576,6 +577,26 @@ func (e *Engine) persistInitialGeneration( return true, nil } +// persistExpectedStatus atomically updates a same-generation route status. +// Persistence without status-fencing support fails closed. +func (e *Engine) persistExpectedStatus(ctx context.Context, expectedGeneration int64, expectedStatus string, state RouteState) error { + if e.persistence == nil { + return nil + } + claimer, ok := e.persistence.(GenerationStatusClaimer) + if !ok { + return ErrStatusClaimUnsupported + } + claimed, err := claimer.ClaimRouteStateFrom(ctx, expectedGeneration, expectedStatus, state) + if err != nil { + return err + } + if !claimed { + return ErrStaleGeneration + } + return nil +} + func mustJSON(value PolicyState) []byte { payload, err := json.Marshal(value) if err != nil { @@ -725,6 +746,27 @@ func (e *Engine) MarkActionRequired( }, nil } +// MarkRecoveryActionRequired returns an in-flight route to manual recovery. +// Stale generations and states that already moved on are no-ops. +func (e *Engine) MarkRecoveryActionRequired(ctx context.Context, sessionID string, expectedGeneration int64) error { + state, exists, err := e.stateForFailure(ctx, sessionID) + if err != nil { + return err + } + if !exists || state.Generation != expectedGeneration || state.Status != routeStatusRetrying { + return nil + } + state.Status = routeStatusActionRequired + state.UpdatedAt = e.now() + if err := e.persistExpectedStatus(ctx, expectedGeneration, routeStatusRetrying, state); err != nil { + return err + } + e.mu.Lock() + e.states[sessionID] = state + e.mu.Unlock() + return nil +} + func (e *Engine) resumePending( ctx context.Context, sessionID string, @@ -741,12 +783,13 @@ func (e *Engine) resumePending( if state.Generation != expectedGeneration { return RouteDecision{}, ErrStaleGeneration } - if force && state.Status == "retrying" { + if force && state.Status == routeStatusRetrying { return RouteDecision{}, ErrRecoveryPending } if !force && state.Status != string(routingpolicy.DecisionRetry) && state.Status != string(routingpolicy.DecisionWaitForReset) { return RouteDecision{}, ErrRecoveryPending } + observedStatus := state.Status var policyState PolicyState if state.PolicyStateJSON != "" { if err := json.Unmarshal([]byte(state.PolicyStateJSON), &policyState); err != nil { @@ -757,10 +800,9 @@ func (e *Engine) resumePending( if !force && policyState.Deadline != nil && now.Before(*policyState.Deadline) { return RouteDecision{}, ErrRecoveryNotDue } - expectedStatus := state.Status - state.Status = "retrying" + state.Status = routeStatusRetrying state.UpdatedAt = now - if err := e.persistSameGeneration(ctx, expectedGeneration, expectedStatus, state); err != nil { + if err := e.persistExpectedStatus(ctx, expectedGeneration, observedStatus, state); err != nil { return RouteDecision{}, err } e.mu.Lock() diff --git a/apps/backend/internal/agent/runtime/dynamic/engine_test.go b/apps/backend/internal/agent/runtime/dynamic/engine_test.go index cce722554e..221a50c81f 100644 --- a/apps/backend/internal/agent/runtime/dynamic/engine_test.go +++ b/apps/backend/internal/agent/runtime/dynamic/engine_test.go @@ -404,6 +404,86 @@ func TestEngineMarkActionRequiredIsNoOpOnceRouteIsActive(t *testing.T) { } } +func TestEngineMarkRecoveryActionRequiredUnsticksManualRecoveryAfterFailedLaunch(t *testing.T) { + engine := NewEngine() + document := routingpolicy.DefaultDocument() + document.Transient.Retry = routingpolicy.RetryPolicy{Enabled: true, MaxRetries: 1, InitialIntervalSeconds: 60} + profile := Profile{ID: "launch-fail-policy", Version: 1, Candidates: []Candidate{ + {ID: "first", Enabled: true, Policies: document}, + }} + initial, err := engine.Select("launch-fail-session", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + if _, err := engine.ApplyFailure( + "launch-fail-session", profile, initial.Generation, initial.ExecutionProfileID, + &routingerr.Error{Code: routingerr.CodeRateLimited, Class: routingerr.ClassTransient, FallbackAllowed: true}, + ); !errors.Is(err, ErrRecoveryPending) { + t.Fatalf("ApplyFailure error = %v, want recovery pending", err) + } + ctx := context.Background() + if _, err := engine.ResumePendingNow(ctx, "launch-fail-session", initial.Generation); err != nil { + t.Fatalf("ResumePendingNow (simulated timer/manual claim): %v", err) + } + + // The durable state is now "retrying" — as if a launch was claimed but + // then failed before it could advance the state further. Every manual + // recovery path must reject it until the launch-failure handler syncs it. + if _, err := engine.ResumePendingNow(ctx, "launch-fail-session", initial.Generation); !errors.Is(err, ErrRecoveryPending) { + t.Fatalf("ResumePendingNow while stuck at retrying = %v, want recovery pending", err) + } + if _, err := engine.CancelPending(ctx, "launch-fail-session", initial.Generation, "manual_stop"); !errors.Is(err, ErrRecoveryPending) { + t.Fatalf("CancelPending while stuck at retrying = %v, want recovery pending", err) + } + + if err := engine.MarkRecoveryActionRequired(ctx, "launch-fail-session", initial.Generation); err != nil { + t.Fatalf("MarkRecoveryActionRequired: %v", err) + } + state, exists := engine.State("launch-fail-session") + if !exists || state.Status != routeStatusActionRequired || state.Generation != initial.Generation { + t.Fatalf("state after MarkRecoveryActionRequired = %#v, exists=%v", state, exists) + } + + // Manual retry and stop are both unblocked now that the durable state + // reflects action_required instead of a launch that will never resolve. + if stopped, err := engine.CancelPending(ctx, "launch-fail-session", initial.Generation, "manual_stop"); err != nil { + t.Fatalf("CancelPending after sync: %v", err) + } else if stopped.Status != routeStatusActionRequired { + t.Fatalf("stopped decision = %#v", stopped) + } +} + +func TestEngineMarkRecoveryActionRequiredIsNoOpWhenStateMovedOn(t *testing.T) { + engine := NewEngine() + profile := Profile{ID: "no-op-policy", Version: 1, Candidates: []Candidate{ + {ID: "first", Enabled: true}, + }} + ctx := context.Background() + initial, err := engine.Select("no-op-session", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + + // Never entered "retrying" — no-op, not an error. + if err := engine.MarkRecoveryActionRequired(ctx, "no-op-session", initial.Generation); err != nil { + t.Fatalf("MarkRecoveryActionRequired on non-retrying state: %v", err) + } + state, _ := engine.State("no-op-session") + if state.Status == routeStatusActionRequired { + t.Fatalf("state unexpectedly advanced to action_required: %#v", state) + } + + // Stale generation — no-op, not an error. + if err := engine.MarkRecoveryActionRequired(ctx, "no-op-session", initial.Generation+1); err != nil { + t.Fatalf("MarkRecoveryActionRequired on stale generation: %v", err) + } + + // Unknown session — no-op, not an error. + if err := engine.MarkRecoveryActionRequired(ctx, "unknown-session", 1); err != nil { + t.Fatalf("MarkRecoveryActionRequired on unknown session: %v", err) + } +} + func TestBindingFingerprinterUsesOpaqueStableHMACKeys(t *testing.T) { key := []byte("installation-secret") fingerprinter := NewBindingFingerprinter(key) diff --git a/apps/backend/internal/agent/runtime/dynamic/types.go b/apps/backend/internal/agent/runtime/dynamic/types.go index 6112eab65d..8283f7dbae 100644 --- a/apps/backend/internal/agent/runtime/dynamic/types.go +++ b/apps/backend/internal/agent/runtime/dynamic/types.go @@ -27,6 +27,11 @@ var ( ErrRouteStateNotFound = errors.New("dynamic route state not found") ErrRecoveryPending = errors.New("dynamic route recovery is pending") ErrRecoveryNotDue = errors.New("dynamic route recovery is not due") + // ErrStatusClaimUnsupported means the persistence implementation cannot + // fence a same-generation status transition. Status-fenced claims must + // fail closed rather than silently degrade to a generation-only claim, + // which would let two concurrent callers both win the same transition. + ErrStatusClaimUnsupported = errors.New("dynamic route persistence does not support status-fenced claims") ) type Candidate struct { @@ -134,7 +139,8 @@ type GenerationClaimer interface { // GenerationStatusClaimer updates one generation only while its status still // matches the caller's observation. This closes the same-generation race -// between a launch becoming active and a concurrent recovery transition. +// between a launch becoming active and a concurrent recovery transition, and +// between two concurrent callers observing the same due retry/wait state. type GenerationStatusClaimer interface { ClaimRouteStateFrom(context.Context, int64, string, RouteState) (bool, error) } diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver.go b/apps/backend/internal/agent/runtime/dynamic_resolver.go index f5f947852c..c1accb099f 100644 --- a/apps/backend/internal/agent/runtime/dynamic_resolver.go +++ b/apps/backend/internal/agent/runtime/dynamic_resolver.go @@ -363,7 +363,7 @@ func (r *ProfileExecutionResolver) resolveRetryRouteAction( if resumeErr == nil { return r.executionFromDecision(ctx, profileID, sessionID, decision) } - if !errors.Is(resumeErr, dynamic.ErrRecoveryPending) && !errors.Is(resumeErr, dynamic.ErrRouteStateNotFound) { + if !errors.Is(resumeErr, dynamic.ErrRouteStateNotFound) { return ProfileExecution{}, resumeErr } } @@ -375,6 +375,11 @@ func (r *ProfileExecutionResolver) resolveSkipRouteAction( sessionID, profileID, currentExecutionProfileID string, expectedGeneration int64, ) (ProfileExecution, error) { + if state, exists, err := r.engine.LoadState(ctx, sessionID); err != nil { + return ProfileExecution{}, err + } else if exists && state.Generation == expectedGeneration && state.Status == "retrying" { + return ProfileExecution{}, dynamic.ErrRecoveryPending + } profileConfig, err := r.loadDynamicProfile(ctx, profileID) if err != nil { return ProfileExecution{}, err @@ -442,6 +447,19 @@ func (r *ProfileExecutionResolver) ResumePendingRoute( return r.executionFromDecision(ctx, state.LogicalProfileID, sessionID, decision) } +// MarkRouteRecoveryActionRequired returns a claimed "retrying" route to +// manual recovery after its resumed launch failed. +func (r *ProfileExecutionResolver) MarkRouteRecoveryActionRequired( + ctx context.Context, + sessionID string, + expectedGeneration int64, +) error { + if r.engine == nil { + return nil + } + return r.engine.MarkRecoveryActionRequired(ctx, sessionID, expectedGeneration) +} + func (r *ProfileExecutionResolver) resolve( ctx context.Context, sessionID, profileID string, diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver_test.go b/apps/backend/internal/agent/runtime/dynamic_resolver_test.go new file mode 100644 index 0000000000..87e14bf658 --- /dev/null +++ b/apps/backend/internal/agent/runtime/dynamic_resolver_test.go @@ -0,0 +1,132 @@ +package runtime + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/agent/agents" + "github.com/kandev/kandev/internal/agent/runtime/dynamic" + "github.com/kandev/kandev/internal/agent/runtime/routingerr" + "github.com/kandev/kandev/internal/agent/runtime/routingpolicy" + agentsettingsmodels "github.com/kandev/kandev/internal/agent/settings/models" + "github.com/kandev/kandev/internal/agent/settings/store" +) + +func TestResolveRouteActionRejectsManualRetryAfterRecoveryClaim(t *testing.T) { + ctx := context.Background() + resolver, engine, generation := newRetryingRouteActionResolver(t) + _, err := resolver.ResolveRouteAction( + ctx, "session-retry", "dynamic-profile", "concrete-profile", generation, "retry", + ) + if !errors.Is(err, dynamic.ErrRecoveryPending) { + t.Fatalf("ResolveRouteAction error = %v, want recovery pending", err) + } + state, ok := engine.State("session-retry") + if !ok || state.Generation != generation || state.Status != "retrying" { + t.Fatalf("route state = %#v, ok=%v, want unchanged claimed state", state, ok) + } +} + +func TestResolveRouteActionRejectsSkipAfterRecoveryClaim(t *testing.T) { + resolver, _, generation := newRetryingRouteActionResolver(t) + _, err := resolver.ResolveRouteAction( + context.Background(), "session-retry", "dynamic-profile", "concrete-profile", generation, "skip", + ) + if !errors.Is(err, dynamic.ErrRecoveryPending) { + t.Fatalf("ResolveRouteAction error = %v, want recovery pending", err) + } +} + +func TestMarkRouteActionRequiredUnblocksRetryAndSkipAfterFailedLaunch(t *testing.T) { + ctx := context.Background() + resolver, engine, generation := newRetryingRouteActionResolver(t) + + // Simulate the launch-failure handler's sync (dynamic_policy_recovery.go + // and dynamic_routing.go both call this after a resumed launch fails). + if err := resolver.MarkRouteRecoveryActionRequired(ctx, "session-retry", generation); err != nil { + t.Fatalf("MarkRouteRecoveryActionRequired: %v", err) + } + state, ok := engine.State("session-retry") + if !ok || state.Status != "action_required" { + t.Fatalf("route state after sync = %#v, ok=%v, want action_required", state, ok) + } + + if _, err := resolver.ResolveRouteAction( + ctx, "session-retry", "dynamic-profile", "concrete-profile", generation, "retry", + ); err != nil { + t.Fatalf("ResolveRouteAction retry after sync = %v, want success", err) + } +} + +func newRetryingRouteActionResolver(t *testing.T) (*ProfileExecutionResolver, *dynamic.Engine, int64) { + t.Helper() + ctx := context.Background() + engine := dynamic.NewEngine() + document := routingpolicy.DefaultDocument() + document.Transient.Retry = routingpolicy.RetryPolicy{ + Enabled: true, MaxRetries: 1, InitialIntervalSeconds: 60, + } + profile := dynamic.Profile{ID: "dynamic-profile", Version: 1, Candidates: []dynamic.Candidate{ + {ID: "concrete-profile", Enabled: true, Policies: document}, + }} + initial, err := engine.Select("session-retry", profile, 0, "") + if err != nil { + t.Fatalf("initial Select: %v", err) + } + if _, err := engine.ApplyFailure( + "session-retry", profile, initial.Generation, initial.ExecutionProfileID, + &routingerr.Error{Code: routingerr.CodeRateLimited, Class: routingerr.ClassTransient, FallbackAllowed: true}, + ); !errors.Is(err, dynamic.ErrRecoveryPending) { + t.Fatalf("ApplyFailure error = %v, want recovery pending", err) + } + if _, err := engine.ResumePendingNow(ctx, "session-retry", initial.Generation); err != nil { + t.Fatalf("ResumePendingNow: %v", err) + } + + profiles := &dynamicResolverTestProfiles{ + logical: &agentsettingsmodels.AgentProfile{ID: "dynamic-profile", AgentID: agents.DynamicAgentID, Enabled: true}, + concrete: &agentsettingsmodels.AgentProfile{ID: "concrete-profile", AgentID: "concrete", Enabled: true}, + dynamic: &agentsettingsmodels.DynamicAgentProfile{ProfileID: "dynamic-profile", Version: 1}, + routes: []agentsettingsmodels.DynamicAgentRoute{{ + DynamicProfileID: "dynamic-profile", ExecutionProfileID: "concrete-profile", Enabled: true, + }}, + } + return NewProfileExecutionResolver(profiles, engine, true), engine, initial.Generation +} + +type dynamicResolverTestProfiles struct { + store.Repository + store.DynamicProfileRepository + logical *agentsettingsmodels.AgentProfile + concrete *agentsettingsmodels.AgentProfile + dynamic *agentsettingsmodels.DynamicAgentProfile + routes []agentsettingsmodels.DynamicAgentRoute +} + +func (p *dynamicResolverTestProfiles) GetAgentProfile(_ context.Context, id string) (*agentsettingsmodels.AgentProfile, error) { + switch id { + case p.logical.ID: + return p.logical, nil + case p.concrete.ID: + return p.concrete, nil + default: + return nil, errors.New("profile not found") + } +} + +func (p *dynamicResolverTestProfiles) GetAgent(_ context.Context, id string) (*agentsettingsmodels.Agent, error) { + if id == agents.DynamicAgentID { + return &agentsettingsmodels.Agent{ID: id, Name: agents.DynamicAgentID}, nil + } + return &agentsettingsmodels.Agent{ID: id, Name: "concrete"}, nil +} + +func (p *dynamicResolverTestProfiles) GetDynamicAgentProfile( + _ context.Context, profileID string, +) (*agentsettingsmodels.DynamicAgentProfile, []agentsettingsmodels.DynamicAgentRoute, error) { + if profileID != p.dynamic.ProfileID { + return nil, nil, errors.New("dynamic profile not found") + } + return p.dynamic, p.routes, nil +} diff --git a/apps/backend/internal/backendapp/dynamic_routing.go b/apps/backend/internal/backendapp/dynamic_routing.go index cc6b8a0d08..cbebee9a69 100644 --- a/apps/backend/internal/backendapp/dynamic_routing.go +++ b/apps/backend/internal/backendapp/dynamic_routing.go @@ -47,6 +47,11 @@ func applyDynamicRouteAction( if err != nil { return nil, err } + if err := repairDynamicRouteAfterLaunchFailure( + ctx, session, request.ExpectedGeneration, resolver.MarkRouteRecoveryActionRequired, + ); err != nil { + return nil, routeActionError(ctx, repo, session, err) + } decision, err := resolveDynamicRouteAction(ctx, resolver, request, session) if err != nil { return handleDynamicRouteActionError(ctx, repo, session, err) @@ -57,7 +62,7 @@ func applyDynamicRouteAction( if request.Action == orchestrator.RouteActionCancelWait || request.Action == orchestrator.RouteActionStop { return routeActionResult(ctx, repo, session), nil } - return finishDynamicRouteAction(ctx, repo, session.ID, launchSuccessor) + return finishDynamicRouteAction(ctx, repo, resolver, session.ID, decision.Generation, launchSuccessor) } func loadDynamicRouteActionSession( @@ -155,7 +160,9 @@ func persistDynamicRouteSelection( func finishDynamicRouteAction( ctx context.Context, repo *sqliterepo.Repository, + resolver *agentruntime.ProfileExecutionResolver, sessionID string, + expectedGeneration int64, launchSuccessor func(context.Context, string) error, ) (*orchestrator.RouteActionResult, error) { if launchSuccessor == nil { @@ -166,7 +173,7 @@ func finishDynamicRouteAction( return routeActionResult(ctx, repo, session), nil } if err := launchSuccessor(ctx, sessionID); err != nil { - return recoverDynamicRouteAction(ctx, repo, sessionID, err) + return recoverDynamicRouteAction(ctx, repo, resolver, sessionID, expectedGeneration, err) } session, err := repo.GetTaskSession(ctx, sessionID) if err != nil { @@ -178,15 +185,23 @@ func finishDynamicRouteAction( func recoverDynamicRouteAction( ctx context.Context, repo *sqliterepo.Repository, + resolver *agentruntime.ProfileExecutionResolver, sessionID string, + expectedGeneration int64, launchErr error, ) (*orchestrator.RouteActionResult, error) { + if resolver != nil { + _ = resolver.MarkRouteRecoveryActionRequired(ctx, sessionID, expectedGeneration) + } session, err := repo.GetTaskSession(ctx, sessionID) if err != nil { return nil, err } + if session.RouteGeneration != expectedGeneration { + return routeActionResult(ctx, repo, session), nil + } session.RouteState = "action_required" - session.RouteReason = "route_action_launch_failed" + session.RouteReason = orchestrator.RouteActionLaunchFailedReason session.State = models.TaskSessionStateWaitingForInput session.ErrorMessage = launchErr.Error() session.DownstreamACPSessionID = "" @@ -196,6 +211,23 @@ func recoverDynamicRouteAction( return routeActionResult(ctx, repo, session), nil } +func repairDynamicRouteAfterLaunchFailure( + ctx context.Context, + session *models.TaskSession, + expectedGeneration int64, + mark func(context.Context, string, int64) error, +) error { + if session == nil || mark == nil || + session.RouteReason != orchestrator.RouteActionLaunchFailedReason || + session.RouteGeneration != expectedGeneration { + return nil + } + if err := mark(ctx, session.ID, expectedGeneration); err != nil { + return fmt.Errorf("restore dynamic route after failed launch: %w", err) + } + return nil +} + func routeActionResult( ctx context.Context, repo *sqliterepo.Repository, diff --git a/apps/backend/internal/backendapp/dynamic_routing_test.go b/apps/backend/internal/backendapp/dynamic_routing_test.go new file mode 100644 index 0000000000..93afafea50 --- /dev/null +++ b/apps/backend/internal/backendapp/dynamic_routing_test.go @@ -0,0 +1,67 @@ +package backendapp + +import ( + "context" + "errors" + "testing" + + "github.com/kandev/kandev/internal/orchestrator" + "github.com/kandev/kandev/internal/task/models" +) + +func TestRepairDynamicRouteAfterLaunchFailureSurfacesMarkerError(t *testing.T) { + session := &models.TaskSession{ + ID: "session-recovery", + RouteGeneration: 7, + RouteReason: orchestrator.RouteActionLaunchFailedReason, + } + markerErr := errors.New("database unavailable") + calls := 0 + err := repairDynamicRouteAfterLaunchFailure( + context.Background(), session, session.RouteGeneration, + func(context.Context, string, int64) error { + calls++ + return markerErr + }, + ) + if !errors.Is(err, markerErr) { + t.Fatalf("repair error = %v, want marker error", err) + } + if calls != 1 { + t.Fatalf("marker calls = %d, want one retry", calls) + } +} + +func TestRepairDynamicRouteAfterLaunchFailureIgnoresUnrelatedProjection(t *testing.T) { + tests := []struct { + name string + reason string + generation int64 + }{ + {name: "different reason", reason: "manual_retry", generation: 7}, + {name: "stale generation", reason: orchestrator.RouteActionLaunchFailedReason, generation: 8}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + calls := 0 + session := &models.TaskSession{ + ID: "session-recovery", + RouteGeneration: 7, + RouteReason: tt.reason, + } + err := repairDynamicRouteAfterLaunchFailure( + context.Background(), session, tt.generation, + func(context.Context, string, int64) error { + calls++ + return nil + }, + ) + if err != nil { + t.Fatalf("repair error = %v, want nil", err) + } + if calls != 0 { + t.Fatalf("marker calls = %d, want zero", calls) + } + }) + } +} diff --git a/apps/backend/internal/orchestrator/dynamic_policy_recovery.go b/apps/backend/internal/orchestrator/dynamic_policy_recovery.go index e3c872815b..eeb8691348 100644 --- a/apps/backend/internal/orchestrator/dynamic_policy_recovery.go +++ b/apps/backend/internal/orchestrator/dynamic_policy_recovery.go @@ -157,6 +157,18 @@ func (s *Service) runDynamicPolicyRecovery(ctx context.Context, sessionID string if !ok { return } + // Serialize route recovery with manual actions through the durable + // projection. Release before lifecycle launch because launch takes the + // session lifecycle lock. + lock, release := s.acquireCancelInFlightGuard(sessionID) + lock.Lock() + locked := true + defer func() { + if locked { + lock.Unlock() + release() + } + }() state, ok := loadDueDynamicPolicyState(ctx, loader, sessionID, generation) if !ok { return @@ -168,7 +180,15 @@ func (s *Service) runDynamicPolicyRecovery(ctx context.Context, sessionID string if s.handleDynamicPolicyResumeError(ctx, loader, sessionID, generation, err) { return } - s.persistDynamicPolicyRecovery(ctx, sessionID, generation, resolved) + if !s.persistDynamicPolicyRecovery(ctx, sessionID, generation, resolved) { + return + } + lock.Unlock() + release() + locked = false + if err := s.LaunchDynamicRouteAction(ctx, sessionID); err != nil { + s.markDynamicPolicyRecoveryActionRequired(ctx, sessionID, generation, err) + } } func (s *Service) removeDynamicPolicyRecoveryTimer(sessionID string) { @@ -234,10 +254,10 @@ func (s *Service) persistDynamicPolicyRecovery( sessionID string, generation int64, resolved agentruntime.ProfileExecution, -) { +) bool { session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil || session == nil || session.RouteGeneration != generation || session.ExecutionProfileID != resolved.ExecutionProfileID { - return + return false } applyDynamicRouteDecisionProjection(session, resolved.Decision) session.RouteState = resolved.Decision.Status @@ -246,11 +266,9 @@ func (s *Service) persistDynamicPolicyRecovery( if err := s.repo.UpdateTaskSession(ctx, session); err != nil { s.logger.Warn("failed to persist dynamic policy retry state", zap.String("session_id", sessionID), zap.Error(err)) - return - } - if err := s.LaunchDynamicRouteAction(ctx, sessionID); err != nil { - s.markDynamicPolicyRecoveryActionRequired(ctx, sessionID, err) + return false } + return true } func dynamicPolicyDeadline(rawPolicyState string) *time.Time { @@ -262,13 +280,24 @@ func dynamicPolicyDeadline(rawPolicyState string) *time.Time { return &deadline } -func (s *Service) markDynamicPolicyRecoveryActionRequired(ctx context.Context, sessionID string, launchErr error) { +func (s *Service) markDynamicPolicyRecoveryActionRequired( + ctx context.Context, sessionID string, generation int64, launchErr error, +) { + if s.profileExecutionResolver != nil { + if err := s.profileExecutionResolver.MarkRouteRecoveryActionRequired(ctx, sessionID, generation); err != nil { + s.logger.Warn("failed to sync durable route state to action_required after launch failure", + zap.String("session_id", sessionID), zap.Error(err)) + } + } session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil || session == nil { return } + if session.RouteGeneration != generation { + return + } session.RouteState = "action_required" - session.RouteReason = "route_action_launch_failed" + session.RouteReason = RouteActionLaunchFailedReason session.State = models.TaskSessionStateWaitingForInput session.ErrorMessage = launchErr.Error() session.DownstreamACPSessionID = "" diff --git a/apps/backend/internal/orchestrator/dynamic_policy_recovery_race_test.go b/apps/backend/internal/orchestrator/dynamic_policy_recovery_race_test.go new file mode 100644 index 0000000000..e736050cd4 --- /dev/null +++ b/apps/backend/internal/orchestrator/dynamic_policy_recovery_race_test.go @@ -0,0 +1,171 @@ +package orchestrator + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/jmoiron/sqlx" + _ "github.com/mattn/go-sqlite3" + + agentruntime "github.com/kandev/kandev/internal/agent/runtime" + dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic" + agentsettingsmodels "github.com/kandev/kandev/internal/agent/settings/models" + agentsettingsstore "github.com/kandev/kandev/internal/agent/settings/store" + "github.com/kandev/kandev/internal/orchestrator/executor" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + wfmodels "github.com/kandev/kandev/internal/workflow/models" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestConcurrentDynamicPolicyRecoveryLaunchesExactlyOnce reproduces two +// recovery-timer fires racing on the same due generation (e.g. a duplicate +// schedule or a backend-restart reschedule overlapping a still-live timer). +// Before this fix, runDynamicPolicyRecovery held no per-session guard and +// Engine.resumePending's durable claim predicate did not include the observed +// status, so both callers could claim the same retry_wait -> retrying +// transition and both launch a successor, delivering the same prompt twice. +// With the exact-status CAS (ClaimRouteStatus) plus the shared +// acquireCancelInFlightGuard held through the claim and session projection, +// exactly one caller claims the transition and only it reaches the launch. +func TestConcurrentDynamicPolicyRecoveryLaunchesExactlyOnce(t *testing.T) { + ctx := context.Background() + const ( + taskID = "task-policy-race" + sessionID = "session-policy-race" + executionID = "execution-policy-race" + dynamicProfileID = "profile-dynamic-race" + concreteProfileID = "profile-concrete-race" + ) + + repo := setupTestRepo(t) + seedSession(t, repo, taskID, sessionID, "step-race") + + resolver := newDynamicPolicyRaceResolver(t, repo, dynamicProfileID, concreteProfileID) + + session, err := repo.GetTaskSession(ctx, sessionID) + if err != nil { + t.Fatalf("get session: %v", err) + } + session.AgentProfileID = dynamicProfileID + session.ExecutionProfileID = concreteProfileID + session.RouteGeneration = 4 + if err := repo.UpdateTaskSession(ctx, session); err != nil { + t.Fatalf("update session: %v", err) + } + seedExecutorRunning(t, repo, sessionID, taskID, executionID) + + elapsedDeadline := time.Now().UTC().Add(-time.Minute) + if err := repo.SaveRouteState(ctx, dynamicruntime.RouteState{ + SessionID: sessionID, LogicalProfileID: dynamicProfileID, + ExecutionProfileID: concreteProfileID, Generation: 4, ProfileVersion: 1, + Status: "retry_wait", + PolicyStateJSON: `{"deadline":"` + elapsedDeadline.Format(time.RFC3339Nano) + `"}`, + UpdatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("SaveRouteState: %v", err) + } + if _, _, err := resolver.EngineForTest().LoadState(ctx, sessionID); err != nil { + t.Fatalf("warm engine cache: %v", err) + } + + var launches int32 + agentManager := &mockAgentManager{ + launchAgentFunc: func(_ context.Context, req *executor.LaunchAgentRequest) (*executor.LaunchAgentResponse, error) { + atomic.AddInt32(&launches, 1) + return &executor.LaunchAgentResponse{AgentExecutionID: "relaunch-" + req.SessionID}, nil + }, + } + taskRepo := newMockTaskRepo() + seedMockTaskState(taskRepo, taskID, v1.TaskStateInProgress) + stepGetter := newMockStepGetter() + stepGetter.steps["step-race"] = &wfmodels.WorkflowStep{ID: "step-race", WorkflowID: "wf1", Name: "Work"} + + svc := createTestServiceWithScheduler(repo, stepGetter, taskRepo, agentManager) + svc.SetProfileExecutionResolver(resolver.ProfileExecutionResolver) + svc.lastTurnPrompt.Store(sessionID, capturedPrompt{text: "retry the task"}) + + var wg sync.WaitGroup + wg.Add(2) + for i := 0; i < 2; i++ { + go func() { + defer wg.Done() + svc.runDynamicPolicyRecovery(ctx, sessionID, 4) + }() + } + wg.Wait() + + if got := atomic.LoadInt32(&launches); got != 1 { + t.Fatalf("successor launches = %d, want exactly 1 (duplicate prompt delivery)", got) + } + agentManager.mu.Lock() + stopCalls := len(agentManager.stopAgentWithReasonArgs) + agentManager.mu.Unlock() + if stopCalls != 1 { + t.Fatalf("predecessor stop calls = %d, want exactly 1", stopCalls) + } +} + +// dynamicPolicyRaceResolver exposes the underlying engine so the test can +// warm its in-memory cache exactly as production restart-recovery does via +// LoadState, before firing the two racing claims. +type dynamicPolicyRaceResolver struct { + *agentruntime.ProfileExecutionResolver + engine *dynamicruntime.Engine +} + +func (r *dynamicPolicyRaceResolver) EngineForTest() *dynamicruntime.Engine { return r.engine } + +func newDynamicPolicyRaceResolver( + t *testing.T, repo *sqliterepo.Repository, dynamicProfileID, concreteProfileID string, +) *dynamicPolicyRaceResolver { + t.Helper() + dbPath := t.TempDir() + "/agent-settings.db" + db, err := sqlx.Open("sqlite3", dbPath) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = db.Close() }) + profileRepo, cleanup, err := agentsettingsstore.Provide(db, db, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = cleanup() }) + ctx := context.Background() + for _, agent := range []*agentsettingsmodels.Agent{ + {ID: "dynamic", Name: "dynamic"}, + {ID: "concrete-agent", Name: "concrete-agent"}, + } { + if err := profileRepo.CreateAgent(ctx, agent); err != nil { + t.Fatal(err) + } + } + for _, profile := range []*agentsettingsmodels.AgentProfile{ + {ID: dynamicProfileID, AgentID: "dynamic", Name: "Cascade", Enabled: true}, + {ID: concreteProfileID, AgentID: "concrete-agent", Name: "Concrete", Enabled: true}, + } { + if err := profileRepo.CreateAgentProfile(ctx, profile); err != nil { + t.Fatal(err) + } + } + if err := profileRepo.CreateDynamicAgentProfile(ctx, + &agentsettingsmodels.DynamicAgentProfile{ProfileID: dynamicProfileID, Version: 1}, + []agentsettingsmodels.DynamicAgentRoute{{ + DynamicProfileID: dynamicProfileID, + ExecutionProfileID: concreteProfileID, + Enabled: true, + }}, + ); err != nil { + t.Fatal(err) + } + engine := dynamicruntime.NewEngine( + dynamicruntime.WithPersistence(repo), + dynamicruntime.WithStateLoader(repo), + ) + return &dynamicPolicyRaceResolver{ + ProfileExecutionResolver: agentruntime.NewProfileExecutionResolver(profileRepo, engine, true), + engine: engine, + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index c3aea6b485..c069404037 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -1176,6 +1176,10 @@ const ( RouteActionSkip RouteAction = "skip" RouteActionCancelWait RouteAction = "cancel_wait" RouteActionStop RouteAction = "stop" + + // RouteActionLaunchFailedReason marks a task-session projection whose + // successor launch failed before durable recovery could be confirmed. + RouteActionLaunchFailedReason = "route_action_launch_failed" ) type RouteActionRequest struct { diff --git a/apps/backend/internal/task/repository/sqlite/dynamic_route_concurrency_test.go b/apps/backend/internal/task/repository/sqlite/dynamic_route_concurrency_test.go new file mode 100644 index 0000000000..d69d25237d --- /dev/null +++ b/apps/backend/internal/task/repository/sqlite/dynamic_route_concurrency_test.go @@ -0,0 +1,85 @@ +package sqlite + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + dynamicruntime "github.com/kandev/kandev/internal/agent/runtime/dynamic" + "github.com/kandev/kandev/internal/task/models" +) + +// TestConcurrentResumePendingClaimsExactlyOneGeneration reproduces the +// duplicate-launch race: two callers that both observe a due retry_wait state +// at the same generation must not both receive a decision. The durable claim +// predicate is the authority, so exactly one caller succeeds and the other +// gets ErrStaleGeneration. +func TestConcurrentResumePendingClaimsExactlyOneGeneration(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + if err := repo.CreateTask(ctx, &models.Task{ID: "task-race", Title: "Race"}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "session-race", TaskID: "task-race", State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + elapsedDeadline := time.Now().UTC().Add(-time.Minute) + state := dynamicruntime.RouteState{ + SessionID: "session-race", LogicalProfileID: "dynamic-1", + ExecutionProfileID: "concrete-1", Generation: 4, ProfileVersion: 1, + Status: "retry_wait", + PolicyStateJSON: `{"deadline":"` + elapsedDeadline.Format(time.RFC3339Nano) + `"}`, + UpdatedAt: time.Now().UTC(), + } + if err := repo.SaveRouteState(ctx, state); err != nil { + t.Fatalf("SaveRouteState: %v", err) + } + + const callers = 2 + engines := make([]*dynamicruntime.Engine, callers) + for i := range engines { + engines[i] = dynamicruntime.NewEngine( + dynamicruntime.WithPersistence(repo), + dynamicruntime.WithStateLoader(repo), + ) + if _, _, err := engines[i].LoadState(ctx, "session-race"); err != nil { + t.Fatalf("warm engine cache %d: %v", i, err) + } + } + + var wg sync.WaitGroup + decisions := make([]dynamicruntime.RouteDecision, callers) + errs := make([]error, callers) + wg.Add(callers) + for i := 0; i < callers; i++ { + go func(i int) { + defer wg.Done() + decisions[i], errs[i] = engines[i].ResumePending(ctx, "session-race", 4) + }(i) + } + wg.Wait() + + successes := 0 + staleCount := 0 + for _, err := range errs { + if err == nil { + successes++ + continue + } + if errors.Is(err, dynamicruntime.ErrStaleGeneration) || errors.Is(err, dynamicruntime.ErrRecoveryPending) { + staleCount++ + continue + } + t.Fatalf("unexpected ResumePending error: %v", err) + } + if successes != 1 { + t.Fatalf("successful claims = %d, errors = %v (want exactly 1 success)", successes, errs) + } + if staleCount != callers-1 { + t.Fatalf("stale/rejected claims = %d, want %d", staleCount, callers-1) + } +} 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 30911ae72a..ad6cc374e6 100644 --- a/apps/backend/internal/task/repository/sqlite/dynamic_route_test.go +++ b/apps/backend/internal/task/repository/sqlite/dynamic_route_test.go @@ -265,6 +265,60 @@ func TestClaimRouteStateFromRequiresExpectedStatus(t *testing.T) { } } +func TestClaimRouteStatusRejectsStateThatAlreadyAdvancedAtSameGeneration(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + if err := repo.CreateTask(ctx, &models.Task{ID: "task-claim-status", Title: "Claim status"}); err != nil { + t.Fatalf("CreateTask: %v", err) + } + if err := repo.CreateTaskSession(ctx, &models.TaskSession{ + ID: "claim-status-session", TaskID: "task-claim-status", State: models.TaskSessionStateWaitingForInput, + }); err != nil { + t.Fatalf("CreateTaskSession: %v", err) + } + now := time.Now().UTC() + seed := dynamicruntime.RouteState{ + SessionID: "claim-status-session", LogicalProfileID: "dynamic-1", + ExecutionProfileID: "concrete-1", Generation: 4, ProfileVersion: 1, + Status: "retry_wait", UpdatedAt: now, + } + if err := repo.SaveRouteState(ctx, seed); err != nil { + t.Fatalf("SaveRouteState: %v", err) + } + + // First claim observes "retry_wait" and transitions to "retrying" at the + // same generation: the durable row still matches, so it succeeds. + winner := seed + winner.Status = "retrying" + winner.UpdatedAt = now.Add(time.Second) + claimed, err := repo.ClaimRouteStateFrom(ctx, 4, "retry_wait", winner) + if err != nil || !claimed { + t.Fatalf("winning ClaimRouteStateFrom = %v, %v", claimed, err) + } + + // A second caller that also observed "retry_wait" at generation 4 loses: + // the durable state column has already advanced to "retrying", even + // though the generation itself has not changed. + loser := seed + loser.Status = "retrying" + loser.UpdatedAt = now.Add(2 * time.Second) + claimed, err = repo.ClaimRouteStateFrom(ctx, 4, "retry_wait", loser) + if err != nil { + t.Fatalf("losing ClaimRouteStateFrom: %v", err) + } + if claimed { + t.Fatal("second caller claimed a status transition already applied by the first") + } + + loaded, err := repo.LoadRouteState(ctx, "claim-status-session") + if err != nil { + t.Fatalf("LoadRouteState: %v", err) + } + if loaded == nil || loaded.UpdatedAt.Unix() != winner.UpdatedAt.Unix() { + t.Fatalf("loaded state = %#v, want the winner's write", loaded) + } +} + func TestRecordRouteDecisionClaimsInitialGenerationAndWritesAttemptAtomically(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background()