From 3d306812798df68bf402c521f1c74340afec51cd Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 01:52:39 +0300 Subject: [PATCH] feat: persist verified graph synthesis runtime --- adapters/postgres/work_repository.go | 20 +++++++ cmd/work-graph-runtime/main.go | 13 +++++ .../036_work_graph_result_synthesis.sql | 58 +++++++++++++++++++ dev/ACTIVE_TASK.json | 4 +- ...00189-durable-graph-synthesis-runtime.json | 23 ++++++++ modules/work/application/graph_runtime.go | 22 +++++++ modules/work/ports/graph.go | 4 ++ .../graph_persistence_sql_test.go | 13 +++++ tests/agent_execution/graph_runtime_test.go | 23 ++++++++ 9 files changed, 178 insertions(+), 2 deletions(-) create mode 100644 database/migrations/036_work_graph_result_synthesis.sql create mode 100644 dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 47669d6..8d225d5 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -199,6 +199,26 @@ func (r *WorkRepository) LoadWorkGraph(ctx context.Context, graphID string, revi return graph, nil } +func (r *WorkRepository) SaveGraphSynthesis(ctx context.Context, conflict work.ConflictSet, candidates []work.ResultCandidate, synthesis work.GraphSynthesisCandidate) error { + tx, err := r.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + for _, candidate := range candidates { + if _, err := tx.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); err != nil { + return err + } + } + if _, err := tx.Exec(ctx, `INSERT INTO work.graph_conflicts(conflict_id,graph_id,graph_revision,node_id,candidate_ids,resolved) VALUES($1,$2,$3,$4,$5,$6) ON CONFLICT(conflict_id) DO NOTHING`, conflict.ID, conflict.GraphID, conflict.GraphRevision, conflict.NodeID, conflict.CandidateIDs, conflict.Resolved); err != nil { + return err + } + if _, err := tx.Exec(ctx, `INSERT INTO work.graph_synthesis_candidates(synthesis_id,graph_id,graph_revision,node_id,candidate_ids,selected_candidate_id,dissent_candidate_ids,verifier_run_id,verifier_identity,evidence_refs,summary,state) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12) ON CONFLICT(synthesis_id) DO NOTHING`, synthesis.ID, synthesis.GraphID, synthesis.GraphRevision, synthesis.NodeID, synthesis.CandidateIDs, synthesis.SelectedCandidateID, synthesis.DissentCandidateIDs, synthesis.VerifierRunID, synthesis.VerifierIdentity, synthesis.EvidenceRefs, synthesis.Summary, synthesis.State); err != nil { + return err + } + return tx.Commit(ctx) +} + func (r *WorkRepository) RecoverStaleClaims(ctx context.Context, nowUnixMillis, staleAfterMillis int64) (int64, error) { command, err := r.pool.Exec(ctx, `UPDATE work.agent_dispatch_queue SET state='QUEUED',claimed_by='',updated_at_unix_millis=$1 WHERE state='CLAIMED' AND updated_at_unix_millis <= $2`, nowUnixMillis, nowUnixMillis-staleAfterMillis) if err != nil { diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index 62ee482..9c5fc15 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -55,6 +55,19 @@ func main() { } 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 { + http.Error(w, "invalid graph synthesis request", http.StatusBadRequest) + return + } + conflict, synthesis, err := application.SynthesizeGraph(r.Context(), repository, request) + if err != nil { + http.Error(w, err.Error(), http.StatusUnprocessableEntity) + return + } + writeJSON(w, map[string]any{"conflict": conflict, "synthesis": synthesis}) + }) addr := os.Getenv("LISTEN_ADDR") if addr == "" { addr = ":8088" diff --git a/database/migrations/036_work_graph_result_synthesis.sql b/database/migrations/036_work_graph_result_synthesis.sql new file mode 100644 index 0000000..fa39150 --- /dev/null +++ b/database/migrations/036_work_graph_result_synthesis.sql @@ -0,0 +1,58 @@ +BEGIN; + +SET ROLE work_role; + +CREATE TABLE IF NOT EXISTS work.graph_result_candidates ( + result_candidate_id TEXT PRIMARY KEY, + graph_id TEXT NOT NULL, + graph_revision BIGINT NOT NULL, + node_id TEXT NOT NULL, + author_run_id TEXT NOT NULL, + artifact_refs TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[], + evidence_summary TEXT NOT NULL, + summary TEXT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (result_candidate_id <> ''), CHECK (graph_id <> ''), CHECK (graph_revision > 0), + CHECK (node_id <> ''), CHECK (author_run_id <> ''), CHECK (evidence_summary <> ''), CHECK (summary <> ''), + FOREIGN KEY (graph_id, graph_revision) REFERENCES work.graphs(graph_id, revision) +); + +CREATE TABLE IF NOT EXISTS work.graph_conflicts ( + conflict_id TEXT PRIMARY KEY, + graph_id TEXT NOT NULL, + graph_revision BIGINT NOT NULL, + node_id TEXT NOT NULL, + candidate_ids TEXT[] NOT NULL, + resolved BOOLEAN NOT NULL DEFAULT FALSE, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (conflict_id <> ''), CHECK (graph_id <> ''), CHECK (graph_revision > 0), CHECK (node_id <> ''), + CHECK (cardinality(candidate_ids) >= 2), + FOREIGN KEY (graph_id, graph_revision) REFERENCES work.graphs(graph_id, revision) +); + +CREATE TABLE IF NOT EXISTS work.graph_synthesis_candidates ( + synthesis_id TEXT PRIMARY KEY, + graph_id TEXT NOT NULL, + graph_revision BIGINT NOT NULL, + node_id TEXT NOT NULL, + candidate_ids TEXT[] NOT NULL, + selected_candidate_id TEXT NOT NULL, + dissent_candidate_ids TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[], + verifier_run_id TEXT NOT NULL, + verifier_identity TEXT NOT NULL, + evidence_refs TEXT[] NOT NULL, + summary TEXT NOT NULL, + state TEXT NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, + CHECK (synthesis_id <> ''), CHECK (graph_id <> ''), CHECK (graph_revision > 0), CHECK (node_id <> ''), + CHECK (cardinality(candidate_ids) >= 2), CHECK (selected_candidate_id <> ''), CHECK (verifier_run_id <> ''), + CHECK (verifier_identity <> ''), CHECK (cardinality(evidence_refs) >= 1), CHECK (summary <> ''), CHECK (state = 'VERIFIED'), + FOREIGN KEY (graph_id, graph_revision) REFERENCES work.graphs(graph_id, revision) +); + +CREATE INDEX IF NOT EXISTS graph_result_candidates_node_idx ON work.graph_result_candidates(graph_id, graph_revision, node_id); +CREATE INDEX IF NOT EXISTS graph_conflicts_node_idx ON work.graph_conflicts(graph_id, graph_revision, node_id); +CREATE INDEX IF NOT EXISTS graph_synthesis_node_idx ON work.graph_synthesis_candidates(graph_id, graph_revision, node_id); + +RESET ROLE; +COMMIT; diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index b3729c7..1130bef 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000188-graph-synthesis-independent-verifier.json", - "task_id": "DEV-000188", + "task_file": "dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json", + "task_id": "DEV-000189", "task_revision": 1 } diff --git a/dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json b/dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json new file mode 100644 index 0000000..f74aab6 --- /dev/null +++ b/dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000189", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "41bc4b5", + "risk_tier": "E", + "intent": "Persist Work Graph result candidates, conflicts and independently verified synthesis through the work owner and expose the governed synthesis runtime path.", + "requirement_ids": ["WP-403", "Durable graph result evidence", "Durable conflict resolution", "Synthesis runtime endpoint"], + "allowed_change_scope": {"modules": ["work-agent-execution", "work-storage", "work-graph-runtime"], "paths": ["database/migrations/036_work_graph_result_synthesis.sql", "adapters/postgres/work_repository.go", "modules/work/ports/graph.go", "modules/work/application/graph_runtime.go", "cmd/work-graph-runtime/main.go", "tests/agent_execution/graph_persistence_sql_test.go", "tests/agent_execution/graph_runtime_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "contracts/authority/**", "modules/epistemics/**", "deploy/**", ".gitea/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, + "contracts": {"consume": ["Existing Work Graph ResultCandidate/ConflictSet/GraphSynthesisCandidate contracts", "WP-403 independent verifier gate"], "may_change": ["GraphStore persistence port"]}, + "semantic_owner": "work-agent-execution", + "canonical_state": ["work.graph_result_candidates", "work.graph_conflicts", "work.graph_synthesis_candidates"], + "state_transitions": ["result candidate append -> conflict append -> independently verified synthesis append", "failed verification does not persist a resolved synthesis"], + "side_effects_and_privileges": ["Work schema writes only; no external mutation, provider call, secret expansion, or authority change"], + "persistence_and_migrations": ["Append-only migration 036; one owner-local transaction persists candidates, conflict and synthesis"], + "failure_retry_restart": ["Idempotent exact-record inserts; transaction rollback preserves no partial resolution; restart reloads durable evidence"], + "assumptions": ["Production migration runner applies migration before runtime endpoint use", "Authenticated upstream boundary already protects runtime access"], + "acceptance_scenarios": ["SQL is revision scoped and work-owned", "runtime rejects invalid/self-verifying synthesis", "successful synthesis persists all candidates, dissent and verifier identity atomically", "production canary returns durable success without external side effect"], + "required_tests": ["go test ./...", "go vet ./...", "go run ./cmd/devctl preflight -task dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json", "go run ./cmd/devctl check-diff -task dev/tasks/DEV-000189-durable-graph-synthesis-runtime.json", "go run ./cmd/devctl check-architecture", "go run ./cmd/devctl independent", "git diff --check", "CI", "production migration/readiness and authenticated synthesis canary"], + "terminal_gap_states": ["BLOCKED_BY_MIGRATION_DECISION", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] +} diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index f50fde6..3b8281d 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -25,6 +25,28 @@ type GraphCompleteRequest struct { Succeeded bool `json:"succeeded"` } +type GraphSynthesisRequest struct { + Conflict work.ConflictSet `json:"conflict"` + Candidates []work.ResultCandidate `json:"candidates"` + SelectedCandidate string `json:"selected_candidate_id"` + VerifierRunID work.AgentRunID `json:"verifier_run_id"` + VerifierIdentity string `json:"verifier_identity"` + EvidenceRefs []work.EvidenceRef `json:"evidence_refs"` + Summary string `json:"summary"` + Verified bool `json:"verified"` +} + +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 { + return work.ConflictSet{}, work.GraphSynthesisCandidate{}, err + } + if err := store.SaveGraphSynthesis(ctx, conflict, request.Candidates, synthesis); err != nil { + return work.ConflictSet{}, work.GraphSynthesisCandidate{}, err + } + return conflict, synthesis, 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 1bb4612..5d0b3a8 100644 --- a/modules/work/ports/graph.go +++ b/modules/work/ports/graph.go @@ -12,3 +12,7 @@ type GraphStore interface { SaveDispatchIntent(context.Context, work.AgentDispatchIntent) error EnqueueDispatch(context.Context, work.AgentDispatchQueueItem) error } + +type GraphResultStore interface { + SaveGraphSynthesis(context.Context, work.ConflictSet, []work.ResultCandidate, work.GraphSynthesisCandidate) error +} diff --git a/tests/agent_execution/graph_persistence_sql_test.go b/tests/agent_execution/graph_persistence_sql_test.go index 15756c0..3862af8 100644 --- a/tests/agent_execution/graph_persistence_sql_test.go +++ b/tests/agent_execution/graph_persistence_sql_test.go @@ -18,3 +18,16 @@ func TestGraphPersistenceMigrationIsRevisionScoped(t *testing.T) { } } } + +func TestGraphSynthesisMigrationIsWorkOwnedAndAtomicInputsExist(t *testing.T) { + data, err := os.ReadFile("../../database/migrations/036_work_graph_result_synthesis.sql") + if err != nil { + t.Fatal(err) + } + sql := string(data) + for _, marker := range []string{"work.graph_result_candidates", "work.graph_conflicts", "work.graph_synthesis_candidates", "verifier_run_id", "dissent_candidate_ids", "FOREIGN KEY (graph_id, graph_revision)", "SET ROLE work_role"} { + if !strings.Contains(sql, marker) { + t.Fatalf("graph synthesis persistence marker missing: %s", marker) + } + } +} diff --git a/tests/agent_execution/graph_runtime_test.go b/tests/agent_execution/graph_runtime_test.go index 28f2880..278f4b9 100644 --- a/tests/agent_execution/graph_runtime_test.go +++ b/tests/agent_execution/graph_runtime_test.go @@ -58,3 +58,26 @@ func TestGraphRuntimeSchedulesFanOutAndCompletesFanIn(t *testing.T) { t.Fatalf("completed=%+v err=%v", completed, err) } } + +type graphResultStoreStub struct { + conflict work.ConflictSet + count int +} + +func (s *graphResultStoreStub) SaveGraphSynthesis(_ context.Context, conflict work.ConflictSet, candidates []work.ResultCandidate, _ work.GraphSynthesisCandidate) error { + s.conflict = conflict + s.count = len(candidates) + return nil +} + +func TestSynthesizeGraphPersistsVerifiedDissent(t *testing.T) { + store := &graphResultStoreStub{} + conflict, synthesis, err := application.SynthesizeGraph(context.Background(), store, application.GraphSynthesisRequest{ + Conflict: work.ConflictSet{ID: "conflict-1", GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", CandidateIDs: []string{"candidate-a", "candidate-b"}}, + Candidates: []work.ResultCandidate{{ID: "candidate-a", GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", AuthorRunID: "producer-a", EvidenceSummary: "a", Summary: "a"}, {ID: "candidate-b", GraphID: "graph-1", GraphRevision: 1, NodeID: "node-1", AuthorRunID: "producer-b", EvidenceSummary: "b", Summary: "b"}}, + SelectedCandidate: "candidate-a", VerifierRunID: "verifier-1", VerifierIdentity: "independent", EvidenceRefs: []work.EvidenceRef{"tests:1"}, Summary: "verified", Verified: true, + }) + if err != nil || !conflict.Resolved || synthesis.State != work.GraphSynthesisVerified || store.count != 2 || !store.conflict.Resolved { + t.Fatalf("unexpected runtime synthesis: conflict=%+v synthesis=%+v stored=%+v err=%v", conflict, synthesis, store.conflict, err) + } +} -- 2.54.0