From b1a79ede91672c344be8120b752768e717d8c710 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Mon, 24 Aug 2026 00:27:10 +0300 Subject: [PATCH] feat: add local agent execution bridge --- cmd/agent-execution-bridge/main.go | 162 ++++++++++++++++++ cmd/agent-execution-bridge/main_test.go | 55 ++++++ cmd/agent-worker/main.go | 15 +- deploy/docker-compose.prod.yml | 20 +++ dev/ACTIVE_TASK.json | 4 +- ...V-000266-local-agent-execution-bridge.json | 26 +++ 6 files changed, 279 insertions(+), 3 deletions(-) create mode 100644 cmd/agent-execution-bridge/main.go create mode 100644 cmd/agent-execution-bridge/main_test.go create mode 100644 dev/tasks/DEV-000266-local-agent-execution-bridge.json diff --git a/cmd/agent-execution-bridge/main.go b/cmd/agent-execution-bridge/main.go new file mode 100644 index 0000000..2b06611 --- /dev/null +++ b/cmd/agent-execution-bridge/main.go @@ -0,0 +1,162 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "strings" + "time" + + "git.cin.su/m/Assistant/contracts/artifact" + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" + capabilitydomain "git.cin.su/m/Assistant/modules/capability/domain" +) + +type bridge struct { + cognitiveURL string + callbackURL string + verifierURL string + client *http.Client +} + +func main() { + b := bridge{ + cognitiveURL: strings.TrimRight(envOr("COGNITIVE_WORKER_URL", "http://cognitive-worker:8083"), "/"), + callbackURL: strings.TrimRight(envOr("WORK_GRAPH_RUNTIME_URL", "http://work-graph-runtime:8088"), "/"), + verifierURL: strings.TrimRight(envOr("TASK_VERIFIER_URL", "http://task-verifier:8091"), "/"), + client: &http.Client{Timeout: 10 * time.Minute}, + } + mux := http.NewServeMux() + mux.HandleFunc("/healthz", healthHandler) + mux.HandleFunc("/readyz", healthHandler) + mux.HandleFunc("/v1/execute", b.executeHandler) + addr := envOr("LISTEN_ADDR", ":8092") + if err := http.ListenAndServe(addr, mux); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } +} + +func (b bridge) executeHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + var request work.AgentExecutionRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&request); err != nil { + http.Error(w, "invalid execution request", http.StatusBadRequest) + return + } + result, err := b.execute(r.Context(), request) + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(result) +} + +func (b bridge) execute(ctx context.Context, request work.AgentExecutionRequest) (work.AgentExecutionResult, error) { + now := time.Now().UnixMilli() + intent := request.Intent + if intent.TaskEvaluationExecution == nil || intent.TaskEvaluationProfile == nil || len(intent.TaskEvaluationCases) == 0 { + return work.AgentExecutionResult{State: work.AgentExecutionWaiting, Summary: "локальный bridge не поддерживает это намерение; работа сохранена и ожидает подходящего исполнителя", ObservedAtUnixMillis: now}, nil + } + execution := intent.TaskEvaluationExecution + if execution.GraphID == "" || execution.NodeID == "" || execution.BindingRevision < 1 || execution.ActorID == "" || execution.PrincipalID == "" || execution.ScopeID == "" { + return work.AgentExecutionResult{State: work.AgentExecutionWaiting, Summary: "контекст task-evaluation неполон; работа сохранена для повторной маршрутизации", ObservedAtUnixMillis: now}, nil + } + payload := capability.TaskEvaluationExecutionRequest{Profile: *intent.TaskEvaluationProfile, CapabilityID: intent.TaskEvaluationProfile.CapabilityID, BindingID: intent.TaskEvaluationBindingID, BindingRevision: execution.BindingRevision, ActorID: execution.ActorID, PrincipalID: execution.PrincipalID, ScopeID: execution.ScopeID, Cases: intent.TaskEvaluationCases, MaxOutputBytes: 4096} + response, err := postJSON[capability.TaskEvaluationExecutionResponse](ctx, b.client, b.cognitiveURL+"/v1/task-evaluations/execute", payload) + if err != nil { + return work.AgentExecutionResult{}, err + } + if len(response.Responses) == 0 || response.ObservedAtUnixMillis < 1 { + return work.AgentExecutionResult{}, fmt.Errorf("task evaluation returned no observations") + } + caseByID := make(map[string]capability.TaskEvaluationCase, len(intent.TaskEvaluationCases)) + for _, item := range intent.TaskEvaluationCases { + caseByID[item.CaseID] = item + } + observations := response.Observations + if !hasCriticalFailure(observations) { + observations = make([]capability.TaskCaseObservation, 0, len(response.Responses)) + for _, item := range response.Responses { + caseValue, ok := caseByID[item.CaseID] + if !ok || caseValue.Verification.Mode == "" { + return work.AgentExecutionResult{State: work.AgentExecutionWaiting, Summary: "task-evaluation case has no bounded verifier; work remains waiting", ObservedAtUnixMillis: time.Now().UnixMilli()}, nil + } + verified, verifyErr := postJSON[capability.TaskCaseVerificationResponse](ctx, b.client, b.verifierURL+"/v1/task-evaluations/verify", capability.TaskCaseVerificationRequest{CaseID: item.CaseID, ResponseText: item.ResponseText, LatencyMicros: item.LatencyMicros, EvidenceArtifactRef: item.EvidenceArtifactRef, Verification: caseValue.Verification}) + if verifyErr != nil { + return work.AgentExecutionResult{}, verifyErr + } + observations = append(observations, capability.TaskCaseObservation{CaseID: verified.CaseID, Result: verified.Result, QualityScoreMillis: verified.QualityScoreMillis, LatencyMicros: verified.LatencyMicros, CriticalFailure: verified.CriticalFailure, EvidenceArtifactRef: verified.EvidenceArtifactRef, VerificationEvidenceRef: verified.VerificationEvidenceRef, VerifierIdentity: verified.VerifierIdentity}) + } + } + evaluation, err := capabilitydomain.BuildTaskEvaluation(*intent.TaskEvaluationProfile, capability.TaskEvaluationID("evaluation:"+intent.DispatchID), response.BindingID, response.BindingRevision, intent.TaskEvaluationProfile.CapabilityID, observations, artifact.ArtifactID(intent.TaskEvaluationCases[0].EvidenceArtifactRef), response.ObservedAtUnixMillis) + if err != nil { + return work.AgentExecutionResult{}, err + } + callback := work.TaskEvaluationEvidenceCallback{GraphID: execution.GraphID, Revision: execution.GraphRevision, NodeID: execution.NodeID, Evaluation: evaluation, Succeeded: true} + if _, err := postJSON[json.RawMessage](ctx, b.client, b.callbackURL+"/v1/work-graphs/task-evaluations/callback", callback); err != nil { + return work.AgentExecutionResult{}, err + } + return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "local bridge executed task evaluation and persisted verifier callback", OutcomeRef: string(evaluation.ID), ArtifactRefs: []string{string(evaluation.ArtifactID)}, ObservedAtUnixMillis: time.Now().UnixMilli()}, nil +} + +func hasCriticalFailure(observations []capability.TaskCaseObservation) bool { + for _, observation := range observations { + if observation.CriticalFailure { + return true + } + } + return false +} + +func postJSON[T any](ctx context.Context, client *http.Client, url string, payload any) (T, error) { + var result T + body, err := json.Marshal(payload) + if err != nil { + return result, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return result, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := client.Do(req) + if err != nil { + return result, err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return result, fmt.Errorf("POST %s: status %s", url, resp.Status) + } + if resp.StatusCode == http.StatusNoContent { + return result, nil + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return result, err + } + return result, nil +} + +func healthHandler(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"runtime":"agent-execution-bridge","status":"ok","capabilities":["task-evaluation"]}`)) +} + +func envOr(name, fallback string) string { + if value := strings.TrimSpace(os.Getenv(name)); value != "" { + return value + } + return fallback +} diff --git a/cmd/agent-execution-bridge/main_test.go b/cmd/agent-execution-bridge/main_test.go new file mode 100644 index 0000000..34bef5e --- /dev/null +++ b/cmd/agent-execution-bridge/main_test.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" +) + +func TestBridgeWaitsForUnsupportedIntent(t *testing.T) { + b := bridge{client: http.DefaultClient} + result, err := b.execute(context.Background(), work.AgentExecutionRequest{Intent: work.AgentDispatchIntent{DispatchID: "dispatch-unsupported"}}) + if err != nil { + t.Fatal(err) + } + if result.State != work.AgentExecutionWaiting || result.ObservedAtUnixMillis < 1 { + t.Fatalf("result=%+v", result) + } +} + +func TestHealthHandlerExposesBoundedCapability(t *testing.T) { + recorder := httptest.NewRecorder() + request := httptest.NewRequest(http.MethodGet, "/healthz", nil) + healthHandler(recorder, request) + if recorder.Code != http.StatusOK || recorder.Body.String() == "" { + t.Fatalf("status=%d body=%q", recorder.Code, recorder.Body.String()) + } +} + +func TestBridgeRunsTaskEvaluationThroughVerifierAndCallback(t *testing.T) { + profile := capability.TaskEvaluationProfile{ID: "profile-1", Revision: 1, CapabilityID: "capability-1", Stage: capability.TaskEvaluationGeneral, BenchmarkVersion: "general-v1", RequiredCaseIDs: []string{"case-1"}, QualityWeightMillis: 600, ReliabilityWeightMillis: 300, LatencyWeightMillis: 100} + verification := capability.TaskCaseVerification{Mode: capability.TaskCaseVerificationExactText, ExpectedText: "answer", MinimumQualityMillis: 900} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/v1/task-evaluations/execute": + _ = json.NewEncoder(w).Encode(capability.TaskEvaluationExecutionResponse{ProfileID: profile.ID, BindingID: "binding-1", BindingRevision: 3, Responses: []capability.TaskCaseResponse{{CaseID: "case-1", ResponseText: "answer", LatencyMicros: 2, EvidenceArtifactRef: "artifact-1"}}, ObservedAtUnixMillis: 10}) + case "/v1/task-evaluations/verify": + _ = json.NewEncoder(w).Encode(capability.TaskCaseVerificationResponse{CaseID: "case-1", Result: capability.EvaluationSupported, QualityScoreMillis: 900, LatencyMicros: 2, EvidenceArtifactRef: "artifact-1", VerificationEvidenceRef: "verify-1", VerifierIdentity: "verifier-1"}) + case "/v1/work-graphs/task-evaluations/callback": + w.WriteHeader(http.StatusNoContent) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + b := bridge{cognitiveURL: server.URL, callbackURL: server.URL, verifierURL: server.URL, client: server.Client()} + result, err := b.execute(context.Background(), work.AgentExecutionRequest{Intent: work.AgentDispatchIntent{DispatchID: "dispatch-1", TaskEvaluationBindingID: "binding-1", TaskEvaluationProfile: &profile, TaskEvaluationCases: []capability.TaskEvaluationCase{{CaseID: "case-1", Prompt: "answer", EvidenceArtifactRef: "artifact-1", Verification: verification}}, TaskEvaluationExecution: &work.TaskEvaluationDispatchContext{GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", BindingRevision: 3, ActorID: "actor-1", PrincipalID: "principal-1", ScopeID: "scope-1"}}}) + if err != nil || result.State != work.AgentExecutionCompleted || result.OutcomeRef == "" { + t.Fatalf("result=%+v err=%v", result, err) + } +} diff --git a/cmd/agent-worker/main.go b/cmd/agent-worker/main.go index 37024dd..d08c707 100644 --- a/cmd/agent-worker/main.go +++ b/cmd/agent-worker/main.go @@ -42,7 +42,7 @@ func main() { defer repository.Close() var executor ports.AgentExecutor localFallback := false - if bridgeURL, bridgeConfigured := loadProtectedBridgeURL(); bridgeConfigured { + if bridgeURL, bridgeConfigured := loadConfiguredBridgeURL(); bridgeConfigured { executor, err = agentbridge.NewHTTPExecutor(bridgeURL, &http.Client{Timeout: *leaseTTL}) if err != nil { fmt.Fprintln(os.Stderr, err) @@ -99,6 +99,19 @@ func main() { } } +func loadConfiguredBridgeURL() (string, bool) { + if bridgeURL, configured := loadProtectedBridgeURL(); configured { + return bridgeURL, true + } + // The URL is routing configuration, not a credential. A protected file + // remains preferred when an independently operated bridge is supplied. + value := strings.TrimSpace(os.Getenv("AGENT_EXECUTION_BRIDGE_URL")) + if value == "" || !strings.HasPrefix(value, "http://") && !strings.HasPrefix(value, "https://") { + return "", false + } + return value, true +} + func loadProtectedBridgeURL() (string, bool) { return loadProtectedBridgeURLFrom("/run/secrets", "/run/secrets/agent_execution_bridge_url") } diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index 7151afe..038df91 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -119,12 +119,32 @@ services: WORK_POSTGRES_PASSWORD_FILE: /run/secrets/work_db_password SECRET_BACKEND_ROOT: /run/secrets TASK_VERIFIER_URL: http://task-verifier:8091 + AGENT_EXECUTION_BRIDGE_URL: http://agent-execution-bridge:8092 depends_on: postgres: { condition: service_healthy } task-verifier: { condition: service_healthy } + agent-execution-bridge: { condition: service_healthy } volumes: - assistant_core_secrets:/run/secrets:ro + agent-execution-bridge: + build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: agent-execution-bridge, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } + restart: unless-stopped + environment: + LISTEN_ADDR: :8092 + COGNITIVE_WORKER_URL: http://cognitive-worker:8083 + WORK_GRAPH_RUNTIME_URL: http://work-graph-runtime:8088 + TASK_VERIFIER_URL: http://task-verifier:8091 + depends_on: + cognitive-worker: { condition: service_started } + task-verifier: { condition: service_healthy } + work-graph-runtime: { condition: service_started } + healthcheck: + test: ["CMD-SHELL", "wget -qO- http://localhost:8092/healthz >/dev/null"] + interval: 10s + timeout: 3s + retries: 10 + task-verifier: build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: task-verifier, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } restart: unless-stopped diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index c5a58f9..35f0016 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000265-agent-worker-capability-readiness-evidence.json", - "task_id": "DEV-000265", + "task_file": "dev/tasks/DEV-000266-local-agent-execution-bridge.json", + "task_id": "DEV-000266", "task_revision": 1 } diff --git a/dev/tasks/DEV-000266-local-agent-execution-bridge.json b/dev/tasks/DEV-000266-local-agent-execution-bridge.json new file mode 100644 index 0000000..41469d9 --- /dev/null +++ b/dev/tasks/DEV-000266-local-agent-execution-bridge.json @@ -0,0 +1,26 @@ +{ + "task_id": "DEV-000266", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "c4f3f02", + "risk_tier": "E", + "intent": "Provide an independently running local Agent Execution bridge for the bounded task-evaluation capability, while keeping unsupported generic work durably waiting.", + "requirement_ids": ["WP-301", "Agent Execution runtime and recovery"], + "allowed_change_scope": { + "modules": ["agent-worker", "agentbridge"], + "paths": ["adapters/agentbridge/**", "cmd/agent-worker/**", "cmd/agent-execution-bridge/**", "deploy/docker-compose.prod.yml", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000266-local-agent-execution-bridge.json", "docs/evidence/**", "docs/PROJECT_STATUS.md"] + }, + "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "security/**", "contracts/authority/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 0}, + "contracts": {"consume": ["AgentExecutionRequest", "AgentExecutionResult", "TaskEvaluationExecutionResponse", "TaskCaseVerificationResponse", "TaskEvaluationEvidenceCallback"], "may_change": []}, + "semantic_owner": "work dispatches; capability owns evaluation evidence", + "canonical_state": ["existing work lease/callback state", "existing capability evaluation evidence"], + "state_transitions": ["durable dispatch -> bridge request -> bounded evaluation/verifier/callback -> completed or explicit waiting"], + "side_effects_and_privileges": ["outbound HTTP only to cognitive-worker, task-verifier and work-graph-runtime; no database access; no credentials forwarded"], + "persistence_and_migrations": ["reuse existing owners; no migration"], + "failure_retry_restart": ["bridge or dependency failure returns an executor error; worker retains durable retry/recovery evidence", "unsupported intent returns WAITING and is never completed"], + "assumptions": ["local bridge is a bounded fallback, not a universal Codex executor", "generic work remains pending until a separately qualified capability is installed"], + "acceptance_scenarios": ["bridge receives typed request", "task-evaluation reaches cognitive worker, verifier and graph callback", "unsupported generic intent returns WAITING", "worker can use bridge-backed execution without changing lease/recovery semantics"], + "required_tests": ["unit", "runtime HTTP", "architecture", "go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "docker compose config", "git diff --check"], + "terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] +} -- 2.54.0