work_graph_execute.py
2.6 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
"""c_work_graph_execute (03-C §3.5): dynamic-workflow invocation skeleton.
U-006 scope: write an orchestration script into the target project's
`.claude/workflows/<name>.js` and invoke it via `claude -p 'Run /<name> with
args {...}'`, reusing the exact code path the A-1 spike validated
(spikes/a1-dynamic-workflows/, confirmed 2026-07-29). No budget-usage
bookkeeping or CLASSIFIER_BLOCKED handling yet -- that's U-012's job
("骨架" here, per development-plan.md).
"""
import json
import os
import subprocess
class WorkGraphExecuteError(Exception):
"""Raised when the claude subprocess exits non-zero, reports is_error, or
produces no structured_output."""
def write_workflow_script(target_folder_path, name, script_content):
"""Writes `<target>/.claude/workflows/<name>.js` -- versioned with the
target project's own git history (03-C §3.5 step 1), not graphlemon's."""
workflows_dir = os.path.join(target_folder_path, ".claude", "workflows")
os.makedirs(workflows_dir, exist_ok=True)
path = os.path.join(workflows_dir, name + ".js")
with open(path, "w", encoding="utf-8") as f:
f.write(script_content)
return path
def execute_work_graph(target_folder_path, name, args, output_schema, timeout=600):
"""Runs the workflow already written to `.claude/workflows/<name>.js`
(write_workflow_script) via `claude -p 'Run /<name> with args {...}'`.
No `--bare` (A-1 spike: skips credential/login discovery, unrelated to
workflow presence). `--permission-mode auto`, never `bypassPermissions`/
`--dangerously-skip-permissions` (05 §5.3.1). Returns the parsed
`structured_output` dict.
"""
prompt = "Run /{} with args {}".format(name, json.dumps(args))
result = subprocess.run(
[
"claude", "-p", prompt,
"--output-format", "json",
"--json-schema", json.dumps(output_schema),
"--permission-mode", "auto",
],
cwd=target_folder_path,
stdin=subprocess.DEVNULL,
capture_output=True,
text=True,
timeout=timeout,
)
if result.returncode != 0:
raise WorkGraphExecuteError(
"workflow {} exited {}: {}".format(name, result.returncode, result.stderr)
)
payload = json.loads(result.stdout)
if payload.get("is_error"):
raise WorkGraphExecuteError("workflow {} reported is_error: {}".format(name, payload))
structured_output = payload.get("structured_output")
if structured_output is None:
raise WorkGraphExecuteError("workflow {} produced no structured_output: {}".format(name, payload))
return structured_output