From 7afe3ff48f86b25b06fae48b6186e054321cba52 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 12:28:42 +0300 Subject: [PATCH] feat(work): accept task evaluation evidence callbacks --- cmd/work-graph-runtime/main.go | 26 ++++++ .../task_evaluation_sink.go | 49 ++++++++++++ contracts/work/graph.go | 11 +++ dev/ACTIVE_TASK.json | 4 +- .../DEV-000211-task-evaluator-callback.json | 23 ++++++ modules/work/application/graph_runtime.go | 26 ++++++ .../task_evaluation_callback_test.go | 80 +++++++++++++++++++ modules/work/ports/graph.go | 7 ++ 8 files changed, 224 insertions(+), 2 deletions(-) create mode 100644 cmd/work-graph-runtime/task_evaluation_sink.go create mode 100644 dev/tasks/DEV-000211-task-evaluator-callback.json create mode 100644 modules/work/application/task_evaluation_callback_test.go diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index 5d4e90a..1ee20f8 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -7,9 +7,11 @@ import ( "log" "net/http" "os" + "time" "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/platform/secrets" "git.cin.su/m/Assistant/runtime/health" @@ -26,6 +28,10 @@ func main() { log.Fatal(err) } defer repository.Close() + evidenceSink, err := newCoreTaskEvaluationSink(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), &http.Client{Timeout: 12 * time.Second}) + if err != nil { + log.Fatal(err) + } mux := http.NewServeMux() mux.Handle("/healthz", health.Handler{Runtime: "work-graph-runtime", SemanticGeneration: 1, ArchitectureGeneration: 1}) mux.Handle("/readyz", health.ReadinessHandler{Runtime: "work-graph-runtime", Probes: []health.Probe{{Name: "work-postgres", Check: func(ctx context.Context) error { return repository.Ping(ctx) }}}}) @@ -68,6 +74,19 @@ func main() { } writeJSON(w, graph) }) + mux.HandleFunc("POST /v1/work-graphs/task-evaluations/callback", func(w http.ResponseWriter, r *http.Request) { + var request work.TaskEvaluationEvidenceCallback + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&request); err != nil { + http.Error(w, "invalid task evaluation callback", http.StatusBadRequest) + return + } + graph, err := application.CompleteTaskEvaluationNode(r.Context(), repository, evidenceSink, request) + if err != nil { + http.Error(w, err.Error(), http.StatusUnprocessableEntity) + return + } + writeJSON(w, graph) + }) mux.HandleFunc("POST /v1/work-graphs/synthesis", func(w http.ResponseWriter, r *http.Request) { var request application.GraphSynthesisRequest if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&request); err != nil { @@ -89,6 +108,13 @@ func main() { log.Fatal(http.ListenAndServe(addr, mux)) } +func envOr(name, fallback string) string { + if value := os.Getenv(name); value != "" { + return value + } + return fallback +} + func writeJSON(w http.ResponseWriter, value any) { w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(value) diff --git a/cmd/work-graph-runtime/task_evaluation_sink.go b/cmd/work-graph-runtime/task_evaluation_sink.go new file mode 100644 index 0000000..5ff5803 --- /dev/null +++ b/cmd/work-graph-runtime/task_evaluation_sink.go @@ -0,0 +1,49 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + + "git.cin.su/m/Assistant/contracts/capability" +) + +type coreTaskEvaluationSink struct { + baseURL string + client *http.Client +} + +func newCoreTaskEvaluationSink(baseURL string, client *http.Client) (*coreTaskEvaluationSink, error) { + baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/") + if baseURL == "" { + return nil, fmt.Errorf("core runtime URL is required") + } + if client == nil { + client = http.DefaultClient + } + return &coreTaskEvaluationSink{baseURL: baseURL, client: client}, nil +} + +func (s *coreTaskEvaluationSink) SaveTaskEvaluation(ctx context.Context, value capability.TaskEvaluation) error { + payload, err := json.Marshal(value) + if err != nil { + return err + } + request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.baseURL+"/v1/cognitive-registry/task-evaluations", bytes.NewReader(payload)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + response, err := s.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("save task evaluation: status %s", response.Status) + } + return nil +} diff --git a/contracts/work/graph.go b/contracts/work/graph.go index 137eb30..b4b9623 100644 --- a/contracts/work/graph.go +++ b/contracts/work/graph.go @@ -81,6 +81,17 @@ type TaskEvaluationGraphRequest struct { TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"` } +// TaskEvaluationEvidenceCallback is emitted by a specialist after it has +// completed the bounded case set. Capability remains the evidence owner; +// Work only advances its matching graph node after the evidence is accepted. +type TaskEvaluationEvidenceCallback struct { + GraphID string `json:"graph_id"` + Revision int64 `json:"revision"` + NodeID GraphNodeID `json:"node_id"` + Evaluation capability.TaskEvaluation `json:"evaluation"` + Succeeded bool `json:"succeeded"` +} + type ResultCandidate struct { ID string `json:"id"` GraphID string `json:"graph_id"` diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 0a82b24..3a6fe8e 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000210-task-evaluator-worker.json", - "task_id": "DEV-000210", + "task_file": "dev/tasks/DEV-000211-task-evaluator-callback.json", + "task_id": "DEV-000211", "task_revision": 1 } diff --git a/dev/tasks/DEV-000211-task-evaluator-callback.json b/dev/tasks/DEV-000211-task-evaluator-callback.json new file mode 100644 index 0000000..9f4c972 --- /dev/null +++ b/dev/tasks/DEV-000211-task-evaluator-callback.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000211", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "6dec47c79163f456cf6b18085a268ebb2619c7bb", + "risk_tier": "E", + "intent": "Accept specialist task-evaluation evidence through a Work Graph callback: persist evidence through the Capability owner and complete the matching graph node only after typed lineage and revision checks pass.", + "requirement_ids": ["evidence callback", "fan-in completion", "owner-local persistence", "revision safety", "no cross-owner DB access"], + "allowed_change_scope": {"modules": ["work"], "paths": ["contracts/work/**", "modules/work/**", "cmd/work-graph-runtime/**", "tests/model_qualification/**", "tests/agent_execution/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000211-task-evaluator-callback.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "cmd/devctl/**", "modules/identity/**", "database/**", "modules/capability/**"]}, + "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": ["capability.TaskEvaluation", "existing Work Graph revision/completion contract", "Core Runtime capability evaluation HTTP boundary"], "may_change": ["typed task-evaluation callback contract"]}, + "semantic_owner": "work for graph lifecycle; capability for evaluation evidence", + "canonical_state": ["Capability owns evaluation evidence; Work owns graph/node lifecycle and callback lineage"], + "state_transitions": ["specialist result -> callback received", "callback -> Capability evidence persisted", "persisted evidence -> matching node completion and graph fan-in"], + "side_effects_and_privileges": ["remote Core Runtime call through existing internal boundary; no direct foreign DB access and no new secrets"], + "persistence_and_migrations": ["uses existing capability task-evaluation and work graph persistence; no migration"], + "failure_retry_restart": ["Capability write failure leaves node incomplete and callback retryable", "stale graph revision is rejected; unknown/unsupported evaluation is not promoted by callback"], + "assumptions": ["specialist executor has already produced evidence and artifact references", "the callback is asynchronous and never blocks the chat request"], + "acceptance_scenarios": ["callback rejects non-task graph node", "callback rejects profile/binding lineage mismatch", "Capability write occurs before node completion", "sink failure leaves graph node unfinished", "valid callback advances graph fan-in", "duplicate evidence remains idempotent at Capability owner"], + "required_tests": ["go test", "go vet", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production public health check"], + "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY"] +} diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index 204e041..7e3289b 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -11,6 +11,7 @@ import ( ) var ErrGraphDispatchBinding = errors.New("graph node dispatch binding is missing") +var ErrTaskEvaluationCallbackMismatch = errors.New("task evaluation callback does not match graph node") type GraphScheduleRequest struct { Graph work.WorkGraph `json:"graph"` @@ -42,6 +43,31 @@ type GraphSynthesisRequest struct { Verified bool `json:"verified"` } +func CompleteTaskEvaluationNode(ctx context.Context, graphStore ports.GraphStore, evidenceSink ports.TaskEvaluationSink, request work.TaskEvaluationEvidenceCallback) (work.WorkGraph, error) { + if evidenceSink == nil || request.GraphID == "" || request.Revision < 1 || request.NodeID == "" || !request.Succeeded { + return work.WorkGraph{}, ErrTaskEvaluationCallbackMismatch + } + graph, err := graphStore.LoadWorkGraph(ctx, request.GraphID, request.Revision) + if err != nil { + return work.WorkGraph{}, err + } + matched := false + for _, node := range graph.Nodes { + if node.ID != request.NodeID { + continue + } + matched = node.Purpose == work.GraphNodePurposeTaskEvaluation && node.TaskProfileID == request.Evaluation.ProfileID && node.BindingID == request.Evaluation.BindingID + break + } + if !matched || request.Evaluation.BindingRevision < 1 || request.Evaluation.CapabilityID == "" { + return work.WorkGraph{}, ErrTaskEvaluationCallbackMismatch + } + if err := evidenceSink.SaveTaskEvaluation(ctx, request.Evaluation); err != nil { + return work.WorkGraph{}, err + } + return CompleteGraphNode(ctx, graphStore, request.GraphID, request.Revision, request.NodeID, true) +} + func SynthesizeGraph(ctx context.Context, store ports.GraphResultStore, request GraphSynthesisRequest) (work.ConflictSet, work.GraphSynthesisCandidate, error) { conflict, synthesis, err := domain.BuildVerifiedGraphSynthesis(request.Conflict, request.Candidates, request.SelectedCandidate, request.VerifierRunID, request.VerifierIdentity, request.EvidenceRefs, request.Summary, request.Verified) if err != nil { diff --git a/modules/work/application/task_evaluation_callback_test.go b/modules/work/application/task_evaluation_callback_test.go new file mode 100644 index 0000000..68a8019 --- /dev/null +++ b/modules/work/application/task_evaluation_callback_test.go @@ -0,0 +1,80 @@ +package application + +import ( + "context" + "errors" + "testing" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/work" +) + +type callbackGraphStore struct { + graph work.WorkGraph + saves int +} + +func (s *callbackGraphStore) SaveWorkGraph(_ context.Context, graph work.WorkGraph) error { + s.graph = graph + s.saves++ + return nil +} +func (s *callbackGraphStore) LoadWorkGraph(context.Context, string, int64) (work.WorkGraph, error) { + return s.graph, nil +} +func (s *callbackGraphStore) SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error { + return nil +} +func (s *callbackGraphStore) EnqueueDispatch(context.Context, work.AgentDispatchQueueItem) error { + return nil +} + +type callbackSink struct { + evaluation capability.TaskEvaluation + err error +} + +func (s *callbackSink) SaveTaskEvaluation(_ context.Context, value capability.TaskEvaluation) error { + s.evaluation = value + return s.err +} + +func callbackRequest() (work.WorkGraph, work.TaskEvaluationEvidenceCallback) { + graph := work.WorkGraph{ID: "graph-1", Revision: 1, State: work.GraphRunning, Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "node-1", DispatchID: "dispatch-1", IdempotencyKey: "idem-1", State: work.GraphNodeRunning, Purpose: work.GraphNodePurposeTaskEvaluation, TaskProfileID: "profile-1", BindingID: "binding-1"}}} + evaluation := capability.TaskEvaluation{ID: "evaluation-1", ProfileID: "profile-1", BindingID: "binding-1", BindingRevision: 2, CapabilityID: "capability-1", Result: capability.EvaluationSupported} + return graph, work.TaskEvaluationEvidenceCallback{GraphID: "graph-1", Revision: 1, NodeID: "node-1", Evaluation: evaluation, Succeeded: true} +} + +func TestCompleteTaskEvaluationPersistsEvidenceBeforeGraphCompletion(t *testing.T) { + graph, callback := callbackRequest() + store := &callbackGraphStore{graph: graph} + sink := &callbackSink{} + completed, err := CompleteTaskEvaluationNode(context.Background(), store, sink, callback) + if err != nil { + t.Fatal(err) + } + if sink.evaluation.ID != "evaluation-1" || store.saves != 1 || completed.State != work.GraphCompleted || completed.Nodes[0].State != work.GraphNodeSuccess { + t.Fatalf("callback did not complete in order: evaluation=%+v saves=%d graph=%+v", sink.evaluation, store.saves, completed) + } +} + +func TestCompleteTaskEvaluationLeavesNodeOpenWhenCapabilityRejectsEvidence(t *testing.T) { + graph, callback := callbackRequest() + store := &callbackGraphStore{graph: graph} + sink := &callbackSink{err: errors.New("capability unavailable")} + if _, err := CompleteTaskEvaluationNode(context.Background(), store, sink, callback); err == nil { + t.Fatal("expected capability failure") + } + if store.saves != 0 { + t.Fatalf("graph was advanced after capability failure: saves=%d", store.saves) + } +} + +func TestCompleteTaskEvaluationRejectsLineageMismatch(t *testing.T) { + graph, callback := callbackRequest() + callback.Evaluation.ProfileID = "profile-other" + store := &callbackGraphStore{graph: graph} + if _, err := CompleteTaskEvaluationNode(context.Background(), store, &callbackSink{}, callback); !errors.Is(err, ErrTaskEvaluationCallbackMismatch) { + t.Fatalf("expected lineage rejection, got %v", err) + } +} diff --git a/modules/work/ports/graph.go b/modules/work/ports/graph.go index 5d0b3a8..acd6b24 100644 --- a/modules/work/ports/graph.go +++ b/modules/work/ports/graph.go @@ -3,6 +3,7 @@ package ports import ( "context" + "git.cin.su/m/Assistant/contracts/capability" "git.cin.su/m/Assistant/contracts/work" ) @@ -16,3 +17,9 @@ type GraphStore interface { type GraphResultStore interface { SaveGraphSynthesis(context.Context, work.ConflictSet, []work.ResultCandidate, work.GraphSynthesisCandidate) error } + +// TaskEvaluationSink is a remote-capable boundary to Capability's owner. +// Work never writes the capability schema directly. +type TaskEvaluationSink interface { + SaveTaskEvaluation(context.Context, capability.TaskEvaluation) error +} -- 2.54.0