From 284ea88553dbf594a282a69d7b63fb9ee5556815 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 13:10:18 +0300 Subject: [PATCH 1/2] feat(work): make agent execution recovery durable --- adapters/postgres/work_repository.go | 30 ++++++++- cmd/agent-canary/main.go | 3 +- .../042_agent_execution_recovery.sql | 15 +++++ dev/ACTIVE_TASK.json | 4 +- .../DEV-000214-agent-execution-recovery.json | 23 +++++++ modules/work/application/agent_dispatch.go | 67 ++++++++++++++++++- .../work/application/agent_dispatch_test.go | 32 +++++++++ modules/work/application/worker.go | 17 +++-- modules/work/application/worker_test.go | 6 ++ modules/work/ports/execution.go | 2 + modules/work/ports/worker.go | 2 + tests/agent_execution/canary_test.go | 3 +- .../dry_run_orchestration_test.go | 3 + tests/agent_execution/recovery_schema_test.go | 20 ++++++ 14 files changed, 214 insertions(+), 13 deletions(-) create mode 100644 database/migrations/042_agent_execution_recovery.sql create mode 100644 dev/tasks/DEV-000214-agent-execution-recovery.json create mode 100644 tests/agent_execution/recovery_schema_test.go diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 8c5b6f4..ba429d2 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -176,7 +176,7 @@ func (r *WorkRepository) ClaimNextDispatch(ctx context.Context, workerID string, } defer tx.Rollback(ctx) var item work.AgentDispatchQueueItem - err = tx.QueryRow(ctx, `SELECT dispatch_id,state,available_at_unix_millis,attempts,claimed_by,updated_at_unix_millis FROM work.agent_dispatch_queue WHERE state='QUEUED' AND available_at_unix_millis <= $1 ORDER BY available_at_unix_millis,dispatch_id FOR UPDATE SKIP LOCKED LIMIT 1`, nowUnixMillis).Scan(&item.DispatchID, &item.State, &item.AvailableAtUnixMillis, &item.Attempts, &item.ClaimedBy, &item.UpdatedAtUnixMillis) + err = tx.QueryRow(ctx, `SELECT dispatch_id,state,available_at_unix_millis,attempts,claimed_by,updated_at_unix_millis FROM work.agent_dispatch_queue WHERE state='QUEUED' AND available_at_unix_millis <= $1 AND attempts < retry_limit ORDER BY available_at_unix_millis,dispatch_id FOR UPDATE SKIP LOCKED LIMIT 1`, nowUnixMillis).Scan(&item.DispatchID, &item.State, &item.AvailableAtUnixMillis, &item.Attempts, &item.ClaimedBy, &item.UpdatedAtUnixMillis) if err != nil { if errors.Is(err, pgx.ErrNoRows) { return work.AgentDispatchQueueItem{}, nil @@ -219,6 +219,21 @@ func (r *WorkRepository) CompleteDispatch(ctx context.Context, dispatchID, worke return nil } +func (r *WorkRepository) RetryDispatch(ctx context.Context, dispatchID, workerID string, nowUnixMillis int64, delay time.Duration) error { + if delay < 0 { + return fmt.Errorf("invalid retry delay") + } + availableAt := nowUnixMillis + delay.Milliseconds() + command, err := r.pool.Exec(ctx, `UPDATE work.agent_dispatch_queue SET state=CASE WHEN attempts >= retry_limit THEN 'FAILED' ELSE 'QUEUED' END,available_at_unix_millis=$1,claimed_by='',updated_at_unix_millis=$2 WHERE dispatch_id=$3 AND state='CLAIMED' AND claimed_by=$4`, availableAt, nowUnixMillis, dispatchID, workerID) + if err != nil { + return err + } + if command.RowsAffected() != 1 { + return ErrAgentCallbackRejected + } + return nil +} + func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph) error { tx, err := r.pool.Begin(ctx) if err != nil { @@ -314,10 +329,21 @@ func (r *WorkRepository) SaveGraphSynthesis(ctx context.Context, conflict work.C } func (r *WorkRepository) RecoverStaleClaims(ctx context.Context, nowUnixMillis, staleAfterMillis int64) (int64, error) { - command, err := r.pool.Exec(ctx, `UPDATE work.agent_dispatch_queue SET state='QUEUED',claimed_by='',updated_at_unix_millis=$1 WHERE state='CLAIMED' AND updated_at_unix_millis <= $2`, nowUnixMillis, nowUnixMillis-staleAfterMillis) + tx, err := r.pool.Begin(ctx) if err != nil { return 0, err } + defer tx.Rollback(ctx) + if _, err := tx.Exec(ctx, `UPDATE work.agent_leases AS l SET state='EXPIRED' FROM work.agent_dispatch_queue AS q JOIN work.agent_dispatch_intents AS i ON i.dispatch_id=q.dispatch_id WHERE q.state='CLAIMED' AND q.updated_at_unix_millis <= $1 AND l.run_id=i.run_id AND l.state='ACTIVE'`, nowUnixMillis-staleAfterMillis); err != nil { + return 0, err + } + command, err := tx.Exec(ctx, `UPDATE work.agent_dispatch_queue SET state=CASE WHEN attempts >= retry_limit THEN 'FAILED' ELSE 'QUEUED' END,available_at_unix_millis=$1,claimed_by='',updated_at_unix_millis=$2 WHERE state='CLAIMED' AND updated_at_unix_millis <= $3`, nowUnixMillis, nowUnixMillis, nowUnixMillis-staleAfterMillis) + if err != nil { + return 0, err + } + if err := tx.Commit(ctx); err != nil { + return 0, err + } return command.RowsAffected(), nil } diff --git a/cmd/agent-canary/main.go b/cmd/agent-canary/main.go index 42cd850..81b7dba 100644 --- a/cmd/agent-canary/main.go +++ b/cmd/agent-canary/main.go @@ -24,7 +24,8 @@ func (s *memoryStore) SaveDispatchIntent(context.Context, work.AgentDispatchInte s.writes++ return nil } -func (s *memoryStore) SaveAgentLease(context.Context, work.AgentLease) error { s.writes++; return nil } +func (s *memoryStore) SaveAgentLease(context.Context, work.AgentLease) error { s.writes++; return nil } +func (s *memoryStore) RenewAgentLease(context.Context, work.AgentLease, time.Time) error { return nil } func (s *memoryStore) AcceptAgentCallback(context.Context, work.AgentCallback) error { s.writes++ return nil diff --git a/database/migrations/042_agent_execution_recovery.sql b/database/migrations/042_agent_execution_recovery.sql new file mode 100644 index 0000000..c5ad9ef --- /dev/null +++ b/database/migrations/042_agent_execution_recovery.sql @@ -0,0 +1,15 @@ +BEGIN; + +SET ROLE work_role; + +ALTER TABLE work.agent_dispatch_queue + ADD COLUMN IF NOT EXISTS retry_limit INTEGER NOT NULL DEFAULT 3; + +ALTER TABLE work.agent_dispatch_queue + DROP CONSTRAINT IF EXISTS agent_dispatch_queue_retry_limit_check; + +ALTER TABLE work.agent_dispatch_queue + ADD CONSTRAINT agent_dispatch_queue_retry_limit_check CHECK (retry_limit > 0); + +RESET ROLE; +COMMIT; diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index eb32e6f..6d3b366 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000213-worker-evaluator-link.json", - "task_id": "DEV-000213", + "task_file": "dev/tasks/DEV-000214-agent-execution-recovery.json", + "task_id": "DEV-000214", "task_revision": 1 } diff --git a/dev/tasks/DEV-000214-agent-execution-recovery.json b/dev/tasks/DEV-000214-agent-execution-recovery.json new file mode 100644 index 0000000..d129638 --- /dev/null +++ b/dev/tasks/DEV-000214-agent-execution-recovery.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000214", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "f9b9765100df02bead16280d2db9c1af478a4589", + "risk_tier": "E", + "intent": "Make Agent Execution recovery restart-safe: renew active leases with managed heartbeats, fence stale workers, cap retries, and requeue transient execution failures without losing durable dispatch state.", + "requirement_ids": ["durable heartbeat", "lease fencing", "bounded retry", "restart recovery", "transient failure requeue"], + "allowed_change_scope": {"modules": ["work", "agent-worker"], "paths": ["contracts/work/**", "modules/work/**", "adapters/postgres/work_repository.go", "cmd/agent-worker/**", "database/migrations/042_agent_execution_recovery.sql", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000214-agent-execution-recovery.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "modules/capability/**", "LLM-local/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, + "contracts": {"consume": ["existing AgentExecutor", "existing lease/callback contract", "durable dispatch queue"], "may_change": ["WorkerStore retry/recovery port"]}, + "semantic_owner": "work owns dispatch, lease, callback and retry lifecycle", + "canonical_state": ["work agent dispatch queue and lease tables remain canonical; heartbeat is durable evidence of liveness"], + "state_transitions": ["CLAIMED -> QUEUED on transient failure or stale recovery", "CLAIMED -> FAILED after retry limit", "ACTIVE lease -> EXPIRED before stale claim requeue", "heartbeat renews lease and callback sequence"], + "side_effects_and_privileges": ["work_role writes only work-owned queue and lease state; no external mutation"], + "persistence_and_migrations": ["042 adds bounded retry limit to dispatch queue"], + "failure_retry_restart": ["stale recovery expires the old lease before requeue", "transient execution failure uses durable delayed retry", "retry limit prevents hot-loop poison dispatch", "heartbeat failure cancels execution and leaves a retryable dispatch"], + "assumptions": ["operational heartbeat ticker is lifecycle-managed by the bounded worker call and durable state remains authoritative"], + "acceptance_scenarios": ["long execution renews lease and emits ordered heartbeats", "stale claim is fenced and requeued", "third failed attempt becomes FAILED", "waiting/error dispatch is not left CLAIMED", "legacy dispatch behavior remains compatible"], + "required_tests": ["go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "migration/schema checks", "CI", "production public health check"], + "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY"] +} diff --git a/modules/work/application/agent_dispatch.go b/modules/work/application/agent_dispatch.go index 1817e74..cc8bc75 100644 --- a/modules/work/application/agent_dispatch.go +++ b/modules/work/application/agent_dispatch.go @@ -2,6 +2,8 @@ package application import ( "context" + "fmt" + "sync" "time" "git.cin.su/m/Assistant/contracts/work" @@ -33,16 +35,75 @@ func ExecuteAgentDispatch(ctx context.Context, store ports.ExecutionStore, execu if err := store.AcceptAgentCallback(ctx, heartbeat); err != nil { return work.AgentExecutionResult{}, err } - result, err := executor.Execute(ctx, work.AgentExecutionRequest{Intent: intent, WorkerID: workerID, FencingToken: token, LeaseExpiresAt: lease.ExpiresAt}) + executionCtx, cancel := context.WithCancel(ctx) + defer cancel() + var stateMu sync.Mutex + currentLease := lease + lastSequence := heartbeat.Sequence + heartbeatDone := make(chan struct{}) + heartbeatStopped := make(chan struct{}) + heartbeatFailure := make(chan error, 1) + interval := ttl / 3 + if interval <= 0 { + return work.AgentExecutionResult{}, fmt.Errorf("invalid heartbeat interval") + } + go func() { + defer close(heartbeatStopped) + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-heartbeatDone: + return + case observedAt := <-ticker.C: + stateMu.Lock() + nextLease, renewErr := domain.RenewAgentLease(currentLease, workerID, token, observedAt, ttl) + if renewErr == nil { + renewErr = store.RenewAgentLease(executionCtx, nextLease, nextLease.ExpiresAt) + } + if renewErr == nil { + lastSequence++ + heartbeat = work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: lastSequence, Kind: work.AgentCallbackHeartbeat, ObservedAt: observedAt} + renewErr = domain.ValidateAgentCallback(heartbeat, nextLease, lastSequence-1) + } + if renewErr == nil { + renewErr = store.AcceptAgentCallback(executionCtx, heartbeat) + } + if renewErr == nil { + currentLease = nextLease + } + stateMu.Unlock() + if renewErr != nil { + select { + case heartbeatFailure <- renewErr: + default: + } + cancel() + return + } + } + } + }() + result, err := executor.Execute(executionCtx, work.AgentExecutionRequest{Intent: intent, WorkerID: workerID, FencingToken: token, LeaseExpiresAt: lease.ExpiresAt}) + close(heartbeatDone) + <-heartbeatStopped + select { + case heartbeatErr := <-heartbeatFailure: + return work.AgentExecutionResult{}, heartbeatErr + default: + } if err != nil { return work.AgentExecutionResult{}, err } + stateMu.Lock() + lease = currentLease + stateMu.Unlock() observedAt := time.UnixMilli(result.ObservedAtUnixMillis) - callback := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 2, Kind: work.AgentCallbackResult, ObservedAt: observedAt} + callback := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: lastSequence + 1, Kind: work.AgentCallbackResult, ObservedAt: observedAt} if result.State == work.AgentExecutionWaiting { callback.Kind = work.AgentCallbackCheckpoint } - if err := domain.ValidateAgentCallback(callback, lease, heartbeat.Sequence); err != nil { + if err := domain.ValidateAgentCallback(callback, lease, lastSequence); err != nil { return work.AgentExecutionResult{}, err } if err := store.AcceptAgentCallback(ctx, callback); err != nil { diff --git a/modules/work/application/agent_dispatch_test.go b/modules/work/application/agent_dispatch_test.go index f28e4f0..a56cdcf 100644 --- a/modules/work/application/agent_dispatch_test.go +++ b/modules/work/application/agent_dispatch_test.go @@ -18,6 +18,10 @@ func (s *bridgeStore) SaveAgentLease(_ context.Context, lease work.AgentLease) e s.lease = lease return nil } +func (s *bridgeStore) RenewAgentLease(_ context.Context, lease work.AgentLease, _ time.Time) error { + s.lease = lease + return nil +} func (s *bridgeStore) AcceptAgentCallback(_ context.Context, callback work.AgentCallback) error { s.callbacks = append(s.callbacks, callback) return nil @@ -29,6 +33,17 @@ func (bridgeExecutor) Execute(context.Context, work.AgentExecutionRequest) (work return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "done", ObservedAtUnixMillis: 2}, nil } +type slowBridgeExecutor struct{} + +func (slowBridgeExecutor) Execute(ctx context.Context, request work.AgentExecutionRequest) (work.AgentExecutionResult, error) { + select { + case <-time.After(45 * time.Millisecond): + return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "done", ObservedAtUnixMillis: time.Now().UnixMilli()}, nil + case <-ctx.Done(): + return work.AgentExecutionResult{}, ctx.Err() + } +} + func TestExecuteAgentDispatchPersistsLeaseAndBridgeCallbacks(t *testing.T) { store := &bridgeStore{} now := time.UnixMilli(1) @@ -40,3 +55,20 @@ func TestExecuteAgentDispatchPersistsLeaseAndBridgeCallbacks(t *testing.T) { t.Fatalf("callbacks=%+v", store.callbacks) } } + +func TestExecuteAgentDispatchRenewsLeaseAndOrdersHeartbeats(t *testing.T) { + store := &bridgeStore{} + now := time.Now().UTC() + _, err := ExecuteAgentDispatch(context.Background(), store, slowBridgeExecutor{}, work.AgentDispatchIntent{DispatchID: "d-slow", IdempotencyKey: "i-slow", DelegationID: "del-slow", RunID: "run-slow", ImplementationID: "impl", ImplementationRevision: 1, SourceRevision: "source", WorktreeID: "tree", VerificationPlanRef: "verify", CreatedAtUnixMillis: now.UnixMilli()}, "worker", 1, now, 30*time.Millisecond) + if err != nil { + t.Fatal(err) + } + if len(store.callbacks) < 3 { + t.Fatalf("expected heartbeat callbacks, got %d", len(store.callbacks)) + } + for i, callback := range store.callbacks { + if callback.Sequence != uint64(i+1) { + t.Fatalf("callback sequence is not ordered: %#v", store.callbacks) + } + } +} diff --git a/modules/work/application/worker.go b/modules/work/application/worker.go index 59e2402..96fd37c 100644 --- a/modules/work/application/worker.go +++ b/modules/work/application/worker.go @@ -15,6 +15,12 @@ type WorkerCycleResult struct { Idle bool } +const transientDispatchRetryDelay = time.Second + +func retryClaimedDispatch(ctx context.Context, store ports.WorkerStore, dispatchID, workerID string, now time.Time) error { + return store.RetryDispatch(ctx, dispatchID, workerID, now.UnixMilli(), transientDispatchRetryDelay) +} + // RecoverOnlyWorkerCycle keeps durable recovery alive while the real executor // capability is absent. It intentionally does not claim a dispatch: a missing // executor must never be represented as successful work. @@ -37,11 +43,11 @@ func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID s } intent, err := store.LoadDispatchIntent(ctx, item.DispatchID) if err != nil { - _ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli()) + _ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now) return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err } if _, err := ExecuteDryRunDispatch(ctx, store, intent, workerID, token, now, leaseTTL); err != nil { - _ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli()) + _ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now) return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err } if err := store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchCompleted, now.UnixMilli()); err != nil { @@ -64,15 +70,18 @@ func ExecuteWorkerCycleWithExecutor(ctx context.Context, store ports.WorkerStore } intent, err := store.LoadDispatchIntent(ctx, item.DispatchID) if err != nil { - _ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli()) + _ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now) return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err } result, err := ExecuteAgentDispatch(ctx, store, executor, intent, workerID, token, now, leaseTTL) if err != nil { - _ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli()) + _ = retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now) return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err } if result.State != work.AgentExecutionCompleted { + if err := retryClaimedDispatch(ctx, store, item.DispatchID, workerID, now); err != nil { + return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err + } return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, nil } if err := store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchCompleted, now.UnixMilli()); err != nil { diff --git a/modules/work/application/worker_test.go b/modules/work/application/worker_test.go index 4acb0ae..54d364c 100644 --- a/modules/work/application/worker_test.go +++ b/modules/work/application/worker_test.go @@ -26,10 +26,16 @@ func (s *recoveryOnlyStore) LoadDispatchIntent(context.Context, string) (work.Ag func (s *recoveryOnlyStore) CompleteDispatch(context.Context, string, string, work.AgentDispatchQueueState, int64) error { return nil } +func (s *recoveryOnlyStore) RetryDispatch(context.Context, string, string, int64, time.Duration) error { + return nil +} func (s *recoveryOnlyStore) SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error { return nil } func (s *recoveryOnlyStore) SaveAgentLease(context.Context, work.AgentLease) error { return nil } +func (s *recoveryOnlyStore) RenewAgentLease(context.Context, work.AgentLease, time.Time) error { + return nil +} func (s *recoveryOnlyStore) AcceptAgentCallback(context.Context, work.AgentCallback) error { return nil } diff --git a/modules/work/ports/execution.go b/modules/work/ports/execution.go index 51f1a79..d47aab4 100644 --- a/modules/work/ports/execution.go +++ b/modules/work/ports/execution.go @@ -2,6 +2,7 @@ package ports import ( "context" + "time" "git.cin.su/m/Assistant/contracts/work" ) @@ -13,5 +14,6 @@ type AgentExecutor interface { type ExecutionStore interface { SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error SaveAgentLease(context.Context, work.AgentLease) error + RenewAgentLease(context.Context, work.AgentLease, time.Time) error AcceptAgentCallback(context.Context, work.AgentCallback) error } diff --git a/modules/work/ports/worker.go b/modules/work/ports/worker.go index aaa44aa..ee65992 100644 --- a/modules/work/ports/worker.go +++ b/modules/work/ports/worker.go @@ -3,6 +3,7 @@ package ports import ( "context" "errors" + "time" "git.cin.su/m/Assistant/contracts/work" ) @@ -14,5 +15,6 @@ type WorkerStore interface { LoadDispatchIntent(context.Context, string) (work.AgentDispatchIntent, error) ClaimNextDispatch(context.Context, string, int64) (work.AgentDispatchQueueItem, error) CompleteDispatch(context.Context, string, string, work.AgentDispatchQueueState, int64) error + RetryDispatch(context.Context, string, string, int64, time.Duration) error RecoverStaleClaims(context.Context, int64, int64) (int64, error) } diff --git a/tests/agent_execution/canary_test.go b/tests/agent_execution/canary_test.go index 8463bb2..9f08aa5 100644 --- a/tests/agent_execution/canary_test.go +++ b/tests/agent_execution/canary_test.go @@ -15,7 +15,8 @@ func (s *canaryStore) SaveDispatchIntent(context.Context, work.AgentDispatchInte s.writes++ return nil } -func (s *canaryStore) SaveAgentLease(context.Context, work.AgentLease) error { s.writes++; return nil } +func (s *canaryStore) SaveAgentLease(context.Context, work.AgentLease) error { s.writes++; return nil } +func (s *canaryStore) RenewAgentLease(context.Context, work.AgentLease, time.Time) error { return nil } func (s *canaryStore) AcceptAgentCallback(context.Context, work.AgentCallback) error { s.writes++ return nil diff --git a/tests/agent_execution/dry_run_orchestration_test.go b/tests/agent_execution/dry_run_orchestration_test.go index 282669b..c3bb173 100644 --- a/tests/agent_execution/dry_run_orchestration_test.go +++ b/tests/agent_execution/dry_run_orchestration_test.go @@ -21,6 +21,9 @@ func (s *recordingExecutionStore) SaveAgentLease(context.Context, work.AgentLeas s.operations = append(s.operations, "lease") return nil } +func (s *recordingExecutionStore) RenewAgentLease(context.Context, work.AgentLease, time.Time) error { + return nil +} func (s *recordingExecutionStore) AcceptAgentCallback(context.Context, work.AgentCallback) error { s.operations = append(s.operations, "callback") return nil diff --git a/tests/agent_execution/recovery_schema_test.go b/tests/agent_execution/recovery_schema_test.go new file mode 100644 index 0000000..291e363 --- /dev/null +++ b/tests/agent_execution/recovery_schema_test.go @@ -0,0 +1,20 @@ +package agent_execution_test + +import ( + "os" + "strings" + "testing" +) + +func TestRecoveryMigrationDefinesBoundedRetryState(t *testing.T) { + payload, err := os.ReadFile("../../database/migrations/042_agent_execution_recovery.sql") + if err != nil { + t.Fatal(err) + } + source := strings.ToLower(string(payload)) + for _, fragment := range []string{"set role work_role", "retry_limit integer not null default 3", "retry_limit > 0"} { + if !strings.Contains(source, fragment) { + t.Fatalf("migration is missing %q", fragment) + } + } +} -- 2.54.0 From c9eb3988576f2c3ee0024d07d0419b0fac344b3c Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 13:12:25 +0300 Subject: [PATCH 2/2] chore(governance): declare canary scope for recovery --- dev/tasks/DEV-000214-agent-execution-recovery.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dev/tasks/DEV-000214-agent-execution-recovery.json b/dev/tasks/DEV-000214-agent-execution-recovery.json index d129638..33d5007 100644 --- a/dev/tasks/DEV-000214-agent-execution-recovery.json +++ b/dev/tasks/DEV-000214-agent-execution-recovery.json @@ -6,7 +6,7 @@ "risk_tier": "E", "intent": "Make Agent Execution recovery restart-safe: renew active leases with managed heartbeats, fence stale workers, cap retries, and requeue transient execution failures without losing durable dispatch state.", "requirement_ids": ["durable heartbeat", "lease fencing", "bounded retry", "restart recovery", "transient failure requeue"], - "allowed_change_scope": {"modules": ["work", "agent-worker"], "paths": ["contracts/work/**", "modules/work/**", "adapters/postgres/work_repository.go", "cmd/agent-worker/**", "database/migrations/042_agent_execution_recovery.sql", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000214-agent-execution-recovery.json"]}, + "allowed_change_scope": {"modules": ["work", "agent-worker"], "paths": ["contracts/work/**", "modules/work/**", "adapters/postgres/work_repository.go", "cmd/agent-worker/**", "cmd/agent-canary/**", "database/migrations/042_agent_execution_recovery.sql", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000214-agent-execution-recovery.json"]}, "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "modules/capability/**", "LLM-local/**"]}, "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, "contracts": {"consume": ["existing AgentExecutor", "existing lease/callback contract", "durable dispatch queue"], "may_change": ["WorkerStore retry/recovery port"]}, -- 2.54.0