work_graph_orchestrate.py
8.51 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
"""Bridges `plan_work_graph()`'s `graph_manifest` to real per-node execution
via `execute_work_graph()` + `run_work_node()` -- the missing integration
between three previously isolated, individually-tested modules (U-011/
U-012/U-013), closed during U-031's regression audit (2026-08-01): each of
the three had its own unit tests (and even real-API validation) but none of
them had ever actually been called by production code, and the one
genuinely-validated closed loop (`e2e_pipeline.py`, U-014) used a much
simpler, separate mechanism (flat `org_node_trigger.trigger_org_node()`
calls) that bypassed all three.
Design: each node's actual work happens via its OWN one-node dynamic
workflow script (`write_workflow_script` + `execute_work_graph`), not one
big multi-node script for the whole graph. This is a deliberate trade-off,
not the "amortize orchestration cost across many nodes in one `claude -p`
call" pattern 03-C §3.5 originally motivated `execute_work_graph` with --
REQ-06's breakpoint-recovery requirement (resume after a process restart
without re-running already-`done` nodes) is only achievable with per-node,
Python-level persisted state (`work_node_run.py`'s whole design), which is
fundamentally incompatible with wrapping the WHOLE graph in one opaque
`claude` subprocess call Python has no visibility into mid-flight.
Validated for real (2026-08-01): a one-node workflow script with a custom
`agentType` + `schema` genuinely loads the target org_node's migrated
persona (confirmed via a real call asking the agent to self-identify) and
returns real `structured_output` -- see 05 §5.4.1.5.
"""
import json
import time
import uuid
from src import work_graph_execute, work_node_run
def topological_order(graph_manifest):
"""Kahn's algorithm. Dependencies are read from BOTH `node['depends_on']`
and `graph_manifest['edges']` and merged -- `plan_work_graph()`'s own
output isn't schema-validated (it can't be, `--json-schema` doesn't
work with `--agent`, see work_graph_plan.py), so a real LLM response
might populate one and not the other; trusting only one would silently
mis-order nodes rather than raise. Raises ValueError on a cycle or a
dependency referencing an unknown node_id.
"""
nodes = {n["node_id"]: n for n in graph_manifest["nodes"]}
deps = {node_id: set(n.get("depends_on") or []) for node_id, n in nodes.items()}
for edge in graph_manifest.get("edges", []):
to_id, from_id = edge.get("to"), edge.get("from")
if to_id in deps and from_id in nodes:
deps[to_id].add(from_id)
in_degree = {node_id: len(d) for node_id, d in deps.items()}
dependents = {node_id: [] for node_id in nodes}
for node_id, d in deps.items():
for dep in d:
if dep in dependents:
dependents[dep].append(node_id)
ready = sorted(node_id for node_id, degree in in_degree.items() if degree == 0)
ordered = []
while ready:
node_id = ready.pop(0)
ordered.append(nodes[node_id])
for dependent in sorted(dependents[node_id]):
in_degree[dependent] -= 1
if in_degree[dependent] == 0:
ready.append(dependent)
ready.sort()
if len(ordered) != len(nodes):
raise ValueError(
"graph_manifest has a dependency cycle or a depends_on/edge references an unknown node_id"
)
return ordered
def ensure_placeholder_skill(conn, skill_id, owner_org_node):
"""No Skill content has actually been migrated from the Hermes profiles
yet (skill_update.py's own documented gap) -- work_node_execution.
skill_ref is a required FK, so a real row has to exist regardless.
Idempotent."""
exists = conn.execute("SELECT 1 FROM skill WHERE skill_id = ?", (skill_id,)).fetchone()
if exists is None:
conn.execute(
"INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
"VALUES (?, ?, ?, 'unknown', 'new', 'active')",
(skill_id, owner_org_node, "skills/" + skill_id + "/"),
)
conn.commit()
def _ensure_node_row(conn, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, input_payload):
existing = conn.execute(
"SELECT node_exec_id FROM work_node_execution WHERE run_id = ? AND node_id = ?", (run_id, node_id)
).fetchone()
if existing:
return existing[0]
node_exec_id = str(uuid.uuid4())
conn.execute(
"INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, "
"input_schema_ref, output_schema_ref, input_payload, status) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending')",
(node_exec_id, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, json.dumps(input_payload)),
)
conn.commit()
return node_exec_id
def _render_node_script(script_name, org_node, prompt, output_schema, model):
model_opt = ", model: " + json.dumps(model) if model else ""
return (
"export const meta = {{ name: {name}, description: 'graphlemon Work Node ({org_node})' }}\n\n"
"const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema}{model_opt} }})\n\n"
"return result\n"
).format(
name=json.dumps(script_name), org_node=org_node, org_node_json=json.dumps(org_node),
prompt=json.dumps(prompt), schema=json.dumps(output_schema), model_opt=model_opt,
)
def _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout):
script_name = "graphlemon-node-" + node_exec_id.replace("-", "")
def run_fn(cwd, model=None):
script_content = _render_node_script(script_name, org_node, prompt, output_schema, model)
work_graph_execute.write_workflow_script(cwd, script_name, script_content)
try:
return work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout)
except work_graph_execute.UsageLimitReachedError as e:
work_node_run.raise_usage_limit_signal(e.quota_limit_type)
# ClassifierBlockedError / generic WorkGraphExecuteError propagate
# as-is -- run_work_node's own `except Exception` clause marks the
# node 'blocked' and re-raises, no special handling needed here.
return run_fn
def run_graph(
conn, run_id, org_node, skill_id, target_folder_path, graph_manifest,
build_prompt, input_schema, output_schema, timeout=600, needs_worktree=False,
sleep=time.sleep, now=time.time,
):
"""Runs every node in `graph_manifest` (topologically ordered) through
`run_work_node()`, each backed by a real one-node dynamic-workflow
`agent()` call using `org_node`'s migrated persona.
`build_prompt(node, outputs)` is called right before running that
node -- `outputs` contains every already-completed node's
output_payload so far (topological order guarantees a node's
dependencies are always already present), so prompts needing upstream
content (e.g. "here is the approved BRD...") can be built just-in-time
instead of requiring the whole graph's prompts precomputed before
anything runs. This also makes resume-after-restart work correctly:
`run_work_node` short-circuits already-`done` nodes and returns their
persisted output, which lands in `outputs` the same as a freshly-run
one would.
`needs_worktree=True` (03-C §3.6: "并行写代码的节点必须先申请
git_worktree_ref") isolates each node's actual file changes in its own
git worktree, merged back only once the node succeeds -- callers
writing code (Coder) should pass this; callers only producing document
content (Planner) don't need it.
Returns {node_id: output_payload}.
"""
ensure_placeholder_skill(conn, skill_id, org_node)
ordered = topological_order(graph_manifest)
outputs = {}
for node in ordered:
node_id = node["node_id"]
node_exec_id = _ensure_node_row(
conn, run_id, node_id, org_node, skill_id,
"work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
{"node_id": node_id},
)
prompt = build_prompt(node, outputs)
run_fn = _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout)
output = work_node_run.run_work_node(
conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn,
needs_worktree=needs_worktree, repo_path=target_folder_path,
sleep=sleep, now=now,
)
outputs[node_id] = output
return outputs