diff --git a/src/org_node_trigger.py b/src/org_node_trigger.py new file mode 100644 index 0000000..edce431 --- /dev/null +++ b/src/org_node_trigger.py @@ -0,0 +1,54 @@ +"""c_org_node_trigger (03-C §3.3): subprocess wrapper skeleton. + +U-005 scope: shell out `claude --agent -p ...` and return the text +response -- "骨架，不含完整状态机" per development-plan.md. No task-lifecycle +state machine yet: TASK_NOT_READY/CONCURRENCY_LIMIT_REACHED/USAGE_LIMIT_REACHED +handling is U-010's job, not this one. + +Identity comes entirely from `.claude/agents/.md` in the target +project (see src/agent_profile.py) -- this module only assembles the prompt +and parses the response, no role-isolation logic of its own (03-C §3.3). +""" + +import json +import subprocess + +from src.agent_profile import ORG_NODES + + +class OrgNodeTriggerError(Exception): + """Raised when the claude subprocess exits non-zero or reports is_error.""" + + +def trigger_org_node(org_node, prompt, target_folder_path, timeout=300): + """Runs `claude --agent -p ` with cwd=target_folder_path. + + `.claude/agents/.md` must already be deployed in target_folder_path + (agent_profile.deploy_agent_definitions) -- this function does not deploy it. + Returns the response text (the JSON `result` field). + """ + if org_node not in ORG_NODES: + raise ValueError(org_node) + + 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 OrgNodeTriggerError( + "claude --agent {} exited {}: {}".format(org_node, result.returncode, result.stderr) + ) + + payload = json.loads(result.stdout) + if payload.get("is_error"): + raise OrgNodeTriggerError("claude --agent {} reported is_error: {}".format(org_node, payload)) + return payload["result"] diff --git a/src/test_agent_e2e.py b/src/test_agent_e2e.py index c6ab634..2dc0fd5 100644 --- a/src/test_agent_e2e.py +++ b/src/test_agent_e2e.py @@ -1,17 +1,18 @@ -"""U-004's actual A-2 validation: a REAL `claude --agent planner -p` call against -a deployed target project. Gated behind RUN_REAL_AGENT_E2E=1 (CLAUDE.md Testing +"""A-2's actual validation (U-004) + U-005's real-call coverage: a REAL +`trigger_org_node("planner", ...)` call against a deployed target project, +using the same production code path U-005 built rather than a one-off +subprocess call. Gated behind RUN_REAL_AGENT_E2E=1 (CLAUDE.md Testing constraint) -- costs real API usage, skipped by default in normal `python3 -m unittest` runs, matching graphloop's RUN_REAL_AGENT_E2E precedent. Run: RUN_REAL_AGENT_E2E=1 python3 -m unittest src.test_agent_e2e -v """ -import json import subprocess import tempfile import unittest -from src import agent_profile, project_scaffold, schema +from src import agent_profile, org_node_trigger, project_scaffold, schema from src.test_support import skip_unless_env @@ -25,24 +26,14 @@ class RealPlannerInvocationTest(unittest.TestCase): project_scaffold.create_project(conn, tmp, init_git_if_missing=False) agent_profile.deploy_agent_definitions(tmp) - result = subprocess.run( - [ - "claude", "--agent", "planner", - "-p", "In one short sentence: what is your job, and what do you never do? Answer only, take no other action.", - "--output-format", "json", - "--permission-mode", "auto", - ], - cwd=tmp, - stdin=subprocess.DEVNULL, - capture_output=True, - text=True, + answer = org_node_trigger.trigger_org_node( + "planner", + "In one short sentence: what is your job, and what do you never do? " + "Answer only, take no other action.", + tmp, timeout=120, ) - self.assertEqual(result.returncode, 0, msg=result.stderr) - payload = json.loads(result.stdout) - self.assertFalse(payload.get("is_error"), msg=payload) - answer = payload["result"] self.assertTrue(answer, "planner returned an empty result") # A-2's actual question: did the identity come through, not the default # assistant persona? "不写业务代码" / "不实现" is the Planner's own red line. diff --git a/src/test_org_node_trigger.py b/src/test_org_node_trigger.py new file mode 100644 index 0000000..52a37e1 --- /dev/null +++ b/src/test_org_node_trigger.py @@ -0,0 +1,66 @@ +"""U-005 tests: c_org_node_trigger's subprocess wrapper, mocked at the +subprocess boundary (CLAUDE.md Testing constraint). Real end-to-end coverage +is in src/test_agent_e2e.py (RUN_REAL_AGENT_E2E=1). + +Run: python3 -m unittest src.test_org_node_trigger -v +""" + +import json +import subprocess +import unittest +from unittest.mock import patch + +from src import org_node_trigger + + +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 TriggerOrgNodeTest(unittest.TestCase): + def test_rejects_unknown_org_node_without_shelling_out(self): + with patch("src.org_node_trigger.subprocess.run") as run: + with self.assertRaises(ValueError): + org_node_trigger.trigger_org_node("not-a-real-node", "hi", "/tmp/some-project") + run.assert_not_called() + + def test_happy_path_returns_result_text(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False, "result": "hello from planner"}) + answer = org_node_trigger.trigger_org_node("planner", "say hi", "/tmp/some-project") + self.assertEqual(answer, "hello from planner") + + def test_invokes_claude_with_expected_args_and_cwd(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False, "result": "ok"}) + org_node_trigger.trigger_org_node("coder", "do the thing", "/tmp/some-project") + + args, kwargs = run.call_args + argv = args[0] + self.assertEqual(argv[:3], ["claude", "--agent", "coder"]) + self.assertIn("-p", argv) + self.assertIn("do the thing", 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.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(returncode=1, stdout_obj=None, stderr="boom") + with self.assertRaises(org_node_trigger.OrgNodeTriggerError): + org_node_trigger.trigger_org_node("planner", "hi", "/tmp/some-project") + + def test_is_error_true_raises(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": True, "result": ""}) + with self.assertRaises(org_node_trigger.OrgNodeTriggerError): + org_node_trigger.trigger_org_node("validator", "hi", "/tmp/some-project") + + +if __name__ == "__main__": + unittest.main()