From 86fc0bbcf7d197cea813929a56782198437cb30e Mon Sep 17 00:00:00 2001 From: bboonnee Date: Wed, 12 Aug 2026 20:48:12 +0800 Subject: [PATCH] Phase 2a: node-level subagent/skill binding + dynamic_loop_back (Reflection-review + Fix-Skill) --- roles/validator/role.json | 3 ++- skills/skill-defect-remediation/SKILL.md | 37 +++++++++++++++++++++++++++++++++++++ src/test_work_graph_orchestrate.py | 314 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- src/work_graph_orchestrate.py | 410 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------- 4 files changed, 753 insertions(+), 11 deletions(-) create mode 100644 skills/skill-defect-remediation/SKILL.md diff --git a/roles/validator/role.json b/roles/validator/role.json index cdd6ec5..46c1309 100644 --- a/roles/validator/role.json +++ b/roles/validator/role.json @@ -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": [] diff --git a/skills/skill-defect-remediation/SKILL.md b/skills/skill-defect-remediation/SKILL.md new file mode 100644 index 0000000..d31a063 --- /dev/null +++ b/skills/skill-defect-remediation/SKILL.md @@ -0,0 +1,37 @@ +--- +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 迭代唯一的人类可读说明,写清楚比写简短更重要。 diff --git a/src/test_work_graph_orchestrate.py b/src/test_work_graph_orchestrate.py index 07e2599..ee1a095 100644 --- a/src/test_work_graph_orchestrate.py +++ b/src/test_work_graph_orchestrate.py @@ -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() diff --git a/src/work_graph_orchestrate.py b/src/work_graph_orchestrate.py index e5c0166..1501e38 100644 --- a/src/work_graph_orchestrate.py +++ b/src/work_graph_orchestrate.py @@ -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) -- libgit2 0.21.4