diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 1e9af75..05c6fb0 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -68,6 +68,67 @@ func (r *WorkRepository) SaveDispatchIntent(ctx context.Context, intent work.Age return tx.Commit(ctx) } +func (r *WorkRepository) LoadDispatchIntent(ctx context.Context, dispatchID string) (work.AgentDispatchIntent, error) { + var intent work.AgentDispatchIntent + err := r.pool.QueryRow(ctx, `SELECT dispatch_id,idempotency_key,delegation_id,run_id,implementation_id,implementation_revision,source_revision,worktree_id,verification_plan_ref,created_at_unix_millis FROM work.agent_dispatch_intents WHERE dispatch_id=$1`, dispatchID).Scan(&intent.DispatchID, &intent.IdempotencyKey, &intent.DelegationID, &intent.RunID, &intent.ImplementationID, &intent.ImplementationRevision, &intent.SourceRevision, &intent.WorktreeID, &intent.VerificationPlanRef, &intent.CreatedAtUnixMillis) + if err != nil { + return work.AgentDispatchIntent{}, fmt.Errorf("load dispatch intent: %w", err) + } + return intent, nil +} + +func (r *WorkRepository) EnqueueDispatch(ctx context.Context, item work.AgentDispatchQueueItem) error { + _, err := r.pool.Exec(ctx, `INSERT INTO work.agent_dispatch_queue(dispatch_id,state,available_at_unix_millis,attempts,claimed_by,updated_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT(dispatch_id) DO NOTHING`, item.DispatchID, item.State, item.AvailableAtUnixMillis, item.Attempts, item.ClaimedBy, item.UpdatedAtUnixMillis) + return err +} + +func (r *WorkRepository) ClaimNextDispatch(ctx context.Context, workerID string, nowUnixMillis int64) (work.AgentDispatchQueueItem, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return work.AgentDispatchQueueItem{}, err + } + 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) + if err != nil { + return work.AgentDispatchQueueItem{}, err + } + item.State = work.AgentDispatchClaimed + item.Attempts++ + item.ClaimedBy = workerID + item.UpdatedAtUnixMillis = nowUnixMillis + _, err = tx.Exec(ctx, `UPDATE work.agent_dispatch_queue SET state=$1,attempts=$2,claimed_by=$3,updated_at_unix_millis=$4 WHERE dispatch_id=$5 AND state='QUEUED'`, item.State, item.Attempts, item.ClaimedBy, item.UpdatedAtUnixMillis, item.DispatchID) + if err != nil { + return work.AgentDispatchQueueItem{}, err + } + if err := tx.Commit(ctx); err != nil { + return work.AgentDispatchQueueItem{}, err + } + return item, nil +} + +func (r *WorkRepository) CompleteDispatch(ctx context.Context, dispatchID, workerID string, state work.AgentDispatchQueueState, nowUnixMillis int64) error { + if state != work.AgentDispatchCompleted && state != work.AgentDispatchFailed { + return fmt.Errorf("invalid terminal dispatch state: %s", state) + } + command, err := r.pool.Exec(ctx, `UPDATE work.agent_dispatch_queue SET state=$1,updated_at_unix_millis=$2 WHERE dispatch_id=$3 AND state='CLAIMED' AND claimed_by=$4`, state, nowUnixMillis, dispatchID, workerID) + if err != nil { + return err + } + if command.RowsAffected() != 1 { + return ErrAgentCallbackRejected + } + return nil +} + +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) + if err != nil { + return 0, err + } + return command.RowsAffected(), nil +} + func (r *WorkRepository) SaveAgentLease(ctx context.Context, lease work.AgentLease) error { err := r.pool.QueryRow(ctx, `INSERT INTO work.agent_leases(lease_id,run_id,worker_id,fencing_token,state,acquired_at_unix_millis,expires_at_unix_millis) VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT(run_id) DO UPDATE SET lease_id=EXCLUDED.lease_id,worker_id=EXCLUDED.worker_id,fencing_token=EXCLUDED.fencing_token,state=EXCLUDED.state,acquired_at_unix_millis=EXCLUDED.acquired_at_unix_millis,expires_at_unix_millis=EXCLUDED.expires_at_unix_millis WHERE work.agent_leases.state='EXPIRED' OR work.agent_leases.expires_at_unix_millis <= EXCLUDED.acquired_at_unix_millis RETURNING run_id`, lease.ID, lease.RunID, lease.WorkerID, lease.FencingToken, lease.State, lease.AcquiredAt.UnixMilli(), lease.ExpiresAt.UnixMilli()).Scan(new(string)) if err != nil { diff --git a/cmd/agent-worker/main.go b/cmd/agent-worker/main.go new file mode 100644 index 0000000..d793637 --- /dev/null +++ b/cmd/agent-worker/main.go @@ -0,0 +1,37 @@ +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "os" + "time" + + "git.cin.su/m/Assistant/adapters/postgres" + "git.cin.su/m/Assistant/contracts/work" + "git.cin.su/m/Assistant/modules/work/application" +) + +func main() { + workerID := flag.String("worker-id", "agent-worker-1", "durable worker identity") + staleAfter := flag.Duration("stale-after", 10*time.Minute, "age after which a claimed item is recovered") + leaseTTL := flag.Duration("lease-ttl", 15*time.Minute, "dry-run lease duration") + flag.Parse() + + repository, err := postgres.OpenWork(context.Background(), postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("WORK_POSTGRES_USER"), Password: os.Getenv("WORK_POSTGRES_PASSWORD")}) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + defer repository.Close() + result, err := application.ExecuteWorkerCycle(context.Background(), repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + if err := json.NewEncoder(os.Stdout).Encode(result); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} diff --git a/contracts/work/dispatch_queue.go b/contracts/work/dispatch_queue.go new file mode 100644 index 0000000..563c2d0 --- /dev/null +++ b/contracts/work/dispatch_queue.go @@ -0,0 +1,19 @@ +package work + +type AgentDispatchQueueState string + +const ( + AgentDispatchQueued AgentDispatchQueueState = "QUEUED" + AgentDispatchClaimed AgentDispatchQueueState = "CLAIMED" + AgentDispatchCompleted AgentDispatchQueueState = "COMPLETED" + AgentDispatchFailed AgentDispatchQueueState = "FAILED" +) + +type AgentDispatchQueueItem struct { + DispatchID string + State AgentDispatchQueueState + AvailableAtUnixMillis int64 + Attempts int + ClaimedBy string + UpdatedAtUnixMillis int64 +} diff --git a/database/migrations/030_work_agent_dispatch_queue.sql b/database/migrations/030_work_agent_dispatch_queue.sql new file mode 100644 index 0000000..d9788cd --- /dev/null +++ b/database/migrations/030_work_agent_dispatch_queue.sql @@ -0,0 +1,22 @@ +BEGIN; + +SET ROLE work_role; + +CREATE TABLE IF NOT EXISTS work.agent_dispatch_queue ( + dispatch_id TEXT PRIMARY KEY REFERENCES work.agent_dispatch_intents(dispatch_id), + state TEXT NOT NULL, + available_at_unix_millis BIGINT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + claimed_by TEXT NOT NULL DEFAULT '', + updated_at_unix_millis BIGINT NOT NULL, + CHECK (state IN ('QUEUED','CLAIMED','COMPLETED','FAILED')), + CHECK (available_at_unix_millis > 0), + CHECK (attempts >= 0), + CHECK (updated_at_unix_millis > 0) +); + +CREATE INDEX IF NOT EXISTS agent_dispatch_queue_claim_idx + ON work.agent_dispatch_queue(state, available_at_unix_millis, dispatch_id); + +RESET ROLE; +COMMIT; diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index aeaa1e8..7a1c4a2 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000128-agent-canary-compose-profile.json", - "task_id": "DEV-000128", + "task_file": "dev/tasks/DEV-000130-agent-worker-recovery.json", + "task_id": "DEV-000130", "task_revision": 1 } diff --git a/dev/tasks/DEV-000129-agent-dispatch-queue.json b/dev/tasks/DEV-000129-agent-dispatch-queue.json new file mode 100644 index 0000000..df415d1 --- /dev/null +++ b/dev/tasks/DEV-000129-agent-dispatch-queue.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000129", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "bab96b7", + "risk_tier": "E", + "intent": "Add durable Work-owned dispatch queue with atomic claim, retry attempt count and terminal state transitions as the first real dispatcher runtime foundation.", + "requirement_ids": ["WP-301", "Durable dispatcher queue", "Atomic claim"], + "allowed_change_scope": {"modules": ["work-agent-execution", "work-storage"], "paths": ["database/migrations/030_work_agent_dispatch_queue.sql", "contracts/work/dispatch_queue.go", "adapters/postgres/work_repository.go", "tests/agent_execution/dispatch_queue_sql_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000129-agent-dispatch-queue.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/**", "runtime/**", ".gitea/**"]}, + "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": ["DEV-000121 durable persistence", "DEV-000120 dispatch intent"], "may_change": ["Work dispatch queue contract and repository methods"]}, + "semantic_owner": "work-agent-execution", + "canonical_state": ["work.agent_dispatch_queue"], + "state_transitions": ["QUEUED -> CLAIMED -> COMPLETED/FAILED"], + "side_effects_and_privileges": ["Work schema only; no cross-owner SQL; no runtime entrypoint"], + "persistence_and_migrations": ["Append-only migration 030; work_role only; production apply is a later canary step"], + "failure_retry_restart": ["SKIP LOCKED prevents duplicate claim; attempts are durable; terminal transition is worker-fenced"], + "assumptions": ["A later worker task will enqueue intents and execute claimed work"], + "acceptance_scenarios": ["Queue schema is durable", "Concurrent claim uses row lock and SKIP LOCKED", "Wrong worker cannot complete another claim"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check"], + "terminal_gap_states": ["BLOCKED_BY_MIGRATION_DECISION", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] +} diff --git a/dev/tasks/DEV-000130-agent-worker-recovery.json b/dev/tasks/DEV-000130-agent-worker-recovery.json new file mode 100644 index 0000000..a9141b1 --- /dev/null +++ b/dev/tasks/DEV-000130-agent-worker-recovery.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000130", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "323a67a", + "risk_tier": "E", + "intent": "Implement a one-shot durable agent worker cycle that recovers stale claims, claims one queue item, executes the deterministic worker path and completes it with worker fencing.", + "requirement_ids": ["WP-301", "Agent worker", "Stale-run/claim recovery"], + "allowed_change_scope": {"modules": ["work-agent-execution", "work-storage"], "paths": ["adapters/postgres/work_repository.go", "modules/work/ports/worker.go", "modules/work/application/worker.go", "cmd/agent-worker/main.go", "tests/agent_execution/worker_sql_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000130-agent-worker-recovery.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "database/**", "runtime/**", ".gitea/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0}, + "contracts": {"consume": ["DEV-000129 durable queue", "DEV-000122 deterministic worker", "DEV-000123 orchestration"], "may_change": ["WorkerStore port and one-shot worker application"]}, + "semantic_owner": "work-agent-execution", + "canonical_state": ["Queue state and dispatch intent remain Work-owned"], + "state_transitions": ["stale CLAIMED -> QUEUED -> CLAIMED -> COMPLETED/FAILED"], + "side_effects_and_privileges": ["Explicit one-shot command; no unmanaged goroutine or in-memory queue"], + "persistence_and_migrations": ["Reuses migration 030; production apply follows separate canary task"], + "failure_retry_restart": ["Stale recovery is durable; worker completion requires claimed_by match; failures become FAILED"], + "assumptions": ["Dispatcher enqueue path is supplied by a later integration task", "Deterministic worker remains the only implementation in this canary stage"], + "acceptance_scenarios": ["Worker recovers stale claims", "Worker claim is row-locked", "Wrong worker cannot complete claim", "Worker cycle is one-shot and restartable"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check"], + "terminal_gap_states": ["BLOCKED_BY_MIGRATION_DECISION", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] +} diff --git a/modules/work/application/worker.go b/modules/work/application/worker.go new file mode 100644 index 0000000..62638b9 --- /dev/null +++ b/modules/work/application/worker.go @@ -0,0 +1,39 @@ +package application + +import ( + "context" + "time" + + "git.cin.su/m/Assistant/contracts/work" + "git.cin.su/m/Assistant/modules/work/ports" +) + +type WorkerCycleResult struct { + RecoveredClaims int64 + DispatchID string + Completed bool +} + +func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID string, token work.AgentFencingToken, now time.Time, staleAfter, leaseTTL time.Duration) (WorkerCycleResult, error) { + recovered, err := store.RecoverStaleClaims(ctx, now.UnixMilli(), staleAfter.Milliseconds()) + if err != nil { + return WorkerCycleResult{}, err + } + item, err := store.ClaimNextDispatch(ctx, workerID, now.UnixMilli()) + if err != nil { + return WorkerCycleResult{RecoveredClaims: recovered}, err + } + intent, err := store.LoadDispatchIntent(ctx, item.DispatchID) + if err != nil { + _ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli()) + 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()) + return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err + } + if err := store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchCompleted, now.UnixMilli()); err != nil { + return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err + } + return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID, Completed: true}, nil +} diff --git a/modules/work/ports/worker.go b/modules/work/ports/worker.go new file mode 100644 index 0000000..5f63b44 --- /dev/null +++ b/modules/work/ports/worker.go @@ -0,0 +1,15 @@ +package ports + +import ( + "context" + + "git.cin.su/m/Assistant/contracts/work" +) + +type WorkerStore interface { + ExecutionStore + LoadDispatchIntent(context.Context, string) (work.AgentDispatchIntent, error) + ClaimNextDispatch(context.Context, string, int64) (work.AgentDispatchQueueItem, error) + CompleteDispatch(context.Context, string, string, work.AgentDispatchQueueState, int64) error + RecoverStaleClaims(context.Context, int64, int64) (int64, error) +} diff --git a/tests/agent_execution/dispatch_queue_sql_test.go b/tests/agent_execution/dispatch_queue_sql_test.go new file mode 100644 index 0000000..b40dff8 --- /dev/null +++ b/tests/agent_execution/dispatch_queue_sql_test.go @@ -0,0 +1,20 @@ +package agent_execution_test + +import ( + "os" + "strings" + "testing" +) + +func TestDispatchQueueMigrationDefinesClaimableDurableState(t *testing.T) { + data, err := os.ReadFile("../../database/migrations/030_work_agent_dispatch_queue.sql") + if err != nil { + t.Fatal(err) + } + sql := string(data) + for _, marker := range []string{"work.agent_dispatch_queue", "state IN ('QUEUED','CLAIMED','COMPLETED','FAILED')", "CREATE INDEX IF NOT EXISTS", "REFERENCES work.agent_dispatch_intents"} { + if !strings.Contains(sql, marker) { + t.Fatalf("queue contract marker missing: %s", marker) + } + } +} diff --git a/tests/agent_execution/worker_sql_test.go b/tests/agent_execution/worker_sql_test.go new file mode 100644 index 0000000..a8d6b21 --- /dev/null +++ b/tests/agent_execution/worker_sql_test.go @@ -0,0 +1,20 @@ +package agent_execution_test + +import ( + "os" + "strings" + "testing" +) + +func TestWorkerRecoveryAndClaimUseDurableQueueStatements(t *testing.T) { + data, err := os.ReadFile("../../adapters/postgres/work_repository.go") + if err != nil { + t.Fatal(err) + } + source := string(data) + for _, marker := range []string{"FOR UPDATE SKIP LOCKED", "state='CLAIMED'", "state='QUEUED'", "RecoverStaleClaims"} { + if !strings.Contains(source, marker) { + t.Fatalf("worker persistence marker missing: %s", marker) + } + } +}