Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
da251e1aa8 | ||
|
|
eac39a3fd4 | ||
|
|
1f8af98f31 | ||
|
|
ee16103e03 |
@@ -263,7 +263,7 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback(ctx)
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO work.graphs(graph_id,revision,state,max_parallel,max_attempts,parent_graph_id,parent_graph_revision,trigger_evidence_ref) VALUES($1,$2,$3,$4,$5,$6,$7,$8) ON CONFLICT(graph_id,revision) DO NOTHING`, graph.ID, graph.Revision, graph.State, graph.Budget.MaxParallel, graph.Budget.MaxAttempts, graph.ParentGraphID, graph.ParentGraphRevision, graph.TriggerEvidenceRef); err != nil {
|
||||
if _, err := tx.Exec(ctx, `INSERT INTO work.graphs(graph_id,revision,state,max_parallel,max_attempts,parent_graph_id,parent_graph_revision,trigger_evidence_ref,orchestration_bound,orchestration_goal_id,orchestration_goal_state,orchestration_search_round,orchestration_max_round,orchestration_cancel_requested,orchestration_verification_ready,orchestration_unresolved) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16) ON CONFLICT(graph_id,revision) DO NOTHING`, graph.ID, graph.Revision, graph.State, graph.Budget.MaxParallel, graph.Budget.MaxAttempts, graph.ParentGraphID, graph.ParentGraphRevision, graph.TriggerEvidenceRef, graph.SearchInputBound, graph.SearchInput.GoalID, graph.SearchInput.GoalState, graph.SearchInput.SearchRound, graph.SearchInput.MaxRound, graph.SearchInput.CancelRequested, graph.SearchInput.VerificationReady, graph.SearchInput.Unresolved); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, node := range graph.Nodes {
|
||||
@@ -285,7 +285,7 @@ func (r *WorkRepository) SaveWorkGraph(ctx context.Context, graph work.WorkGraph
|
||||
|
||||
func (r *WorkRepository) LoadWorkGraph(ctx context.Context, graphID string, revision int64) (work.WorkGraph, error) {
|
||||
var graph work.WorkGraph
|
||||
err := r.pool.QueryRow(ctx, `SELECT graph_id,revision,state,max_parallel,max_attempts,parent_graph_id,parent_graph_revision,trigger_evidence_ref FROM work.graphs WHERE graph_id=$1 AND revision=$2`, graphID, revision).Scan(&graph.ID, &graph.Revision, &graph.State, &graph.Budget.MaxParallel, &graph.Budget.MaxAttempts, &graph.ParentGraphID, &graph.ParentGraphRevision, &graph.TriggerEvidenceRef)
|
||||
err := r.pool.QueryRow(ctx, `SELECT graph_id,revision,state,max_parallel,max_attempts,parent_graph_id,parent_graph_revision,trigger_evidence_ref,orchestration_bound,orchestration_goal_id,orchestration_goal_state,orchestration_search_round,orchestration_max_round,orchestration_cancel_requested,orchestration_verification_ready,orchestration_unresolved FROM work.graphs WHERE graph_id=$1 AND revision=$2`, graphID, revision).Scan(&graph.ID, &graph.Revision, &graph.State, &graph.Budget.MaxParallel, &graph.Budget.MaxAttempts, &graph.ParentGraphID, &graph.ParentGraphRevision, &graph.TriggerEvidenceRef, &graph.SearchInputBound, &graph.SearchInput.GoalID, &graph.SearchInput.GoalState, &graph.SearchInput.SearchRound, &graph.SearchInput.MaxRound, &graph.SearchInput.CancelRequested, &graph.SearchInput.VerificationReady, &graph.SearchInput.Unresolved)
|
||||
if err != nil {
|
||||
return work.WorkGraph{}, fmt.Errorf("load graph: %w", err)
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -87,18 +87,31 @@ 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 {
|
||||
http.Error(w, "invalid graph completion request", http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
graph, err := application.CompleteGraphNode(r.Context(), repository, request.GraphID, request.Revision, request.NodeID, request.Succeeded)
|
||||
result, err := application.CompleteGraphNodeWithReplan(r.Context(), repository, request)
|
||||
if err != nil {
|
||||
http.Error(w, err.Error(), http.StatusUnprocessableEntity)
|
||||
return
|
||||
}
|
||||
writeJSON(w, graph)
|
||||
writeJSON(w, result)
|
||||
})
|
||||
mux.HandleFunc("POST /v1/work-graphs/task-evaluations/callback", func(w http.ResponseWriter, r *http.Request) {
|
||||
var request work.TaskEvaluationEvidenceCallback
|
||||
@@ -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 == "" {
|
||||
|
||||
+29
-17
@@ -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 {
|
||||
@@ -56,15 +57,17 @@ type GraphEdge struct {
|
||||
}
|
||||
|
||||
type WorkGraph struct {
|
||||
ID string `json:"id"`
|
||||
Revision int64 `json:"revision"`
|
||||
State GraphState `json:"state"`
|
||||
Nodes []GraphNode `json:"nodes"`
|
||||
Edges []GraphEdge `json:"edges"`
|
||||
Budget GraphBudget `json:"budget"`
|
||||
ParentGraphID string `json:"parent_graph_id,omitempty"`
|
||||
ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"`
|
||||
TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"`
|
||||
ID string `json:"id"`
|
||||
Revision int64 `json:"revision"`
|
||||
State GraphState `json:"state"`
|
||||
Nodes []GraphNode `json:"nodes"`
|
||||
Edges []GraphEdge `json:"edges"`
|
||||
Budget GraphBudget `json:"budget"`
|
||||
ParentGraphID string `json:"parent_graph_id,omitempty"`
|
||||
ParentGraphRevision int64 `json:"parent_graph_revision,omitempty"`
|
||||
TriggerEvidenceRef string `json:"trigger_evidence_ref,omitempty"`
|
||||
SearchInputBound bool `json:"search_input_bound,omitempty"`
|
||||
SearchInput SolutionSearchInput `json:"search_input,omitempty"`
|
||||
}
|
||||
|
||||
// TaskEvaluationGraphRequest describes a durable fan-out. The graph contains
|
||||
@@ -110,6 +113,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"`
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
BEGIN;
|
||||
|
||||
SET ROLE work_role;
|
||||
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_bound BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_goal_id TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_goal_state TEXT NOT NULL DEFAULT '';
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_search_round INTEGER NOT NULL DEFAULT 0;
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_max_round INTEGER NOT NULL DEFAULT 0;
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_cancel_requested BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_verification_ready BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
ALTER TABLE work.graphs ADD COLUMN IF NOT EXISTS orchestration_unresolved BOOLEAN NOT NULL DEFAULT FALSE;
|
||||
|
||||
ALTER TABLE work.graphs DROP CONSTRAINT IF EXISTS work_graphs_orchestration_snapshot_check;
|
||||
ALTER TABLE work.graphs ADD CONSTRAINT work_graphs_orchestration_snapshot_check CHECK (
|
||||
(NOT orchestration_bound AND orchestration_goal_id = '' AND orchestration_search_round = 0 AND orchestration_max_round = 0)
|
||||
OR (orchestration_bound AND orchestration_goal_id <> '' AND orchestration_search_round >= 1 AND orchestration_max_round >= 1)
|
||||
);
|
||||
|
||||
RESET ROLE;
|
||||
COMMIT;
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"task_file": "dev/tasks/DEV-000232-durable-solution-wave-dispatch.json",
|
||||
"task_id": "DEV-000232",
|
||||
"task_file": "dev/tasks/DEV-000234-orchestration-snapshot-replan.json",
|
||||
"task_id": "DEV-000234",
|
||||
"task_revision": 1
|
||||
}
|
||||
|
||||
@@ -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"]
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"task_id": "DEV-000234",
|
||||
"task_revision": 1,
|
||||
"architecture_revision": "ADR-0018",
|
||||
"base_repo_revision": "eac39a3fd40e9f8d0752af6f9e7893cb0b03055b",
|
||||
"risk_tier": "E",
|
||||
"intent": "Persist the solution-search snapshot with every orchestration Work Graph revision and return the next durable search decision after verified graph completion.",
|
||||
"requirement_ids": ["durable goal continuity", "automatic replan signal", "restart recovery", "no invented execution authority"],
|
||||
"allowed_change_scope": {"modules": ["work-graph", "postgres-work-adapter"], "paths": ["contracts/work/graph.go", "modules/work/domain/orchestration.go", "modules/work/application/graph_runtime.go", "adapters/postgres/work_repository.go", "database/migrations/044_work_graph_orchestration_snapshot.sql", "cmd/work-graph-runtime/main.go", "tests/agent_execution/graph_runtime_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000234-orchestration-snapshot-replan.json", "docs/PROJECT_STATUS.md"]},
|
||||
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "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": 1},
|
||||
"contracts": {"consume": ["Work Graph immutable revision store", "SolutionSearchDecision policy"], "may_change": ["orchestration snapshot fields on WorkGraph", "graph completion result with next search decision"]},
|
||||
"semantic_owner": "Work Graph orchestration continuity",
|
||||
"canonical_state": ["work.graphs revision snapshot including GoalID and search round"],
|
||||
"state_transitions": ["wave materialization persists search snapshot", "verified graph completion -> next search decision", "next decision remains nonterminal when unresolved"],
|
||||
"side_effects_and_privileges": ["Work-owned graph persistence and queue orchestration only; no model/provider call, secret access, external mutation or promotion"],
|
||||
"persistence_and_migrations": ["migration 044 adds nullable-by-contract orchestration snapshot columns with backward-compatible defaults"],
|
||||
"failure_retry_restart": ["restarting from any immutable graph revision reconstructs the same search input; missing snapshot does not invent a replan"],
|
||||
"assumptions": ["dispatch-intent discovery/materialization is supplied by the next wave planner; this stage returns the durable hand-off rather than fabricating agent authority"],
|
||||
"acceptance_scenarios": ["new solution wave persists goal and round", "verified completion returns next round decision", "cancelled or verified terminal goal does not replan", "legacy graph without snapshot completes without invented replan"],
|
||||
"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", "REQUIRES_MIGRATION_DECISION"]
|
||||
}
|
||||
@@ -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-000234-orchestration-snapshot-replan.json revision 1`; orchestration waves now persist their GoalID/search snapshot in immutable Work Graph revisions, and verified graph completion returns the next durable search decision. Dispatch-intent discovery/materialization remains an explicit planner hand-off rather than fabricated authority.
|
||||
- `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 = connect the returned replan decision to capability/agent-intent discovery and automatic next-wave materialization; then add bounded stale-wave recovery and continue remaining audit gaps.`
|
||||
|
||||
@@ -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"`
|
||||
@@ -35,6 +36,12 @@ type GraphCompleteRequest struct {
|
||||
Succeeded bool `json:"succeeded"`
|
||||
}
|
||||
|
||||
type GraphCompletionResult struct {
|
||||
Graph work.WorkGraph `json:"graph"`
|
||||
NextSearchDecision work.SolutionSearchDecision `json:"next_search_decision"`
|
||||
ReplanRequired bool `json:"replan_required"`
|
||||
}
|
||||
|
||||
type GraphSynthesisRequest struct {
|
||||
Conflict work.ConflictSet `json:"conflict"`
|
||||
Candidates []work.ResultCandidate `json:"candidates"`
|
||||
@@ -82,6 +89,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()
|
||||
@@ -231,3 +283,27 @@ func CompleteGraphNode(ctx context.Context, store ports.GraphStore, graphID stri
|
||||
}
|
||||
return graph, nil
|
||||
}
|
||||
|
||||
// CompleteGraphNodeWithReplan evaluates the persisted search snapshot after a
|
||||
// verified graph completion. It does not silently invent dispatch intents;
|
||||
// the returned decision is the durable hand-off to the next wave planner.
|
||||
func CompleteGraphNodeWithReplan(ctx context.Context, store ports.GraphStore, request GraphCompleteRequest) (GraphCompletionResult, error) {
|
||||
graph, err := CompleteGraphNode(ctx, store, request.GraphID, request.Revision, request.NodeID, request.Succeeded)
|
||||
if err != nil {
|
||||
return GraphCompletionResult{}, err
|
||||
}
|
||||
result := GraphCompletionResult{Graph: graph}
|
||||
if !graph.SearchInputBound || graph.State != work.GraphCompleted {
|
||||
return result, nil
|
||||
}
|
||||
input := graph.SearchInput
|
||||
input.ActiveOptionCount = 0
|
||||
input.Unresolved = true
|
||||
decision, err := DecideSolutionSearch(input)
|
||||
if err != nil {
|
||||
return GraphCompletionResult{}, err
|
||||
}
|
||||
result.NextSearchDecision = decision
|
||||
result.ReplanRequired = decision.ContinueSearching
|
||||
return result, nil
|
||||
}
|
||||
|
||||
@@ -78,7 +78,9 @@ func BuildSolutionSearchWave(decision work.SolutionSearchDecision, request work.
|
||||
if len(nodes) == 0 {
|
||||
return work.WorkGraph{}, nil, ErrInvalidSolutionSearchWave
|
||||
}
|
||||
return work.WorkGraph{ID: request.GraphID, Revision: 1, State: work.GraphDraft, Nodes: nodes, Budget: request.Budget, ParentGraphID: request.ParentGraphID, ParentGraphRevision: request.ParentGraphRevision, TriggerEvidenceRef: request.TriggerEvidenceRef}, intents, nil
|
||||
searchInput := request.Input
|
||||
searchInput.SearchRound = decision.NextSearchRound
|
||||
return work.WorkGraph{ID: request.GraphID, Revision: 1, State: work.GraphDraft, Nodes: nodes, Budget: request.Budget, ParentGraphID: request.ParentGraphID, ParentGraphRevision: request.ParentGraphRevision, TriggerEvidenceRef: request.TriggerEvidenceRef, SearchInputBound: true, SearchInput: searchInput}, intents, nil
|
||||
}
|
||||
|
||||
func containsSearchOption(options []work.SearchOptionKind, target work.SearchOptionKind) bool {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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,34 @@ 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 TestCompletedOrchestrationGraphReturnsDurableNextWaveDecision(t *testing.T) {
|
||||
store := &graphStore{}
|
||||
graph, _, err := application.AdmitAndSchedule(context.Background(), store, application.GraphScheduleRequest{Graph: work.WorkGraph{ID: "replan-graph", Revision: 1, Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 1}, SearchInputBound: true, SearchInput: work.SolutionSearchInput{GoalID: "goal-1", GoalState: work.GoalInProgress, SearchRound: 1, MaxRound: 3}, 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)
|
||||
}
|
||||
result, err := application.CompleteGraphNodeWithReplan(context.Background(), store, application.GraphCompleteRequest{GraphID: graph.ID, Revision: graph.Revision, NodeID: "node-1", Succeeded: true})
|
||||
if err != nil || !result.ReplanRequired || !result.NextSearchDecision.ContinueSearching || result.NextSearchDecision.NextSearchRound != 2 {
|
||||
t.Fatalf("missing next-wave decision: result=%+v err=%v", result, 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"`) {
|
||||
|
||||
Reference in New Issue
Block a user