From b2f922cbbc99a1d7d0e160dcb1f3875a81b4e72c Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 13:29:43 +0300 Subject: [PATCH] feat(work): add deterministic graph harness --- contracts/work/graph.go | 72 +++++++++++++ dev/ACTIVE_TASK.json | 4 +- dev/tasks/DEV-000131-work-graph-harness.json | 23 ++++ modules/work/domain/graph.go | 107 +++++++++++++++++++ tests/agent_execution/graph_test.go | 35 ++++++ 5 files changed, 239 insertions(+), 2 deletions(-) create mode 100644 contracts/work/graph.go create mode 100644 dev/tasks/DEV-000131-work-graph-harness.json create mode 100644 modules/work/domain/graph.go create mode 100644 tests/agent_execution/graph_test.go diff --git a/contracts/work/graph.go b/contracts/work/graph.go new file mode 100644 index 0000000..f9c83b9 --- /dev/null +++ b/contracts/work/graph.go @@ -0,0 +1,72 @@ +package work + +type GraphNodeID string + +type GraphState string +type GraphNodeState string + +const ( + GraphDraft GraphState = "DRAFT" + GraphAdmitted GraphState = "ADMITTED" + GraphRunning GraphState = "RUNNING" + GraphPartial GraphState = "PARTIAL" + GraphVerifying GraphState = "VERIFYING" + GraphCompleted GraphState = "COMPLETED" + 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" +) + +type GraphBudget struct { + MaxParallel int + MaxAttempts int +} + +type GraphNode struct { + ID GraphNodeID + IdempotencyKey string + State GraphNodeState + Attempts int +} + +type GraphEdge struct { + From GraphNodeID + To GraphNodeID +} + +type WorkGraph struct { + ID string + Revision int64 + State GraphState + Nodes []GraphNode + Edges []GraphEdge + Budget GraphBudget +} + +type ResultCandidate struct { + ID string + GraphID string + GraphRevision int64 + NodeID GraphNodeID + AuthorRunID AgentRunID + ArtifactRefs []EvidenceRef + EvidenceSummary string + Summary string +} + +type ConflictSet struct { + ID string + GraphID string + GraphRevision int64 + NodeID GraphNodeID + CandidateIDs []string + Resolved bool +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 7a1c4a2..14d4146 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000130-agent-worker-recovery.json", - "task_id": "DEV-000130", + "task_file": "dev/tasks/DEV-000131-work-graph-harness.json", + "task_id": "DEV-000131", "task_revision": 1 } diff --git a/dev/tasks/DEV-000131-work-graph-harness.json b/dev/tasks/DEV-000131-work-graph-harness.json new file mode 100644 index 0000000..2019b27 --- /dev/null +++ b/dev/tasks/DEV-000131-work-graph-harness.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000131", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "a445034", + "risk_tier": "E", + "intent": "Implement the first deterministic Multi-Agent Work Graph harness: typed graph revision, DAG validation, budgets and dependency-aware ready-node selection.", + "requirement_ids": ["WP-400", "DAG validation", "Fan-out/fan-in foundation"], + "allowed_change_scope": {"modules": ["work-agent-execution"], "paths": ["contracts/work/graph.go", "modules/work/domain/graph.go", "tests/agent_execution/graph_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000131-work-graph-harness.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "database/**", "adapters/**", "cmd/**", "runtime/**", ".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": 0}, + "contracts": {"consume": ["ADR-0021 graph proposal", "DEV-000130 worker/recovery contracts"], "may_change": ["Work graph typed contracts and pure domain harness"]}, + "semantic_owner": "work-agent-execution", + "canonical_state": ["Typed WorkGraph value only; no persistence in this harness"], + "state_transitions": ["DRAFT/ADMITTED/RUNNING graph with dependency-aware READY nodes"], + "side_effects_and_privileges": ["Pure deterministic validation; no dispatch or external mutation"], + "persistence_and_migrations": ["Persistence/scheduler implementation follows after harness"], + "failure_retry_restart": ["Cycles, duplicate IDs/keys and invalid budgets reject before scheduling"], + "assumptions": ["Graph scheduler must not bypass Agent Execution leases or verifier policy"], + "acceptance_scenarios": ["Cycle rejected", "Ready nodes deterministic", "Dependent node remains blocked until prerequisite succeeds", "Budget is validated"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check"], + "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] +} diff --git a/modules/work/domain/graph.go b/modules/work/domain/graph.go new file mode 100644 index 0000000..039002d --- /dev/null +++ b/modules/work/domain/graph.go @@ -0,0 +1,107 @@ +package domain + +import ( + "errors" + "sort" + + "git.cin.su/m/Assistant/contracts/work" +) + +var ( + ErrInvalidGraph = errors.New("invalid work graph") + ErrGraphCycle = errors.New("work graph contains a cycle") +) + +func ValidateWorkGraph(graph work.WorkGraph) error { + if graph.ID == "" || graph.Revision < 1 || graph.Budget.MaxParallel < 1 || graph.Budget.MaxAttempts < 1 || len(graph.Nodes) == 0 { + return ErrInvalidGraph + } + nodes := make(map[work.GraphNodeID]struct{}, len(graph.Nodes)) + keys := make(map[string]struct{}, len(graph.Nodes)) + for _, node := range graph.Nodes { + if node.ID == "" || node.IdempotencyKey == "" || node.Attempts < 0 { + return ErrInvalidGraph + } + if _, exists := nodes[node.ID]; exists { + return ErrInvalidGraph + } + if _, exists := keys[node.IdempotencyKey]; exists { + return ErrInvalidGraph + } + nodes[node.ID] = struct{}{} + keys[node.IdempotencyKey] = struct{}{} + } + adjacency := make(map[work.GraphNodeID][]work.GraphNodeID, len(nodes)) + for _, edge := range graph.Edges { + if edge.From == "" || edge.To == "" || edge.From == edge.To { + return ErrInvalidGraph + } + if _, ok := nodes[edge.From]; !ok { + return ErrInvalidGraph + } + if _, ok := nodes[edge.To]; !ok { + return ErrInvalidGraph + } + adjacency[edge.From] = append(adjacency[edge.From], edge.To) + } + visiting := map[work.GraphNodeID]bool{} + visited := map[work.GraphNodeID]bool{} + var visit func(work.GraphNodeID) error + visit = func(node work.GraphNodeID) error { + if visiting[node] { + return ErrGraphCycle + } + if visited[node] { + return nil + } + visiting[node] = true + for _, next := range adjacency[node] { + if err := visit(next); err != nil { + return err + } + } + delete(visiting, node) + visited[node] = true + return nil + } + for node := range nodes { + if err := visit(node); err != nil { + return err + } + } + return nil +} + +func ReadyGraphNodes(graph work.WorkGraph) ([]work.GraphNodeID, error) { + if err := ValidateWorkGraph(graph); err != nil { + return nil, err + } + completed := map[work.GraphNodeID]bool{} + for _, node := range graph.Nodes { + if node.State == work.GraphNodeSuccess { + completed[node.ID] = true + } + } + blockedBy := map[work.GraphNodeID][]work.GraphNodeID{} + for _, edge := range graph.Edges { + blockedBy[edge.To] = append(blockedBy[edge.To], edge.From) + } + ready := make([]work.GraphNodeID, 0) + for _, node := range graph.Nodes { + if node.State != work.GraphNodeReady && node.State != work.GraphNodeBlocked { + continue + } + dependenciesReady := true + for _, dependency := range blockedBy[node.ID] { + if !completed[dependency] { + dependenciesReady = false + break + } + } + if dependenciesReady { + ready = append(ready, node.ID) + } + } + sort.Slice(ready, func(i, j int) bool { return ready[i] < ready[j] }) + return ready, nil +} diff --git a/tests/agent_execution/graph_test.go b/tests/agent_execution/graph_test.go new file mode 100644 index 0000000..034c252 --- /dev/null +++ b/tests/agent_execution/graph_test.go @@ -0,0 +1,35 @@ +package agent_execution_test + +import ( + "testing" + + "git.cin.su/m/Assistant/contracts/work" + workdomain "git.cin.su/m/Assistant/modules/work/domain" +) + +func TestWorkGraphRejectsCycleAndOrdersReadyNodes(t *testing.T) { + graph := work.WorkGraph{ID: "graph-1", Revision: 1, State: work.GraphAdmitted, Budget: work.GraphBudget{MaxParallel: 2, MaxAttempts: 2}, Nodes: []work.GraphNode{ + {ID: "node-b", IdempotencyKey: "key-b", State: work.GraphNodeReady}, + {ID: "node-a", IdempotencyKey: "key-a", State: work.GraphNodeReady}, + {ID: "node-c", IdempotencyKey: "key-c", State: work.GraphNodeBlocked}, + }, Edges: []work.GraphEdge{{From: "node-a", To: "node-c"}}} + ready, err := workdomain.ReadyGraphNodes(graph) + if err != nil || len(ready) != 2 || ready[0] != "node-a" || ready[1] != "node-b" { + t.Fatalf("unexpected ready nodes: %#v %v", ready, err) + } + graph.Edges = append(graph.Edges, work.GraphEdge{From: "node-c", To: "node-a"}) + if err := workdomain.ValidateWorkGraph(graph); err != workdomain.ErrGraphCycle { + t.Fatalf("expected cycle rejection, got %v", err) + } +} + +func TestWorkGraphKeepsDependentNodeBlocked(t *testing.T) { + graph := work.WorkGraph{ID: "graph-1", Revision: 1, State: work.GraphRunning, Budget: work.GraphBudget{MaxParallel: 1, MaxAttempts: 1}, Nodes: []work.GraphNode{ + {ID: "source", IdempotencyKey: "source-key", State: work.GraphNodeRunning}, + {ID: "dependent", IdempotencyKey: "dependent-key", State: work.GraphNodeBlocked}, + }, Edges: []work.GraphEdge{{From: "source", To: "dependent"}}} + ready, err := workdomain.ReadyGraphNodes(graph) + if err != nil || len(ready) != 0 { + t.Fatalf("dependent node became ready early: %#v %v", ready, err) + } +}