diff --git a/adapters/agentbridge/http.go b/adapters/agentbridge/http.go new file mode 100644 index 0000000..aed5214 --- /dev/null +++ b/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 +} diff --git a/adapters/agentbridge/http_test.go b/adapters/agentbridge/http_test.go new file mode 100644 index 0000000..c83bce2 --- /dev/null +++ b/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) + } +} diff --git a/cmd/agent-worker/main.go b/cmd/agent-worker/main.go index 55a5bd4..4635317 100644 --- a/cmd/agent-worker/main.go +++ b/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 { diff --git a/contracts/work/bridge.go b/contracts/work/bridge.go new file mode 100644 index 0000000..cbade58 --- /dev/null +++ b/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"` +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 1a587d0..d371789 100644 --- a/dev/ACTIVE_TASK.json +++ b/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 } diff --git a/dev/tasks/DEV-000145-agent-execution-bridge-runtime.json b/dev/tasks/DEV-000145-agent-execution-bridge-runtime.json new file mode 100644 index 0000000..d8f452c --- /dev/null +++ b/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"] +} diff --git a/modules/work/application/agent_dispatch.go b/modules/work/application/agent_dispatch.go new file mode 100644 index 0000000..1817e74 --- /dev/null +++ b/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 +} diff --git a/modules/work/application/agent_dispatch_test.go b/modules/work/application/agent_dispatch_test.go new file mode 100644 index 0000000..f28e4f0 --- /dev/null +++ b/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) + } +} diff --git a/modules/work/application/worker.go b/modules/work/application/worker.go index b687a94..9331ea3 100644 --- a/modules/work/application/worker.go +++ b/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 +} diff --git a/modules/work/ports/execution.go b/modules/work/ports/execution.go index 0dd711f..51f1a79 100644 --- a/modules/work/ports/execution.go +++ b/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 diff --git a/modules/work/ports/worker.go b/modules/work/ports/worker.go index 5f63b44..aaa44aa 100644 --- a/modules/work/ports/worker.go +++ b/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)