diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 1b27b73..423371d 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -351,6 +351,14 @@ func (r *WorkRepository) SaveGraphSynthesis(ctx context.Context, conflict work.C return tx.Commit(ctx) } +func (r *WorkRepository) SaveGraphResultCandidate(ctx context.Context, candidate work.ResultCandidate) error { + if candidate.ID == "" || candidate.GraphID == "" || candidate.GraphRevision < 1 || candidate.NodeID == "" || candidate.AuthorRunID == "" || candidate.EvidenceSummary == "" || candidate.Summary == "" { + return errors.New("invalid graph result candidate") + } + _, err := r.pool.Exec(ctx, `INSERT INTO work.graph_result_candidates(result_candidate_id,graph_id,graph_revision,node_id,author_run_id,artifact_refs,evidence_summary,summary) VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(result_candidate_id) DO NOTHING`, candidate.ID, candidate.GraphID, candidate.GraphRevision, candidate.NodeID, candidate.AuthorRunID, candidate.ArtifactRefs, candidate.EvidenceSummary, candidate.Summary) + return err +} + func (r *WorkRepository) RecoverStaleClaims(ctx context.Context, nowUnixMillis, staleAfterMillis int64) (int64, error) { tx, err := r.pool.Begin(ctx) if err != nil { diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index 63acab6..48cdc97 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -87,6 +87,19 @@ func main() { } writeJSON(w, result) }) + mux.HandleFunc("POST /v1/work-graphs/result-callback", func(w http.ResponseWriter, r *http.Request) { + var request work.GraphResultCallback + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&request); err != nil { + http.Error(w, "invalid graph result callback", http.StatusBadRequest) + return + } + graph, err := application.RecordGraphResultCandidate(r.Context(), repository, repository, request) + if err != nil { + http.Error(w, "invalid graph result callback", http.StatusUnprocessableEntity) + return + } + writeJSON(w, graph) + }) mux.HandleFunc("POST /v1/work-graphs/complete", func(w http.ResponseWriter, r *http.Request) { var request application.GraphCompleteRequest if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&request); err != nil { @@ -119,12 +132,12 @@ func main() { http.Error(w, "invalid graph synthesis request", http.StatusBadRequest) return } - conflict, synthesis, err := application.SynthesizeGraph(r.Context(), repository, request) + conflict, synthesis, graph, err := application.SynthesizeAndCompleteGraph(r.Context(), repository, repository, request) if err != nil { http.Error(w, err.Error(), http.StatusUnprocessableEntity) return } - writeJSON(w, map[string]any{"conflict": conflict, "synthesis": synthesis}) + writeJSON(w, map[string]any{"conflict": conflict, "synthesis": synthesis, "graph": graph}) }) addr := os.Getenv("LISTEN_ADDR") if addr == "" { diff --git a/contracts/work/graph.go b/contracts/work/graph.go index 0ee1dd9..cffc085 100644 --- a/contracts/work/graph.go +++ b/contracts/work/graph.go @@ -21,14 +21,15 @@ const ( GraphFailed GraphState = "FAILED" GraphUnknown GraphState = "UNKNOWN" - GraphNodeBlocked GraphNodeState = "BLOCKED_PREREQUISITE" - GraphNodeReady GraphNodeState = "READY" - GraphNodeLeased GraphNodeState = "LEASED" - GraphNodeRunning GraphNodeState = "RUNNING" - GraphNodeSuccess GraphNodeState = "SUCCEEDED" - GraphNodeFailed GraphNodeState = "FAILED" - GraphNodeCancel GraphNodeState = "CANCELLED" - GraphNodeUnknown GraphNodeState = "UNKNOWN" + GraphNodeBlocked GraphNodeState = "BLOCKED_PREREQUISITE" + GraphNodeReady GraphNodeState = "READY" + GraphNodeLeased GraphNodeState = "LEASED" + GraphNodeRunning GraphNodeState = "RUNNING" + GraphNodeVerifying GraphNodeState = "VERIFYING" + GraphNodeSuccess GraphNodeState = "SUCCEEDED" + GraphNodeFailed GraphNodeState = "FAILED" + GraphNodeCancel GraphNodeState = "CANCELLED" + GraphNodeUnknown GraphNodeState = "UNKNOWN" ) type GraphBudget struct { @@ -110,6 +111,15 @@ type ResultCandidate struct { Summary string `json:"summary"` } +// GraphResultCallback carries one agent's attributable candidate to Work. +// Recording a candidate never makes the node successful; the node enters +// VERIFYING until an independent synthesis/verifier accepts it. +type GraphResultCallback struct { + GraphID string `json:"graph_id"` + Revision int64 `json:"revision"` + Candidate ResultCandidate `json:"candidate"` +} + type ConflictSet struct { ID string `json:"id"` GraphID string `json:"graph_id"` diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 3b29de4..e15d5b7 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000232-durable-solution-wave-dispatch.json", - "task_id": "DEV-000232", + "task_file": "dev/tasks/DEV-000233-graph-result-fanin-verification.json", + "task_id": "DEV-000233", "task_revision": 1 } diff --git a/dev/tasks/DEV-000233-graph-result-fanin-verification.json b/dev/tasks/DEV-000233-graph-result-fanin-verification.json new file mode 100644 index 0000000..7a58fb8 --- /dev/null +++ b/dev/tasks/DEV-000233-graph-result-fanin-verification.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000233", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "ee16103e03a5fd911119e2bc45e5904255ab7752", + "risk_tier": "E", + "intent": "Close the multi-agent Work Graph result lifecycle: persist attributable agent candidates, prevent candidate text from completing a node, and complete only after independently verified synthesis.", + "requirement_ids": ["fan-in", "independent verifier", "result provenance", "no false completion"], + "allowed_change_scope": {"modules": ["work-graph", "agent-execution", "postgres-work-adapter"], "paths": ["contracts/work/graph.go", "modules/work/ports/graph.go", "modules/work/application/graph_runtime.go", "adapters/postgres/work_repository.go", "cmd/work-graph-runtime/main.go", "tests/agent_execution/graph_runtime_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000233-graph-result-fanin-verification.json", "docs/PROJECT_STATUS.md"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "deploy/**", "LLM-local/**", "modules/work/secrets/**"]}, + "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": ["existing Work Graph revision store", "existing graph result candidate/synthesis tables", "agent callback evidence"], "may_change": ["typed graph result callback", "Work result callback route", "verified synthesis completion path"]}, + "semantic_owner": "Work Graph result and verification lifecycle", + "canonical_state": ["work.graphs, work.graph_nodes, work.graph_result_candidates, work.graph_conflicts and work.graph_synthesis_candidates"], + "state_transitions": ["agent result -> durable ResultCandidate", "candidate -> no completion", "independent verified synthesis -> graph node SUCCEEDED", "all verified nodes -> graph COMPLETED"], + "side_effects_and_privileges": ["Work-owned candidate/graph persistence only; no provider call, secret access, external mutation, promotion or deployment"], + "persistence_and_migrations": ["none; existing result candidate and synthesis tables are reused"], + "failure_retry_restart": ["immutable graph revisions remain attributable; duplicate candidate IDs are idempotent at storage; missing or unverified synthesis cannot mark success"], + "assumptions": ["the verifier supplies all competing candidates and evidence in the synthesis request", "automatic replan selection remains the next stage because a graph does not yet own a persisted goal-search snapshot"], + "acceptance_scenarios": ["result callback persists candidate", "result callback leaves graph running and not completed", "verified independent synthesis advances the node", "verifier cannot be a producer", "fan-in completion requires all graph nodes successful"], + "required_tests": ["go test ./modules/work/domain ./modules/work/application ./tests/agent_execution ./cmd/work-graph-runtime/...", "go vet ./modules/work/domain ./modules/work/application ./tests/agent_execution ./cmd/work-graph-runtime/...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "CI", "production health smoke"], + "terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP"] +} diff --git a/docs/PROJECT_STATUS.md b/docs/PROJECT_STATUS.md index 61aab46..a000306 100644 --- a/docs/PROJECT_STATUS.md +++ b/docs/PROJECT_STATUS.md @@ -24,8 +24,8 @@ - `COMPLETION_MATRIX = docs/evidence/audit-completion-matrix-2026-08-23-v3.json`; all seven objective areas are mapped to current repository/production proof, with external bridge, GPU promotion/quality and deep-acceptance gaps explicit. -- `ACTIVE_TASK = dev/tasks/DEV-000232-durable-solution-wave-dispatch.json revision 1`; the solution-search decision now materializes the next unresolved wave into the existing durable Work Graph and dispatch queue, preserving independent agent/capability/research/reflection directions. Provider selection, execution and result fan-in verification remain worker responsibilities. +- `ACTIVE_TASK = dev/tasks/DEV-000233-graph-result-fanin-verification.json revision 1`; agent result callbacks now persist attributable candidates without false completion, and only independently verified synthesis advances a graph node to success and allows fan-in completion. Automatic goal-snapshot replan after a completed wave remains the next integration gap. - `LINEAGE_DECISION = docs/GOVERNANCE_LINEAGE_DECISION_2026-08-22.md`; historical packets remain provenance and are not current authority. - `REGISTRY_STATUS = YAML and JSON semantic mirrors agree` - `PRODUCTION_DOMAIN = docker.cin.su`; shared docker-test is not used for this project. -- `NEXT_WORK = connect completed-wave callbacks to automatic replan/fan-in verification, then add bounded recovery of stale waves; retain local fallback, truth research and governed development gates.` +- `NEXT_WORK = persist the goal/search snapshot with each graph revision and connect verified graph completion to automatic replan/materialization; then add bounded recovery of stale waves.` diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index fdfa795..7ce6180 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -15,6 +15,7 @@ import ( var ErrGraphDispatchBinding = errors.New("graph node dispatch binding is missing") var ErrTaskEvaluationCallbackMismatch = errors.New("task evaluation callback does not match graph node") +var ErrGraphResultCallbackMismatch = errors.New("graph result callback does not match graph node") type GraphScheduleRequest struct { Graph work.WorkGraph `json:"graph"` @@ -82,6 +83,51 @@ func SynthesizeGraph(ctx context.Context, store ports.GraphResultStore, request return conflict, synthesis, nil } +// RecordGraphResultCandidate persists attributable agent output. The graph is +// immutable by revision, so candidate recording does not rewrite the node; +// only verified synthesis may create the next successful graph revision. +func RecordGraphResultCandidate(ctx context.Context, candidateStore ports.GraphCandidateStore, graphStore ports.GraphStore, request work.GraphResultCallback) (work.WorkGraph, error) { + candidate := request.Candidate + if candidateStore == nil || request.GraphID == "" || request.Revision < 1 || candidate.ID == "" || candidate.GraphID != request.GraphID || candidate.GraphRevision != request.Revision || candidate.NodeID == "" || candidate.AuthorRunID == "" || candidate.EvidenceSummary == "" || candidate.Summary == "" { + return work.WorkGraph{}, ErrGraphResultCallbackMismatch + } + graph, err := graphStore.LoadWorkGraph(ctx, request.GraphID, request.Revision) + if err != nil { + return work.WorkGraph{}, err + } + found := false + for i := range graph.Nodes { + if graph.Nodes[i].ID != candidate.NodeID { + continue + } + found = true + if graph.Nodes[i].State == work.GraphNodeSuccess || graph.Nodes[i].State == work.GraphNodeCancel { + return work.WorkGraph{}, ErrGraphResultCallbackMismatch + } + } + if !found { + return work.WorkGraph{}, ErrGraphResultCallbackMismatch + } + if err := candidateStore.SaveGraphResultCandidate(ctx, candidate); err != nil { + return work.WorkGraph{}, err + } + return graph, nil +} + +// SynthesizeAndCompleteGraph is the only path that turns a verifying node +// into success: synthesis must be independently verified first. +func SynthesizeAndCompleteGraph(ctx context.Context, resultStore ports.GraphResultStore, graphStore ports.GraphStore, request GraphSynthesisRequest) (work.ConflictSet, work.GraphSynthesisCandidate, work.WorkGraph, error) { + conflict, synthesis, err := SynthesizeGraph(ctx, resultStore, request) + if err != nil { + return work.ConflictSet{}, work.GraphSynthesisCandidate{}, work.WorkGraph{}, err + } + graph, err := CompleteGraphNode(ctx, graphStore, conflict.GraphID, conflict.GraphRevision, conflict.NodeID, true) + if err != nil { + return work.ConflictSet{}, work.GraphSynthesisCandidate{}, work.WorkGraph{}, err + } + return conflict, synthesis, graph, nil +} + func AdmitAndSchedule(ctx context.Context, store ports.GraphStore, request GraphScheduleRequest) (work.WorkGraph, []work.GraphNodeID, error) { if request.Now < 1 { request.Now = time.Now().UnixMilli() diff --git a/modules/work/ports/graph.go b/modules/work/ports/graph.go index acd6b24..9c5e532 100644 --- a/modules/work/ports/graph.go +++ b/modules/work/ports/graph.go @@ -18,6 +18,10 @@ type GraphResultStore interface { SaveGraphSynthesis(context.Context, work.ConflictSet, []work.ResultCandidate, work.GraphSynthesisCandidate) error } +type GraphCandidateStore interface { + SaveGraphResultCandidate(context.Context, work.ResultCandidate) error +} + // TaskEvaluationSink is a remote-capable boundary to Capability's owner. // Work never writes the capability schema directly. type TaskEvaluationSink interface { diff --git a/tests/agent_execution/graph_runtime_test.go b/tests/agent_execution/graph_runtime_test.go index c35aef1..dc1a1a7 100644 --- a/tests/agent_execution/graph_runtime_test.go +++ b/tests/agent_execution/graph_runtime_test.go @@ -12,9 +12,11 @@ import ( ) type graphStore struct { - graphs map[string]work.WorkGraph - queued []string - intents []string + graphs map[string]work.WorkGraph + queued []string + intents []string + candidates []work.ResultCandidate + syntheses int } func (s *graphStore) SaveWorkGraph(_ context.Context, graph work.WorkGraph) error { @@ -35,6 +37,14 @@ func (s *graphStore) EnqueueDispatch(_ context.Context, item work.AgentDispatchQ s.queued = append(s.queued, item.DispatchID) return nil } +func (s *graphStore) SaveGraphResultCandidate(_ context.Context, candidate work.ResultCandidate) error { + s.candidates = append(s.candidates, candidate) + return nil +} +func (s *graphStore) SaveGraphSynthesis(_ context.Context, _ work.ConflictSet, _ []work.ResultCandidate, _ work.GraphSynthesisCandidate) error { + s.syntheses++ + return nil +} func itoa(value int64) string { return strconv.FormatInt(value, 10) @@ -84,6 +94,22 @@ func TestSynthesizeGraphPersistsVerifiedDissent(t *testing.T) { } } +func TestGraphResultCallbackRequiresIndependentSynthesisBeforeSuccess(t *testing.T) { + store := &graphStore{} + graph, _, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "result-graph", Revision: 1, Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 1}, Nodes: []work.GraphNode{{ID: "node-1", DispatchID: "dispatch-1", IdempotencyKey: "node-1", State: work.GraphNodeReady}}}, Intents: []work.AgentDispatchIntent{{DispatchID: "dispatch-1"}}, Now: 1}) + if err != nil { + t.Fatal(err) + } + returned, err := application.RecordGraphResultCandidate(context.Background(), store, store, work.GraphResultCallback{GraphID: graph.ID, Revision: graph.Revision, Candidate: work.ResultCandidate{ID: "candidate-1", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", AuthorRunID: "producer-1", EvidenceSummary: "evidence", Summary: "candidate"}}) + if err != nil || len(store.candidates) != 1 || returned.State != work.GraphRunning { + t.Fatalf("candidate callback changed completion state: returned=%+v candidates=%d err=%v", returned, len(store.candidates), err) + } + conflict, synthesis, completed, err := application.SynthesizeAndCompleteGraph(context.Background(), store, store, application.GraphSynthesisRequest{Conflict: work.ConflictSet{ID: "conflict-1", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", CandidateIDs: []string{"candidate-1", "candidate-2"}}, Candidates: []work.ResultCandidate{{ID: "candidate-1", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", AuthorRunID: "producer-1", EvidenceSummary: "evidence", Summary: "candidate"}, {ID: "candidate-2", GraphID: graph.ID, GraphRevision: graph.Revision, NodeID: "node-1", AuthorRunID: "producer-2", EvidenceSummary: "evidence-2", Summary: "candidate-2"}}, SelectedCandidate: "candidate-1", VerifierRunID: "verifier-1", VerifierIdentity: "independent", EvidenceRefs: []work.EvidenceRef{"test:evidence"}, Summary: "verified", Verified: true}) + if err != nil || !conflict.Resolved || synthesis.State != work.GraphSynthesisVerified || completed.State != work.GraphCompleted { + t.Fatalf("verified fan-in did not complete graph: conflict=%+v synthesis=%+v graph=%+v err=%v", conflict, synthesis, completed, err) + } +} + func TestGraphSynthesisJSONUsesStableSnakeCaseContract(t *testing.T) { payload, err := json.Marshal(application.GraphSynthesisRequest{Conflict: work.ConflictSet{ID: "c", GraphID: "g", GraphRevision: 1, NodeID: "n", CandidateIDs: []string{"a", "b"}}, SelectedCandidate: "a", VerifierRunID: "v", VerifierIdentity: "independent", EvidenceRefs: []work.EvidenceRef{"e"}, Summary: "s", Verified: true}) if err != nil || !strings.Contains(string(payload), `"graph_id":"g"`) || !strings.Contains(string(payload), `"selected_candidate_id":"a"`) {