feat: persist verified graph synthesis runtime #162
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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;
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"]
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user