fix(backend): stop disabled dynamic routing from resuming pending retries (#3361)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
2026-09-04 13:40:37 +00:00
parent 15c0c35df6
commit 541f37b3d2
6 changed files with 310 additions and 2 deletions
@@ -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
@@ -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)
+	}
+}
@@ -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 {
@@ -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(
@@ -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)
+	}
+}
@@ -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