Guard orphan sweep when routing is disabled
apps/backend/internal/orchestrator/dynamic_policy_recovery.go ↗Adds Enabled check so the sweep does nothing when dynamic routing is off, like startDynamicPolicyRecovery.
Guard
// healthy idling dynamic session. Dynamic routing disabled means any
// recovery action this sweep produces is guaranteed to fail
// (LaunchDynamicRouteAction returns ErrDynamicRoutingDisabled), so it must
// stay a no-op exactly like startDynamicPolicyRecovery.
func (s *Service) reconcileOrphanedDynamicStartingRoutes(ctx context.Context) {
if s.profileExecutionResolver == nil {
if s.profileExecutionResolver == nil || !s.profileExecutionResolver.Enabled() {
return
}
lister, ok := s.repo.(dynamicStartingRouteLister)
One-time legacy backfill migration
apps/backend/internal/task/repository/sqlite/dynamic_route_legacy_active_migration.go ↗Moves legacy IDLE starting rows to active once, gated on marker column not on value shape.
Backfill
func (r *Repository) backfillLegacyActiveDynamicRoutes() error {
exists, err := db.ColumnExists(r.db, "dynamic_route_states", dynamicRouteLegacyActiveBackfillColumn)
if err != nil {
return fmt.Errorf("dynamic route legacy backfill: probe marker column: %w", err)
}
if exists {
return nil
}
tx, err := r.db.Beginx()
if err != nil {
return fmt.Errorf("dynamic route legacy backfill: begin tx: %w", err)
}
defer func() { _ = tx.Rollback() }()
if err := r.acquireDynamicRouteLegacyActiveBackfillLock(tx); err != nil {
return err
}
exists, err = r.columnExists(tx, "dynamic_route_states", dynamicRouteLegacyActiveBackfillColumn)
if err != nil {
return fmt.Errorf("dynamic route legacy backfill: re-probe marker column: %w", err)
}
if exists {
return nil
}
if _, err := tx.Exec(`
UPDATE dynamic_route_states
SET state = 'active'
WHERE state = 'starting'
AND session_id IN (SELECT id FROM task_sessions WHERE state = 'IDLE')
`); err != nil {
return fmt.Errorf("dynamic route legacy backfill: backfill dynamic_route_states: %w", err)
}
if _, err := tx.Exec(`
UPDATE task_sessions
SET route_state = 'active'
WHERE state = 'IDLE' AND route_state = 'starting'
AND id IN (SELECT session_id FROM dynamic_route_states WHERE state = 'active')
`); err != nil {
return fmt.Errorf("dynamic route legacy backfill: backfill task_sessions: %w", err)
}
if _, err := tx.Exec(`ALTER TABLE dynamic_route_states ADD COLUMN ` + dynamicRouteLegacyActiveBackfillColumn + ` INTEGER NOT NULL DEFAULT 1`); err != nil {
return fmt.Errorf("dynamic route legacy backfill: add marker column: %w", err)
}
if err := tx.Commit(); err != nil {
return fmt.Errorf("dynamic route legacy backfill: commit: %w", err)
}
return nil
}
Advisory lock
func (r *Repository) acquireDynamicRouteLegacyActiveBackfillLock(tx *sqlx.Tx) error {
if !dialect.IsPostgres(r.db.DriverName()) {
return nil
}
if _, err := tx.Exec(`SET LOCAL lock_timeout = '30s'`); err != nil {
return fmt.Errorf("dynamic route legacy backfill: set lock timeout: %w", err)
}
if _, err := tx.Exec(`SELECT pg_advisory_xact_lock($1)`, dynamicRouteLegacyActiveBackfillLockID); err != nil {
return fmt.Errorf("dynamic route legacy backfill: acquire migration advisory lock: %w", err)
}
return nil
}
Fresh-install schema marker
apps/backend/internal/task/repository/sqlite/base_schema.go ↗Declares marker column inline so a fresh database never runs the backfill.
DDL
continuation_json TEXT NOT NULL DEFAULT '',
policy_state_json TEXT NOT NULL DEFAULT '',
legacy_active_backfill_applied INTEGER NOT NULL DEFAULT 1,
updated_at TIMESTAMP NOT NULL,
FOREIGN KEY (session_id) REFERENCES task_sessions(id) ON DELETE CASCADE
);
Wire backfill into runMigrations
apps/backend/internal/task/repository/sqlite/base_migrations.go ↗Calls the backfill early in the migration chain so upgrade repairs before the sweep.
Wiring
r.migrate.Apply("dynamic_route_states.continuation_json", `ALTER TABLE dynamic_route_states ADD COLUMN continuation_json TEXT NOT NULL DEFAULT ''`)
r.migrate.Apply("dynamic_route_states.policy_state_json", `ALTER TABLE dynamic_route_states ADD COLUMN policy_state_json TEXT NOT NULL DEFAULT ''`)
if err := r.backfillLegacyActiveDynamicRoutes(); err != nil {
return err
}
r.migrate.Apply("executors_running.execution_profile_id", `ALTER TABLE executors_running ADD COLUMN execution_profile_id TEXT NOT NULL DEFAULT ''`)
Two-dialect upgrade regression test
apps/backend/internal/persistence/storeconformance/upgrade_test.go ↗Seeds legacy IDLE starting shape on v0.93.0 fixture and proves backfill works on SQLite and Postgres.
Test
func TestPreviousStableUpgrade_BackfillsLegacyActiveDynamicRoutes(t *testing.T) {
manifest := loadUpgradeManifest(t)
for _, engine := range []testconformance.EngineName{testconformance.EngineSQLite, testconformance.EnginePostgres} {
engine := engine
t.Run(string(engine)+"/"+previousStableTag, func(t *testing.T) {
fixture := fixtureForEngine(t, manifest, engine)
database := testconformance.OpenEngine(t, engine, "")
if err := applyFixture(t, database, fixture); err != nil {
t.Fatalf("apply %s fixture: %v", engine, err)
}
seedLegacyDynamicRouteFixtureRows(t, database)
if err := runCurrentInitialization(database); err != nil {
t.Fatalf("current initialization: %v", err)
}
assertLegacyDynamicRouteBackfilled(t, database)
})
}
}
func seedLegacyDynamicRouteFixtureRows(t *testing.T, engine testconformance.Engine) {
t.Helper()
ctx := context.Background()
now := "2026-01-01 00:00:00"
if _, err := engine.DB.ExecContext(ctx, engine.DB.Rebind(`
INSERT INTO task_sessions (id, task_id, state, route_generation, route_state, started_at, updated_at)
VALUES (?, ?, 'IDLE', 1, 'starting', ?, ?)
`), "fixture-v0930-dynamic-session", "fixture-v0930-task", now, now); err != nil {
t.Fatalf("seed legacy IDLE dynamic session: %v", err)
}
if _, err := engine.DB.ExecContext(ctx, engine.DB.Rebind(`
INSERT INTO dynamic_route_states (
session_id, logical_profile_id, execution_profile_id, route_generation, profile_version, state, updated_at
) VALUES (?, 'dynamic-logical', 'candidate-1', 1, 1, 'starting', ?)
`), "fixture-v0930-dynamic-session", now); err != nil {
t.Fatalf("seed legacy starting dynamic_route_states row: %v", err)
}
}
Orphan sweep regression tests
apps/backend/internal/orchestrator/dynamic_route_recovery_test.go ↗Proves disabled routing leaves starting rows untouched and ordering still recovers real orphans.
Disabled routing
func TestReconcileOrphanedDynamicStartingRoutes_SkipsWhenRoutingDisabled(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, taskID, sessionID, models.TaskSessionStateIdle)
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, false))
svc.reconcileOrphanedDynamicStartingRoutes(ctx)
routeState, _ := repo.LoadRouteState(ctx, sessionID)
if routeState.Status != "starting" {
t.Fatalf("durable route state = %#v, want untouched 'starting' with routing disabled", routeState)
}
}
