diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index af95d26..d58ef16 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -53,7 +53,7 @@ import uuid from src import ( artifact_write, org_node_lifecycle, org_node_trigger, paths, task_create, - work_graph_execute, work_graph_orchestrate, work_graph_plan, work_node_run, + work_graph_orchestrate, work_graph_plan, work_node_run, ) _READY_PROMPT = "Confirm you're ready to begin. Take no other action yet." @@ -378,16 +378,9 @@ def _check_prd_consistency(target_folder_path, prd_content, timeout=180): worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None cwd = worktree_path or target_folder_path try: - script_name = "graphlemon-prd-consistency-" + uuid.uuid4().hex prompt = _PRD_CONSISTENCY_PROMPT.format(prd_content=prd_content) - script_content = ( - "export const meta = {{ name: {name}, description: 'graphlemon PRD consistency check' }}\n\n" - "const result = await agent({prompt}, {{ agentType: 'planner', schema: {schema} }})\n\n" - "return result\n" - ).format(name=json.dumps(script_name), prompt=json.dumps(prompt), schema=json.dumps(_PRD_CONSISTENCY_SCHEMA)) - work_graph_execute.write_workflow_script(cwd, script_name, script_content) - result = work_graph_execute.execute_work_graph( - cwd, script_name, {}, _PRD_CONSISTENCY_SCHEMA, timeout=timeout + result = org_node_trigger.trigger_org_node_structured( + "planner", prompt, cwd, _PRD_CONSISTENCY_SCHEMA, timeout=timeout ) if worktree_path: work_node_run.ensure_worktree_committed(worktree_path, node_exec_id) diff --git a/src/org_node_trigger.py b/src/org_node_trigger.py index 786d0ba..e8d5d38 100644 --- a/src/org_node_trigger.py +++ b/src/org_node_trigger.py @@ -8,18 +8,155 @@ 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). + +**`trigger_org_node_structured` (2026-08, work-graph-node-execution- +deblackbox-requirements.md)**: a sibling of `trigger_org_node` above, not a +replacement -- same flat `claude --agent` call, but also extracts a JSON +payload matching a caller-supplied output_schema from the free-text +response (05 §5.4.1.1's documented workaround: `--json-schema` silently +drops `structured_output` when combined with `--agent`, so ask for a single +```json fenced block in the prompt instead and parse it out, exactly the +convention `work_graph_plan.py::plan_work_graph()` already validated in +production). This is what closes the three remaining call sites that used +to render a one-off Dynamic Workflow `.js` script and run it via +`work_graph_execute.execute_work_graph()` purely to get "agent identity + +schema-shaped output" -- `plan_work_graph()` proves flat calls already do +both, so the Workflow indirection was never load-bearing for that need. """ import json import subprocess from src.agent_profile import ORG_NODES +from src.work_graph_plan import extract_json_payload class OrgNodeTriggerError(Exception): """Raised when the claude subprocess exits non-zero or reports is_error.""" +class StructuredOutputExtractionError(OrgNodeTriggerError): + """trigger_org_node_structured got a well-formed claude response, but + couldn't find any JSON payload in it matching output_schema's top-level + required keys (bare or ```json-fenced) -- see extract_json_payload.""" + + +# REQ-14's happy case cites "You've hit your session limit" as the real +# signal text (session/weekly/opus phrase matching, ported verbatim from +# work_graph_execute.py's own `_quota_limit_type` -- see this module's +# docstring for why trigger_org_node_structured needs THIS granular version +# (with a quota_limit_type of "session"/"weekly"/"opus") rather than +# org_node_lifecycle.py's simpler boolean `_looks_like_usage_limit`: +# work_node_run.raise_usage_limit_signal(quota_limit_type) and +# _make_run_fn's own quota_limit_type-keyed logging both need the specific +# type, not just a yes/no. org_node_lifecycle.py's own detection is a +# different call path (trigger_org_node_full, task-lifecycle transitions, +# untouched by this migration) and is deliberately left alone rather than +# unified here -- merging it would touch that unrelated flow's established, +# tested behavior for no benefit to this migration. work_graph_execute.py +# keeps its own copy too (frozen/historical per this migration's FR-6, not +# worth coupling a soon-to-be-unused module to this one) -- both copies +# implement the identical algorithm, so there is no *inconsistent* duplicate +# detection logic in play, only a frozen historical copy next to the live one. +_USAGE_LIMIT_PHRASES = ("hit your session limit", "hit your weekly limit", "usage limit") + + +def _quota_limit_type(text): + lowered = text.lower() + if "weekly" in lowered and "hit your" in lowered: + return "weekly" + if "opus" in lowered and ("limit" in lowered or "quota" in lowered): + return "opus" + if any(phrase in lowered for phrase in _USAGE_LIMIT_PHRASES): + return "session" + return None + + +class UsageLimitReachedError(OrgNodeTriggerError): + """USAGE_LIMIT_REACHED -- 04-P p_usage_limit_wait. `quota_limit_type` is + "session"/"weekly"/"opus" (see _quota_limit_type above); callers that + already have a downgrade-retry mechanism (work_node_run.py's U-020 + model-downgrade retry, driven via work_node_run.raise_usage_limit_signal) + consume it directly.""" + + def __init__(self, message, quota_limit_type): + super().__init__(message) + self.quota_limit_type = quota_limit_type + + +_JSON_FENCE_INSTRUCTION = ( + "Your final answer MUST be a single JSON object wrapped in exactly one " + "```json fenced code block, matching this schema: {schema}. Do not " + "describe what you did or are about to do -- the fenced block's content " + "IS the answer, not a summary of it." +) + + +def trigger_org_node_structured(org_node, prompt, target_folder_path, output_schema, model=None, timeout=300): + """Like trigger_org_node(), but also extracts a JSON payload matching + output_schema's top-level required keys from the free-text response -- + see this module's docstring. `_JSON_FENCE_INSTRUCTION` is appended to + `prompt` automatically, callers don't repeat this boilerplate themselves. + + `model`: real CLI `--model` override (FR-4) -- empirically verified for + real (RUN_REAL_AGENT_E2E=1, see this module's own test file) that + `claude --agent -p ... --model ` (1) still loads the + target org_node's persona, (2) genuinely runs the requested model (the + JSON response's own `modelUsage` key names it), and (3) the ```json + fence-extraction convention above still works on that real output -- + 05 §5.4.1.5's "CLI top-level args and DSL-internal options are two + different paths, don't assume one from the other" lesson, checked here + rather than assumed. + """ + if org_node not in ORG_NODES: + raise ValueError(org_node) + + args = [ + "claude", "--agent", org_node, + "-p", prompt + "\n\n" + _JSON_FENCE_INSTRUCTION.format(schema=json.dumps(output_schema)), + ] + if model: + args += ["--model", model] + args += ["--output-format", "json", "--permission-mode", "auto"] + + result = subprocess.run( + args, cwd=target_folder_path, stdin=subprocess.DEVNULL, + capture_output=True, text=True, timeout=timeout, + ) + if result.returncode != 0: + quota_type = _quota_limit_type(result.stderr) or _quota_limit_type(result.stdout) + detail = "stderr: {}\nstdout (truncated): {}".format(result.stderr, result.stdout[:4000]) + if quota_type: + raise UsageLimitReachedError( + "trigger_org_node_structured for {} exited {}\n{}".format(org_node, result.returncode, detail), + quota_type, + ) + raise OrgNodeTriggerError( + "trigger_org_node_structured for {} exited {}\n{}".format(org_node, result.returncode, detail) + ) + + payload = json.loads(result.stdout) + if payload.get("is_error"): + quota_type = _quota_limit_type(str(payload.get("result", ""))) + if quota_type: + raise UsageLimitReachedError( + "trigger_org_node_structured for {} hit a usage limit: {}".format(org_node, payload), quota_type + ) + raise OrgNodeTriggerError( + "trigger_org_node_structured for {} reported is_error: {}".format(org_node, payload) + ) + + required_keys = list(output_schema.get("required", [])) + extracted = extract_json_payload(payload.get("result", ""), required_keys=required_keys) + if extracted is None: + raise StructuredOutputExtractionError( + "could not extract a payload matching required keys {} from org_node={}'s response: {!r}".format( + required_keys, org_node, payload.get("result", "")[:2000] + ) + ) + return extracted + + def trigger_org_node(org_node, prompt, target_folder_path, timeout=300): """Runs `claude --agent -p ` with cwd=target_folder_path. diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 7760068..5aa4d1f 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -8,16 +8,18 @@ src/test_e2e_pipeline_e2e.py (RUN_REAL_AGENT_E2E=1). **Rewired 2026-08-01 (U-031 regression audit)**: plan_documents/ implement_code now go through work_graph_plan.plan_work_graph + -work_graph_orchestrate.run_graph (which calls work_graph_execute. -execute_work_graph per node) instead of a flat org_node_trigger. -trigger_org_node call -- see e2e_pipeline.py's module docstring for why. -plan_work_graph is mocked at its own subprocess boundary (same pattern as -test_work_graph_plan.py) so its real graph_manifest-parsing logic and its -real work_graph_run INSERT both still run for real; execute_work_graph is -mocked directly (one level higher than its own subprocess boundary) since -this file's job is testing e2e_pipeline's wiring, not re-proving -execute_work_graph's own mechanics (covered in test_work_graph_execute.py) -or work_graph_orchestrate's own mechanics (test_work_graph_orchestrate.py). +work_graph_orchestrate.run_graph (which calls org_node_trigger. +trigger_org_node_structured per node, 2026-08 deblackbox migration -- +originally work_graph_execute.execute_work_graph, see that module's own +docstring) instead of a flat org_node_trigger.trigger_org_node call -- see +e2e_pipeline.py's module docstring for why. plan_work_graph is mocked at +its own subprocess boundary (same pattern as test_work_graph_plan.py) so +its real graph_manifest-parsing logic and its real work_graph_run INSERT +both still run for real; trigger_org_node_structured is mocked directly +(one level higher than its own subprocess boundary) since this file's job +is testing e2e_pipeline's wiring, not re-proving trigger_org_node_ +structured's own mechanics (covered in test_org_node_trigger.py) or +work_graph_orchestrate's own mechanics (test_work_graph_orchestrate.py). Run: python3 -m unittest src.test_e2e_pipeline -v """ @@ -30,7 +32,7 @@ import tempfile import unittest from unittest.mock import patch -from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute, work_graph_orchestrate +from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_orchestrate _FAKE_DOCS = { "brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake", @@ -303,9 +305,8 @@ class PlanDocumentsTest(PipelineTestBase): self.conn.commit() with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = [ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = [ {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]}, {"consistent": True, "issues": []}, # PRD consistency check (REQ-02 boundary) {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, @@ -374,9 +375,8 @@ class PlanDocumentsTest(PipelineTestBase): return real_subprocess_run(*args, **kwargs) with patch("src.work_graph_plan.subprocess.run", side_effect=_discriminating_fake), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = [ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = [ {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]}, {"consistent": True, "issues": []}, # PRD consistency check (REQ-02 boundary) {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, @@ -397,10 +397,9 @@ class RunEndToEndTest(PipelineTestBase): def test_full_pipeline_wiring(self): with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_ready_trigger), \ patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."): - mock_execute.side_effect = [ + mock_trigger.side_effect = [ {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]}, {"consistent": True, "issues": []}, # PRD consistency check (REQ-02 boundary) {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, @@ -653,11 +652,11 @@ class PrdConsistencyVerifyFnTest(unittest.TestCase): def test_non_prd_node_passes_through_without_a_call(self): node_artifact_types = {"brd": "brd"} verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1") - with patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: ok, feedback = verify_fn({"node_id": "brd"}, {"content": "brd text"}) self.assertTrue(ok) self.assertIsNone(feedback) - mock_execute.assert_not_called() + mock_trigger.assert_not_called() def test_string_false_from_a_real_call_is_not_treated_as_truthy(self): # regression: a real call (2026-08-02) returned "consistent": "false" @@ -665,39 +664,37 @@ class PrdConsistencyVerifyFnTest(unittest.TestCase): # Python, silently inverting the check until a real test caught it. node_artifact_types = {"prd": "prd"} verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1") - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - return_value={"consistent": "false", "issues": ["contradiction found"]}, - ): + with patch.object( + org_node_trigger, "trigger_org_node_structured", + return_value={"consistent": "false", "issues": ["contradiction found"]}, + ): ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"}) self.assertFalse(ok) def test_string_true_from_a_real_call_is_treated_as_consistent(self): node_artifact_types = {"prd": "prd"} verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1") - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - return_value={"consistent": "true", "issues": []}, - ): + with patch.object( + org_node_trigger, "trigger_org_node_structured", + return_value={"consistent": "true", "issues": []}, + ): ok, feedback = verify_fn({"node_id": "prd"}, {"content": "consistent prd"}) self.assertTrue(ok) def test_prd_node_triggers_a_real_check_call(self): node_artifact_types = {"prd": "prd"} verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1") - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - return_value={"consistent": False, "issues": ["contradiction found"]}, - ) as mock_execute: + with patch.object( + org_node_trigger, "trigger_org_node_structured", + return_value={"consistent": False, "issues": ["contradiction found"]}, + ) as mock_trigger: ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"}) self.assertFalse(ok) self.assertEqual(feedback, ["contradiction found"]) - mock_execute.assert_called_once() - args = mock_execute.call_args[0] - self.assertEqual(args[0], "/tmp/proj-1") + mock_trigger.assert_called_once() + args = mock_trigger.call_args[0] + self.assertEqual(args[0], "planner") + self.assertEqual(args[2], "/tmp/proj-1") class CheckScriptCompilesTest(unittest.TestCase): @@ -796,14 +793,13 @@ class CombinedPlannerVerifyFnTest(unittest.TestCase): def test_prd_node_dispatches_to_prd_check(self): node_artifact_types = {"n1": "prd"} verify_fn = e2e_pipeline._combined_planner_verify_fn(node_artifact_types, "/tmp/proj-1") - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - return_value={"consistent": True, "issues": []}, - ) as mock_execute: + with patch.object( + org_node_trigger, "trigger_org_node_structured", + return_value={"consistent": True, "issues": []}, + ) as mock_trigger: ok, detail = verify_fn({"node_id": "n1"}, {"content": "a prd"}) self.assertTrue(ok) - mock_execute.assert_called_once() + mock_trigger.assert_called_once() def test_test_script_node_dispatches_to_compile_check(self): node_artifact_types = {"n1": "test_script"} @@ -839,19 +835,18 @@ class PrdConsistencySelfTargetWorktreeTest(unittest.TestCase): patch.object(e2e_pipeline.work_node_run, "ensure_worktree_committed") as mock_ensure, \ patch.object(e2e_pipeline.work_node_run, "merge_worktree") as mock_merge, \ patch.object(e2e_pipeline.work_node_run, "remove_worktree") as mock_remove, \ - patch.object(work_graph_execute, "write_workflow_script") as mock_write, \ patch.object( - work_graph_execute, "execute_work_graph", + org_node_trigger, "trigger_org_node_structured", return_value={"consistent": True, "issues": []}, - ) as mock_execute: + ) as mock_trigger: ok, feedback = verify_fn({"node_id": "prd"}, {"content": "some prd"}) self.assertTrue(ok) mock_create.assert_called_once_with("/repo", unittest.mock.ANY) - # both the script write and the real execute call must use the - # WORKTREE path as cwd, never the main checkout, when self-targeting - self.assertEqual(mock_write.call_args[0][0], "/repo/.graphlemon-worktrees/x") - self.assertEqual(mock_execute.call_args[0][0], "/repo/.graphlemon-worktrees/x") + # the real trigger call must use the WORKTREE path as cwd (3rd + # positional arg: org_node, prompt, target_folder_path), never the + # main checkout, when self-targeting + self.assertEqual(mock_trigger.call_args[0][2], "/repo/.graphlemon-worktrees/x") # called twice: once for the worktree itself, once for the main # checkout (worktrees share project identity, so a real memory # write can land on the main checkout regardless of cwd -- see @@ -868,16 +863,14 @@ class PrdConsistencySelfTargetWorktreeTest(unittest.TestCase): verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1") with patch.object(e2e_pipeline.paths, "is_self_target", return_value=False), \ patch.object(e2e_pipeline.work_node_run, "create_worktree") as mock_create, \ - patch.object(work_graph_execute, "write_workflow_script") as mock_write, \ patch.object( - work_graph_execute, "execute_work_graph", + org_node_trigger, "trigger_org_node_structured", return_value={"consistent": True, "issues": []}, - ) as mock_execute: + ) as mock_trigger: verify_fn({"node_id": "prd"}, {"content": "some prd"}) mock_create.assert_not_called() - self.assertEqual(mock_write.call_args[0][0], "/tmp/proj-1") - self.assertEqual(mock_execute.call_args[0][0], "/tmp/proj-1") + self.assertEqual(mock_trigger.call_args[0][2], "/tmp/proj-1") class ValidateImplementationTest(unittest.TestCase): @@ -1104,9 +1097,8 @@ class ValidateImplementationWithOwnGraphTest(unittest.TestCase): mock_responses.append({"content": script_content}) mock_responses.append({"content": verdict_content}) with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_validator_plan_subprocess_run(manifest)), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = mock_responses + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = mock_responses return e2e_pipeline.validate_implementation_with_own_graph( self.conn, "task-1", "run-shared", "sdd content", "prd content", "goal text", self.tmp.name, ) diff --git a/src/test_org_level_loop.py b/src/test_org_level_loop.py index 2d54200..a87e436 100644 --- a/src/test_org_level_loop.py +++ b/src/test_org_level_loop.py @@ -6,9 +6,10 @@ Testing constraint). Real coverage is in src/test_org_level_loop_e2e.py **F (2026-08-06)**: root-cause judgment now runs through root_cause_trace.judge_root_cause_via_work_graph (a real Work Graph fan-out, see docs/done/verifier-fan-out-jury-design.md), so the 3 perspective calls -are mocked at work_graph_execute.write_workflow_script/execute_work_graph -(same boundary as test_work_graph_orchestrate.py), not at -org_node_trigger.trigger_org_node -- that mock is now only used for the +are mocked at org_node_trigger.trigger_org_node_structured (2026-08 +deblackbox migration -- originally work_graph_execute.write_workflow_script/ +execute_work_graph, same boundary as test_work_graph_orchestrate.py), not +at org_node_trigger.trigger_org_node -- that mock is now only used for the fixer/verify calls, which are unchanged. Run: python3 -m unittest src.test_org_level_loop -v @@ -17,16 +18,16 @@ Run: python3 -m unittest src.test_org_level_loop -v import unittest from unittest.mock import patch -from src import org_level_loop, root_cause_trace, schema, work_graph_execute +from src import org_level_loop, org_node_trigger, root_cause_trace, schema -def _make_perspective_execute(perspective_category): - """Fakes work_graph_execute.execute_work_graph for the 3 perspective - Work Nodes -- all agree on perspective_category, real schema-shaped - output (no more 'CATEGORY:'/'DETAIL:' text parsing).""" - def fake_execute(cwd, name, args, output_schema, timeout=300): +def _make_perspective_trigger(perspective_category): + """Fakes org_node_trigger.trigger_org_node_structured for the 3 + perspective Work Nodes -- all agree on perspective_category, real + schema-shaped output (no more 'CATEGORY:'/'DETAIL:' text parsing).""" + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): return {"category": perspective_category, "detail": "because reasons"} - return fake_execute + return fake_trigger def _make_trigger(fix_ok=True): @@ -61,11 +62,10 @@ class RunFixLoopTestBase(unittest.TestCase): self.conn.close() def _run_fix_loop(self, perspective_category, fix_ok=True): - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - side_effect=_make_perspective_execute(perspective_category), - ), \ + with patch.object( + org_node_trigger, "trigger_org_node_structured", + side_effect=_make_perspective_trigger(perspective_category), + ), \ patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=fix_ok)): return org_level_loop.run_fix_loop( self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it" @@ -118,11 +118,10 @@ class InconclusiveDisagreementTest(RunFixLoopTestBase): def test_three_way_split_raises_and_creates_no_bug_record(self): categories = iter(["planning", "coder", "skill_defect"]) - def fake_execute(cwd, name, args, output_schema, timeout=300): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): return {"category": next(categories), "detail": "d"} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute), \ + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger), \ patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=True)): with self.assertRaises(root_cause_trace.InconclusiveEscalateError): org_level_loop.run_fix_loop( diff --git a/src/test_org_node_trigger.py b/src/test_org_node_trigger.py index 52a37e1..2ae16e9 100644 --- a/src/test_org_node_trigger.py +++ b/src/test_org_node_trigger.py @@ -62,5 +62,166 @@ class TriggerOrgNodeTest(unittest.TestCase): org_node_trigger.trigger_org_node("validator", "hi", "/tmp/some-project") +_SCHEMA = {"type": "object", "required": ["ok", "issues"], "properties": { + "ok": {"type": "boolean"}, "issues": {"type": "array"}, +}} + + +def _fenced(payload): + return "Here you go:\n\n```json\n" + json.dumps(payload) + "\n```\n\nDone." + + +class TriggerOrgNodeStructuredTest(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_structured( + "not-a-real-node", "hi", "/tmp/some-project", _SCHEMA + ) + run.assert_not_called() + + def test_extracts_bare_json_result(self): + payload = {"ok": True, "issues": []} + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False, "result": json.dumps(payload)}) + extracted = org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + self.assertEqual(extracted, payload) + + def test_extracts_a_single_fenced_json_block_with_prose_around_it(self): + payload = {"ok": False, "issues": ["bug 1"]} + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False, "result": _fenced(payload)}) + extracted = org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + self.assertEqual(extracted, payload) + + def test_multiple_fenced_blocks_returns_first_matching_one(self): + unrelated = "```json\n" + json.dumps({"unrelated": True}) + "\n```\n\n" + payload = {"ok": True, "issues": []} + text = unrelated + _fenced(payload) + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": False, "result": text}) + extracted = org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + self.assertEqual(extracted, payload) + + def test_no_matching_json_anywhere_raises_extraction_error(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": "sorry, I can't produce that"} + ) + with self.assertRaises(org_node_trigger.StructuredOutputExtractionError): + org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + + def test_appends_json_fence_instruction_and_schema_to_prompt(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": json.dumps({"ok": True, "issues": []})} + ) + org_node_trigger.trigger_org_node_structured("validator", "check this", "/tmp/some-project", _SCHEMA) + + argv = run.call_args[0][0] + prompt_arg = argv[argv.index("-p") + 1] + self.assertIn("check this", prompt_arg) + self.assertIn("```json fenced code block", prompt_arg) + self.assertIn(json.dumps(_SCHEMA), prompt_arg) + + def test_model_override_passed_through_as_cli_flag(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": json.dumps({"ok": True, "issues": []})} + ) + org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA, model="claude-fable-5", + ) + + argv = run.call_args[0][0] + self.assertIn("--model", argv) + self.assertEqual(argv[argv.index("--model") + 1], "claude-fable-5") + + def test_no_model_means_no_model_flag(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": json.dumps({"ok": True, "issues": []})} + ) + org_node_trigger.trigger_org_node_structured("validator", "check this", "/tmp/some-project", _SCHEMA) + + argv = run.call_args[0][0] + self.assertNotIn("--model", argv) + + def test_nonzero_exit_with_usage_limit_phrase_raises_usage_limit_error_with_type(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + returncode=1, stdout_obj=None, stderr="Claude AI usage limit reached, you've hit your session limit" + ) + with self.assertRaises(org_node_trigger.UsageLimitReachedError) as ctx: + org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + self.assertEqual(ctx.exception.quota_limit_type, "session") + + def test_nonzero_exit_without_usage_limit_phrase_raises_generic_error(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_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + + def test_is_error_with_weekly_limit_phrase_raises_usage_limit_error(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": True, "result": "You've hit your weekly limit"} + ) + with self.assertRaises(org_node_trigger.UsageLimitReachedError) as ctx: + org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + self.assertEqual(ctx.exception.quota_limit_type, "weekly") + + def test_is_error_with_opus_quota_phrase_raises_usage_limit_error(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": True, "result": "You have reached your opus quota limit for this period"} + ) + with self.assertRaises(org_node_trigger.UsageLimitReachedError) as ctx: + org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + self.assertEqual(ctx.exception.quota_limit_type, "opus") + + def test_is_error_without_usage_limit_phrase_raises_generic_error(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process(stdout_obj={"is_error": True, "result": "something else failed"}) + with self.assertRaises(org_node_trigger.OrgNodeTriggerError): + org_node_trigger.trigger_org_node_structured( + "validator", "check this", "/tmp/some-project", _SCHEMA + ) + + def test_invokes_claude_with_expected_cwd_and_stdin(self): + with patch("src.org_node_trigger.subprocess.run") as run: + run.return_value = _completed_process( + stdout_obj={"is_error": False, "result": json.dumps({"ok": True, "issues": []})} + ) + org_node_trigger.trigger_org_node_structured("coder", "do the thing", "/tmp/some-project", _SCHEMA) + + args, kwargs = run.call_args + argv = args[0] + self.assertEqual(argv[:3], ["claude", "--agent", "coder"]) + 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) + + if __name__ == "__main__": unittest.main() diff --git a/src/test_root_cause_trace.py b/src/test_root_cause_trace.py index 6b5e16b..4f56cb5 100644 --- a/src/test_root_cause_trace.py +++ b/src/test_root_cause_trace.py @@ -14,7 +14,7 @@ import time import unittest from unittest.mock import patch -from src import bug_record, root_cause_trace, schema, work_graph_execute +from src import bug_record, org_node_trigger, root_cause_trace, schema def _fixed(category, detail="detail"): @@ -169,12 +169,11 @@ class JudgeRootCauseViaWorkGraphTestBase(unittest.TestCase): def tearDown(self): self.conn.close() - def _perspective_for_script(self, prompt_by_script, script_name): - content = prompt_by_script[script_name] + def _perspective_for_prompt(self, prompt): for name, framing in root_cause_trace._PERSPECTIVE_FRAMINGS.items(): - if framing in content: + if framing in prompt: return name - raise AssertionError("could not identify perspective for script " + script_name) + raise AssertionError("could not identify perspective for prompt " + prompt) class RealWorkNodesPerPerspectiveTest(JudgeRootCauseViaWorkGraphTestBase): @@ -183,17 +182,11 @@ class RealWorkNodesPerPerspectiveTest(JudgeRootCauseViaWorkGraphTestBase): calls inside one function.""" def test_three_perspectives_are_three_real_work_node_rows(self): - prompt_by_script = {} - - def fake_write_script(cwd, script_name, script_content): - prompt_by_script[script_name] = script_content - - def fake_execute(cwd, name, args, output_schema, timeout=600): - perspective = self._perspective_for_script(prompt_by_script, name) + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + perspective = self._perspective_for_prompt(prompt) return {"category": "coder", "detail": "agreed via " + perspective} - with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): result = root_cause_trace.judge_root_cause_via_work_graph( self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", ) @@ -221,17 +214,11 @@ class JudgmentRuleEquivalenceTest(JudgeRootCauseViaWorkGraphTestBase): same shared _tally_perspectives.""" def _run_with_categories(self, categories_by_perspective): - prompt_by_script = {} - - def fake_write_script(cwd, script_name, script_content): - prompt_by_script[script_name] = script_content - - def fake_execute(cwd, name, args, output_schema, timeout=600): - perspective = self._perspective_for_script(prompt_by_script, name) + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + perspective = self._perspective_for_prompt(prompt) return {"category": categories_by_perspective[perspective], "detail": "d-" + perspective} - with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): return root_cause_trace.judge_root_cause_via_work_graph( self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", ) @@ -261,11 +248,10 @@ class IndependentCleanContextTest(JudgeRootCauseViaWorkGraphTestBase): carries what it needs to judge -- not other perspectives' reasoning.""" def test_input_payload_has_no_cross_perspective_content(self): - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - return_value={"category": "coder", "detail": "d"}, - ): + with patch.object( + org_node_trigger, "trigger_org_node_structured", + return_value={"category": "coder", "detail": "d"}, + ): root_cause_trace.judge_root_cause_via_work_graph( self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", ) @@ -286,7 +272,7 @@ class ConcurrentPerspectiveExecutionTest(JudgeRootCauseViaWorkGraphTestBase): events = [] events_lock = threading.Lock() - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): start = time.monotonic() time.sleep(0.05) end = time.monotonic() @@ -294,8 +280,7 @@ class ConcurrentPerspectiveExecutionTest(JudgeRootCauseViaWorkGraphTestBase): events.append((start, end)) return {"category": "coder", "detail": "d"} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): root_cause_trace.judge_root_cause_via_work_graph( self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", ) diff --git a/src/test_scheduler.py b/src/test_scheduler.py index d376752..03c0264 100644 --- a/src/test_scheduler.py +++ b/src/test_scheduler.py @@ -19,8 +19,8 @@ import unittest from unittest.mock import patch from src import ( - bug_record, e2e_pipeline, org_level_loop, org_loop_route, paths, schema, scheduler, settings_config, - work_graph_execute, work_node_run, + bug_record, e2e_pipeline, org_level_loop, org_loop_route, org_node_trigger, paths, schema, scheduler, + settings_config, work_graph_execute, work_node_run, ) @@ -417,9 +417,9 @@ def _fix_loop_trigger(org_node, prompt, target_folder_path, timeout=300): raise AssertionError("unexpected call: " + org_node) -def _fix_loop_perspective_execute(cwd, name, args, output_schema, timeout=300): - """work_graph_execute.execute_work_graph fake for run_fix_loop's 3 - root-cause perspective Work Nodes (F) -- all agree 'coder'.""" +def _fix_loop_perspective_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + """org_node_trigger.trigger_org_node_structured fake for run_fix_loop's + 3 root-cause perspective Work Nodes (F) -- all agree 'coder'.""" return {"category": "coder", "detail": "because reasons"} @@ -466,8 +466,7 @@ class SchedulerContinuesPastBlockedTaskTest(unittest.TestCase): with patch.object(scheduler.e2e_pipeline, "run_task_lifecycle", side_effect=fake_run_task_lifecycle), \ patch.object(scheduler.e2e_pipeline, "_read_artifact_content", return_value="benchmark content"), \ patch("src.org_node_trigger.trigger_org_node", side_effect=_fix_loop_trigger), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=_fix_loop_perspective_execute): + patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=_fix_loop_perspective_trigger): results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=3) by_goal = {r["goal_text"]: r for r in results} diff --git a/src/test_task_execution.py b/src/test_task_execution.py index 92e0cb0..50aa739 100644 --- a/src/test_task_execution.py +++ b/src/test_task_execution.py @@ -16,7 +16,7 @@ from unittest.mock import patch from src import ( e2e_pipeline, org_node_lifecycle, org_node_trigger, project_scaffold, schema, scheduler, - task_execution, work_graph_execute, work_graph_orchestrate, work_graph_plan, + task_execution, work_graph_orchestrate, work_graph_plan, ) @@ -185,8 +185,7 @@ class ResumeFromCodingTest(TaskExecutionTestBase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: # validator's self-planned final_verdict node (Phase 2c) mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") @@ -218,8 +217,7 @@ class ResumeFromPlanningTest(TaskExecutionTestBase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \ patch.object(e2e_pipeline, "plan_documents") as mock_plan: mock_execute.side_effect = [ @@ -263,8 +261,7 @@ class ResumeFromReviewingTest(TaskExecutionTestBase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=tracking_ready_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS") as mock_trigger, \ patch.object(e2e_pipeline, "plan_documents") as mock_plan: mock_execute.side_effect = [ @@ -313,8 +310,7 @@ class ResumeFromPlanningWithZeroArtifactsTest(TaskExecutionTestBase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute, \ patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS"), \ patch.object(e2e_pipeline, "plan_documents", side_effect=fake_plan_documents) as mock_plan: mock_execute.side_effect = [ @@ -349,8 +345,7 @@ class ResumeFromValidatingTest(TaskExecutionTestBase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger) as mock_ready, \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node result = task_execution.run_or_resume_task_lifecycle(self.conn, "task-1", self.tmp.name, "a goal") @@ -567,8 +562,7 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", @@ -592,8 +586,8 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): # a second thread that would run the same task twice. # # Fully mocks the same real work the OTHER StartTaskExecutionIn - # BackgroundTest tests mock (write_workflow_script/execute_work_graph/ - # org_node_trigger.trigger_org_node), not just the readiness trigger + # BackgroundTest tests mock (org_node_trigger.trigger_org_node_ + # structured/trigger_org_node), not just the readiness trigger # this test blocks on -- an earlier version of this test only # patched the readiness trigger, so once unblocked the thread went on # to attempt REAL subprocess/file work, outlived this test's own @@ -618,8 +612,7 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=blocking_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + patch.object(org_node_trigger, "trigger_org_node_structured") as mock_execute: mock_execute.side_effect = [{"content": "VERDICT: PASS\nlooks good"}] # validator's self-planned final_verdict node thread = task_execution.start_task_execution_in_background( self.db_path, "task-1", self.repo.name, "a goal", @@ -644,8 +637,7 @@ class StartTaskExecutionInBackgroundTest(unittest.TestCase): with patch.object(org_node_lifecycle, "trigger_org_node", side_effect=_fake_ready_trigger), \ patch.object(work_graph_plan.subprocess, "run", side_effect=_fake_plan_subprocess_run_dispatch), \ - patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=RuntimeError("boom")): + patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=RuntimeError("boom")): # Phase 2c: Validator's own verdict is now a self-planned node, # not a flat org_node_trigger.trigger_org_node call -- the # failure this test cares about (a real exception surfacing as diff --git a/src/test_work_graph_fork.py b/src/test_work_graph_fork.py index 605dde5..a119d58 100644 --- a/src/test_work_graph_fork.py +++ b/src/test_work_graph_fork.py @@ -12,7 +12,7 @@ import tempfile import unittest from unittest.mock import patch -from src import schema, work_graph_execute, work_graph_fork +from src import org_node_trigger, schema, work_graph_fork _BRD_PRD_SDD_BENCHMARK_MANIFEST = { @@ -100,12 +100,11 @@ class ForkFromPrdAcceptanceCriterion1Test(ForkTestBase): execute_calls = [] - def fake_execute(cwd, name, args, output_schema, timeout=600): - execute_calls.append(name) - return {"content": "regenerated for " + name} + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + execute_calls.append(prompt) + return {"content": "regenerated for " + prompt} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): result = work_graph_fork.fork_work_graph_run( self.conn, node_exec_ids["prd"], {"content": "a completely different prd"}, "/tmp/proj-1", ) @@ -165,12 +164,11 @@ class ForkFromMixedShapeGraphTest(ForkTestBase): execute_calls = [] - def fake_execute(cwd, name, args, output_schema, timeout=600): - execute_calls.append(name) - return {"content": "regenerated for " + name} + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + execute_calls.append(prompt) + return {"content": "regenerated for " + prompt} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): result = work_graph_fork.fork_work_graph_run( self.conn, node_exec_ids["sdd"], {"content": "a completely different sdd"}, "/tmp/proj-1", ) @@ -199,8 +197,7 @@ class OriginalRunUnaffectedTest(ForkTestBase): "SELECT * FROM work_node_execution WHERE run_id = 'run-orig' ORDER BY node_id" ).fetchall() - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"}): work_graph_fork.fork_work_graph_run( self.conn, node_exec_ids["prd"], {"content": "different"}, "/tmp/proj-1", ) @@ -226,13 +223,11 @@ class MultipleForksFromSameNodeTest(ForkTestBase): "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents ) - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "branch-a"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "branch-a"}): result_a = work_graph_fork.fork_work_graph_run( self.conn, node_exec_ids["prd"], {"content": "prd variant A"}, "/tmp/proj-1", ) - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "branch-b"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "branch-b"}): result_b = work_graph_fork.fork_work_graph_run( self.conn, node_exec_ids["prd"], {"content": "prd variant B"}, "/tmp/proj-1", ) @@ -345,8 +340,7 @@ class StartForkInBackgroundTest(unittest.TestCase): os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home def test_returns_run_id_immediately_and_completes_in_background(self): - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "regenerated"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "regenerated"}): run_id, thread = work_graph_fork.start_fork_in_background( self.db_path, "run-orig-prd", {"content": "new prd"}, "/tmp/proj-1", ) @@ -375,9 +369,8 @@ class ForkOfLeafNodeTest(ForkTestBase): node_exec_ids = self._seed_completed_run( "run-orig", "planner", _BRD_PRD_SDD_BENCHMARK_MANIFEST, contents ) - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = AssertionError("should not be called -- nothing downstream of benchmark") + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = AssertionError("should not be called -- nothing downstream of benchmark") result = work_graph_fork.fork_work_graph_run( self.conn, node_exec_ids["benchmark"], {"content": "different benchmark"}, "/tmp/proj-1", ) diff --git a/src/test_work_graph_orchestrate.py b/src/test_work_graph_orchestrate.py index db049eb..be9b9f6 100644 --- a/src/test_work_graph_orchestrate.py +++ b/src/test_work_graph_orchestrate.py @@ -17,8 +17,8 @@ import unittest from unittest.mock import patch from src import ( - execution_log, paths, schema, skill_query, skill_update, - work_graph_execute, work_graph_orchestrate as wgo, work_node_run, + execution_log, org_node_trigger, paths, schema, skill_query, skill_update, + work_graph_orchestrate as wgo, work_node_run, ) @@ -268,9 +268,8 @@ class RunGraphHappyPathTest(RunGraphTestBase): calls.append((node["node_id"], dict(outputs))) return "prompt for " + node["node_id"] - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = [{"content": "brd text"}, {"content": "prd text"}] + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = [{"content": "brd text"}, {"content": "prd text"}] outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, @@ -282,8 +281,7 @@ class RunGraphHappyPathTest(RunGraphTestBase): def test_creates_work_node_execution_rows_and_placeholder_skill(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"}): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -308,9 +306,10 @@ class RunGraphHappyPathTest(RunGraphTestBase): def build_prompt(node, outputs): return "p-" + node["node_id"] - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = lambda cwd, name, args, schema, timeout=600: calls.append(name) or {"content": "c-" + name} + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = ( + lambda org_node, prompt, cwd, schema, model=None, timeout=300: calls.append(prompt) or {"content": "c-" + prompt} + ) wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, @@ -318,9 +317,8 @@ class RunGraphHappyPathTest(RunGraphTestBase): self.assertEqual(len(calls), 2) # simulate a restart: run_graph called again for the same run_id - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute2: - mock_execute2.side_effect = AssertionError("should not be called -- both nodes are already done") + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger2: + mock_trigger2.side_effect = AssertionError("should not be called -- both nodes are already done") outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, @@ -333,15 +331,14 @@ class RunGraphUsageLimitTest(RunGraphTestBase): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} attempts = {"n": 0} - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): attempts["n"] += 1 if attempts["n"] == 1: - raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session") + raise org_node_trigger.UsageLimitReachedError("hit your session limit", "session") return {"content": "recovered"} sleeps = [] - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -353,13 +350,18 @@ class RunGraphUsageLimitTest(RunGraphTestBase): class RunGraphClassifierBlockedTest(RunGraphTestBase): def test_classifier_blocked_marks_node_blocked_and_propagates(self): + # trigger_org_node_structured doesn't replicate execute_work_graph's + # ClassifierBlockedError/permission_denials-threshold detection (no + # call site ever branched on that specific type -- _make_run_fn's + # own except-clause always treated it the same as any other generic + # failure) -- any non-usage-limit OrgNodeTriggerError exercises the + # exact same "mark node blocked and propagate" path. manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - side_effect=work_graph_execute.ClassifierBlockedError("blocked by classifier"), - ): - with self.assertRaises(work_graph_execute.ClassifierBlockedError): + with patch.object( + org_node_trigger, "trigger_org_node_structured", + side_effect=org_node_trigger.OrgNodeTriggerError("blocked by classifier"), + ): + with self.assertRaises(org_node_trigger.OrgNodeTriggerError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -389,8 +391,7 @@ class RunGraphUnanticipatedExceptionMarksRunBlockedTest(RunGraphTestBase): def bad_build_prompt(node, outputs): raise KeyError("prd_content") - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph"): + with patch.object(org_node_trigger, "trigger_org_node_structured"): with self.assertRaises(KeyError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -410,8 +411,7 @@ class RunGraphUnanticipatedExceptionMarksRunBlockedTest(RunGraphTestBase): def bad_build_prompt(node, outputs): raise RuntimeError("boom") - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph"): + with patch.object(org_node_trigger, "trigger_org_node_structured"): with self.assertRaises(RuntimeError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -427,15 +427,16 @@ class VerifyFnTest(RunGraphTestBase): def test_verify_fn_passing_immediately_does_not_regenerate(self): manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "consistent prd"}) as mock_execute: + with patch.object( + org_node_trigger, "trigger_org_node_structured", return_value={"content": "consistent prd"} + ) as mock_trigger: outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, verify_fn=lambda node, output: (True, None), ) self.assertEqual(outputs["prd"], {"content": "consistent prd"}) - self.assertEqual(mock_execute.call_count, 1) + self.assertEqual(mock_trigger.call_count, 1) run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0] self.assertEqual(run_status, "running") # untouched -- verify_fn never failed @@ -448,8 +449,9 @@ class VerifyFnTest(RunGraphTestBase): checked.append(output["content"]) return output["content"] == "fixed prd", ["scenario A contradicts scenario B"] - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=lambda *a, **k: {"content": next(contents)}): + with patch.object( + org_node_trigger, "trigger_org_node_structured", side_effect=lambda *a, **k: {"content": next(contents)} + ): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, verify_fn=verify_fn, @@ -465,8 +467,9 @@ class VerifyFnTest(RunGraphTestBase): def test_verify_fn_failing_twice_raises_and_blocks_run(self): manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "still contradictory"}): + with patch.object( + org_node_trigger, "trigger_org_node_structured", return_value={"content": "still contradictory"} + ): with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -480,9 +483,9 @@ class VerifyFnTest(RunGraphTestBase): class SelfTargetForcesWorktreeTest(RunGraphTestBase): """U-035 (05 §5.8): self-target unconditionally forces needs_worktree=True, regardless of what the caller passed. run_work_node itself is mocked here - (not just execute_work_graph) so this only asserts what needs_worktree - value run_graph computes and passes down -- it doesn't exercise real git - worktree commands against this actual checkout.""" + (not just trigger_org_node_structured) so this only asserts what + needs_worktree value run_graph computes and passes down -- it doesn't + exercise real git worktree commands against this actual checkout.""" def test_self_target_forces_worktree_even_when_caller_passed_false(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} @@ -490,8 +493,7 @@ class SelfTargetForcesWorktreeTest(RunGraphTestBase): "UPDATE project SET target_folder_path = ? WHERE project_id = 'proj-1'", (paths.repo_root(),) ) self.conn.commit() - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: + with patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", paths.repo_root(), manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=False, @@ -500,8 +502,7 @@ class SelfTargetForcesWorktreeTest(RunGraphTestBase): def test_non_self_target_respects_caller_supplied_false(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: + with patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=False, @@ -510,8 +511,7 @@ class SelfTargetForcesWorktreeTest(RunGraphTestBase): def test_non_self_target_still_respects_caller_supplied_true(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: + with patch.object(work_node_run, "run_work_node", return_value={"content": "x"}) as mock_run: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, needs_worktree=True, @@ -522,8 +522,9 @@ class SelfTargetForcesWorktreeTest(RunGraphTestBase): class RunFnLogsExecutionEventsTest(RunGraphTestBase): """DU-01 (bug A1/A2 root cause): `_make_run_fn()`'s `run_fn` closure must actually call `execution_log.append_log()` around the real - `execute_work_graph()` call, so `execution_log_file`'s backing file on - disk genuinely has content -- not just a DB row tracking its lifecycle.""" + `trigger_org_node_structured()` call, so `execution_log_file`'s backing + file on disk genuinely has content -- not just a DB row tracking its + lifecycle.""" def _node_exec_id(self, node_id="brd"): return self.conn.execute( @@ -536,8 +537,7 @@ class RunFnLogsExecutionEventsTest(RunGraphTestBase): def test_success_writes_node_start_then_node_completed(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "brd text"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "brd text"}): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -556,12 +556,11 @@ class RunFnLogsExecutionEventsTest(RunGraphTestBase): def test_classifier_blocked_writes_node_failed_and_still_propagates(self): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object( - work_graph_execute, "execute_work_graph", - side_effect=work_graph_execute.ClassifierBlockedError("blocked by classifier"), - ): - with self.assertRaises(work_graph_execute.ClassifierBlockedError): + with patch.object( + org_node_trigger, "trigger_org_node_structured", + side_effect=org_node_trigger.OrgNodeTriggerError("blocked by classifier"), + ): + with self.assertRaises(org_node_trigger.OrgNodeTriggerError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -575,14 +574,13 @@ class RunFnLogsExecutionEventsTest(RunGraphTestBase): manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} attempts = {"n": 0} - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): attempts["n"] += 1 if attempts["n"] == 1: - raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session") + raise org_node_trigger.UsageLimitReachedError("hit your session limit", "session") return {"content": "recovered"} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -610,16 +608,15 @@ class RunGraphConcurrencyTest(RunGraphTestBase): events = [] events_lock = threading.Lock() - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): start = time.monotonic() time.sleep(0.05) end = time.monotonic() with events_lock: events.append((start, end)) - return {"content": name} + return {"content": prompt} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p-" + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, @@ -642,7 +639,7 @@ class RunGraphConcurrencyTest(RunGraphTestBase): events = [] events_lock = threading.Lock() - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): start = time.monotonic() time.sleep(0.03) end = time.monotonic() @@ -650,8 +647,7 @@ class RunGraphConcurrencyTest(RunGraphTestBase): events.append((start, end)) return {"content": "x"} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -681,8 +677,7 @@ class RunGraphLayerErrorTest(RunGraphTestBase): raise RuntimeError("boom on a") return {"content": "b done"} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_node_run, "run_work_node", side_effect=fake_run_work_node): + with patch.object(work_node_run, "run_work_node", side_effect=fake_run_work_node): with self.assertRaises(wgo.WorkGraphLayerError) as ctx: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -707,21 +702,19 @@ class RunGraphConcurrentResumeTest(RunGraphTestBase): } calls = [] - def fake_execute(cwd, name, args, output_schema, timeout=600): - calls.append(name) - return {"content": "c-" + name} + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + calls.append(prompt) + return {"content": "c-" + prompt} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) self.assertEqual(len(calls), 2) - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute2: - mock_execute2.side_effect = AssertionError("should not be called -- both nodes already done") + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger2: + mock_trigger2.side_effect = AssertionError("should not be called -- both nodes already done") outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -740,17 +733,9 @@ class RunGraphQuotaCoexistsWithNormalCompletionTest(RunGraphTestBase): "nodes": [{"node_id": "a", "depends_on": []}, {"node_id": "b", "depends_on": []}], "edges": [], } - prompt_by_script = {} - prompt_lock = threading.Lock() - - def fake_write_script(cwd, script_name, script_content): - with prompt_lock: - prompt_by_script[script_name] = script_content - - def fake_execute(cwd, name, args, output_schema, timeout=600): - content = prompt_by_script[name] - if '"p-a"' in content: - raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session") + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + if prompt == "p-a": + raise org_node_trigger.UsageLimitReachedError("hit your session limit", "session") return {"content": "b done"} now_calls = {"n": 0} @@ -764,8 +749,7 @@ class RunGraphQuotaCoexistsWithNormalCompletionTest(RunGraphTestBase): now_calls["n"] += 1 return 0.0 if now_calls["n"] == 1 else 999999.0 - with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): with self.assertRaises(wgo.WorkGraphLayerError) as ctx: wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -812,9 +796,9 @@ class RunGraphRealSqliteConcurrencyTest(unittest.TestCase): "edges": [], } - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): time.sleep(0.005) - return {"content": name} + return {"content": prompt} for round_n in range(5): db_path = os.path.join(self.db_tmp.name, "test_{}.db".format(round_n)) @@ -835,8 +819,7 @@ class RunGraphRealSqliteConcurrencyTest(unittest.TestCase): conn.commit() try: - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -932,24 +915,22 @@ class LlmJudgmentVerifyFnTest(unittest.TestCase): judgment = {"judge_prompt": "Review: {content}"} verify_fn = wgo._llm_judgment_verify_fn(judgment, "planner", "/repo") with patch.object(wgo.paths, "is_self_target", return_value=False), \ - patch.object(work_graph_execute, "write_workflow_script") as mock_write, \ patch.object( - work_graph_execute, "execute_work_graph", + org_node_trigger, "trigger_org_node_structured", return_value={"ok": True, "issues": []}, - ) as mock_execute: + ) as mock_trigger: ok, issues = verify_fn({"node_id": "x"}, {"content": "some content"}) self.assertTrue(ok) self.assertEqual(issues, []) - self.assertEqual(mock_write.call_args[0][0], "/repo") - self.assertEqual(mock_execute.call_args[0][0], "/repo") + self.assertEqual(mock_trigger.call_args[0][0], "planner") + self.assertEqual(mock_trigger.call_args[0][2], "/repo") def test_string_ok_is_coerced(self): judgment = {"judge_prompt": "Review: {content}"} verify_fn = wgo._llm_judgment_verify_fn(judgment, "planner", "/repo") with patch.object(wgo.paths, "is_self_target", return_value=False), \ - patch.object(work_graph_execute, "write_workflow_script"), \ patch.object( - work_graph_execute, "execute_work_graph", + org_node_trigger, "trigger_org_node_structured", return_value={"ok": "false", "issues": ["bad"]}, ): ok, issues = verify_fn({"node_id": "x"}, {"content": "some content"}) @@ -965,16 +946,14 @@ class LlmJudgmentVerifyFnTest(unittest.TestCase): patch.object(wgo.work_node_run, "ensure_worktree_committed") as mock_ensure, \ patch.object(wgo.work_node_run, "merge_worktree") as mock_merge, \ patch.object(wgo.work_node_run, "remove_worktree") as mock_remove, \ - patch.object(work_graph_execute, "write_workflow_script") as mock_write, \ patch.object( - work_graph_execute, "execute_work_graph", + org_node_trigger, "trigger_org_node_structured", return_value={"ok": True, "issues": []}, - ) as mock_execute: + ) as mock_trigger: ok, issues = verify_fn({"node_id": "x"}, {"content": "some content"}) self.assertTrue(ok) mock_create.assert_called_once_with("/repo", unittest.mock.ANY) - self.assertEqual(mock_write.call_args[0][0], "/repo/.graphlemon-worktrees/x") - self.assertEqual(mock_execute.call_args[0][0], "/repo/.graphlemon-worktrees/x") + self.assertEqual(mock_trigger.call_args[0][2], "/repo/.graphlemon-worktrees/x") self.assertEqual(mock_ensure.call_count, 2) mock_merge.assert_called_once() mock_remove.assert_called_once() @@ -990,9 +969,8 @@ class RunGraphMechanicalJudgmentTest(RunGraphTestBase): "edges": [], } registry = {"compiles": lambda content: (content == "real code", None if content == "real code" else "not code")} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = [{"content": "narrative text"}, {"content": "real code"}] + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = [{"content": "narrative text"}, {"content": "real code"}] outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -1005,15 +983,14 @@ class RunGraphMechanicalJudgmentTest(RunGraphTestBase): "nodes": [{"node_id": "script", "judgment": {"kind": "mechanical", "check_ref": "typo"}}], "edges": [], } - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: with self.assertRaises(ValueError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry={"compiles": lambda c: (True, None)}, ) - mock_execute.assert_not_called() + mock_trigger.assert_not_called() class RunGraphLoopBackTest(RunGraphTestBase): @@ -1037,13 +1014,13 @@ class RunGraphLoopBackTest(RunGraphTestBase): def build_prompt(node, outputs): return "base prompt" - def fake_execute(cwd, name, args, output_schema, timeout=600): + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + prompts_seen.append(prompt) return {"content": contents.pop(0)} registry = {"compiles": lambda content: (content == "good", "not real code: " + content)} - with patch.object(work_graph_execute, "write_workflow_script") as mock_write, \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, @@ -1067,18 +1044,13 @@ class RunGraphLoopBackTest(RunGraphTestBase): self.assertEqual(payloads[1]["attempt_number"], 2) # the THIRD (successful) execution's prompt must carry both prior - # rejections' feedback, not just the most recent one. - third_script_prompt = json.loads( - self.conn.execute( - "SELECT input_payload FROM work_node_execution WHERE node_id = 'benchmark-attempt3'" - ).fetchone()[0] - ) - # the real prompt text lives in the rendered workflow script, not - # input_payload (that's just {"node_id": ...}) -- inspect what got - # passed to write_workflow_script for the third attempt instead. - third_call_script = mock_write.call_args_list[2].args[2] - self.assertIn("not real code: bad v1", third_call_script) - self.assertIn("not real code: bad v2", third_call_script) + # rejections' feedback, not just the most recent one -- inspect what + # got passed as trigger_org_node_structured's own `prompt` argument + # directly (input_payload is just {"node_id": ...}, the real prompt + # text never round-trips through the DB). + third_call_prompt = prompts_seen[2] + self.assertIn("not real code: bad v1", third_call_prompt) + self.assertIn("not real code: bad v2", third_call_prompt) def test_loop_back_to_earlier_node_reruns_target_and_downstream(self): # benchmark declares on_fail.action='loop_back' with an on_failure @@ -1110,8 +1082,8 @@ class RunGraphLoopBackTest(RunGraphTestBase): # sdd v1, bad benchmark, sdd v2 (re-run via loop_back), good benchmark. call_order = [] - def side_effect(cwd, name, args, output_schema, timeout=600): - call_order.append(name) + def side_effect(org_node, prompt, cwd, output_schema, model=None, timeout=300): + call_order.append(prompt) call_index = len(call_order) if call_index == 1: return {"content": sdd_outputs[0]} @@ -1123,8 +1095,7 @@ class RunGraphLoopBackTest(RunGraphTestBase): registry = {"compiles": lambda content: (content == "good benchmark", "rejected: " + content)} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=side_effect): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=side_effect): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, build_prompt, _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, @@ -1150,8 +1121,7 @@ class RunGraphLoopBackTest(RunGraphTestBase): "edges": [], } registry = {"compiles": lambda content: (False, "always rejected")} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "always bad"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "always bad"}): with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -1170,14 +1140,15 @@ class RunGraphLoopBackTest(RunGraphTestBase): "edges": [], } registry = {"compiles": lambda content: (False, "bad")} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}) as mock_execute: + with patch.object( + org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"} + ) as mock_trigger: with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, ) - self.assertEqual(mock_execute.call_count, 1) # no retry at all + self.assertEqual(mock_trigger.call_count, 1) # no retry at all def test_resume_after_restart_continues_from_the_real_attempt_count(self): # Simulates a PRIOR process that ran attempt 1 (rejected), recorded @@ -1221,8 +1192,9 @@ class RunGraphLoopBackTest(RunGraphTestBase): # time succeeding -- must resume as attempt 2 (reading the true # count from execution_event), not restart from attempt 1 or lose # the earlier rejection's history. - with patch.object(work_graph_execute, "write_workflow_script") as mock_write, \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "good"}): + with patch.object( + org_node_trigger, "trigger_org_node_structured", return_value={"content": "good"} + ) as mock_trigger: outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, mechanical_check_registry=registry, @@ -1232,7 +1204,7 @@ class RunGraphLoopBackTest(RunGraphTestBase): "SELECT node_id FROM work_node_execution WHERE run_id = 'run-1' ORDER BY node_id" ).fetchall() self.assertEqual([r[0] for r in rows], ["benchmark", "benchmark-attempt2"]) - self.assertIn("bad: bad v1", mock_write.call_args[0][2]) + self.assertIn("bad: bad v1", mock_trigger.call_args[0][1]) def test_org_node_type_scoped_config_limit_used_when_no_node_level_override(self): from src import settings_config @@ -1246,8 +1218,9 @@ class RunGraphLoopBackTest(RunGraphTestBase): "edges": [], } registry = {"compiles": lambda content: (False, "bad")} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}) as mock_execute: + with patch.object( + org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"} + ) as mock_trigger: with self.assertRaises(wgo.ConsistencyCheckFailedError): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, @@ -1258,7 +1231,7 @@ class RunGraphLoopBackTest(RunGraphTestBase): # attempt_number=1 on the first failure is not > 1, so one retry # happens; attempt_number=2 on THAT failure is > 1 and exceeds. # Two real executions total, not one. - self.assertEqual(mock_execute.call_count, 2) + self.assertEqual(mock_trigger.call_count, 2) class NodeLevelSubagentSkillTest(RunGraphTestBase): @@ -1274,8 +1247,7 @@ class NodeLevelSubagentSkillTest(RunGraphTestBase): ], "edges": [], } - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}): + with patch.object(org_node_trigger, "trigger_org_node_structured", return_value={"content": "x"}): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, @@ -1301,28 +1273,20 @@ class NodeLevelSubagentSkillTest(RunGraphTestBase): }], "edges": [], } - seen_agent_types = [] - - def fake_execute(cwd, name, args, output_schema, timeout=600): - if "ok" in output_schema.get("required", []): - return {"ok": True, "issues": []} - return {"content": "x"} + seen_org_nodes = [] - def fake_write(cwd, name, script_content): - if "graphlemon Work Node" in script_content or "graphlemon node judgment" in script_content: - seen_agent_types.append(script_content) + def fake_trigger(org_node, prompt, cwd, output_schema, model=None, timeout=300): + if "content" in output_schema.get("required", []): + return {"content": "x"} + seen_org_nodes.append(org_node) + return {"ok": True, "issues": []} - with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=fake_trigger): wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, ) - judgment_script = next(s for s in seen_agent_types if "graphlemon node judgment" in s) - # _llm_judgment_verify_fn renders agentType as an unquoted JS - # identifier key (agentType: "coder"), not a JSON key -- only the - # VALUE is json.dumps'd. - self.assertIn('agentType:"coder"', judgment_script.replace(" ", "")) + self.assertEqual(seen_org_nodes, ["coder"]) class DynamicLoopBackContentErrorTest(RunGraphTestBase): @@ -1347,10 +1311,11 @@ class DynamicLoopBackContentErrorTest(RunGraphTestBase): node_call_count = [0] lock = threading.Lock() - def side_effect(cwd, name, args, output_schema, timeout=600): - if name.startswith("graphlemon-reflection-"): + def side_effect(org_node, prompt, cwd, output_schema, model=None, timeout=300): + required = output_schema.get("required", []) + if "target_node_id" in required: return {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "sdd is stale"} - if name.startswith("graphlemon-node-"): + if "content" in required: with lock: node_call_count[0] += 1 idx = node_call_count[0] @@ -1361,12 +1326,11 @@ class DynamicLoopBackContentErrorTest(RunGraphTestBase): if idx == 3: return {"content": sdd_contents[1]} return {"content": "good final review"} - raise AssertionError("unexpected script name: " + name) + raise AssertionError("unexpected output_schema: " + repr(output_schema)) registry = {"is_good": lambda content: (content == "good final review", None if content == "good final review" else "rejected: " + content)} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=side_effect): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=side_effect): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "prompt for " + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, @@ -1486,12 +1450,13 @@ class DynamicLoopBackSkillDefectTest(RunGraphTestBase): node_call_count = [0] lock = threading.Lock() - def side_effect(cwd, name, args, output_schema, timeout=600): - if name.startswith("graphlemon-reflection-"): + def side_effect(org_node, prompt, cwd, output_schema, model=None, timeout=300): + required = output_schema.get("required", []) + if "target_node_id" in required: return {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "systemic issue"} - if name.startswith("graphlemon-fixskill-"): + if "new_content" in required: return {"new_content": "v2 content, fixed", "change_summary": "fixed the systemic issue"} - if name.startswith("graphlemon-node-"): + if "content" in required: with lock: node_call_count[0] += 1 idx = node_call_count[0] @@ -1500,12 +1465,11 @@ class DynamicLoopBackSkillDefectTest(RunGraphTestBase): if idx == 2: return {"content": "bad final review"} return {"content": "good final review"} - raise AssertionError("unexpected script name: " + name) + raise AssertionError("unexpected output_schema: " + repr(output_schema)) registry = {"is_good": lambda content: (content == "good final review", None if content == "good final review" else "rejected: " + content)} - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph", side_effect=side_effect): + with patch.object(org_node_trigger, "trigger_org_node_structured", side_effect=side_effect): outputs = wgo.run_graph( self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, lambda node, outputs: "prompt for " + node["node_id"], _IN_SCHEMA, _OUT_SCHEMA, diff --git a/src/test_work_graph_plan.py b/src/test_work_graph_plan.py index 9ea6ce0..9f15df7 100644 --- a/src/test_work_graph_plan.py +++ b/src/test_work_graph_plan.py @@ -12,7 +12,7 @@ import subprocess import unittest from unittest.mock import patch -from src import schema, work_graph_execute, work_graph_orchestrate, work_graph_plan +from src import org_node_trigger, schema, work_graph_orchestrate, work_graph_plan _MANIFEST = { "nodes": [ @@ -571,9 +571,10 @@ class GenericPathEndToEndViaRunGraphTest(PlanWorkGraphTestBase): # plan_work_graph() already inserted the work_graph_run row # (status='planning') -- run_graph() doesn't require any particular # status on it, so nothing more to set up here. - with patch.object(work_graph_execute, "write_workflow_script"), \ - patch.object(work_graph_execute, "execute_work_graph") as mock_execute: - mock_execute.side_effect = lambda cwd, name, args, schema, timeout=600: {"content": "done: " + name} + with patch.object(org_node_trigger, "trigger_org_node_structured") as mock_trigger: + mock_trigger.side_effect = ( + lambda org_node, prompt, cwd, schema, model=None, timeout=300: {"content": "done: " + prompt} + ) outputs = work_graph_orchestrate.run_graph( self.conn, plan_result["run_id"], "coder", "coder-implementation", "/tmp/proj-1", plan_result["graph_manifest"], work_graph_plan.generic_node_prompt_builder, @@ -581,7 +582,7 @@ class GenericPathEndToEndViaRunGraphTest(PlanWorkGraphTestBase): ) self.assertEqual(set(outputs.keys()), {"test-module-a", "test-module-b", "test-module-c"}) - self.assertEqual(mock_execute.call_count, 3) + self.assertEqual(mock_trigger.call_count, 3) if __name__ == "__main__": diff --git a/src/work_graph_execute.py b/src/work_graph_execute.py index d95f32b..2ab729f 100644 --- a/src/work_graph_execute.py +++ b/src/work_graph_execute.py @@ -1,5 +1,19 @@ """c_work_graph_execute (03-C §3.5): dynamic-workflow invocation, full implementation. +**HISTORICAL, 2026-08 (work-graph-node-execution-deblackbox-requirements.md)**: +this module's `write_workflow_script`/`execute_work_graph`/`execute_work_ +graph_with_usage_limit_retry` are no longer called by any production code +path -- the three call sites that used to render a one-off Dynamic Workflow +`.js` script here purely to get "agent identity + schema-shaped output" now +use `org_node_trigger.trigger_org_node_structured()` (a flat `claude +--agent` call + ```json-fence extraction), the same mechanism `work_graph_ +plan.py::plan_work_graph()` had already been using in production the whole +time. Kept in this file (not deleted) as a historical record and because +`.claude/workflows/*.js` files already generated by real self-hosting runs +are real execution artifacts worth being able to re-read the generating +code for. Do not add new callers -- extend `org_node_trigger.py` instead. +See that module's docstring for the full rationale. + Writes an orchestration script into the target project's `.claude/workflows/ .js` and invokes it via `claude -p 'Run / with args {...}'`, reusing the exact code path the A-1 spike validated (spikes/a1-dynamic-workflows/, @@ -68,7 +82,10 @@ def _quota_limit_type(text): 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.""" + target project's own git history (03-C §3.5 step 1), not graphlemon's. + + Historical mechanism, no longer called by production code as of 2026-08 + (see this module's docstring) -- do not add new callers.""" 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") @@ -87,6 +104,9 @@ def execute_work_graph(target_folder_path, name, args, output_schema, timeout=60 `structured_output` dict. Raises ClassifierBlockedError or UsageLimitReachedError instead of the generic WorkGraphExecuteError when the response matches those specific signals. + + Historical mechanism, no longer called by production code as of 2026-08 + (see this module's docstring) -- do not add new callers. """ prompt = "Run /{} with args {}".format(name, json.dumps(args)) result = subprocess.run( @@ -157,6 +177,13 @@ def execute_work_graph_with_usage_limit_retry( non-opus model" means here. `sleep`/`now` are injectable so tests don't block on real 5-30 minute waits. + + Historical mechanism, no longer called by production code as of 2026-08 + (see this module's docstring) -- do not add new callers. FR-5's + equivalent retry-on-usage-limit behavior for the flat-call mechanism is + `work_node_run.py`'s own model-downgrade retry (U-020), driven via + `org_node_trigger.UsageLimitReachedError`/`work_node_run. + raise_usage_limit_signal` -- not a direct port of this function. """ start = now() backoff_index = 0 diff --git a/src/work_graph_orchestrate.py b/src/work_graph_orchestrate.py index 1501e38..1ca7e5b 100644 --- a/src/work_graph_orchestrate.py +++ b/src/work_graph_orchestrate.py @@ -1,28 +1,48 @@ """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 -- +via `org_node_trigger.trigger_org_node_structured()` + `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 blocking +`trigger_org_node_structured()` call, one per node, not one call amortized +across 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 a whole-graph Dynamic Workflow script 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. +**2026-08 (work-graph-node-execution-deblackbox-requirements.md)**: this +module used to render each node's call as a one-off Dynamic Workflow `.js` +script (`work_graph_execute.write_workflow_script` + `.execute_work_graph`, +`claude -p 'Run /...'`) purely to get "agent identity + schema-shaped +output" -- 05 §5.4.1.1's finding that `--json-schema` silently drops +`structured_output` when combined with `--agent` was, at the time (U-031, +2026-08-01), only known to be worked around inside a Workflow script's +`agent(prompt, {agentType, schema})` call. `work_graph_plan.py:: +plan_work_graph()` had already been doing the same "agent identity + schema +output" job with a flat `claude --agent` call + ```json-fence extraction the +whole time, in production, without ever needing Workflow -- so the +Workflow indirection was never actually load-bearing here either, just +unnoticed. `org_node_trigger.trigger_org_node_structured()` generalizes +that flat-call mechanism for any org_node/output_schema/model combination; +`work_graph_execute.py`'s `write_workflow_script`/`execute_work_graph` are +kept (not deleted) as a historical mechanism with no new callers -- see +that module's own docstring. + +Validated for real (2026-08-01, one-node Dynamic Workflow script; +re-validated 2026-08, flat `trigger_org_node_structured` call -- see +src/test_org_node_trigger_e2e.py): a custom `agentType`/`--agent` + +schema-extraction genuinely loads the target org_node's migrated persona +(confirmed via a real call asking the agent to self-identify) and returns +a real schema-shaped payload -- see 05 §5.4.1.5. **`verify_fn` (2026-08-02, PH4-closure audit before starting PH5)**: closes REQ-02's boundary case (PRD internal-contradiction check must intercept @@ -43,17 +63,17 @@ doesn't describe either. **DU-01 (2026-08-04, bug A1/A2 root-cause fix)**: `_make_run_fn()`'s `run_fn` closure is the natural place to call `execution_log.append_log()` -- it's the one spot that both has `node_exec_id` and wraps the actual blocking -`execute_work_graph()` call, and `work_node_run.run_work_node()` already -guarantees an `execution_log_file` row (status='active') exists before -`run_fn` is ever invoked. Before this, `append_log()` was fully implemented -(U-013/U-027) but never called by any production code path, so +`trigger_org_node_structured()` call, and `work_node_run.run_work_node()` +already guarantees an `execution_log_file` row (status='active') exists +before `run_fn` is ever invoked. Before this, `append_log()` was fully +implemented (U-013/U-027) but never called by any production code path, so `execution_log_file` rows existed with zero bytes of real content forever -- the console's "暂无日志内容" was a correct rendering of genuinely empty data, and downloading it crashed `backend_server._send_file_download()` with an uncaught `FileNotFoundError` (also fixed here, see that function). -This is one blocking `agent()` call per node (03-C's own dynamic-workflow -design, see this module's own docstring above) with no mid-flight -progress to report, so logging is 4 discrete events around that one call, +This is one blocking `agent()` call per node (see this module's own +docstring above) with no mid-flight progress to report, so logging is 4 +discrete events around that one call, not a real character-by-character stream: `node_start` (before the call), `node_completed` (after a successful return), `node_failed` (any other exception, re-raised unchanged after logging), `quota_limit` (the @@ -72,7 +92,10 @@ import time import uuid from collections import Counter -from src import bug_record, execution_log, paths, settings_config, skill_query, skill_update, work_graph_execute, work_node_run +from src import ( + bug_record, execution_log, org_node_trigger, paths, settings_config, skill_query, skill_update, + work_node_run, +) def _log_event(conn, node_exec_id, event_type, **payload): @@ -360,28 +383,14 @@ def _ensure_node_row(conn, run_id, node_id, subagent_type, skill_ref, input_sche 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(conn, 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): _log_event(conn, node_exec_id, "node_start", org_node=org_node, model=model) - 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: - output = work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout) - except work_graph_execute.UsageLimitReachedError as e: + output = org_node_trigger.trigger_org_node_structured( + org_node, prompt, cwd, output_schema, model=model, timeout=timeout, + ) + except org_node_trigger.UsageLimitReachedError as e: _log_event(conn, node_exec_id, "quota_limit", quota_limit_type=e.quota_limit_type) work_node_run.raise_usage_limit_signal(e.quota_limit_type) except Exception as e: @@ -445,17 +454,7 @@ def _llm_judgment_verify_fn(judgment, org_node, target_folder_path, timeout=180) cwd = worktree_path or target_folder_path try: prompt = judgment["judge_prompt"].format(content=output["content"]) - script_name = "graphlemon-judgment-" + uuid.uuid4().hex - script_content = ( - "export const meta = {{ name: {name}, description: 'graphlemon node judgment' }}\n\n" - "const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema} }})\n\n" - "return result\n" - ).format( - name=json.dumps(script_name), prompt=json.dumps(prompt), - org_node_json=json.dumps(org_node), schema=json.dumps(_JUDGMENT_SCHEMA), - ) - work_graph_execute.write_workflow_script(cwd, script_name, script_content) - result = work_graph_execute.execute_work_graph(cwd, script_name, {}, _JUDGMENT_SCHEMA, timeout=timeout) + result = org_node_trigger.trigger_org_node_structured(org_node, prompt, cwd, _JUDGMENT_SCHEMA, timeout=timeout) if worktree_path: work_node_run.ensure_worktree_committed(worktree_path, node_exec_id) work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift") @@ -610,17 +609,9 @@ def _run_reflection_perspective(framing_name, prompt, target_folder_path, timeou worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None cwd = worktree_path or target_folder_path try: - script_name = "graphlemon-reflection-" + uuid.uuid4().hex - script_content = ( - "export const meta = {{ name: {name}, description: 'graphlemon Reflection-review ({framing})' }}\n\n" - "const result = await agent({prompt}, {{ agentType: 'validator', schema: {schema} }})\n\n" - "return result\n" - ).format( - name=json.dumps(script_name), framing=framing_name, prompt=json.dumps(prompt), - schema=json.dumps(_REFLECTION_OUTPUT_SCHEMA), + result = org_node_trigger.trigger_org_node_structured( + "validator", prompt, cwd, _REFLECTION_OUTPUT_SCHEMA, timeout=timeout ) - work_graph_execute.write_workflow_script(cwd, script_name, script_content) - result = work_graph_execute.execute_work_graph(cwd, script_name, {}, _REFLECTION_OUTPUT_SCHEMA, timeout=timeout) if worktree_path: work_node_run.ensure_worktree_committed(worktree_path, node_exec_id) work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift") @@ -792,14 +783,9 @@ def _fix_skill_for_node(conn, run_id, target_node_id, graph_manifest, build_prom worktree_path = work_node_run.create_worktree(target_folder_path, node_exec_id) if is_self else None cwd = worktree_path or target_folder_path try: - script_name = "graphlemon-fixskill-" + uuid.uuid4().hex - script_content = ( - "export const meta = {{ name: {name}, description: 'graphlemon Fix-Skill' }}\n\n" - "const result = await agent({prompt}, {{ agentType: 'validator', schema: {schema} }})\n\n" - "return result\n" - ).format(name=json.dumps(script_name), prompt=json.dumps(prompt), schema=json.dumps(_SKILL_REWRITE_SCHEMA)) - work_graph_execute.write_workflow_script(cwd, script_name, script_content) - rewrite_result = work_graph_execute.execute_work_graph(cwd, script_name, {}, _SKILL_REWRITE_SCHEMA, timeout=timeout) + rewrite_result = org_node_trigger.trigger_org_node_structured( + "validator", prompt, cwd, _SKILL_REWRITE_SCHEMA, timeout=timeout + ) if worktree_path: work_node_run.ensure_worktree_committed(worktree_path, node_exec_id) work_node_run.ensure_worktree_committed(target_folder_path, node_exec_id + "-main-checkout-drift") @@ -964,8 +950,8 @@ def run_graph( mechanical_check_registry=None, ): """Runs every node in `graph_manifest` through `run_work_node()`, each - backed by a real one-node dynamic-workflow `agent()` call using - `org_node`'s migrated persona. Nodes are grouped into topological + backed by a real `trigger_org_node_structured()` call using `org_node`'s + migrated persona. Nodes are grouped into topological layers (`topological_layers()`) and run layer-by-layer with a barrier in between; within a layer -- nodes that are mutually independent by construction -- a layer with 2+ nodes runs them concurrently via a diff --git a/src/work_graph_plan.py b/src/work_graph_plan.py index 073534f..ad523ca 100644 --- a/src/work_graph_plan.py +++ b/src/work_graph_plan.py @@ -339,22 +339,44 @@ class PlanningFailedError(Exception): """PLANNING_FAILED -- 03-C §3.4.""" -def _extract_graph_manifest(text): +def extract_json_payload(text, required_keys=()): """--json-schema doesn't constrain output when --agent is set (see module - docstring) -- pulls the graph_manifest out of free-form response text + docstring) -- pulls a JSON payload out of free-form response text instead: try the whole text as JSON, then each ```json fenced block; - accept either {"nodes":...,"edges":...} directly or wrapped one level - deep under a "graph_manifest" key (models sometimes add that wrapper - despite being asked not to).""" + accept the first candidate that parses to a dict containing every key in + required_keys. Generalized (2026-08, work-graph-node-execution- + deblackbox) from what used to be `_extract_graph_manifest`'s own inline + algorithm, so `org_node_trigger.trigger_org_node_structured` can reuse it + for arbitrary output_schemas, not just graph_manifest's {nodes, edges} + shape. Wrapper-key fallbacks (like _extract_graph_manifest's own + "graph_manifest" key below) are caller-specific and deliberately left + out of this shared layer.""" candidates = [text] + _JSON_FENCE_RE.findall(text) for candidate in candidates: try: parsed = json.loads(candidate) except (json.JSONDecodeError, TypeError): continue + if isinstance(parsed, dict) and all(key in parsed for key in required_keys): + return parsed + return None + + +def _extract_graph_manifest(text): + """Thin wrapper around extract_json_payload: also accepts the payload + wrapped one level deep under a "graph_manifest" key (models sometimes + add that wrapper despite being asked not to) -- kept as its own function + rather than folding the wrapper-key check into extract_json_payload + itself, since no other caller needs this specific wrapper-key name.""" + result = extract_json_payload(text, required_keys=["nodes", "edges"]) + if result is not None: + return result + for candidate in [text] + _JSON_FENCE_RE.findall(text): + try: + parsed = json.loads(candidate) + except (json.JSONDecodeError, TypeError): + continue if isinstance(parsed, dict): - if "nodes" in parsed and "edges" in parsed: - return parsed inner = parsed.get("graph_manifest") if isinstance(inner, dict) and "nodes" in inner and "edges" in inner: return inner