From 09d53adfd10c98c79995f4ce85be5caa7747307d Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sun, 23 Aug 2026 16:56:54 +0300 Subject: [PATCH] feat(work): materialize durable solution waves --- cmd/work-graph-runtime/main.go | 13 ++++++ contracts/work/orchestration.go | 24 +++++++++++ dev/ACTIVE_TASK.json | 4 +- ...000232-durable-solution-wave-dispatch.json | 23 +++++++++++ docs/PROJECT_STATUS.md | 4 +- modules/work/application/orchestration.go | 29 ++++++++++++++ modules/work/domain/orchestration.go | 40 +++++++++++++++++++ modules/work/domain/orchestration_test.go | 29 ++++++++++++++ 8 files changed, 162 insertions(+), 4 deletions(-) create mode 100644 dev/tasks/DEV-000232-durable-solution-wave-dispatch.json diff --git a/cmd/work-graph-runtime/main.go b/cmd/work-graph-runtime/main.go index dd69603..63acab6 100644 --- a/cmd/work-graph-runtime/main.go +++ b/cmd/work-graph-runtime/main.go @@ -74,6 +74,19 @@ func main() { } writeJSON(w, decision) }) + mux.HandleFunc("POST /v1/work-graphs/orchestration/wave", func(w http.ResponseWriter, r *http.Request) { + var request work.SolutionSearchWaveRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)).Decode(&request); err != nil { + http.Error(w, "invalid orchestration wave request", http.StatusBadRequest) + return + } + result, err := application.OrchestrateSolutionSearch(r.Context(), repository, request) + if err != nil { + http.Error(w, "invalid orchestration wave request", http.StatusUnprocessableEntity) + return + } + writeJSON(w, result) + }) 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 { diff --git a/contracts/work/orchestration.go b/contracts/work/orchestration.go index a83ef3c..41bb6c1 100644 --- a/contracts/work/orchestration.go +++ b/contracts/work/orchestration.go @@ -40,3 +40,27 @@ type SolutionSearchDecision struct { Options []SearchOptionKind Reason string } + +type SolutionSearchWaveRequest struct { + Input SolutionSearchInput `json:"input"` + GraphID string `json:"graph_id"` + IdempotencyPrefix string `json:"idempotency_prefix"` + Dispatches []SolutionSearchDispatch `json:"dispatches"` + 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"` + NowUnixMillis int64 `json:"now_unix_millis"` +} + +type SolutionSearchDispatch struct { + Option SearchOptionKind `json:"option"` + Intent AgentDispatchIntent `json:"intent"` +} + +type SolutionSearchWaveResult struct { + Decision SolutionSearchDecision `json:"decision"` + Graph WorkGraph `json:"graph"` + ScheduledNodeIDs []GraphNodeID `json:"scheduled_node_ids,omitempty"` + GraphMaterialized bool `json:"graph_materialized"` +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 6e5bd80..3b29de4 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000231-solution-orchestrator.json", - "task_id": "DEV-000231", + "task_file": "dev/tasks/DEV-000232-durable-solution-wave-dispatch.json", + "task_id": "DEV-000232", "task_revision": 1 } diff --git a/dev/tasks/DEV-000232-durable-solution-wave-dispatch.json b/dev/tasks/DEV-000232-durable-solution-wave-dispatch.json new file mode 100644 index 0000000..4d43b32 --- /dev/null +++ b/dev/tasks/DEV-000232-durable-solution-wave-dispatch.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000232", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "89694047b1e6c7729b00c20a52acb5a2487e6dd6", + "risk_tier": "E", + "intent": "Connect the solution-search policy to the existing durable Work Graph scheduler so each unresolved wave is materialized as parallel dispatch intents and queued nodes.", + "requirement_ids": ["durable solution waves", "parallel fan-out", "agent/reflex orchestration", "restartable scheduling"], + "allowed_change_scope": {"modules": ["work-graph", "work-domain"], "paths": ["contracts/work/orchestration.go", "modules/work/domain/orchestration.go", "modules/work/domain/orchestration_test.go", "modules/work/application/orchestration.go", "cmd/work-graph-runtime/main.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000232-durable-solution-wave-dispatch.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": ["typed solution-search decision", "existing Work Graph and dispatch queue ports"], "may_change": ["solution-search wave materialization request/result", "Work Graph orchestration wave route"]}, + "semantic_owner": "Work solution orchestration and Work Graph", + "canonical_state": ["existing work.graphs, graph_nodes, dispatch_intents and dispatch_queue stores"], + "state_transitions": ["policy SEARCHING/WAITING -> graph ADMITTED/RUNNING", "each option -> one queued dispatch node", "human attention remains decision metadata"], + "side_effects_and_privileges": ["durable Work-owned graph and queue writes only; no provider call, secret access, external mutation, promotion or deployment"], + "persistence_and_migrations": ["none; existing graph and queue tables are reused"], + "failure_retry_restart": ["existing queue idempotency and graph revision persistence govern restart; invalid wave shape is rejected; no unresolved goal is marked failed by wave exhaustion"], + "assumptions": ["dispatch intents are supplied by the upstream authority-aware planner", "provider/model selection and execution remain worker responsibilities"], + "acceptance_scenarios": ["four independent solution avenues materialize as four graph nodes", "all ready nodes are queued up to graph MaxParallel", "human attention is not silently executed as an agent", "terminal policy decisions do not create a graph", "invalid or duplicate options are rejected"], + "required_tests": ["go test ./modules/work/domain ./modules/work/application ./cmd/work-graph-runtime/...", "go vet ./modules/work/domain ./modules/work/application ./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 ecac54a..61aab46 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-000231-solution-orchestrator.json revision 1`; Work Graph now exposes a typed parallel-first solution-search decision: active waves execute, unresolved goals open agent/capability/research/reflection avenues, and bounded exhaustion advances to a durable next wave with human attention instead of failure. Automatic dispatcher fan-out and persistence of the returned decision remain the next integration step. +- `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. - `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 this decision policy to the durable Work Graph dispatcher/worker so every unresolved goal persists its next wave, fans out independent options and joins verified evidence; retain local fallback, truth research and governed development gates.` +- `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.` diff --git a/modules/work/application/orchestration.go b/modules/work/application/orchestration.go index 1e0cff2..d79a65e 100644 --- a/modules/work/application/orchestration.go +++ b/modules/work/application/orchestration.go @@ -1,10 +1,39 @@ package application import ( + "context" "git.cin.su/m/Assistant/contracts/work" "git.cin.su/m/Assistant/modules/work/domain" + "git.cin.su/m/Assistant/modules/work/ports" + "time" ) func DecideSolutionSearch(input work.SolutionSearchInput) (work.SolutionSearchDecision, error) { return domain.DecideSolutionSearch(input) } + +func OrchestrateSolutionSearch(ctx context.Context, store ports.GraphStore, request work.SolutionSearchWaveRequest) (work.SolutionSearchWaveResult, error) { + decision, err := domain.DecideSolutionSearch(request.Input) + if err != nil { + return work.SolutionSearchWaveResult{}, err + } + result := work.SolutionSearchWaveResult{Decision: decision} + if !decision.ContinueSearching || decision.State == work.OrchestrationExecuting { + return result, nil + } + graph, intents, err := domain.BuildSolutionSearchWave(decision, request) + if err != nil { + return work.SolutionSearchWaveResult{}, err + } + if request.NowUnixMillis < 1 { + request.NowUnixMillis = time.Now().UnixMilli() + } + graph, ready, err := AdmitAndSchedule(ctx, store, GraphScheduleRequest{Graph: graph, Intents: intents, Now: request.NowUnixMillis}) + if err != nil { + return work.SolutionSearchWaveResult{}, err + } + result.Graph = graph + result.ScheduledNodeIDs = ready + result.GraphMaterialized = true + return result, nil +} diff --git a/modules/work/domain/orchestration.go b/modules/work/domain/orchestration.go index a5b6ab1..c6b27bb 100644 --- a/modules/work/domain/orchestration.go +++ b/modules/work/domain/orchestration.go @@ -2,6 +2,7 @@ package domain import ( "errors" + "fmt" "git.cin.su/m/Assistant/contracts/work" ) @@ -49,3 +50,42 @@ func DecideSolutionSearch(input work.SolutionSearchInput) (work.SolutionSearchDe } return decision, nil } + +var ErrInvalidSolutionSearchWave = errors.New("invalid solution search wave") + +// BuildSolutionSearchWave turns a policy decision into a durable Work Graph +// fan-out. Human attention remains in the decision but is not an agent dispatch. +func BuildSolutionSearchWave(decision work.SolutionSearchDecision, request work.SolutionSearchWaveRequest) (work.WorkGraph, []work.AgentDispatchIntent, error) { + if !decision.ContinueSearching || (decision.State != work.OrchestrationSearching && decision.State != work.OrchestrationWaiting) || request.GraphID == "" || request.IdempotencyPrefix == "" || request.Budget.MaxParallel < 1 || request.Budget.MaxAttempts < 1 { + return work.WorkGraph{}, nil, ErrInvalidSolutionSearchWave + } + seen := make(map[work.SearchOptionKind]struct{}, len(request.Dispatches)) + nodes := make([]work.GraphNode, 0, len(request.Dispatches)) + intents := make([]work.AgentDispatchIntent, 0, len(request.Dispatches)) + for index, dispatch := range request.Dispatches { + if dispatch.Option == work.SearchOptionHuman || dispatch.Option == "" || dispatch.Intent.DispatchID == "" || dispatch.Intent.IdempotencyKey == "" { + return work.WorkGraph{}, nil, ErrInvalidSolutionSearchWave + } + if _, exists := seen[dispatch.Option]; exists || !containsSearchOption(decision.Options, dispatch.Option) { + return work.WorkGraph{}, nil, ErrInvalidSolutionSearchWave + } + seen[dispatch.Option] = struct{}{} + intent := dispatch.Intent + intent.SpecialistPurpose = string(dispatch.Option) + intents = append(intents, intent) + nodes = append(nodes, work.GraphNode{ID: work.GraphNodeID(fmt.Sprintf("%s-node-%d", request.GraphID, index+1)), DispatchID: intent.DispatchID, IdempotencyKey: fmt.Sprintf("%s:%s:%d", request.IdempotencyPrefix, dispatch.Option, decision.NextSearchRound), State: work.GraphNodeBlocked, Purpose: work.GraphNodePurposeGeneric}) + } + 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 +} + +func containsSearchOption(options []work.SearchOptionKind, target work.SearchOptionKind) bool { + for _, option := range options { + if option == target { + return true + } + } + return false +} diff --git a/modules/work/domain/orchestration_test.go b/modules/work/domain/orchestration_test.go index b9c6c63..4f30f3b 100644 --- a/modules/work/domain/orchestration_test.go +++ b/modules/work/domain/orchestration_test.go @@ -1,6 +1,7 @@ package domain import ( + "fmt" "git.cin.su/m/Assistant/contracts/work" "testing" ) @@ -28,3 +29,31 @@ func TestSolutionSearchStopsOnlyForCancelOrVerifiedCompletion(t *testing.T) { t.Fatalf("cancel was not terminal: %#v %v", cancelled, err) } } + +func TestBuildSolutionSearchWaveMaterializesParallelAgentDirections(t *testing.T) { + decision, err := DecideSolutionSearch(work.SolutionSearchInput{GoalID: "goal-1", GoalState: work.GoalInProgress, MaxRound: 1}) + if err != nil { + t.Fatal(err) + } + options := []work.SearchOptionKind{work.SearchOptionAgents, work.SearchOptionCapability, work.SearchOptionResearch, work.SearchOptionReflection} + dispatches := make([]work.SolutionSearchDispatch, 0, len(options)) + for i, option := range options { + dispatches = append(dispatches, work.SolutionSearchDispatch{Option: option, Intent: work.AgentDispatchIntent{DispatchID: workDispatchID(i), IdempotencyKey: workDispatchID(i) + ":idem"}}) + } + graph, intents, err := BuildSolutionSearchWave(decision, work.SolutionSearchWaveRequest{GraphID: "graph-1", IdempotencyPrefix: "goal-1", Dispatches: dispatches, Budget: work.GraphBudget{MaxParallel: 4, MaxAttempts: 2}}) + if err != nil { + t.Fatal(err) + } + if len(graph.Nodes) != 4 || len(intents) != 4 || graph.State != work.GraphDraft { + t.Fatalf("unexpected materialized wave: %#v %#v", graph, intents) + } + for i, node := range graph.Nodes { + if node.State != work.GraphNodeBlocked || intents[i].SpecialistPurpose != string(options[i]) { + t.Fatalf("wave direction not preserved: node=%#v intent=%#v", node, intents[i]) + } + } +} + +func workDispatchID(index int) string { + return fmt.Sprintf("dispatch-%d", index) +} -- 2.54.0