work_graph_plan.py
13.2 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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
"""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", "test_dataset", "test_script"],
},
"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, Benchmark, "
"Test Dataset, and Test Script artifacts for this goal: {goal_text}\n"
"Output a graph_manifest with one node per artifact_type (brd, prd, sdd, benchmark, "
"test_dataset, test_script) and edges expressing their dependency chain (sdd depends "
"on prd depends on brd, benchmark depends on sdd, test_dataset depends on benchmark, "
"test_script depends on test_dataset). 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."
)
# E (2026-08-06): the generic, cross-Org-Agent path -- used for every
# org_node OTHER than 'planner'. Planner's own doc-writing chain stays on
# _PROMPT_TEMPLATE above unchanged (zero regression risk, see
# docs/todo/dynamic-work-graph-design-skill-design.md §3): that chain is a
# fixed architectural constant (brd->prd->sdd->benchmark->...), not
# something that should vary call to call, whereas Coder/Validator's
# sub-tasks have no such fixed small vocabulary of node types. Each node
# here carries its own self-contained `task_prompt` (see the
# design-dynamic-work-graph Skill this prompt points at) instead of an
# `artifact_type` that a fixed lookup table like _DOC_PROMPTS could resolve.
_GENERIC_PROMPT_TEMPLATE = (
"Plan (do not execute) a Work Graph for this goal: {goal_text}\n"
"Follow the design-dynamic-work-graph Skill's methodology: only add a depends_on edge "
"when a node's task_prompt genuinely needs another node's output content -- nodes with "
"no real data dependency between them will be scheduled to run CONCURRENTLY, so adding "
"an edge just to make the graph look ordered actively slows execution down. If this goal "
"is small enough to finish in one call, output a single-node graph_manifest rather than "
"forcing a split.\n"
"Output a graph_manifest whose nodes each have: node_id (a short readable identifier), "
"depends_on (list of other node_ids), and task_prompt (the FULL, self-contained prompt "
"for that node's own execution -- assume the agent running it sees nothing except this "
"task_prompt and its dependencies' outputs, not this planning conversation). 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
_CANONICAL_DOC_CHAIN = ("brd", "prd", "sdd", "benchmark", "test_dataset", "test_script")
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`.
**Canonical doc-chain enforcement (2026-08-03, real self-hosting run)**:
a real call was observed to omit the brd->prd->sdd->benchmark
dependency entirely -- no `depends_on`, no `edges` connecting them at
all, not just the wrong key names the edge-derived backfill above can
already recover from. With no dependency info to normalize from,
`topological_order()` scheduled `benchmark` before `sdd` even ran
(alphabetically first among nodes with an empty in-degree), and
`e2e_pipeline.build_prompt()` crashed with `KeyError: 'sdd_content'`
trying to format a template that needed sdd's not-yet-produced output.
`_PROMPT_TEMPLATE` always asks for exactly this fixed 4-node linear
chain -- it's a constant architectural fact about what the Planner's
doc-writing graph looks like, not something that genuinely varies call
to call, so don't leave it to chance: whenever all four canonical
artifact_types are present, deterministically enforce the known linear
order regardless of what edges/depends_on the LLM did or didn't
provide. Same "机制用代码" principle already applied elsewhere
(`work_node_run.ensure_worktree_committed`, etc.).
"""
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)
by_artifact_type = {n.get("artifact_type"): n for n in nodes}
if all(t in by_artifact_type for t in _CANONICAL_DOC_CHAIN):
for prev_type, cur_type in zip(_CANONICAL_DOC_CHAIN, _CANONICAL_DOC_CHAIN[1:]):
prev_id = by_artifact_type[prev_type]["node_id"]
cur_node = by_artifact_type[cur_type]
if prev_id not in cur_node["depends_on"]:
cur_node["depends_on"].append(prev_id)
# 2026-08-06: test_script's own prompt needs {benchmark_content}
# directly (not just what survives into test_dataset_content, which
# can be a terse 'N/A: ...' with none of the original scenario
# detail) -- the plain linear chain above only gives it
# test_dataset as an immediate dependency, so add benchmark as a
# second one. e2e_pipeline._planner_prompt_builder already threads
# every *immediate* dependency's content through generically, this
# is the one canonical node that genuinely needs two.
benchmark_id = by_artifact_type["benchmark"]["node_id"]
script_node = by_artifact_type["test_script"]
if benchmark_id not in script_node["depends_on"]:
script_node["depends_on"].append(benchmark_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.
E (2026-08-06): `org_node == 'planner'` still gets the fixed
brd/prd/sdd/benchmark/... chain prompt (_PROMPT_TEMPLATE, unchanged).
Every other org_node gets _GENERIC_PROMPT_TEMPLATE -- the cross-Org-
Agent dynamic planning path described in the design-dynamic-work-graph
Skill, whose nodes carry their own task_prompt instead of a fixed
artifact_type. Function signature/return shape are unchanged either way."""
template = _PROMPT_TEMPLATE if org_node == "planner" else _GENERIC_PROMPT_TEMPLATE
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}
def generic_node_prompt_builder(node, outputs):
"""E (2026-08-06): the `build_prompt(node, outputs)` run_graph() needs
for a graph_manifest produced via _GENERIC_PROMPT_TEMPLATE -- lives
here, not in e2e_pipeline.py, because it's tightly coupled to the
`task_prompt` field convention that prompt asks for (keeping both in
the same module means a field rename can't silently desync producer
and consumer). No factory/closure needed: every node is self-contained
per the design-dynamic-work-graph Skill's contract, so `outputs`
(already-completed dependency content) isn't needed here."""
return node["task_prompt"]