feat(work): add durable agent dispatch queue

Reviewed-on: #97

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 {

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
+}

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;

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
 }

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"]
+}

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)
+		}
+	}
+}