From b7324fcb670831962235f8c03f079f3e7d71f24d Mon Sep 17 00:00:00 2001 From: bboonnee Date: Fri, 31 Jul 2026 11:22:18 +0800 Subject: [PATCH] U-006: c_work_graph_execute dynamic-workflow wrapper (skeleton) --- src/test_work_graph_execute.py | 86 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/test_work_graph_execute_e2e.py | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/work_graph_execute.py | 67 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 205 insertions(+), 0 deletions(-) create mode 100644 src/test_work_graph_execute.py create mode 100644 src/test_work_graph_execute_e2e.py create mode 100644 src/work_graph_execute.py diff --git a/src/test_work_graph_execute.py b/src/test_work_graph_execute.py new file mode 100644 index 0000000..6f17e06 --- /dev/null +++ b/src/test_work_graph_execute.py @@ -0,0 +1,86 @@ +"""U-006 tests: c_work_graph_execute's dynamic-workflow wrapper, mocked at the +subprocess boundary (CLAUDE.md Testing constraint). Real end-to-end coverage +is in src/test_work_graph_execute_e2e.py (RUN_REAL_AGENT_E2E=1), reusing the +exact spike script the A-1 assumption was validated against. + +Run: python3 -m unittest src.test_work_graph_execute -v +""" + +import json +import os +import subprocess +import tempfile +import unittest +from unittest.mock import patch + +from src import work_graph_execute + + +def _completed_process(returncode=0, stdout_obj=None, stderr=""): + stdout = json.dumps(stdout_obj) if stdout_obj is not None else "" + return subprocess.CompletedProcess(args=["claude"], returncode=returncode, stdout=stdout, stderr=stderr) + + +class WriteWorkflowScriptTest(unittest.TestCase): + def test_writes_into_target_claude_workflows_not_graphlemon_repo(self): + with tempfile.TemporaryDirectory() as tmp: + path = work_graph_execute.write_workflow_script(tmp, "run-123", "export const meta = {}") + self.assertEqual(path, os.path.join(tmp, ".claude", "workflows", "run-123.js")) + with open(path, encoding="utf-8") as f: + self.assertEqual(f.read(), "export const meta = {}") + + +class ExecuteWorkGraphTest(unittest.TestCase): + SCHEMA = {"type": "object", "required": ["sum"], "properties": {"sum": {"type": "number"}}} + + def test_happy_path_returns_structured_output(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "structured_output": {"sum": 29}} + ) + output = work_graph_execute.execute_work_graph( + "/tmp/some-project", "run-123", {"numbers": [2, 3, 4]}, self.SCHEMA + ) + self.assertEqual(output, {"sum": 29}) + + def test_invokes_claude_with_expected_args_and_cwd(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False, "structured_output": {"sum": 1}}) + work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {"n": 1}, self.SCHEMA) + + args, kwargs = run.call_args + argv = args[0] + self.assertEqual(argv[0], "claude") + self.assertIn("-p", argv) + self.assertIn("Run /run-123 with args {\"n\": 1}", argv) + self.assertIn("--json-schema", argv) + self.assertIn(json.dumps(self.SCHEMA), argv) + self.assertIn("--permission-mode", argv) + self.assertIn("auto", argv) + self.assertNotIn("--bare", argv) + self.assertNotIn("--dangerously-skip-permissions", argv) + self.assertNotIn("bypassPermissions", argv) + self.assertEqual(kwargs["cwd"], "/tmp/some-project") + self.assertEqual(kwargs["stdin"], subprocess.DEVNULL) + + def test_nonzero_exit_raises(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process(returncode=1, stdout_obj=None, stderr="boom") + with self.assertRaises(work_graph_execute.WorkGraphExecuteError): + work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA) + + def test_is_error_true_raises(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": True}) + with self.assertRaises(work_graph_execute.WorkGraphExecuteError): + work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA) + + def test_missing_structured_output_raises(self): + with patch("src.work_graph_execute.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False}) + with self.assertRaises(work_graph_execute.WorkGraphExecuteError): + work_graph_execute.execute_work_graph("/tmp/some-project", "run-123", {}, self.SCHEMA) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/test_work_graph_execute_e2e.py b/src/test_work_graph_execute_e2e.py new file mode 100644 index 0000000..acabf64 --- /dev/null +++ b/src/test_work_graph_execute_e2e.py @@ -0,0 +1,52 @@ +"""U-006's real-call coverage: reuses the exact script the A-1 spike validated +(spikes/a1-dynamic-workflows/workflow-under-test.js) through the new +write_workflow_script/execute_work_graph wrapper, not a re-typed copy -- +"直接复用...已验证的代码路径" per development-plan.md U-006. Gated behind +RUN_REAL_AGENT_E2E=1 (CLAUDE.md Testing constraint). + +Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_work_graph_execute_e2e -v +""" + +import os +import subprocess +import tempfile +import unittest + +from src import work_graph_execute +from src.test_support import skip_unless_env + +_SPIKE_SCRIPT = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "spikes", "a1-dynamic-workflows", "workflow-under-test.js", +) +_SCHEMA = { + "type": "object", + "required": ["input_numbers", "squares", "sum"], + "properties": { + "input_numbers": {"type": "array"}, + "squares": {"type": "array"}, + "sum": {"type": "number"}, + }, +} + + +@skip_unless_env("RUN_REAL_AGENT_E2E") +class RealWorkGraphExecuteTest(unittest.TestCase): + def test_reused_a1_spike_script_runs_through_the_new_wrapper(self): + with tempfile.TemporaryDirectory() as tmp: + subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True) + with open(_SPIKE_SCRIPT, encoding="utf-8") as f: + script_content = f.read() + + work_graph_execute.write_workflow_script(tmp, "graphlemon-a1-spike", script_content) + output = work_graph_execute.execute_work_graph( + tmp, "graphlemon-a1-spike", {"numbers": [2, 3, 4]}, _SCHEMA, timeout=120 + ) + + self.assertEqual(output["input_numbers"], [2, 3, 4]) + self.assertEqual(output["squares"], [4, 9, 16]) + self.assertEqual(output["sum"], 29) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/work_graph_execute.py b/src/work_graph_execute.py new file mode 100644 index 0000000..fa983a7 --- /dev/null +++ b/src/work_graph_execute.py @@ -0,0 +1,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/.js` and invoke it via `claude -p 'Run / 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 `/.claude/workflows/.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/.js` + (write_workflow_script) via `claude -p 'Run / 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 -- libgit2 0.21.4