Phase 2a: node-level subagent/skill binding + dynamic_loop_back (Reflection-review + Fix-Skill)

Implements org-graph-unification-phase2-universal-self-planning-design.md §4 + §5, the engine-extension sub-phase (Phase 2a) -- no org-flow callers changed yet, this is the underlying capability. - _run_one_node() now resolves subagent_type/skill_ref per-node (node.get with the run-level org_node/skill_id as fallback), threaded through _ensure_node_row/_make_run_fn/_resolve_node_verify_fn -- a node's own judgment.kind="llm" call runs under its own declared identity, not always the graph-level default. Found and fixed a real gap surfaced by writing this: run_graph() only ensures the run-level default skill_id exists as a placeholder skill (work_node_execution.skill_ref's FK target) -- a node declaring its own skill_ref needs the same guarantee or the first INSERT for that skill_id fails FOREIGN KEY constraint. - New on_fail.action="dynamic_loop_back": Reflection-review (_judge_root_cause_node) runs three real parallel perspectives (content/ connectivity/history) under the Validator identity, each picking a target_node_id from the failing node's real ancestors plus a defect_kind (content_error vs skill_defect), majority-voted by _tally_reflection (own small Counter-based implementation, not imported from root_cause_trace.py -- that module already imports this one). A runtime ancestor check in _run_one_node backs up _tally_reflection's own degradation guarantee. - skill_defect triggers Fix-Skill (_fix_skill_for_node) before the loop back fires: reads the target node's declared skill_ref, builds a rewrite prompt from what was asked/produced/why it failed/prior loop history, runs it under the Validator identity (user-confirmed via AskUserQuestion), then calls the existing but -- until now -- never actually invoked by any production path bug_record.create_bug_record() + skill_update.update_skill(), against skill_query._REPO_ROOT (never target_folder_path). Logs a real 'skill_iterated' execution_event (a CHECK-enum value that has existed since the original U-001 schema but was never written before). Rate-limited to _MAX_SKILL_FIXES_PER_RUN_PER_SKILL rewrites of the same Skill per run. - New total-attempts-per-source-node loop_limit guard alongside the existing per-(target,source) one -- dynamic_loop_back can pick a different target each round, which would otherwise dodge the per-pair cap indefinitely. New settings_config key shape "{org_node}:{source_node_id}:total", no interface change needed there. - New skills/skill-defect-remediation/SKILL.md (Validator-only), wired into roles/validator/role.json. python3 -m unittest discover -s src (741 tests) + discover -s supervisor -p "test_*.py" (126 tests) both green, zero regressions. Deferred to Phase 2b (per design doc's own sequencing note): §2.1/§2.2 (work_graph_run.org_node removal + ux_work_graph_run_active index narrowing) -- narrowing that index now, before the append-to-existing-run mechanism lands, would break the current production flow, since Planner's and Coder's work_graph_run rows both stay non-terminal indefinitely on the happy path today (only Validator's own graph ever reaches 'done'). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Phase 2a: node-level subagent/skill binding + dynamic_loop_back (Reflection-review + Fix-Skill)
Implements org-graph-unification-phase2-universal-self-planning-design.md §4 + §5, the engine-extension sub-phase (Phase 2a) -- no org-flow callers changed yet, this is the underlying capability. - _run_one_node() now resolves subagent_type/skill_ref per-node (node.get with the run-level org_node/skill_id as fallback), threaded through _ensure_node_row/_make_run_fn/_resolve_node_verify_fn -- a node's own judgment.kind="llm" call runs under its own declared identity, not always the graph-level default. Found and fixed a real gap surfaced by writing this: run_graph() only ensures the run-level default skill_id exists as a placeholder skill (work_node_execution.skill_ref's FK target) -- a node declaring its own skill_ref needs the same guarantee or the first INSERT for that skill_id fails FOREIGN KEY constraint. - New on_fail.action="dynamic_loop_back": Reflection-review (_judge_root_cause_node) runs three real parallel perspectives (content/ connectivity/history) under the Validator identity, each picking a target_node_id from the failing node's real ancestors plus a defect_kind (content_error vs skill_defect), majority-voted by _tally_reflection (own small Counter-based implementation, not imported from root_cause_trace.py -- that module already imports this one). A runtime ancestor check in _run_one_node backs up _tally_reflection's own degradation guarantee. - skill_defect triggers Fix-Skill (_fix_skill_for_node) before the loop back fires: reads the target node's declared skill_ref, builds a rewrite prompt from what was asked/produced/why it failed/prior loop history, runs it under the Validator identity (user-confirmed via AskUserQuestion), then calls the existing but -- until now -- never actually invoked by any production path bug_record.create_bug_record() + skill_update.update_skill(), against skill_query._REPO_ROOT (never target_folder_path). Logs a real 'skill_iterated' execution_event (a CHECK-enum value that has existed since the original U-001 schema but was never written before). Rate-limited to _MAX_SKILL_FIXES_PER_RUN_PER_SKILL rewrites of the same Skill per run. - New total-attempts-per-source-node loop_limit guard alongside the existing per-(target,source) one -- dynamic_loop_back can pick a different target each round, which would otherwise dodge the per-pair cap indefinitely. New settings_config key shape "{org_node}:{source_node_id}:total", no interface change needed there. - New skills/skill-defect-remediation/SKILL.md (Validator-only), wired into roles/validator/role.json. python3 -m unittest discover -s src (741 tests) + discover -s supervisor -p "test_*.py" (126 tests) both green, zero regressions. Deferred to Phase 2b (per design doc's own sequencing note): §2.1/§2.2 (work_graph_run.org_node removal + ux_work_graph_run_active index narrowing) -- narrowing that index now, before the append-to-existing-run mechanism lands, would break the current production flow, since Planner's and Coder's work_graph_run rows both stay non-terminal indefinitely on the happy path today (only Validator's own graph ever reaches 'done'). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee · 宜博
1 parent 528ec6e1
Showing 4 changed files with 753 additions and 11 deletions
roles/validator/role.json
skills/skill-defect-remediation/SKILL.md
src/test_work_graph_orchestrate.py
src/work_graph_orchestrate.py
@@ -14,7 +14,8 @@
     "qa-task-validation",
     "unattended-agent-operations",
     "design-dynamic-work-graph",
-    "unknown-scope-defect-sweep"
+    "unknown-scope-defect-sweep",
+    "skill-defect-remediation"
   ],
   "applications": {
     "mcp_servers": []
+---
+name: skill-defect-remediation
+description: |
+  当 Reflection-review 判定一次节点失败的根因是系统性 Skill 缺陷(不是这一次内容的偶然错误)时,
+  怎么安全地修订这份 Skill 文件本身——精确定位并只修正导致这次真实失败的那一小部分,不破坏该
+  Skill 对其他场景已经正确的既有指导。产出必须是完整的、可直接替换生效的 Skill 内容,不是
+  diff 或建议清单。
+version: 1.0.0
+tags:
+  - skill-iteration
+  - remediation
+  - self-evolution
+---
+
+# Skill 缺陷修复
+
+## 我在做什么、不做什么
+
+我在改一份 Skill 文件的内容,不是在改这一次任务的输出。触发条件是 Reflection-review(`_judge_root_cause_node`)已经判定:这不是"这次恰好没写好"(`content_error`),而是"这份 Skill 本身的方法论/提示词有系统性偏差,同类任务会反复踩同一个坑"(`skill_defect`)。
+
+我拿到的输入是:这份 Skill 当前的完整内容、使用它的节点这次被要求做什么、它实际产出了什么、为什么被判定失败、这个节点在本次 run 内此前被退回的历史(如果有)。我的输出会被 `skill_update.update_skill()` 直接、原样写入 Skill 文件并提交 Git——不是草案,不是给人看的建议,是最终生效内容。
+
+## 核心原则:手术刀式修改,不是重写
+
+- **只改导致这次失败的那一部分**。这份 Skill 对其他场景的既有指导,大概率是对的——它能走到今天,说明大部分情况下有效。逐字重写整份文件、"顺手优化"看起来不够好的其他段落,是最常见的过度修改陷阱,会引入新的、这次任务范围之外的回归风险。
+- **先定位到具体是哪一句/哪一段导致了偏差,再改**。不要停留在"这份 Skill 整体上不够好"这种笼统判断——不可操作,也无法验证改对了没有。好的定位是这个形状:"第 X 节要求 Y,但没有说明 Z 场景下应该怎么做,导致模型默认走了错误的 W 路径"。
+- **改动要能解释"为什么这次会防住同一类失败"**,不是"看起来更完善了"。如果说不清楚新内容具体怎么拦住了这次复现的失败模式,说明还没找到真正的病灶。
+
+## 一个真实的参考案例:叙述文字冒充结构化产出
+
+这个代码库真实反复复现过的一类系统性 Skill 缺陷(`docs/todo/validator-test-script-narrative-output-issue.md` §6 完整根因记录):要求"先用工具探索、再产出最终内容"的节点(比如需要先读真实代码才能写 SDD/Benchmark 的场景),模型探索完之后的收尾话术天然倾向"汇报我刚才做了/启动了什么",而不是切换成"直接吐出最终内容"模式——即使 Skill 提示词已经写了"不要写叙述文字",这句话附加在提示词末尾的格式规则,不足以覆盖"多轮工具调用之后的收尾习惯"这个更底层的行为模式。
+
+如果这次要修的 Skill 也有类似"先探索、再产出"的结构,而失败症状也是"产出的是进度汇报式散文,不是结构化内容",可以直接参考这个案例的诊断方式,但**不要机械照抄"加一句不要写叙述文字"这种已经被证明不够的修法**——真正管用的方向是把"探索阶段"和"最终作答阶段"在提示词结构上明确分开(比如要求模型先完成探索、再有一个清晰的"现在开始,你的下一段输出必须且只能是最终内容本身"的转折),而不是在末尾追加一句格式提醒。
+
+## 输出
+
+`new_content`:完整的、修订后的 Skill 文件内容(不是 diff、不是片段)。`change_summary`:一句话说清楚改了什么、为什么——这句话会被记录进 `skill_iteration.change_summary`,是这次 Skill 迭代唯一的人类可读说明,写清楚比写简短更重要。
@@ -9,13 +9,17 @@ Run: python3 -m unittest src.test_work_graph_orchestrate -v
  
 import json
 import os
+import subprocess
 import tempfile
 import threading
 import time
 import unittest
 from unittest.mock import patch
  
-from src import execution_log, paths, schema, work_graph_execute, work_graph_orchestrate as wgo, work_node_run
+from src import (
+    execution_log, paths, schema, skill_query, skill_update,
+    work_graph_execute, work_graph_orchestrate as wgo, work_node_run,
+)
  
  
 class TopologicalOrderTest(unittest.TestCase):
@@ -1257,5 +1261,313 @@ class RunGraphLoopBackTest(RunGraphTestBase):
         self.assertEqual(mock_execute.call_count, 2)
  
  
+class NodeLevelSubagentSkillTest(RunGraphTestBase):
+    """Phase 2 §4: a node can declare its own subagent_type/skill_ref,
+    overriding the run-level org_node/skill_id defaults -- SDD §10 Phase 2a
+    acceptance criterion 1."""
+
+    def test_node_level_fields_override_run_level_defaults(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "a", "depends_on": [], "subagent_type": "coder", "skill_ref": "code-task-development"},
+                {"node_id": "b", "depends_on": []},  # no override -- uses run-level org_node/skill_id
+            ],
+            "edges": [],
+        }
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", 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,
+            )
+        rows = {
+            r[0]: (r[1], r[2]) for r in self.conn.execute(
+                "SELECT node_id, subagent_type, skill_ref FROM work_node_execution WHERE run_id = 'run-1'"
+            ).fetchall()
+        }
+        self.assertEqual(rows["a"], ("coder", "code-task-development"))
+        self.assertEqual(rows["b"], ("planner", "planner-doc-writing"))
+
+    def test_node_level_org_node_used_for_llm_judgment_agent_identity(self):
+        # A node's own judgment.kind='llm' call must run under ITS OWN
+        # subagent_type, not the run-level default -- otherwise a node
+        # declaring subagent_type='validator' for judgment purposes would
+        # silently get judged under the wrong persona.
+        manifest = {
+            "nodes": [{
+                "node_id": "a", "depends_on": [], "subagent_type": "coder",
+                "judgment": {"kind": "llm", "judge_prompt": "check {content}"},
+                "on_fail": {"action": "hard_fail"},
+            }],
+            "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"}
+
+        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)
+
+        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):
+            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(" ", ""))
+
+
+class DynamicLoopBackContentErrorTest(RunGraphTestBase):
+    """Phase 2 §5: on_fail.action='dynamic_loop_back' -- Reflection-review
+    picks a target node from the failing node's real ancestors and judges
+    defect_kind. content_error skips Fix-Skill and loops back directly --
+    SDD §10 Phase 2a acceptance criterion 2."""
+
+    def test_unanimous_content_error_loops_back_to_chosen_ancestor(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "sdd", "depends_on": []},
+                {
+                    "node_id": "final_review", "depends_on": ["sdd"],
+                    "judgment": {"kind": "mechanical", "check_ref": "is_good"},
+                    "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
+                },
+            ],
+            "edges": [],
+        }
+        sdd_contents = ["sdd v1", "sdd v2 (fixed)"]
+        node_call_count = [0]
+        lock = threading.Lock()
+
+        def side_effect(cwd, name, args, output_schema, timeout=600):
+            if name.startswith("graphlemon-reflection-"):
+                return {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "sdd is stale"}
+            if name.startswith("graphlemon-node-"):
+                with lock:
+                    node_call_count[0] += 1
+                    idx = node_call_count[0]
+                if idx == 1:
+                    return {"content": sdd_contents[0]}
+                if idx == 2:
+                    return {"content": "bad final review"}
+                if idx == 3:
+                    return {"content": sdd_contents[1]}
+                return {"content": "good final review"}
+            raise AssertionError("unexpected script name: " + name)
+
+        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):
+            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,
+                mechanical_check_registry=registry,
+            )
+
+        self.assertEqual(outputs["sdd"], {"content": sdd_contents[1]})
+        self.assertEqual(outputs["final_review"], {"content": "good final review"})
+
+        rows = [
+            r[0] for r in self.conn.execute(
+                "SELECT node_id FROM work_node_execution WHERE run_id = 'run-1' ORDER BY node_id"
+            ).fetchall()
+        ]
+        self.assertEqual(rows, ["final_review", "final_review-attempt2", "sdd", "sdd-attempt2"])
+
+        events = self.conn.execute(
+            "SELECT payload FROM execution_event WHERE event_type = 'loop_back_triggered'"
+        ).fetchall()
+        self.assertEqual(len(events), 1)
+        payload = json.loads(events[0][0])
+        self.assertEqual(payload["source_node_id"], "final_review")
+        self.assertEqual(payload["target_node_id"], "sdd")
+
+        # no skill_iterated event -- content_error must not touch any Skill.
+        skill_events = self.conn.execute(
+            "SELECT 1 FROM execution_event WHERE event_type = 'skill_iterated'"
+        ).fetchall()
+        self.assertEqual(skill_events, [])
+
+class TallyReflectionTest(unittest.TestCase):
+    """_tally_reflection in isolation -- the majority-vote + degradation
+    logic _judge_root_cause_node relies on. An invalid winner (not among
+    candidate_node_ids) is corrected HERE, before _run_one_node's own
+    'target not in ancestors' runtime check (§5.7 point 1) would ever see
+    it -- that check is defense-in-depth for if this guarantee is ever
+    weakened, not the primary enforcement, so it's tested at this level."""
+
+    def test_unanimous_valid_winner(self):
+        judgments = [
+            {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "a"},
+            {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "b"},
+            {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "c"},
+        ]
+        result = wgo._tally_reflection(judgments, ["sdd", "prd"])
+        self.assertEqual(result["target_node_id"], "sdd")
+        self.assertEqual(result["defect_kind"], "content_error")
+
+    def test_two_thirds_skill_defect_majority(self):
+        judgments = [
+            {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "a"},
+            {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "b"},
+            {"target_node_id": "sdd", "defect_kind": "content_error", "detail": "c"},
+        ]
+        result = wgo._tally_reflection(judgments, ["sdd", "prd"])
+        self.assertEqual(result["target_node_id"], "sdd")
+        self.assertEqual(result["defect_kind"], "skill_defect")
+
+    def test_winner_not_in_candidates_degrades_to_first_candidate_content_error(self):
+        judgments = [
+            {"target_node_id": "not_a_candidate", "defect_kind": "skill_defect", "detail": "a"},
+            {"target_node_id": "not_a_candidate", "defect_kind": "skill_defect", "detail": "b"},
+            {"target_node_id": "not_a_candidate", "defect_kind": "skill_defect", "detail": "c"},
+        ]
+        result = wgo._tally_reflection(judgments, ["sdd", "prd"])
+        self.assertEqual(result["target_node_id"], "sdd")
+        self.assertEqual(result["defect_kind"], "content_error")
+
+    def test_no_two_thirds_agreement_degrades(self):
+        judgments = [
+            {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "a"},
+            {"target_node_id": "prd", "defect_kind": "content_error", "detail": "b"},
+            {"target_node_id": "benchmark", "defect_kind": "content_error", "detail": "c"},
+        ]
+        result = wgo._tally_reflection(judgments, ["sdd", "prd", "benchmark"])
+        self.assertEqual(result["defect_kind"], "content_error")
+        self.assertIn(result["target_node_id"], ["sdd", "prd", "benchmark"])
+
+
+def _init_git_repo(path):
+    subprocess.run(["git", "init"], cwd=path, check=True, capture_output=True)
+
+
+class DynamicLoopBackSkillDefectTest(RunGraphTestBase):
+    """Phase 2 §5.5: defect_kind='skill_defect' triggers Fix-Skill -- a
+    real skill_update.update_skill() call against a real (throwaway) git
+    repo, never graphlemon's own checkout -- before the loop_back fires.
+    SDD §10 Phase 2a acceptance criterion 3."""
+
+    def setUp(self):
+        super().setUp()
+        self.skill_repo = tempfile.TemporaryDirectory()
+        self.addCleanup(self.skill_repo.cleanup)
+        _init_git_repo(self.skill_repo.name)
+        skill_update._write_and_commit(self.skill_repo.name, "skills/my-skill/SKILL.md", "v1 content", "seed skill")
+        self.conn.execute(
+            "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) "
+            "VALUES ('my-skill', 'coder', 'skills/my-skill', 'unknown', 'migrated_from_hermes', 'active')"
+        )
+        self.conn.commit()
+        self.repo_root_patch = patch.object(skill_query, "_REPO_ROOT", self.skill_repo.name)
+        self.repo_root_patch.start()
+        self.addCleanup(self.repo_root_patch.stop)
+
+    def test_skill_defect_calls_real_update_skill_before_loop_back(self):
+        manifest = {
+            "nodes": [
+                {"node_id": "sdd", "depends_on": [], "skill_ref": "my-skill"},
+                {
+                    "node_id": "final_review", "depends_on": ["sdd"],
+                    "judgment": {"kind": "mechanical", "check_ref": "is_good"},
+                    "on_fail": {"action": "dynamic_loop_back", "max_attempts": 3},
+                },
+            ],
+            "edges": [],
+        }
+        node_call_count = [0]
+        lock = threading.Lock()
+
+        def side_effect(cwd, name, args, output_schema, timeout=600):
+            if name.startswith("graphlemon-reflection-"):
+                return {"target_node_id": "sdd", "defect_kind": "skill_defect", "detail": "systemic issue"}
+            if name.startswith("graphlemon-fixskill-"):
+                return {"new_content": "v2 content, fixed", "change_summary": "fixed the systemic issue"}
+            if name.startswith("graphlemon-node-"):
+                with lock:
+                    node_call_count[0] += 1
+                    idx = node_call_count[0]
+                if idx in (1, 3):
+                    return {"content": "sdd content"}
+                if idx == 2:
+                    return {"content": "bad final review"}
+                return {"content": "good final review"}
+            raise AssertionError("unexpected script name: " + name)
+
+        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):
+            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,
+                mechanical_check_registry=registry,
+            )
+
+        self.assertEqual(outputs["final_review"], {"content": "good final review"})
+
+        # a real skill_iteration row was produced, with a real (different)
+        # after_git_commit_sha -- not a stub, an actual git commit.
+        iteration = self.conn.execute(
+            "SELECT skill_id, before_git_commit_sha, after_git_commit_sha, change_summary "
+            "FROM skill_iteration"
+        ).fetchone()
+        self.assertEqual(iteration[0], "my-skill")
+        self.assertNotEqual(iteration[1], iteration[2])
+        self.assertEqual(iteration[3], "fixed the systemic issue")
+
+        with open(os.path.join(self.skill_repo.name, "skills/my-skill/SKILL.md")) as f:
+            self.assertEqual(f.read(), "v2 content, fixed")
+
+        # a real bug_record(root_cause_category='skill_defect') was created
+        # and linked back to the iteration (skill_update.update_skill()'s
+        # own contract).
+        bug_row = self.conn.execute(
+            "SELECT root_cause_category, skill_iteration_suggestion_id FROM bug_record"
+        ).fetchone()
+        self.assertEqual(bug_row[0], "skill_defect")
+        self.assertIsNotNone(bug_row[1])
+
+        # skill_iterated execution_event recorded (existing but previously
+        # never-written CHECK enum value).
+        skill_events = self.conn.execute(
+            "SELECT payload FROM execution_event WHERE event_type = 'skill_iterated'"
+        ).fetchall()
+        self.assertEqual(len(skill_events), 1)
+        payload = json.loads(skill_events[0][0])
+        self.assertEqual(payload["node_id"], "sdd")
+        self.assertEqual(payload["skill_id"], "my-skill")
+
+        # graphlemon's own real checkout was never touched -- the Skill
+        # write landed only in the throwaway skill_repo tempdir.
+        self.assertFalse(
+            os.path.exists(os.path.join(paths.repo_root(), "skills", "my-skill", "SKILL.md.orig"))
+        )
+
+    def test_rate_limit_stops_repeated_rewrites_of_same_skill(self):
+        # §5.7 point 5: after _MAX_SKILL_FIXES_PER_RUN_PER_SKILL rewrites of
+        # the same (run_id, skill_id) this run, Fix-Skill silently skips --
+        # dynamic_loop_back still proceeds as a plain loop_back.
+        for i in range(wgo._MAX_SKILL_FIXES_PER_RUN_PER_SKILL):
+            wgo._log_skill_iterated_event(self.conn, "run-1", "sdd", "my-skill", "iteration-{}".format(i))
+
+        manifest = {"nodes": [{"node_id": "sdd", "depends_on": [], "skill_ref": "my-skill"}], "edges": []}
+        result = wgo._fix_skill_for_node(
+            self.conn, "run-1", "sdd", manifest, lambda node, outputs: "p", {}, ["feedback"], "/tmp/proj-1",
+        )
+        self.assertIsNone(result)
+        # no new skill_iteration row was created beyond what we seeded above.
+        count = self.conn.execute("SELECT COUNT(*) FROM skill_iteration").fetchone()[0]
+        self.assertEqual(count, 0)  # the two seeded events above didn't insert real skill_iteration rows
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -66,11 +66,13 @@ actually happened (N attempts), not a bug.
  
 import concurrent.futures
 import json
+import os
 import threading
 import time
 import uuid
+from collections import Counter
  
-from src import execution_log, paths, settings_config, work_graph_execute, work_node_run
+from src import bug_record, execution_log, paths, settings_config, skill_query, skill_update, work_graph_execute, work_node_run
  
  
 def _log_event(conn, node_exec_id, event_type, **payload):
@@ -506,6 +508,323 @@ def _validate_mechanical_check_refs(graph_manifest, mechanical_check_registry):
             )
  
  
+_REFLECTION_OUTPUT_SCHEMA = {
+    "type": "object", "required": ["target_node_id", "defect_kind", "detail"],
+    "properties": {
+        "target_node_id": {"type": "string"},
+        "defect_kind": {"type": "string", "enum": ["content_error", "skill_defect"]},
+        "detail": {"type": "string"},
+    },
+}
+
+_REFLECTION_FRAMINGS = {
+    "content": (
+        "CONTENT PERSPECTIVE: for each candidate node below, look at what it was asked to produce "
+        "and what it actually produced. Which ONE candidate's output itself is genuinely defective? "
+        "Then decide: is this a one-off mistake for this specific run ('content_error' -- the "
+        "underlying methodology is fine, this particular attempt just didn't execute it well), or "
+        "does it look like a systematic pattern that would recur on other tasks using the same "
+        "approach/prompt style ('skill_defect')? A concrete red flag for skill_defect: the node's "
+        "content reads as a narrative description of an action taken (e.g. 'I launched/produced/"
+        "completed X') instead of the structured content itself -- this is a known recurring failure "
+        "mode in nodes that explore with tools before producing a final answer, not a one-off typo."
+    ),
+    "connectivity": (
+        "CONNECTIVITY PERSPECTIVE: if no single candidate's content is obviously wrong on its own, "
+        "is the real problem how the candidates connect -- e.g. a node that should have depended on "
+        "another candidate's output but didn't, so it worked from stale or missing context? This "
+        "kind of problem is almost always 'content_error' (a one-off planning slip for this run), not "
+        "a systemic Skill defect -- pick whichever candidate's isolation from its true dependency is "
+        "the root issue."
+    ),
+    "history": (
+        "HISTORY PERSPECTIVE: you are given this run's actual loop_back/skill-iteration event history "
+        "for each candidate. If a candidate has already been sent back for revision or had its Skill "
+        "rewritten before in this same run for a SIMILAR reason, that recurrence itself is strong "
+        "evidence of 'skill_defect' -- a Skill that keeps producing the same class of failure even "
+        "after a plain retry has methodology problems, not bad luck. If a candidate has no such prior "
+        "history, default to 'content_error' unless the content itself (see the CONTENT framing's red "
+        "flag) makes skill_defect obvious even on a first occurrence."
+    ),
+}
+
+
+def _candidate_context_block(graph_manifest, build_prompt, outputs, candidate_node_ids):
+    nodes_by_id = {n["node_id"]: n for n in graph_manifest["nodes"]}
+    parts = []
+    for node_id in candidate_node_ids:
+        node = nodes_by_id[node_id]
+        asked = build_prompt(node, outputs)
+        produced = (outputs.get(node_id) or {}).get("content", "")
+        parts.append(
+            "--- candidate node_id={} ---\nWhat it was asked:\n{}\n\nWhat it produced:\n{}".format(
+                node_id, asked, produced
+            )
+        )
+    return "\n\n".join(parts)
+
+
+def _candidate_history_block(conn, run_id, candidate_node_ids):
+    events = _loop_back_events(conn, run_id)
+    skill_events = _skill_iterated_events(conn, run_id)
+    parts = []
+    for node_id in candidate_node_ids:
+        prior_loop_backs = [e for e in events if e["target_node_id"] == node_id]
+        prior_skill_fixes = [e for e in skill_events if e["node_id"] == node_id]
+        parts.append(
+            "--- candidate node_id={} --- prior loop_back count: {}; prior skill_iterated count: {}".format(
+                node_id, len(prior_loop_backs), len(prior_skill_fixes)
+            )
+        )
+    return "\n".join(parts)
+
+
+def _reflection_prompt(framing_name, failed_node_id, feedback, candidate_node_ids, graph_manifest, build_prompt, outputs, conn, run_id):
+    framing = _REFLECTION_FRAMINGS[framing_name]
+    context = (
+        _candidate_history_block(conn, run_id, candidate_node_ids) if framing_name == "history"
+        else _candidate_context_block(graph_manifest, build_prompt, outputs, candidate_node_ids)
+    )
+    return (
+        "{framing}\n\nThe node {failed_node_id} failed judgment with this feedback: {feedback}\n\n"
+        "Candidates (pick target_node_id from EXACTLY these node_ids, nothing else):\n{candidates}\n\n"
+        "Context:\n{context}\n\n"
+        "Reply with target_node_id (one of the candidate node_ids above), defect_kind "
+        "('content_error' or 'skill_defect'), and detail (one sentence justifying both)."
+    ).format(
+        framing=framing, failed_node_id=failed_node_id, feedback="; ".join(feedback or []),
+        candidates=", ".join(candidate_node_ids), context=context,
+    )
+
+
+def _run_reflection_perspective(framing_name, prompt, target_folder_path, timeout):
+    """One perspective's real call, under the Validator identity (Reflection-
+    review is diagnostic, not opinion-laden about the failed content itself
+    -- same 'independent second opinion' reasoning Fix-Skill's own 'who
+    rewrites the Skill' decision uses, see _fix_skill_for_node). Same
+    self-target worktree isolation as _llm_judgment_verify_fn (a real
+    agent() call with memory:project writes an uncommitted change wherever
+    it runs)."""
+    is_self = paths.is_self_target(target_folder_path)
+    node_exec_id = "reflection-" + framing_name + "-" + uuid.uuid4().hex[:12]
+    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),
+        )
+        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")
+            work_node_run.merge_worktree(target_folder_path, node_exec_id)
+        return {
+            "target_node_id": result["target_node_id"],
+            "defect_kind": result["defect_kind"],
+            "detail": result.get("detail", ""),
+        }
+    finally:
+        if worktree_path:
+            work_node_run.remove_worktree(target_folder_path, worktree_path)
+
+
+def _tally_reflection(judgments, candidate_node_ids):
+    """>=2/3 agreement on target_node_id, same threshold root_cause_trace.py's
+    _tally_perspectives uses -- not imported from there (would create a
+    circular import, root_cause_trace.py already imports this module) and
+    small enough to duplicate, same reasoning work_graph_plan.py's own
+    _transitive_ancestors duplication already documents. defect_kind is
+    taken from whichever judgment(s) picked the winning target_node_id
+    (majority among just those, defaulting to 'content_error' on a tie --
+    the conservative choice per §5.7 point 2). <2/3 agreement on
+    target_node_id degrades to the nearest direct ancestor + 'content_error'
+    rather than escalating to a human -- this is one retry inside a single
+    Work Graph run, not the terminal cross-Org-Agent judgment
+    root_cause_trace.py's InconclusiveEscalateError guards."""
+    target_counts = Counter(j["target_node_id"] for j in judgments)
+    winner, count = target_counts.most_common(1)[0]
+    if count * 2 < len(judgments) or winner not in candidate_node_ids:
+        return {"target_node_id": candidate_node_ids[0], "defect_kind": "content_error", "detail": "no >=2/3 agreement"}
+    kind_counts = Counter(j["defect_kind"] for j in judgments if j["target_node_id"] == winner)
+    defect_kind = "skill_defect" if kind_counts.get("skill_defect", 0) > kind_counts.get("content_error", 0) else "content_error"
+    detail = " | ".join(j["detail"] for j in judgments if j["target_node_id"] == winner)
+    return {"target_node_id": winner, "defect_kind": defect_kind, "detail": detail}
+
+
+def _judge_root_cause_node(
+    conn, run_id, failed_node_id, feedback, candidate_node_ids, graph_manifest, build_prompt, outputs,
+    target_folder_path, timeout=180,
+):
+    """Reflection-review (org-graph-unification-and-universal-self-planning-
+    requirements.md §3.1 step 2): three independent real perspectives --
+    content, connectivity, history -- each picks a target_node_id from
+    candidate_node_ids and a defect_kind, run concurrently (own real claude
+    call, own worktree if self-target). Majority vote via _tally_reflection.
+    candidate_node_ids must be non-empty (the caller, _run_one_node's
+    dynamic_loop_back branch, only calls this when the failing node has at
+    least one real ancestor -- a terminal judgment node with zero ancestors
+    has nothing to loop back to and shouldn't declare dynamic_loop_back)."""
+    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
+        futures = [
+            pool.submit(
+                _run_reflection_perspective, name,
+                _reflection_prompt(name, failed_node_id, feedback, candidate_node_ids, graph_manifest, build_prompt, outputs, conn, run_id),
+                target_folder_path, timeout,
+            )
+            for name in _REFLECTION_FRAMINGS
+        ]
+        judgments = [f.result() for f in futures]
+    return _tally_reflection(judgments, candidate_node_ids)
+
+
+_SKILL_REWRITE_SCHEMA = {
+    "type": "object", "required": ["new_content", "change_summary"],
+    "properties": {"new_content": {"type": "string"}, "change_summary": {"type": "string"}},
+}
+
+_SKILL_REWRITE_PROMPT = (
+    "You are revising a Skill file (the reusable methodology/prompt guidance a graphlemon Org Agent "
+    "follows) because it produced a real, repeated failure. Your job is surgical: fix the SPECIFIC "
+    "gap that caused this failure, without breaking the Skill's existing correct guidance for other "
+    "cases it already handles well.\n\n"
+    "Goal this task is working on: {goal_text}\n\n"
+    "Skill being revised (skill_id={skill_id}):\n{current_content}\n\n"
+    "What the node using this Skill was asked to do:\n{what_was_asked}\n\n"
+    "What it actually produced (the defective output):\n{what_it_produced}\n\n"
+    "Why this was judged a failure: {failure_feedback}\n\n"
+    "This node's prior rejection history in this same run (if any): {loop_history}\n\n"
+    "Reply with new_content (the COMPLETE revised Skill file content -- not a diff, not a partial "
+    "excerpt) and change_summary (one sentence describing what changed and why)."
+)
+
+
+def _read_skill_content(skill_id):
+    """Skill library content lives in graphlemon's own repo checkout
+    (skill_query._REPO_ROOT), never in target_folder_path -- same
+    established boundary c_skill_diff/c_skill_rollback's HTTP routes
+    already rely on (05 §5.4.1.4), reused here rather than re-derived."""
+    path = os.path.join(skill_query._REPO_ROOT, "skills", skill_id, "SKILL.md")
+    with open(path, encoding="utf-8") as f:
+        return f.read()
+
+
+def _log_skill_iterated_event(conn, run_id, node_id, skill_id, iteration_id):
+    """Persists a 'skill_iterated' execution_event -- this CHECK-enum value
+    has existed since the original U-001 schema but was never actually
+    written by any code path before Fix-Skill (skill_update.update_skill()
+    itself doesn't log one; it's this caller's job, same division of
+    responsibility loop_back_triggered already has between run_graph() and
+    _record_loop_back_event). No schema migration needed."""
+    row = conn.execute(
+        "SELECT wgr.task_id, t.project_id FROM work_graph_run wgr "
+        "JOIN task t ON wgr.task_id = t.task_id WHERE wgr.run_id = ?", (run_id,)
+    ).fetchone()
+    task_id, project_id = row
+    conn.execute(
+        "INSERT INTO execution_event (event_id, project_id, task_id, run_id, event_type, payload) "
+        "VALUES (?, ?, ?, ?, 'skill_iterated', ?)",
+        (str(uuid.uuid4()), project_id, task_id, run_id, json.dumps({
+            "node_id": node_id, "skill_id": skill_id, "iteration_id": iteration_id,
+        })),
+    )
+    conn.commit()
+
+
+def _skill_iterated_events(conn, run_id):
+    rows = conn.execute(
+        "SELECT payload FROM execution_event WHERE run_id = ? AND event_type = 'skill_iterated' "
+        "ORDER BY created_at", (run_id,)
+    ).fetchall()
+    return [json.loads(p) for (p,) in rows]
+
+
+_MAX_SKILL_FIXES_PER_RUN_PER_SKILL = 2  # §5.7 point 5: cap real Git rewrites of the same Skill within one run
+
+
+def _fix_skill_for_node(conn, run_id, target_node_id, graph_manifest, build_prompt, outputs, feedback, target_folder_path, timeout=300):
+    """Fix-Skill (org-graph-unification-and-universal-self-planning-
+    requirements.md §3.1 step 3): a real, Git-versioned Skill rewrite, not a
+    per-run prompt patch -- calls the existing but (until now) never
+    actually invoked by production code skill_update.py::update_skill()
+    (see design doc §5's 'existing fact B': org_level_loop.py's
+    _ROUTE_FOR_CATEGORY has no 'skill_defect' key, so this capability has
+    sat unused since it was built). Always runs under the Validator
+    identity (user-confirmed 2026-08-12 via AskUserQuestion): letting a
+    node's own org_node rewrite its own Skill risks reproducing the exact
+    blind spot that caused the failure.
+
+    §5.7 point 5 rate limit: if this (run_id, skill_id) pair has already
+    been rewritten _MAX_SKILL_FIXES_PER_RUN_PER_SKILL times this run, skip
+    the rewrite silently (the caller's dynamic_loop_back still proceeds as
+    a plain re-run of target_node_id -- not touching the Skill again is a
+    safe degradation, not a failure)."""
+    target_node = next(n for n in graph_manifest["nodes"] if n["node_id"] == target_node_id)
+    skill_id = target_node.get("skill_ref")
+    if not skill_id:
+        return None  # nothing declared for this node to attribute the fix to -- see caller's docstring
+
+    prior_fixes = sum(1 for e in _skill_iterated_events(conn, run_id) if e["skill_id"] == skill_id)
+    if prior_fixes >= _MAX_SKILL_FIXES_PER_RUN_PER_SKILL:
+        return None
+
+    row = conn.execute("SELECT task_id FROM work_graph_run WHERE run_id = ?", (run_id,)).fetchone()
+    task_id = row[0]
+    goal_text = conn.execute("SELECT goal_text FROM task WHERE task_id = ?", (task_id,)).fetchone()[0]
+    current_content = _read_skill_content(skill_id)
+
+    prompt = _SKILL_REWRITE_PROMPT.format(
+        goal_text=goal_text, skill_id=skill_id, current_content=current_content,
+        what_was_asked=build_prompt(target_node, outputs),
+        what_it_produced=(outputs.get(target_node_id) or {}).get("content", ""),
+        failure_feedback="; ".join(feedback or []),
+        loop_history=_loop_history_prompt_suffix(conn, run_id, target_node_id) or "(none)",
+    )
+
+    is_self = paths.is_self_target(target_folder_path)
+    node_exec_id = "fixskill-" + uuid.uuid4().hex[:12]
+    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)
+        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")
+            work_node_run.merge_worktree(target_folder_path, node_exec_id)
+    finally:
+        if worktree_path:
+            work_node_run.remove_worktree(target_folder_path, worktree_path)
+
+    # skill_query._REPO_ROOT (graphlemon's own checkout), NOT target_folder_path
+    # -- the Skill library is graphlemon's own platform asset (05 §5.4.1.4
+    # precedent), independent of which target project this task is working on.
+    bug_result = bug_record.create_bug_record(
+        conn, task_id=task_id,
+        description="Skill {} produced a repeated defective output at node {}".format(skill_id, target_node_id),
+        root_cause_category="skill_defect", root_cause_detail=rewrite_result["change_summary"],
+        suggested_route="full_replan",
+    )
+    iteration = skill_update.update_skill(
+        conn, skill_id, bug_result["bug_id"], rewrite_result["new_content"],
+        rewrite_result["change_summary"], repo_path=skill_query._REPO_ROOT,
+    )
+    _log_skill_iterated_event(conn, run_id, target_node_id, skill_id, iteration["iteration_id"])
+    return iteration
+
+
 _DEFAULT_LOOP_LIMIT = 3  # only reached if even the "default" config_id row is unseeded
  
  
@@ -528,6 +847,23 @@ def _resolve_loop_limit(conn, org_node, graph_manifest, source_node_id):
     return _DEFAULT_LOOP_LIMIT
  
  
+def _resolve_total_loop_limit(conn, org_node, source_node_id):
+    """Phase 2 §5.7 point 3: total dynamic_loop_back attempts for
+    source_node_id, regardless of which target each attempt picked --
+    guards against dynamic_loop_back dodging _resolve_loop_limit's
+    per-(target,source) cap by picking a different target each round.
+    Lookup: a settings_config.py row keyed "{org_node}:{source_node_id}:total"
+    (config_id has no format constraint, this key shape needs no schema
+    change), then "default" (shared with _resolve_loop_limit's own default
+    row -- deliberately the SAME fallback chain/value, so a source_node_id
+    that never uses dynamic_loop_back sees an unrestrictive default here)."""
+    for config_id in ("{}:{}:total".format(org_node, source_node_id), "default"):
+        config = settings_config.get_loop_limit_config(conn, config_id)
+        if config:
+            return config["max_loop_attempts"]
+    return _DEFAULT_LOOP_LIMIT
+
+
 def _record_loop_back_event(conn, run_id, sig, attempt_number):
     """Persists one 'loop_back_triggered' execution_event row -- this IS
     the loop history (2026-08-08 design §7): no separate table, so a
@@ -567,6 +903,13 @@ def _loop_attempt_count_from_events(conn, run_id, target_node_id, source_node_id
     )
  
  
+def _loop_attempt_count_from_events_total(conn, run_id, source_node_id):
+    """Real count of every prior loop_back triggered by source_node_id,
+    regardless of target -- see _resolve_total_loop_limit's own docstring
+    for why this differs from _loop_attempt_count_from_events."""
+    return sum(1 for e in _loop_back_events(conn, run_id) if e["source_node_id"] == source_node_id)
+
+
 def _loop_history_prompt_suffix(conn, run_id, node_id):
     """When node_id is about to be (re-)executed, append the full history
     of every prior time it was sent back for revision (not just the most
@@ -814,12 +1157,24 @@ def _run_layers(
                 conn, run_id, sig.target_node_id, sig.source_node_id
             ) + 1
             limit = _resolve_loop_limit(conn, org_node, graph_manifest, sig.source_node_id)
-            if attempt_number > limit:
+            # Phase 2 §5.7 point 3: attempt_number/limit above count only
+            # THIS (target, source) pair -- dynamic_loop_back can pick a
+            # DIFFERENT target each round, which would dodge that per-pair
+            # limit indefinitely. total_attempts/total_limit count every
+            # loop_back triggered by this source_node_id regardless of
+            # target. For retry_self/static loop_back (always the same
+            # target every time), total_attempts == attempt_number by
+            # construction, so this is a no-op there -- only dynamic_loop_back
+            # can actually make them diverge.
+            total_attempts = _loop_attempt_count_from_events_total(conn, run_id, sig.source_node_id) + 1
+            total_limit = _resolve_total_loop_limit(conn, org_node, sig.source_node_id)
+            if attempt_number > limit or total_attempts > total_limit:
                 conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
                 conn.commit()
                 raise ConsistencyCheckFailedError(
-                    "loop_back from {} to {} exceeded max_attempts={}".format(
-                        sig.source_node_id, sig.target_node_id, limit
+                    "loop_back from {} to {} exceeded max_attempts={} (per-target={}/{}, total={}/{})".format(
+                        sig.source_node_id, sig.target_node_id, max(limit, total_limit),
+                        attempt_number, limit, total_attempts, total_limit,
                     ),
                     node_id=sig.source_node_id,
                 )
@@ -849,17 +1204,34 @@ def _run_one_node(
     node_id -- see that function's own docstring for why (a loop_back
     target must never collide with, and short-circuit against, its own
     stale pre-loop_back 'done' row; every other node keeps resuming
-    exactly like before)."""
+    exactly like before).
+
+    2026-08-12 (Phase 2 §4, node-level subagent/skill binding): a node can
+    declare its own `subagent_type`/`skill_ref`, overriding the run-level
+    `org_node`/`skill_id` defaults this whole function used to use
+    unconditionally for every node. A node with neither field (every
+    existing graph_manifest -- Planner's fixed chain, Coder's synthetic
+    single-node manifest) behaves identically to before: `node_org_node`/
+    `node_skill_id` just equal the run-level defaults."""
     node_id = node["node_id"]
+    node_org_node = node.get("subagent_type", org_node)
+    node_skill_id = node.get("skill_ref", skill_id)
+    # run_graph() only ensures the RUN-LEVEL default skill_id exists as a
+    # placeholder row (work_node_execution.skill_ref's FK target) -- a
+    # node declaring its own skill_ref needs the same guarantee, or the
+    # INSERT below fails FOREIGN KEY constraint the first time that
+    # specific skill_id is ever seen. Idempotent, cheap to call per node.
+    if node_skill_id != skill_id:
+        ensure_placeholder_skill(conn, node_skill_id, node_org_node)
     exec_node_id_str = _next_exec_node_id(conn, run_id, node_id, graph_manifest)
  
     node_exec_id = _ensure_node_row(
-        conn, run_id, exec_node_id_str, org_node, skill_id,
+        conn, run_id, exec_node_id_str, node_org_node, node_skill_id,
         "work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
         {"node_id": node_id},
     )
     prompt = build_prompt(node, outputs) + _loop_history_prompt_suffix(conn, run_id, node_id)
-    run_fn = _make_run_fn(conn, target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout)
+    run_fn = _make_run_fn(conn, target_folder_path, node_exec_id, node_org_node, prompt, output_schema, timeout)
     output = work_node_run.run_work_node(
         conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn,
         needs_worktree=needs_worktree, repo_path=target_folder_path, sleep=sleep, now=now,
@@ -868,7 +1240,7 @@ def _run_one_node(
     judgment = node.get("judgment")
     has_own_judgment = bool(judgment) and judgment.get("kind", "none") != "none"
     resolved_verify_fn = _resolve_node_verify_fn(
-        node, verify_fn, org_node, target_folder_path, mechanical_check_registry
+        node, verify_fn, node_org_node, target_folder_path, mechanical_check_registry
     )
     if resolved_verify_fn is None:
         return output
@@ -881,7 +1253,7 @@ def _run_one_node(
         # every node, once redundantly), regardless of any on_fail field
         # (legacy nodes don't have one).
         return _verify_and_maybe_regenerate(
-            conn, run_id, org_node, skill_id, target_folder_path, node, node_id, prompt,
+            conn, run_id, node_org_node, node_skill_id, target_folder_path, node, node_id, prompt,
             output, input_schema, output_schema, timeout, needs_worktree, resolved_verify_fn, sleep, now,
         )
  
@@ -899,6 +1271,26 @@ def _run_one_node(
             ),
             node_id=node_id,
         )
+    if action == "dynamic_loop_back":
+        # Phase 2 §5: Reflection-review -> [Fix-Skill] -> Re-loop. See
+        # _judge_root_cause_node/_fix_skill_for_node docstrings for the
+        # full four-step methodology (org-graph-unification-and-universal-
+        # self-planning-requirements.md §3.1).
+        ancestors = sorted(_transitive_ancestors(graph_manifest, node_id) - {node_id})
+        diagnosis = _judge_root_cause_node(
+            conn, run_id, node_id, feedback, ancestors, graph_manifest, build_prompt, outputs,
+            target_folder_path,
+        )
+        target = diagnosis["target_node_id"]
+        if target not in ancestors:
+            raise ValueError(
+                "dynamic_loop_back judge picked {!r}, not a real ancestor of {}".format(target, node_id)
+            )
+        if diagnosis["defect_kind"] == "skill_defect":
+            _fix_skill_for_node(
+                conn, run_id, target, graph_manifest, build_prompt, outputs, feedback, target_folder_path,
+            )
+        raise _LoopBackRequested(source_node_id=node_id, target_node_id=target, feedback=feedback)
     target = node_id if action == "retry_self" else _on_failure_target(graph_manifest, node_id)
     raise _LoopBackRequested(source_node_id=node_id, target_node_id=target, feedback=feedback)