U-005: c_org_node_trigger subprocess wrapper (skeleton)

trigger_org_node(org_node, prompt, target_folder_path): shells out `claude --agent <org_node> -p ... --permission-mode auto` with cwd=target project, parses the JSON result, raises OrgNodeTriggerError on nonzero exit or is_error. Skeleton only, per development-plan.md ("不含完整状态机") -- TASK_NOT_READY/CONCURRENCY_LIMIT_REACHED/USAGE_LIMIT_REACHED handling is U-010's job. Also refactors test_agent_e2e.py (U-004) to call this real function instead of its own ad-hoc subprocess invocation, so the gated real-agent test now exercises U-005's actual production code path -- confirmed still passing against a real `claude --agent planner` call. 5 new mocked stdlib unittest cases (CLAUDE.md Testing constraint: mock the subprocess boundary in normal tests). 46 total across U-001-005, all passing; the 1 gated real-agent test re-verified manually. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-005: c_org_node_trigger subprocess wrapper (skeleton)
trigger_org_node(org_node, prompt, target_folder_path): shells out `claude --agent <org_node> -p ... --permission-mode auto` with cwd=target project, parses the JSON result, raises OrgNodeTriggerError on nonzero exit or is_error. Skeleton only, per development-plan.md ("不含完整状态机") -- TASK_NOT_READY/CONCURRENCY_LIMIT_REACHED/USAGE_LIMIT_REACHED handling is U-010's job. Also refactors test_agent_e2e.py (U-004) to call this real function instead of its own ad-hoc subprocess invocation, so the gated real-agent test now exercises U-005's actual production code path -- confirmed still passing against a real `claude --agent planner` call. 5 new mocked stdlib unittest cases (CLAUDE.md Testing constraint: mock the subprocess boundary in normal tests). 46 total across U-001-005, all passing; the 1 gated real-agent test re-verified manually. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 2a71bd56
Showing 3 changed files with 130 additions and 19 deletions
src/org_node_trigger.py
src/test_agent_e2e.py
src/test_org_node_trigger.py
+"""c_org_node_trigger (03-C §3.3): subprocess wrapper skeleton.
+
+U-005 scope: shell out `claude --agent <org_node> -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/<org_node>.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 <org_node> -p <prompt>` with cwd=target_folder_path.
+
+    `.claude/agents/<org_node>.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"]
-"""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.
+"""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()