work_graph_plan.py
5.75 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
"""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 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:
raise PlanningFailedError("plan failed for task {}: {}".format(task_id, result.stderr))
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))
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}