feat(work): add one-shot agent worker recovery

adapters/postgres/work_repository.go

@@ -68,6 +68,15 @@ 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
@@ -112,6 +121,14 @@ func (r *WorkRepository) CompleteDispatch(ctx context.Context, dispatchID, worke
 	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)
+	}
+}

dev/ACTIVE_TASK.json

@@ -1,5 +1,5 @@
 {
-  "task_file": "dev/tasks/DEV-000129-agent-dispatch-queue.json",
-  "task_id": "DEV-000129",
+  "task_file": "dev/tasks/DEV-000130-agent-worker-recovery.json",
+  "task_id": "DEV-000130",
   "task_revision": 1
 }

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