Work Graph node execution deblackbox: drop native Dynamic Workflow for single-node calls

Implements docs/done/work-graph-node-execution-deblackbox-requirements.md end to end. The three documented call sites (e2e_pipeline.py:: _check_prd_consistency, work_graph_orchestrate.py::_llm_judgment_verify_fn, work_graph_orchestrate.py::_make_run_fn) all used to render a one-off Dynamic Workflow .js script and run it via `claude -p 'Run /<name>...'` purely to get "agent identity + schema-shaped output" -- a workaround for --json-schema silently conflicting with --agent (05 §5.4.1.1). That workaround was never actually necessary: work_graph_plan.py:: plan_work_graph() solved the identical problem from day one with a flat `claude --agent ... -p ...` call + ```json-fence extraction from free text, already validated in production. New org_node_trigger.trigger_org_node_structured() unifies all three call sites onto that same flat-call approach. extract_json_payload() generalizes work_graph_plan.py's fence-extraction algorithm; _extract_graph_manifest is now a thin wrapper over it. Two more call sites not in the original design doc (written before they existed) were found and migrated too: _run_reflection_perspective and _fix_skill_for_node (Phase 2a's Reflection-review/Fix-Skill). work_graph_fork.py needed no changes -- it only reaches _make_run_fn indirectly via run_graph(), so it benefits automatically. FR-4's real-agent verification (RUN_REAL_AGENT_E2E=1, mandatory, not optional -- src/test_org_node_trigger_e2e.py) confirmed persona loading, --model override, and JSON-fence extraction all work against a real model. write_workflow_script/execute_work_graph/execute_work_graph_with_usage_ limit_retry are confirmed to have no remaining production callers; docstrings mark them historical, the file is kept (not deleted). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Work Graph node execution deblackbox: drop native Dynamic Workflow for single-node calls
Implements docs/done/work-graph-node-execution-deblackbox-requirements.md end to end. The three documented call sites (e2e_pipeline.py:: _check_prd_consistency, work_graph_orchestrate.py::_llm_judgment_verify_fn, work_graph_orchestrate.py::_make_run_fn) all used to render a one-off Dynamic Workflow .js script and run it via `claude -p 'Run /<name>...'` purely to get "agent identity + schema-shaped output" -- a workaround for --json-schema silently conflicting with --agent (05 §5.4.1.1). That workaround was never actually necessary: work_graph_plan.py:: plan_work_graph() solved the identical problem from day one with a flat `claude --agent ... -p ...` call + ```json-fence extraction from free text, already validated in production. New org_node_trigger.trigger_org_node_structured() unifies all three call sites onto that same flat-call approach. extract_json_payload() generalizes work_graph_plan.py's fence-extraction algorithm; _extract_graph_manifest is now a thin wrapper over it. Two more call sites not in the original design doc (written before they existed) were found and migrated too: _run_reflection_perspective and _fix_skill_for_node (Phase 2a's Reflection-review/Fix-Skill). work_graph_fork.py needed no changes -- it only reaches _make_run_fn indirectly via run_graph(), so it benefits automatically. FR-4's real-agent verification (RUN_REAL_AGENT_E2E=1, mandatory, not optional -- src/test_org_node_trigger_e2e.py) confirmed persona loading, --model override, and JSON-fence extraction all work against a real model. write_workflow_script/execute_work_graph/execute_work_graph_with_usage_ limit_retry are confirmed to have no remaining production callers; docstrings mark them historical, the file is kept (not deleted). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent a01d1c0e
Showing 14 changed files with 677 additions and 426 deletions
src/e2e_pipeline.py
src/org_node_trigger.py
src/test_e2e_pipeline.py
src/test_org_level_loop.py
src/test_org_node_trigger.py
src/test_root_cause_trace.py
src/test_scheduler.py
src/test_task_execution.py
src/test_work_graph_fork.py
src/test_work_graph_orchestrate.py
src/test_work_graph_plan.py
src/work_graph_execute.py
src/work_graph_orchestrate.py
src/work_graph_plan.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)
@@ -8,18 +8,155 @@ handling is U-010&#39;s job, not this one.
 Identity comes entirely from `.claude/agents/<org_node>.md` in the target
 project (see src/agent_profile.py) -- this module only assembles the prompt
 and parses the response, no role-isolation logic of its own (03-C §3.3).
+
+**`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 <name> -p ... --model <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 <org_node> -p <prompt>` with cwd=target_folder_path.
@@ -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,
             )
@@ -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(
@@ -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()
@@ -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",
             )
@@ -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}
@@ -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
@@ -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",
             )
@@ -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,
@@ -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__":
 """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/
 <name>.js` and invokes it via `claude -p 'Run /<name> 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 `<target>/.claude/workflows/<name>.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
 """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 /<name>...'`) 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&#39;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
@@ -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