diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 1e9af75..5e9063b 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -68,6 +68,50 @@ func (r *WorkRepository) SaveDispatchIntent(ctx context.Context, intent work.Age return tx.Commit(ctx) } +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) 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/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..aeb368e 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-000129-agent-dispatch-queue.json", + "task_id": "DEV-000129", "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/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) + } + } +}