From da251e1aa80c89a289388e08f66682f940db8a15 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 18:12:05 +0300 Subject: [PATCH] feat(work): persist orchestration snapshot for replan --- adapters/postgres/work_repository.go | 4 +-- cmd/work-graph-runtime/main.go | 4 +-- contracts/work/graph.go | 20 +++++++------ .../044_work_graph_orchestration_snapshot.sql | 21 +++++++++++++ dev/ACTIVE_TASK.json | 4 +-- ...-000234-orchestration-snapshot-replan.json | 23 ++++++++++++++ docs/PROJECT_STATUS.md | 4 +-- modules/work/application/graph_runtime.go | 30 +++++++++++++++++++ modules/work/domain/orchestration.go | 4 ++- tests/agent_execution/graph_runtime_test.go | 12 ++++++++ 10 files changed, 108 insertions(+), 18 deletions(-) create mode 100644 database/migrations/044_work_graph_orchestration_snapshot.sql create mode 100644 dev/tasks/DEV-000234-orchestration-snapshot-replan.json diff --git a/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 423371d..82b6b59 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -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) } diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index 48cdc97..74b006a 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -106,12 +106,12 @@ func main() { 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 diff --git a/contracts/work/graph.go b/contracts/work/graph.go index cffc085..cabea08 100644 --- a/contracts/work/graph.go +++ b/contracts/work/graph.go @@ -57,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 diff --git a/database/migrations/044_work_graph_orchestration_snapshot.sql b/database/migrations/044_work_graph_orchestration_snapshot.sql new file mode 100644 index 0000000..4354048 --- /dev/null +++ b/database/migrations/044_work_graph_orchestration_snapshot.sql @@ -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; diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index e15d5b7..f5d636f 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000233-graph-result-fanin-verification.json", - "task_id": "DEV-000233", + "task_file": "dev/tasks/DEV-000234-orchestration-snapshot-replan.json", + "task_id": "DEV-000234", "task_revision": 1 } diff --git a/dev/tasks/DEV-000234-orchestration-snapshot-replan.json b/dev/tasks/DEV-000234-orchestration-snapshot-replan.json new file mode 100644 index 0000000..c4714f5 --- /dev/null +++ b/dev/tasks/DEV-000234-orchestration-snapshot-replan.json @@ -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"] +} diff --git a/docs/PROJECT_STATUS.md b/docs/PROJECT_STATUS.md index a000306..3ab9e65 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-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. +- `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 = 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.` +- `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.` diff --git a/modules/work/application/graph_runtime.go b/modules/work/application/graph_runtime.go index 7ce6180..bf617b9 100644 --- a/modules/work/application/graph_runtime.go +++ b/modules/work/application/graph_runtime.go @@ -36,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"` @@ -277,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 +} diff --git a/modules/work/domain/orchestration.go b/modules/work/domain/orchestration.go index c6b27bb..b3b181f 100644 --- a/modules/work/domain/orchestration.go +++ b/modules/work/domain/orchestration.go @@ -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 { diff --git a/tests/agent_execution/graph_runtime_test.go b/tests/agent_execution/graph_runtime_test.go index dc1a1a7..23345dc 100644 --- a/tests/agent_execution/graph_runtime_test.go +++ b/tests/agent_execution/graph_runtime_test.go @@ -110,6 +110,18 @@ func TestGraphResultCallbackRequiresIndependentSynthesisBeforeSuccess(t *testing } } +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"`) { -- 2.54.0