work_graph_plan.py
7.36 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
"""c_work_graph_plan (03-C §3.4): produces a `graph_manifest`, does not execute.
U-011 scope: a real `claude --agent <org_node> -p` call, recorded as a
`work_graph_run` row with `status='planning'` (04-P `p_work_graph_execution`
start_state).
**Empirical finding (2026-07-31)**: `--json-schema` does NOT constrain output
when combined with `--agent` -- tried it for real, the response came back as
normal prose with an embedded ```json fenced block, `structured_output` was
entirely absent from the payload even though the identity itself came through
correctly (the response referenced "SOUL 原则 #10" by number, i.e. it was
genuinely reasoning as the migrated Planner). So this parses the
graph_manifest out of free-form response text instead of trusting
`structured_output` -- see `_extract_graph_manifest`. Not documented anywhere;
found by actually running it, not assumed.
Does NOT execute the resulting plan or produce artifacts -- that's
U-012 (`c_work_graph_execute` full)/U-013 (`c_work_node_run`), and nothing
today owns `c_artifact_write` either (REQ-02 lists it, no Unit in
development-plan.md builds it -- a real gap, flagged rather than silently
plugged, since it affects more than just this Unit's scope). REQ-02's happy
case (real BRD/PRD/SDD/Benchmark artifacts with lineage) needs all of that;
development-plan.md's U-011 Done line claimed it passes here alone, which
isn't accurate -- corrected in this session's notes, same pattern as
U-009/U-010.
"""
import json
import re
import subprocess
import uuid
GRAPH_MANIFEST_SCHEMA = {
"type": "object",
"required": ["nodes", "edges"],
"properties": {
"nodes": {
"type": "array",
"items": {
"type": "object",
"required": ["node_id", "artifact_type", "depends_on"],
"properties": {
"node_id": {"type": "string"},
"artifact_type": {"type": "string", "enum": ["brd", "prd", "sdd", "benchmark"]},
"depends_on": {"type": "array", "items": {"type": "string"}},
},
},
},
"edges": {
"type": "array",
"items": {
"type": "object",
"required": ["from", "to", "edge_type"],
"properties": {
"from": {"type": "string"},
"to": {"type": "string"},
"edge_type": {"type": "string", "enum": ["zero_token", "needs_judgment"]},
},
},
},
},
}
_PROMPT_TEMPLATE = (
"Plan (do not execute) the Work Graph needed to produce BRD, PRD, SDD, and "
"Benchmark artifacts for this goal: {goal_text}\n"
"Output a graph_manifest with one node per artifact_type (brd, prd, sdd, "
"benchmark) and edges expressing their dependency chain (sdd depends on prd "
"depends on brd, benchmark depends on sdd). Every edge needs edge_type: "
"'zero_token' for pure aggregation/reduce steps, 'needs_judgment' for anything "
"requiring the agent to actually reason about content.\n"
"End your reply with the graph_manifest as a single fenced ```json code block "
"shaped exactly like {{\"nodes\": [...], \"edges\": [...]}} -- no extra wrapper key."
)
_JSON_FENCE_RE = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL)
class PlanningFailedError(Exception):
"""PLANNING_FAILED -- 03-C §3.4."""
def _extract_graph_manifest(text):
"""--json-schema doesn't constrain output when --agent is set (see module
docstring) -- pulls the graph_manifest out of free-form response text
instead: try the whole text as JSON, then each ```json fenced block;
accept either {"nodes":...,"edges":...} directly or wrapped one level
deep under a "graph_manifest" key (models sometimes add that wrapper
despite being asked not to)."""
candidates = [text] + _JSON_FENCE_RE.findall(text)
for candidate in candidates:
try:
parsed = json.loads(candidate)
except (json.JSONDecodeError, TypeError):
continue
if isinstance(parsed, dict):
if "nodes" in parsed and "edges" in parsed:
return parsed
inner = parsed.get("graph_manifest")
if isinstance(inner, dict) and "nodes" in inner and "edges" in inner:
return inner
return None
def _normalize_graph_manifest(manifest):
"""Real LLM output has been observed (2026-08-01, U-031 integration
validation) to use `id` instead of the documented `node_id` key, and to
express dependencies purely via `edges` (from/to) without ever
populating each node's own `depends_on` array -- even though this
schema requires both. Normalizing here, right after parsing and before
persisting, means every downstream consumer (work_graph_orchestrate.py's
topological sort, v_task_detail's to-do tree, v_graph_visualizer's DAG
layering) gets a manifest that actually matches the documented shape,
instead of each one needing its own defensive "id or node_id,
depends_on or edges" fallback logic repeated independently. Mutates and
returns `manifest`.
"""
nodes = manifest.get("nodes", [])
for node in nodes:
if "node_id" not in node and "id" in node:
node["node_id"] = node["id"]
node.setdefault("depends_on", [])
node_ids = {n.get("node_id") for n in nodes}
for edge in manifest.get("edges", []):
from_id, to_id = edge.get("from"), edge.get("to")
if from_id in node_ids and to_id in node_ids:
for node in nodes:
if node.get("node_id") == to_id and from_id not in node["depends_on"]:
node["depends_on"].append(from_id)
return manifest
def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300):
"""Returns {"run_id", "graph_manifest"}. Creates one `work_graph_run` row
(status='planning') -- moving it to 'running'/executing it is out of scope."""
prompt = _PROMPT_TEMPLATE.format(goal_text=goal_text)
result = subprocess.run(
[
"claude", "--agent", org_node,
"-p", prompt,
"--output-format", "json",
"--permission-mode", "auto",
],
cwd=target_folder_path,
stdin=subprocess.DEVNULL,
capture_output=True,
text=True,
timeout=timeout,
)
if result.returncode != 0:
# 2026-07-31 fix (same bug as org_node_trigger.py): include stdout too.
raise PlanningFailedError(
"plan failed for task {}\nstderr: {}\nstdout (truncated): {}".format(
task_id, result.stderr, result.stdout[:4000]
)
)
payload = json.loads(result.stdout)
if payload.get("is_error"):
raise PlanningFailedError("plan failed for task {}: {}".format(task_id, payload))
graph_manifest = _extract_graph_manifest(payload.get("result", ""))
if not graph_manifest:
raise PlanningFailedError("could not extract a graph_manifest for task {}: {}".format(task_id, payload))
graph_manifest = _normalize_graph_manifest(graph_manifest)
run_id = str(uuid.uuid4())
conn.execute(
"INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
"VALUES (?, ?, ?, ?, 'planning')",
(run_id, task_id, org_node, json.dumps(graph_manifest)),
)
conn.commit()
return {"run_id": run_id, "graph_manifest": graph_manifest}