212 lines
11 KiB
Go
212 lines
11 KiB
Go
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/cognitive"
|
|
"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.TaskEvaluationExecution.GenericMessage != "" {
|
|
return b.executeGenericCognitiveWork(ctx, request, *intent.TaskEvaluationExecution)
|
|
}
|
|
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 (b bridge) executeGenericCognitiveWork(ctx context.Context, request work.AgentExecutionRequest, execution work.TaskEvaluationDispatchContext) (work.AgentExecutionResult, error) {
|
|
if execution.GraphID == "" || execution.NodeID == "" || execution.ActorID == "" || execution.PrincipalID == "" || execution.ScopeID == "" {
|
|
return work.AgentExecutionResult{State: work.AgentExecutionWaiting, Summary: "generic cognitive work context is incomplete; durable work remains waiting for rerouting", ObservedAtUnixMillis: time.Now().UnixMilli()}, nil
|
|
}
|
|
maxOutputBytes := execution.GenericMaxOutputBytes
|
|
if maxOutputBytes < 1 {
|
|
maxOutputBytes = 4096
|
|
}
|
|
dataPolicy := capability.DataPolicy(execution.GenericDataPolicy)
|
|
if dataPolicy == "" {
|
|
dataPolicy = capability.DataPolicyLocalOnly
|
|
}
|
|
response, err := postJSON[cognitive.GatewayAdvisoryResponse](ctx, b.client, b.cognitiveURL+"/v1/advisories", cognitive.GatewayAdvisoryRequest{
|
|
Version: cognitive.CreatorAdvisoryV1, RequestID: cognitive.RequestID("agent-work:" + request.Intent.DispatchID), ActorID: execution.ActorID,
|
|
PrincipalID: execution.PrincipalID, ScopeID: execution.ScopeID, Message: execution.GenericMessage, Locale: execution.GenericLocale,
|
|
MaxOutputBytes: maxOutputBytes, DataPolicy: dataPolicy, Meaning: cognitive.TaskMeaning{SpeechAct: "DIRECTIVE", RequestedOutcome: execution.GenericPurpose},
|
|
})
|
|
if err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if response.FinishReason != cognitive.FinishCompleted || strings.TrimSpace(response.AdvisoryText) == "" {
|
|
return work.AgentExecutionResult{}, fmt.Errorf("generic cognitive work did not complete: %s", response.FinishReason)
|
|
}
|
|
candidateID := "candidate:" + request.Intent.DispatchID
|
|
artifactRef := work.EvidenceRef("agent-output:" + request.Intent.DispatchID)
|
|
verification, err := postJSON[work.GenericCandidateVerificationResponse](ctx, b.client, b.verifierURL+"/v1/work-graphs/candidates/verify", work.GenericCandidateVerificationRequest{
|
|
GraphID: execution.GraphID, GraphRevision: execution.GraphRevision, NodeID: execution.NodeID, AuthorRunID: request.Intent.RunID, CandidateID: candidateID, Summary: response.AdvisoryText, EvidenceRefs: []work.EvidenceRef{artifactRef},
|
|
})
|
|
if err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
if !verification.Verified || verification.VerificationEvidenceRef == "" || verification.VerifierIdentity == "" {
|
|
return work.AgentExecutionResult{State: work.AgentExecutionWaiting, Summary: "generic candidate envelope was not independently verified; work remains waiting", ObservedAtUnixMillis: time.Now().UnixMilli()}, nil
|
|
}
|
|
candidate := work.GraphResultCallback{GraphID: execution.GraphID, Revision: execution.GraphRevision, Candidate: work.ResultCandidate{
|
|
ID: candidateID, GraphID: execution.GraphID, GraphRevision: execution.GraphRevision, NodeID: execution.NodeID,
|
|
AuthorRunID: request.Intent.RunID, ArtifactRefs: []work.EvidenceRef{artifactRef, work.EvidenceRef(verification.VerificationEvidenceRef)}, EvidenceSummary: "candidate attribution and evidence envelope independently verified; semantic content remains unverified",
|
|
Summary: response.AdvisoryText,
|
|
}}
|
|
if _, err := postJSON[json.RawMessage](ctx, b.client, b.callbackURL+"/v1/work-graphs/result-callback", candidate); err != nil {
|
|
return work.AgentExecutionResult{}, err
|
|
}
|
|
return work.AgentExecutionResult{State: work.AgentExecutionCompleted, Summary: "local cognitive specialist produced a candidate for independent verification", OutcomeRef: "candidate:" + request.Intent.DispatchID, ArtifactRefs: []string{"agent-output:" + request.Intent.DispatchID}, 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","generic-cognitive-work"]}`))
|
|
}
|
|
|
|
func envOr(name, fallback string) string {
|
|
if value := strings.TrimSpace(os.Getenv(name)); value != "" {
|
|
return value
|
|
}
|
|
return fallback
|
|
}
|