chore: format bridge tests

adapters/agentbridge/http.go

@@ -0,0 +1,61 @@
+package agentbridge
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+var ErrUnavailable = errors.New("agent execution bridge unavailable")
+
+type HTTPExecutor struct {
+	URL    string
+	Client *http.Client
+}
+
+func NewHTTPExecutor(url string, client *http.Client) (*HTTPExecutor, error) {
+	url = strings.TrimSpace(url)
+	if url == "" || !strings.HasPrefix(url, "http://") && !strings.HasPrefix(url, "https://") {
+		return nil, ErrUnavailable
+	}
+	if client == nil {
+		client = http.DefaultClient
+	}
+	return &HTTPExecutor{URL: strings.TrimRight(url, "/"), Client: client}, nil
+}
+
+func (e *HTTPExecutor) Execute(ctx context.Context, request work.AgentExecutionRequest) (work.AgentExecutionResult, error) {
+	body, err := json.Marshal(request)
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	httpRequest, err := http.NewRequestWithContext(ctx, http.MethodPost, e.URL+"/v1/execute", bytes.NewReader(body))
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	httpRequest.Header.Set("Content-Type", "application/json")
+	response, err := e.Client.Do(httpRequest)
+	if err != nil {
+		return work.AgentExecutionResult{}, fmt.Errorf("%w: %v", ErrUnavailable, err)
+	}
+	defer response.Body.Close()
+	if response.StatusCode < 200 || response.StatusCode >= 300 {
+		return work.AgentExecutionResult{}, fmt.Errorf("%w: status=%s", ErrUnavailable, response.Status)
+	}
+	decoder := json.NewDecoder(io.LimitReader(response.Body, 1<<20))
+	var result work.AgentExecutionResult
+	if err := decoder.Decode(&result); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if result.State == "" || result.Summary == "" || result.ObservedAtUnixMillis < 1 {
+		return work.AgentExecutionResult{}, errors.New("agent execution bridge returned invalid result")
+	}
+	return result, nil
+}

adapters/agentbridge/http_test.go

@@ -0,0 +1,43 @@
+package agentbridge
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+func TestHTTPExecutorCarriesTypedDispatchAndDecodesResult(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method != http.MethodPost || r.URL.Path != "/v1/execute" {
+			t.Fatalf("unexpected request: %s %s", r.Method, r.URL.Path)
+		}
+		var request work.AgentExecutionRequest
+		if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+			t.Fatal(err)
+		}
+		if request.WorkerID != "worker-1" || request.FencingToken != 7 || request.Intent.DispatchID != "dispatch-1" {
+			t.Fatalf("request=%+v", request)
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "completed", ObservedAtUnixMillis: 2})
+	}))
+	defer server.Close()
+	executor, err := NewHTTPExecutor(server.URL, server.Client())
+	if err != nil {
+		t.Fatal(err)
+	}
+	result, err := executor.Execute(context.Background(), work.AgentExecutionRequest{Intent: work.AgentDispatchIntent{DispatchID: "dispatch-1"}, WorkerID: "worker-1", FencingToken: 7})
+	if err != nil || result.State != work.AgentExecutionCompleted || result.Summary != "completed" {
+		t.Fatalf("result=%+v err=%v", result, err)
+	}
+}
+
+func TestHTTPExecutorRejectsUnavailableEndpoint(t *testing.T) {
+	if _, err := NewHTTPExecutor("", nil); err != ErrUnavailable {
+		t.Fatalf("expected unavailable, got %v", err)
+	}
+}

cmd/agent-worker/main.go

@@ -5,13 +5,16 @@ import (
 	"encoding/json"
 	"flag"
 	"fmt"
+	"net/http"
 	"os"
 	"time"
 
+	"git.cin.su/m/Assistant/adapters/agentbridge"
 	"git.cin.su/m/Assistant/adapters/postgres"
 	secretcontracts "git.cin.su/m/Assistant/contracts/secrets"
 	"git.cin.su/m/Assistant/contracts/work"
 	"git.cin.su/m/Assistant/modules/work/application"
+	"git.cin.su/m/Assistant/modules/work/ports"
 	platformsecrets "git.cin.su/m/Assistant/platform/secrets"
 )
 
@@ -34,9 +37,25 @@ func main() {
 		os.Exit(1)
 	}
 	defer repository.Close()
+	var executor ports.AgentExecutor
+	if bridgeURL := os.Getenv("AGENT_EXECUTION_BRIDGE_URL"); bridgeURL != "" {
+		executor, err = agentbridge.NewHTTPExecutor(bridgeURL, &http.Client{Timeout: *leaseTTL})
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err)
+			os.Exit(1)
+		}
+	} else {
+		fmt.Fprintln(os.Stderr, "agent execution bridge is not configured; using legacy dry-run dispatch")
+	}
 	ctx := context.Background()
 	for {
-		result, cycleErr := application.ExecuteWorkerCycle(ctx, repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
+		var result application.WorkerCycleResult
+		var cycleErr error
+		if executor == nil {
+			result, cycleErr = application.ExecuteWorkerCycle(ctx, repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
+		} else {
+			result, cycleErr = application.ExecuteWorkerCycleWithExecutor(ctx, repository, executor, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL)
+		}
 		if cycleErr != nil {
 			fmt.Fprintln(os.Stderr, cycleErr)
 			if *once {

contracts/work/bridge.go

@@ -0,0 +1,32 @@
+package work
+
+import "time"
+
+// AgentExecutionRequest is the serializable hand-off from the durable worker
+// to an executor. The bridge receives the already-authorized dispatch context;
+// it does not receive database credentials or ambient authority.
+type AgentExecutionRequest struct {
+	Intent         AgentDispatchIntent `json:"intent"`
+	WorkerID       string              `json:"worker_id"`
+	FencingToken   AgentFencingToken   `json:"fencing_token"`
+	LeaseExpiresAt time.Time           `json:"lease_expires_at"`
+}
+
+type AgentExecutionState string
+
+const (
+	AgentExecutionCompleted AgentExecutionState = "COMPLETED"
+	AgentExecutionWaiting   AgentExecutionState = "WAITING"
+	AgentExecutionFailed    AgentExecutionState = "FAILED"
+)
+
+// AgentExecutionResult is an acknowledgement of the external execution
+// attempt. The worker persists it through the work owner before completing a
+// dispatch, so a bridge timeout never becomes an unrecorded success.
+type AgentExecutionResult struct {
+	State                AgentExecutionState `json:"state"`
+	Summary              string              `json:"summary"`
+	OutcomeRef           string              `json:"outcome_ref"`
+	ArtifactRefs         []string            `json:"artifact_refs"`
+	ObservedAtUnixMillis int64               `json:"observed_at_unix_millis"`
+}

dev/ACTIVE_TASK.json

@@ -1,5 +1,5 @@
 {
-  "task_file": "dev/tasks/DEV-000144-fleet-readiness-runtime.json",
-  "task_id": "DEV-000144",
+  "task_file": "dev/tasks/DEV-000145-agent-execution-bridge-runtime.json",
+  "task_id": "DEV-000145",
   "task_revision": 1
 }

dev/tasks/DEV-000145-agent-execution-bridge-runtime.json

@@ -0,0 +1,23 @@
+{
+  "task_id": "DEV-000145",
+  "task_revision": 1,
+  "architecture_revision": "ADR-0018",
+  "base_repo_revision": "7b06cd64a279db5aec67ada5f3c705f114ffd86b",
+  "risk_tier": "E",
+  "intent": "Connect the durable agent worker to an explicit typed external execution bridge with lease and callback evidence.",
+  "requirement_ids": ["WP-301", "Agent Execution runtime and recovery"],
+  "allowed_change_scope": {"modules": ["work", "agent-worker", "agentbridge"], "paths": ["contracts/work/bridge.go", "adapters/agentbridge/**", "modules/work/application/**", "modules/work/ports/**", "cmd/agent-worker/**", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000145-agent-execution-bridge-runtime.json"]},
+  "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "deploy/**", "security/**", ".gitea/**", "contracts/authority/**"]},
+  "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": ["AgentDispatchIntent", "AgentLease", "AgentCallback"], "may_change": ["serializable bridge request/result contract"]},
+  "semantic_owner": "work",
+  "canonical_state": ["work agent lease and callback head"],
+  "state_transitions": ["durable dispatch -> lease/heartbeat -> bridge result -> result callback -> completed or explicit waiting/failure"],
+  "side_effects_and_privileges": ["outbound HTTP call to configured execution bridge only; no credentials forwarded"],
+  "persistence_and_migrations": ["reuse existing work lease/callback persistence; no migration"],
+  "failure_retry_restart": ["bridge timeout is failure of the current attempt; durable dispatch remains evidence-bearing and stale claims are recoverable"],
+  "assumptions": ["an independently operated bridge will implement POST /v1/execute", "bridge URL is deployment configuration and absent URL preserves legacy dry-run until bridge capability is deployed"],
+  "acceptance_scenarios": ["typed bridge request reaches executor", "completed bridge result persists callback evidence", "invalid/unavailable bridge fails closed", "existing stale-claim recovery remains intact"],
+  "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"],
+  "terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"]
+}

modules/work/application/agent_dispatch.go

@@ -0,0 +1,52 @@
+package application
+
+import (
+	"context"
+	"time"
+
+	"git.cin.su/m/Assistant/contracts/work"
+	"git.cin.su/m/Assistant/modules/work/domain"
+	"git.cin.su/m/Assistant/modules/work/ports"
+)
+
+func ExecuteAgentDispatch(ctx context.Context, store ports.ExecutionStore, executor ports.AgentExecutor, intent work.AgentDispatchIntent, workerID string, token work.AgentFencingToken, now time.Time, ttl time.Duration) (work.AgentExecutionResult, error) {
+	if executor == nil {
+		return work.AgentExecutionResult{}, ports.ErrAgentExecutorUnavailable
+	}
+	if err := domain.ValidateDispatchIntent(intent); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	lease, err := domain.AcquireAgentLease(intent.RunID, workerID, token, now, ttl)
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.SaveDispatchIntent(ctx, intent); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.SaveAgentLease(ctx, lease); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	heartbeat := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 1, Kind: work.AgentCallbackHeartbeat, ObservedAt: now}
+	if err := domain.ValidateAgentCallback(heartbeat, lease, 0); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.AcceptAgentCallback(ctx, heartbeat); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	result, err := executor.Execute(ctx, work.AgentExecutionRequest{Intent: intent, WorkerID: workerID, FencingToken: token, LeaseExpiresAt: lease.ExpiresAt})
+	if err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	observedAt := time.UnixMilli(result.ObservedAtUnixMillis)
+	callback := work.AgentCallback{RunID: intent.RunID, WorkerID: workerID, FencingToken: token, Sequence: 2, Kind: work.AgentCallbackResult, ObservedAt: observedAt}
+	if result.State == work.AgentExecutionWaiting {
+		callback.Kind = work.AgentCallbackCheckpoint
+	}
+	if err := domain.ValidateAgentCallback(callback, lease, heartbeat.Sequence); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	if err := store.AcceptAgentCallback(ctx, callback); err != nil {
+		return work.AgentExecutionResult{}, err
+	}
+	return result, nil
+}

modules/work/application/agent_dispatch_test.go

@@ -0,0 +1,42 @@
+package application
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.cin.su/m/Assistant/contracts/work"
+)
+
+type bridgeStore struct {
+	callbacks []work.AgentCallback
+	lease     work.AgentLease
+}
+
+func (s *bridgeStore) SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error { return nil }
+func (s *bridgeStore) SaveAgentLease(_ context.Context, lease work.AgentLease) error {
+	s.lease = lease
+	return nil
+}
+func (s *bridgeStore) AcceptAgentCallback(_ context.Context, callback work.AgentCallback) error {
+	s.callbacks = append(s.callbacks, callback)
+	return nil
+}
+
+type bridgeExecutor struct{}
+
+func (bridgeExecutor) Execute(context.Context, work.AgentExecutionRequest) (work.AgentExecutionResult, error) {
+	return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "done", ObservedAtUnixMillis: 2}, nil
+}
+
+func TestExecuteAgentDispatchPersistsLeaseAndBridgeCallbacks(t *testing.T) {
+	store := &bridgeStore{}
+	now := time.UnixMilli(1)
+	result, err := ExecuteAgentDispatch(context.Background(), store, bridgeExecutor{}, work.AgentDispatchIntent{DispatchID: "d", IdempotencyKey: "i", DelegationID: "del", RunID: "run", ImplementationID: "impl", ImplementationRevision: 1, SourceRevision: "source", WorktreeID: "tree", VerificationPlanRef: "verify", CreatedAtUnixMillis: 1}, "worker", 1, now, time.Minute)
+	if err != nil || result.State != work.AgentExecutionCompleted {
+		t.Fatalf("result=%+v err=%v", result, err)
+	}
+	if len(store.callbacks) != 2 || store.callbacks[0].Kind != work.AgentCallbackHeartbeat || store.callbacks[1].Kind != work.AgentCallbackResult {
+		t.Fatalf("callbacks=%+v", store.callbacks)
+	}
+}

modules/work/application/worker.go

@@ -41,3 +41,34 @@ func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID s
 	}
 	return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID, Completed: true}, nil
 }
+
+func ExecuteWorkerCycleWithExecutor(ctx context.Context, store ports.WorkerStore, executor ports.AgentExecutor, 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
+	}
+	if item.DispatchID == "" {
+		return WorkerCycleResult{RecoveredClaims: recovered, Idle: true}, nil
+	}
+	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
+	}
+	result, err := ExecuteAgentDispatch(ctx, store, executor, intent, workerID, token, now, leaseTTL)
+	if err != nil {
+		_ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli())
+		return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, err
+	}
+	if result.State != work.AgentExecutionCompleted {
+		return WorkerCycleResult{RecoveredClaims: recovered, DispatchID: item.DispatchID}, nil
+	}
+	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/execution.go

@@ -6,6 +6,10 @@ import (
 	"git.cin.su/m/Assistant/contracts/work"
 )
 
+type AgentExecutor interface {
+	Execute(context.Context, work.AgentExecutionRequest) (work.AgentExecutionResult, error)
+}
+
 type ExecutionStore interface {
 	SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error
 	SaveAgentLease(context.Context, work.AgentLease) error

modules/work/ports/worker.go

@@ -2,10 +2,13 @@ package ports
 
 import (
 	"context"
+	"errors"
 
 	"git.cin.su/m/Assistant/contracts/work"
 )
 
+var ErrAgentExecutorUnavailable = errors.New("agent executor unavailable")
+
 type WorkerStore interface {
 	ExecutionStore
 	LoadDispatchIntent(context.Context, string) (work.AgentDispatchIntent, error)