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 {

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

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-000130-agent-worker-recovery.json",
+  "task_id": "DEV-000130",
   "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"]
+}

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

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

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

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

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