diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver.go b/apps/backend/internal/agent/runtime/dynamic_resolver.go index 1836179c74..03d785fe1b 100644 --- a/apps/backend/internal/agent/runtime/dynamic_resolver.go +++ b/apps/backend/internal/agent/runtime/dynamic_resolver.go @@ -59,6 +59,12 @@ func NewProfileExecutionResolver(profiles store.Repository, engine *dynamic.Engi func (r *ProfileExecutionResolver) SetEnabled(enabled bool) { r.enabled.Store(enabled) } +// Enabled reports the effective dynamic-agent-routing flag value this +// resolver was constructed or last set with. Callers outside this package +// use it to gate durable recovery and manual route-action launch paths that +// do not otherwise pass through a selection method. +func (r *ProfileExecutionResolver) Enabled() bool { return r.enabled.Load() } + // SetCredentialBindingResolver supplies the installation-scoped fingerprint // used to share provider health between concrete profiles that prove the same // credential binding. A missing or incomplete descriptor remains isolated to @@ -394,6 +400,9 @@ func (r *ProfileExecutionResolver) ResumePendingRoute( if r.engine == nil || r.profiles == nil { return ProfileExecution{}, errors.New("dynamic profile execution is not configured") } + if !r.enabled.Load() { + return ProfileExecution{}, ErrDynamicRoutingDisabled + } state, exists, err := r.engine.LoadState(ctx, sessionID) if err != nil { return ProfileExecution{}, err diff --git a/apps/backend/internal/agent/runtime/dynamic_resolver_recovery_test.go b/apps/backend/internal/agent/runtime/dynamic_resolver_recovery_test.go new file mode 100644 index 0000000000..1ba90765bd --- /dev/null +++ b/apps/backend/internal/agent/runtime/dynamic_resolver_recovery_test.go @@ -0,0 +1,116 @@ +package runtime_test + +import ( + "context" + "encoding/json" + "errors" + "testing" + "time" + + agentruntime "github.com/kandev/kandev/internal/agent/runtime" + "github.com/kandev/kandev/internal/agent/runtime/dynamic" + "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" +) + +// fixedProfileRepository is a minimal store.Repository stand-in. The embedded +// interface is left nil; only GetAgentProfile is overridden, since that is +// the only method these tests' code paths reach. +type fixedProfileRepository struct { + store.Repository + profile *agentsettingsmodels.AgentProfile +} + +func (f *fixedProfileRepository) GetAgentProfile(_ context.Context, id string) (*agentsettingsmodels.AgentProfile, error) { + if f.profile != nil && f.profile.ID == id { + return f.profile, nil + } + return nil, errors.New("agent profile not found") +} + +// fixedRouteStateLoader hands the engine one durable route state on first +// access, standing in for a state restored after a backend restart. +type fixedRouteStateLoader struct { + state dynamic.RouteState + loadStateCalls int +} + +func (f *fixedRouteStateLoader) LoadRouteState(_ context.Context, sessionID string) (*dynamic.RouteState, error) { + f.loadStateCalls++ + if sessionID != f.state.SessionID { + return nil, nil + } + state := f.state + return &state, nil +} + +func persistedRetryWaitState(t *testing.T, sessionID string) dynamic.RouteState { + t.Helper() + deadline := time.Now().Add(-time.Minute) + policyState := dynamic.PolicyState{Deadline: &deadline} + raw, err := json.Marshal(policyState) + if err != nil { + t.Fatalf("marshal policy state: %v", err) + } + return dynamic.RouteState{ + SessionID: sessionID, + LogicalProfileID: "dynamic-policy", + ExecutionProfileID: "first", + Generation: 1, + ProfileVersion: 1, + Status: string(routingpolicy.DecisionRetry), + PolicyStateJSON: string(raw), + UpdatedAt: time.Now().Add(-time.Minute), + } +} + +func TestResumePendingRoute_DisabledFlagRefusesWithoutMutatingState(t *testing.T) { + const sessionID = "policy-session" + persisted := persistedRetryWaitState(t, sessionID) + loader := &fixedRouteStateLoader{state: persisted} + engine := dynamic.NewEngine(dynamic.WithStateLoader(loader)) + resolver := agentruntime.NewProfileExecutionResolver(&fixedProfileRepository{}, engine, false) + + ctx := context.Background() + if _, err := resolver.ResumePendingRoute(ctx, sessionID, persisted.Generation); !errors.Is(err, agentruntime.ErrDynamicRoutingDisabled) { + t.Fatalf("ResumePendingRoute with flag off = %v, want ErrDynamicRoutingDisabled", err) + } + if loader.loadStateCalls != 0 { + t.Fatalf("LoadRouteState calls = %d, want 0 when flag disabled", loader.loadStateCalls) + } + + state, exists, err := engine.LoadState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadState after refused resume: %v", err) + } + if !exists || state.Status != string(routingpolicy.DecisionRetry) { + t.Fatalf("route state after refused resume = %+v (exists=%v), want unchanged retry_wait", state, exists) + } +} + +func TestResumePendingRoute_EnabledFlagResumesDueRoute(t *testing.T) { + const sessionID = "policy-session-enabled" + persisted := persistedRetryWaitState(t, sessionID) + loader := &fixedRouteStateLoader{state: persisted} + engine := dynamic.NewEngine(dynamic.WithStateLoader(loader)) + profiles := &fixedProfileRepository{profile: &agentsettingsmodels.AgentProfile{ID: "first", Enabled: true}} + resolver := agentruntime.NewProfileExecutionResolver(profiles, engine, true) + + ctx := context.Background() + execution, err := resolver.ResumePendingRoute(ctx, sessionID, persisted.Generation) + if err != nil { + t.Fatalf("ResumePendingRoute with flag on: %v", err) + } + if execution.ExecutionProfileID != "first" || execution.Generation != persisted.Generation { + t.Fatalf("execution = %+v, want ExecutionProfileID=first Generation=%d", execution, persisted.Generation) + } + + state, exists, err := engine.LoadState(ctx, sessionID) + if err != nil { + t.Fatalf("LoadState after resume: %v", err) + } + if !exists || state.Status != "retrying" { + t.Fatalf("route state after resume = %+v (exists=%v), want status retrying", state, exists) + } +} diff --git a/apps/backend/internal/orchestrator/dynamic_launch.go b/apps/backend/internal/orchestrator/dynamic_launch.go index b32c385358..f079f911e3 100644 --- a/apps/backend/internal/orchestrator/dynamic_launch.go +++ b/apps/backend/internal/orchestrator/dynamic_launch.go @@ -742,6 +742,9 @@ func (s *Service) LaunchDynamicRouteAction(ctx context.Context, sessionID string if s.profileExecutionResolver == nil || sessionID == "" { return errors.New("dynamic route action launch is not configured") } + if !s.profileExecutionResolver.Enabled() { + return agentruntime.ErrDynamicRoutingDisabled + } session, err := s.repo.GetTaskSession(ctx, sessionID) if err != nil || session == nil { if err != nil { diff --git a/apps/backend/internal/orchestrator/dynamic_policy_recovery.go b/apps/backend/internal/orchestrator/dynamic_policy_recovery.go index 82ef5df7d3..c0193732ac 100644 --- a/apps/backend/internal/orchestrator/dynamic_policy_recovery.go +++ b/apps/backend/internal/orchestrator/dynamic_policy_recovery.go @@ -22,7 +22,7 @@ type dynamicRouteStateLister interface { // recovery because a process restart cannot prove whether its launch crossed // the provider boundary. func (s *Service) startDynamicPolicyRecovery(ctx context.Context) { - if s.profileExecutionResolver == nil { + if s.profileExecutionResolver == nil || !s.profileExecutionResolver.Enabled() { return } lister, ok := s.repo.(dynamicRouteStateLister) @@ -120,7 +120,8 @@ func (s *Service) removeDynamicPolicyRecoveryTimer(sessionID string) { } func (s *Service) dynamicPolicyRecoveryActive(ctx context.Context) bool { - return ctx != nil && ctx.Err() == nil && s.profileExecutionResolver != nil + return ctx != nil && ctx.Err() == nil && + s.profileExecutionResolver != nil && s.profileExecutionResolver.Enabled() } func loadDueDynamicPolicyState( diff --git a/apps/backend/internal/orchestrator/dynamic_policy_recovery_flag_test.go b/apps/backend/internal/orchestrator/dynamic_policy_recovery_flag_test.go new file mode 100644 index 0000000000..435d679234 --- /dev/null +++ b/apps/backend/internal/orchestrator/dynamic_policy_recovery_flag_test.go @@ -0,0 +1,170 @@ +package orchestrator + +import ( + "context" + "encoding/json" + "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/routingpolicy" + "github.com/kandev/kandev/internal/agent/settings/store" + "github.com/kandev/kandev/internal/task/models" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" +) + +// fakeSettingsRepo is a minimal store.Repository stand-in for building a +// ProfileExecutionResolver in these tests. None of the flag-gated code paths +// under test reach the embedded interface's methods. +type fakeSettingsRepo struct { + store.Repository +} + +type recordingRouteActionRepo struct { + sessionExecutorStore + getTaskSessionCalls int +} + +func (r *recordingRouteActionRepo) GetTaskSession(context.Context, string) (*models.TaskSession, error) { + r.getTaskSessionCalls++ + return nil, nil +} + +func policyStateJSON(t *testing.T, deadline time.Time) string { + t.Helper() + raw, err := json.Marshal(dynamicruntime.PolicyState{Deadline: &deadline}) + if err != nil { + t.Fatalf("marshal policy state: %v", err) + } + return string(raw) +} + +func seedPendingRouteState(t *testing.T, repo *sqliterepo.Repository, sessionID string, deadline time.Time) { + t.Helper() + ctx := context.Background() + if err := repo.SaveRouteState(ctx, dynamicruntime.RouteState{ + SessionID: sessionID, + LogicalProfileID: "dynamic-policy", + ExecutionProfileID: "first", + Generation: 1, + ProfileVersion: 1, + Status: string(routingpolicy.DecisionRetry), + PolicyStateJSON: policyStateJSON(t, deadline), + UpdatedAt: time.Now().UTC(), + }); err != nil { + t.Fatalf("seed pending route state: %v", err) + } +} + +func TestStartDynamicPolicyRecovery_DisabledFlagArmsNoTimer(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-flag-off", "session-flag-off", "") + seedPendingRouteState(t, repo, "session-flag-off", time.Now().Add(time.Hour)) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo)) + resolver := agentruntime.NewProfileExecutionResolver(&fakeSettingsRepo{}, engine, false) + svc := &Service{logger: testLogger(), repo: repo, profileExecutionResolver: resolver} + + svc.startDynamicPolicyRecovery(ctx) + t.Cleanup(svc.stopDynamicPolicyRecovery) + + svc.dynamicRecoveryMu.Lock() + timers := len(svc.dynamicRecoveryTimers) + svc.dynamicRecoveryMu.Unlock() + if timers != 0 { + t.Fatalf("dynamicRecoveryTimers = %d entries, want 0 when flag disabled", timers) + } +} + +func TestStartDynamicPolicyRecovery_EnabledFlagArmsTimer(t *testing.T) { + ctx := context.Background() + repo := setupTestRepo(t) + seedSession(t, repo, "task-flag-on", "session-flag-on", "") + seedPendingRouteState(t, repo, "session-flag-on", time.Now().Add(time.Hour)) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(repo), dynamicruntime.WithStateLoader(repo)) + resolver := agentruntime.NewProfileExecutionResolver(&fakeSettingsRepo{}, engine, true) + svc := &Service{logger: testLogger(), repo: repo, profileExecutionResolver: resolver} + + svc.startDynamicPolicyRecovery(ctx) + t.Cleanup(svc.stopDynamicPolicyRecovery) + + svc.dynamicRecoveryMu.Lock() + _, armed := svc.dynamicRecoveryTimers["session-flag-on"] + svc.dynamicRecoveryMu.Unlock() + if !armed { + t.Fatal("dynamicRecoveryTimers missing session-flag-on entry when flag enabled") + } +} + +// recordingRouteStateLoader wraps the real repository to count calls to +// LoadRouteState made through the orchestrator's own dynamicRouteStateLoader +// seam. That call happens before the resolver's ResumePendingRoute is ever +// invoked, so it observes the orchestrator's own gate independently of +// ResumePendingRoute's separate, later refusal. +type recordingRouteStateLoader struct { + *sqliterepo.Repository + loadRouteStateCalls int +} + +func (r *recordingRouteStateLoader) LoadRouteState(ctx context.Context, sessionID string) (*dynamicruntime.RouteState, error) { + r.loadRouteStateCalls++ + return r.Repository.LoadRouteState(ctx, sessionID) +} + +func TestRunDynamicPolicyRecovery_FlagFlippedOffAfterArmSkipsLaunch(t *testing.T) { + ctx := context.Background() + baseRepo := setupTestRepo(t) + seedSession(t, baseRepo, "task-flip-off", "session-flip-off", "") + // Deadline already elapsed: firing the timer must reach the resume call, + // not just reschedule for a later due time. + seedPendingRouteState(t, baseRepo, "session-flip-off", time.Now().Add(-time.Minute)) + + engine := dynamicruntime.NewEngine(dynamicruntime.WithPersistence(baseRepo), dynamicruntime.WithStateLoader(baseRepo)) + resolver := agentruntime.NewProfileExecutionResolver(&fakeSettingsRepo{}, engine, true) + repo := &recordingRouteStateLoader{Repository: baseRepo} + svc := &Service{ + logger: testLogger(), repo: repo, profileExecutionResolver: resolver, + dynamicRecoveryTimers: make(map[string]*time.Timer), + } + // Stands in for a recovery context installed by an earlier + // startDynamicPolicyRecovery call whose timer is now firing. + recoveryCtx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + svc.dynamicRecoveryCtx = recoveryCtx + + resolver.SetEnabled(false) + svc.runDynamicPolicyRecovery(recoveryCtx, "session-flip-off", 1) + + if repo.loadRouteStateCalls != 0 { + t.Fatalf("LoadRouteState calls = %d, want 0: the disabled-flag gate must return before "+ + "reading route state, not rely on ResumePendingRoute's own refusal", repo.loadRouteStateCalls) + } + + state, err := baseRepo.LoadRouteState(ctx, "session-flip-off") + if err != nil { + t.Fatalf("LoadRouteState after flipped-off recovery run: %v", err) + } + if state == nil || state.Status != string(routingpolicy.DecisionRetry) { + t.Fatalf("route state after flipped-off recovery run = %+v, want unchanged retry_wait", state) + } +} + +func TestLaunchDynamicRouteAction_DisabledFlagRefuses(t *testing.T) { + ctx := context.Background() + engine := dynamicruntime.NewEngine() + resolver := agentruntime.NewProfileExecutionResolver(&fakeSettingsRepo{}, engine, false) + repo := &recordingRouteActionRepo{} + svc := &Service{logger: testLogger(), repo: repo, profileExecutionResolver: resolver} + + err := svc.LaunchDynamicRouteAction(ctx, "session-launch-disabled") + if !errors.Is(err, agentruntime.ErrDynamicRoutingDisabled) { + t.Fatalf("LaunchDynamicRouteAction with flag off = %v, want ErrDynamicRoutingDisabled", err) + } + if repo.getTaskSessionCalls != 0 { + t.Fatalf("GetTaskSession calls = %d, want 0 when flag disabled", repo.getTaskSessionCalls) + } +} diff --git a/docs/specs/agents/requirements/dynamic-agent-routing-rollout-blockers.md b/docs/specs/agents/requirements/dynamic-agent-routing-rollout-blockers.md index b41032bbe6..4232cfa202 100644 --- a/docs/specs/agents/requirements/dynamic-agent-routing-rollout-blockers.md +++ b/docs/specs/agents/requirements/dynamic-agent-routing-rollout-blockers.md @@ -25,6 +25,7 @@ This repair spec amends [Dynamic Agent Routing](dynamic-agent-routing.md). It tu - **AC-AGENTS-DYNAMIC-AGENT-ROUTING-ROLLOUT-BLOCKERS-001.4:** **GIVEN** a utility call attached to task session `S`, **WHEN** two calls start, **THEN** they use different `utility:` route identities and neither consumes task route state. - **AC-AGENTS-DYNAMIC-AGENT-ROUTING-ROLLOUT-BLOCKERS-001.5:** **GIVEN** two concrete profiles share one proven credential binding, **WHEN** one opens a quota circuit, **THEN** new selections skip that resource and a single expired probe controls recovery. - **AC-AGENTS-DYNAMIC-AGENT-ROUTING-ROLLOUT-BLOCKERS-001.6:** **GIVEN** the flag is disabled, **WHEN** a task picker is opened, **THEN** dynamic profiles are absent while existing dynamic settings and sessions are retained. +- **AC-AGENTS-DYNAMIC-AGENT-ROUTING-ROLLOUT-BLOCKERS-001.7:** **GIVEN** the flag is disabled, **WHEN** the backend restarts with a persisted `retry_wait` or `waiting_for_reset` route, **THEN** no recovery timer is armed and no provider is launched, while the durable route state is retained. ## Migrated source detail @@ -51,6 +52,10 @@ repair. - Every utility invocation receives a unique transient route identity, even when its template is attached to a task session. Utility fallback also requires the same explicit pre-result evidence. +- The flag gates the durable recovery scheduler and the manual route-action + launch path, not only new-route selection. A disabled flag refuses to arm a + recovery timer, refuses to resume a due wait, and refuses a manual launch, + leaving the persisted route state untouched. ## Shared health contract @@ -93,6 +98,10 @@ repair. - **GIVEN** the flag is disabled, **WHEN** a task picker is opened, **THEN** dynamic profiles are absent while existing dynamic settings and sessions are retained. +- **GIVEN** the flag is disabled, **WHEN** the backend restarts with a + persisted `retry_wait` or `waiting_for_reset` route, **THEN** no recovery + timer is armed and no provider is launched, while the durable route state is + retained. ## Scope boundary