From a33aa8cf0d37aca7ddd92fe67e84ca91b738e1b2 Mon Sep 17 00:00:00 2001 From: bboonnee Date: Thu, 6 Aug 2026 13:47:07 +0800 Subject: [PATCH] F: three-perspective jury as a real Work Graph fan-out --- docs/todo/graph-engineering-open-items.md | 6 +++--- docs/todo/verifier-fan-out-jury-design.md | 70 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/org_level_loop.py | 21 ++++++++++++--------- src/root_cause_trace.py | 121 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------- src/scheduler.py | 12 +++++++++++- src/task_execution.py | 8 +++++++- src/test_org_level_loop.py | 80 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++-------------------- src/test_root_cause_trace.py | 183 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- src/test_scheduler.py | 18 +++++++++++++----- 9 files changed, 472 insertions(+), 47 deletions(-) create mode 100644 docs/todo/verifier-fan-out-jury-design.md diff --git a/docs/todo/graph-engineering-open-items.md b/docs/todo/graph-engineering-open-items.md index a2a0fc4..ee64f6f 100644 --- a/docs/todo/graph-engineering-open-items.md +++ b/docs/todo/graph-engineering-open-items.md @@ -90,7 +90,7 @@ ## F. 三视角评审团机制要不要推广到其他验证点 -**状态**：✅ 已确认，且明确了实现方式 +**状态**：✅ 已实现（2026-08-06）→ 需求文档 `docs/todo/verifier-fan-out-jury-requirements.md` + 设计文档 `docs/todo/verifier-fan-out-jury-design.md` **来源**：综述版对照 §7 开放问题 4 + 14 步版对照 §3 开放问题 4（同一主题） @@ -98,7 +98,7 @@ **确认结果（2026-08-05）**：要做，而且明确了实现方式——**不是**继续用现在 `root_cause_trace.judge_root_cause` 那种 Python 层直接顺序调用三个 `perspective_fn` 的写法，而是把三视角评审团**做成一张真正的 Work Graph fan-out/fan-in 菱形拓扑**：fan-out 出 N 个独立的验证节点（每个节点是一次独立的 `claude -p` 调用，带自己的视角提示词），fan-in 节点做投票/汇总。用户原话："三视角评审团可以做，做成 work graph 的 fan out 图来实现吧"。 -**待办**：✅ 需求文档已产出 → `docs/verifier-fan-out-jury-requirements.md`。明确标注**强依赖 G**——没有并行执行能力，把 `judge_root_cause` 改成 Work Graph fan-out 反而更慢（多了 Work Node 调度开销，没有并行收益），建议排在 G 之后。 +**待办**：✅ 已实现（2026-08-06）。汇总节点没有做成第四个 Work Node——投票是纯 `zero_token` 聚合（跟 E 里 `edge_type` 的区分直接对应），不需要 Agent 推理，做成普通 Python 函数在 `run_graph()` 返回后直接对 `outputs` 计数，不产生额外 `claude -p` 调用/token 成本，设计文档第 1 节记录了这个取舍。实现内容：`root_cause_trace.py` 新增 `_tally_perspectives()`（>=2/3 判定规则从 `judge_root_cause` 抽出来，闭包版和 Work Graph 版共用同一份规则，不是两份容易失配的实现）+ `judge_root_cause_via_work_graph()`（N 个视角节点用真实 schema 约束的结构化输出，不再像 `real_validator_perspective` 那样手工解析"CATEGORY:/DETAIL:"文本行）。闭包版 `judge_root_cause`/`real_validator_perspective` **保留**，不废弃（轻量备选，见设计文档第 4 节）。`org_level_loop.run_fix_loop`（唯一真实生产调用方）已迁移到 Work Graph 版本，fixer/verify 两次调用保持不变的 flat `trigger_org_node` 调用。**顺带修了一个真实的潜在 bug**：`judge_root_cause_via_work_graph` 是第一个真正触发 `run_graph()` 多节点并发层的生产代码路径（之前所有真实 `graph_manifest` 要么单节点要么严格串行链，G 的并发调度从未在生产里真正跑过），暴露出 `scheduler.py::run_goal_to_completion`/`task_execution.py::start_task_execution_in_background` 的 `schema.connect(db_path)` 都没传 `check_same_thread=False`——已一并修复。测试改造范围（设计文档第 5 节标注的最高风险点）：`test_org_level_loop.py`/`test_scheduler.py` 的 3 个测试把 mock 边界从 `org_node_trigger.trigger_org_node`（拦截"CATEGORY:"提示词）改成 `work_graph_execute.write_workflow_script`/`execute_work_graph`。新增测试证明验收标准 1-3（三次真实独立 Work Node 执行、判定规则与闭包版等价、每个视角节点的 `input_payload` 不含其他视角内容）+ 一个直接验证并发重叠的测试（`ConcurrentPerspectiveExecutionTest`）。`python3 -m unittest discover -s src`（586 tests）+ `discover -s supervisor`（125 tests）全绿，零回归。FR-4（`verify_fn` 通用对接其他验证点）本轮不实现，标注"应该"优先级，留待后续。 --- @@ -134,7 +134,7 @@ |---|---|---|---|---|---| | **G** 并行执行 | Work Graph 层内并发、层间屏障 | ✅ **已实现**（`bc0eaa5`，575 测试全绿） | M | 已完成 | 无 | | **E** 动态 Work Graph 设计能力 | 让 Coder/Validator 也能规划自己的图，Planning 提示词识别可并行节点 | ✅ **已实现**（2026-08-06，580 测试全绿） | L | 已完成 | G（已满足） | -| **F** 三视角评审团 fan-out | `judge_root_cause` 从 Python 函数改成 Work Graph 菱形拓扑 | 设计已产出 → `docs/todo/verifier-fan-out-jury-design.md`，实现中 | M | P0（G/E 都已落地，正在推进） | 无（G 已满足） | +| **F** 三视角评审团 fan-out | `judge_root_cause` 从 Python 函数改成 Work Graph 菱形拓扑 | ✅ **已实现**（2026-08-06，586 测试全绿） | M | 已完成 | 无（G 已满足） | | **B** loop-until-dry Skill | Validator 未知规模扫描收敛能力 | 仅需求 | S/M | P2 | 无 | | **H** Role Package 抽象层 | 统一 role/context/skills/applications/workflows/permissions/acceptance 打包格式 | 仅需求 | XL | P2（战略级，不阻塞其他项） | workflows 字段依赖 E 的产出 | | **D** Time-travel 调试 | 回到历史节点分叉重跑 | 仅需求 | L | P3 | 无强依赖，但价值随图变复杂（E 落地后）才会显现 | diff --git a/docs/todo/verifier-fan-out-jury-design.md b/docs/todo/verifier-fan-out-jury-design.md new file mode 100644 index 0000000..aac956f --- /dev/null +++ b/docs/todo/verifier-fan-out-jury-design.md @@ -0,0 +1,70 @@ +# 三视角评审团——Work Graph Fan-out 实现——详细设计（SDD） + +**日期**: 2026-08-06 +**性质**: 详细设计，对应需求文档 `docs/todo/verifier-fan-out-jury-requirements.md`（条目 F，已确认）。回答该需求文档第 8 节的开放问题。 +**依赖代码**：`src/root_cause_trace.py`（`judge_root_cause`/`_PERSPECTIVE_FRAMINGS`）、`src/org_level_loop.py`（`run_fix_loop`，唯一真实生产调用方）、`src/work_graph_orchestrate.py`（G 已实现的 `run_graph()`/`topological_layers()`）。 +**状态**: 设计草案，待评审后实施。 + +## 1. 汇总节点不是 Work Node——这是"zero_token 边"原则的直接应用 + +需求文档背景里"1 个汇总节点"的表述容易被误读成"汇总也要是一次 `claude -p` 调用"。实际不需要：`>=2/3` 投票本身是纯粹的计数聚合，不需要 Agent 推理——这正是 E 设计文档里强调的 `edge_type: zero_token` 场景本身。汇总逻辑保持为一个普通 Python 函数，在 `run_graph()` 返回之后，在调用方所在的进程里直接对 `outputs` 做计数，不产生额外的 `claude -p` 调用、不产生额外 token 成本（呼应需求文档第 7 节风险"token 成本上升"的顾虑，此处直接不引入这块新增成本）。 + +验收标准 1 要求"能在 execution_event/日志里看到三次独立的节点执行记录"——这只要求**三个视角节点**是真 Work Node（`run_graph()`/`run_work_node()` 本来就会为每个节点写 `node_start`/`node_completed` 日志），不要求汇总也是。 + +## 2. 判定规则复用，不重写（FR-3） + +`root_cause_trace.judge_root_cause` 现有的 `>=2/3` 计数逻辑被抽成一个共享私有函数 `_tally_perspectives(judgments)`，`judge_root_cause`（保留，闭包版本）和新的 `judge_root_cause_via_work_graph`（Work Graph 版本）都调用它——两套判定规则物理上是同一份代码，不是两份逻辑各自实现、容易失配。`judgments` 的 dict 形状（`{"category", "detail"}`）两个版本完全一致，`_tally_perspectives` 不需要知道调用方是闭包还是 Work Node 输出。 + +## 3. `judge_root_cause_via_work_graph`：新的 Work Graph 版本 + +```python +def judge_root_cause_via_work_graph( + conn, task_id, target_folder_path, bug_description, context, + org_node="validator", skill_id="qa-task-validation", timeout=300, +): +``` + +内部结构： + +1. `_PERSPECTIVE_FRAMINGS`（已存在，三个视角的 framing 文本）驱动构造每个视角的完整提示词（复用现有 framing 文本，不改视角内容本身）。 +2. 构造一个菱形（严格说是"N 节点单层，无边"）`graph_manifest`：`nodes = [{"node_id": "perspective-" + name, "depends_on": []} for name in perspectives]`，`edges = []`——三个节点互不依赖，`topological_layers()` 天然把它们分进同一层，G 自动并发调度，不需要额外的调度代码。 +3. 像 `e2e_pipeline.py::implement_code`/`validate_implementation_with_own_graph` 一样，手动 `INSERT INTO work_graph_run (..., status='running')`（不经过 `plan_work_graph()`——这个图的形状是固定的、代码已知的，不需要一次额外的 LLM 规划调用，同样是"机制用代码"原则的应用）。 +4. `run_graph()` 执行：每个视角节点用**真实 schema 约束的结构化输出**（`{"type": "object", "required": ["category", "detail"]}`），不再像现有 `real_validator_perspective` 那样手工解析"CATEGORY:/DETAIL:"文本行——`work_graph_orchestrate._make_run_fn` 走的是单节点 dynamic-workflow 脚本机制（`agent(prompt, {{ agentType, schema }})`），这条路径已经被 2026-08-01 的真实调用验证过 `structured_output` 真的能拿到（`work_graph_orchestrate.py` 模块 docstring 有记录），跟 `work_graph_plan.py` 那条"`--json-schema` 对 `--agent` 无效"的已知限制是两回事，不受它影响。 +5. `_tally_perspectives([outputs["perspective-" + name] for name in perspectives])`，返回值形状跟现有 `judge_root_cause` 完全一致（`{"root_cause_category", "root_cause_detail", "judgments"}`），调用方不需要区分。 + +每个视角节点获得独立、干净的上下文（验收标准 3）：`_ensure_node_row`/`run_work_node` 本身就是每个 node_exec_id 一次全新的 workflow 脚本渲染 + 独立子进程调用，`input_payload` 只含 `{"node_id": ...}`，视角提示词通过 `build_prompt` 的闭包参数传入，节点之间不共享对话历史——这是 `run_graph()` 现有机制的天然属性，不需要额外设计。 + +## 4. `org_level_loop.run_fix_loop` 迁移（FR-5，唯一真实生产场景） + +现状： + +```python +perspective_fns = [root_cause_trace.real_validator_perspective(...) for name in (...)] +judgment = root_cause_trace.judge_root_cause(perspective_fns) +``` + +改为： + +```python +judgment = root_cause_trace.judge_root_cause_via_work_graph( + conn, task_id, target_folder_path, bug_description, context, +) +``` + +`real_validator_perspective`/闭包版 `judge_root_cause` **不删除**——需求文档第 8 节开放问题已问过"要不要保留作为轻量备选"，答案是保留：`src/test_root_cause_trace.py` 现有单测（`JudgeRootCauseTest`/`RealValidatorPerspectiveParsingTest`）覆盖的是判定规则本身的正确性，这些测试不需要真实 `claude` 子进程调用就能跑（用纯 Python 闭包 mock），继续作为规则正确性的快速回归保障（`_tally_perspectives` 被两边共用，这些测试事实上也在验证 Work Graph 版本的判定逻辑）；闭包版本本身作为"不需要 worktree/Work Graph 开销的场景"的备选实现继续存在，不强制废弃。 + +## 5. 测试改造范围（真实影响面，如实记录） + +`src/test_org_level_loop.py` 现有 3 个测试全部通过 mock `org_node_trigger.trigger_org_node`（拦截"CATEGORY:"提示词）来伪造三视角判断——这个 mock 边界在迁移后失效（新实现走 `work_graph_execute.execute_work_graph`，不再调用 `trigger_org_node` 做视角判断）。必须同步把这三个测试的 mock 边界改成 `work_graph_execute.write_workflow_script`/`execute_work_graph`（跟 `test_work_graph_orchestrate.py` 现有模式一致），`org_node_trigger.trigger_org_node` 的 mock 保留但只用于 fixer/verify 两次调用（迁移范围明确不包括这两处）。这是本次改动风险最高的一步——`run_fix_loop` 是真实生产闭环的一部分，测试改造必须逐条比对新旧断言等价，不能只是"能跑通"就算数。 + +## 6. 不实现的部分（诚实记录，对应需求文档非目标 + FR-4 的"应该"优先级） + +- **FR-4（`verify_fn` 通用对接）不在本轮实现**：把评审团机制包装成一个可以传给 `run_graph(verify_fn=...)` 的工厂函数，用于升级 `_check_prd_consistency` 这类现有单一验证器场景——需求文档标注"应该"而非"必须"，且"哪些验证点该升级成评审团"本身是待定的判定标准（需求文档第 8 节开放问题未回答）。本轮只交付 FR-5（根因溯源场景本身）这一个"必须"级别的真实落地点，`judge_root_cause_via_work_graph` 的函数签名设计上保持通用（视角 framing/数量可参数化），为未来接 FR-4 留出空间，但不在本轮消费它。 +- **视角节点并发时间收益的实测**（验收标准 4）：G 已实现，`topological_layers()` 会自动把三个视角节点分进同一层走 `ThreadPoolExecutor`，机制上必然并发；但"在真实 `claude -p` 调用下测出时间重叠"需要 `RUN_REAL_AGENT_E2E=1` 真实调用，不在 mock 测试范围内验证，留给后续真实运行观察确认（跟 G 自己的 `RunGraphConcurrencyTest` 用 mock+`time.sleep` 验证并发的方式已经覆盖了"机制确实并发"这件事，不需要为 F 重新证明一遍 G 的能力）。 + +## 7. 回答需求文档第 8 节开放问题 + +- `judge_root_cause`（闭包版）保留作为轻量备选，不废弃（第 4 节）。 +- FR-4 接口设计：本轮不实现，见第 6 节。 +- 除根因溯源外还有哪些验证点该升级：本轮不判定，留待 FR-4 一起决定。 +- 视角数量 N 是否可配置：`judge_root_cause_via_work_graph`/`_tally_perspectives` 设计上已经是对任意数量视角通用的（不硬编码 3），根因溯源场景固定传 3 个，未来场景可以传不同数量而不需要改动这两个函数本身。 diff --git a/src/org_level_loop.py b/src/org_level_loop.py index f6367a3..740b479 100644 --- a/src/org_level_loop.py +++ b/src/org_level_loop.py @@ -15,15 +15,18 @@ def run_fix_loop(conn, task_id, target_folder_path, bug_description, context, fi "verify_response"}. Raises root_cause_trace.InconclusiveEscalateError if the 3 perspectives don't reach 2/3 agreement (same as U-015 -- this function doesn't add its own fallback for that, matching 04-P's own - "升级人工" answer to disagreement).""" - perspective_fns = [ - root_cause_trace.real_validator_perspective( - name, bug_description, context, target_folder_path, - trigger_fn=org_node_trigger.trigger_org_node, timeout=timeout, - ) - for name in ("correctness", "requirement_completeness", "skill_defect") - ] - judgment = root_cause_trace.judge_root_cause(perspective_fns) + "升级人工" answer to disagreement). + + **F (2026-08-06, docs/todo/verifier-fan-out-jury-design.md)**: root + cause judgment now runs as a real Work Graph fan-out (each perspective + is its own Work Node with its own clean context) instead of three + sequential Python closures calling org_node_trigger.trigger_org_node -- + see root_cause_trace.judge_root_cause_via_work_graph. The fixer/verify + calls below are unchanged flat trigger_org_node calls; only root-cause + judgment moved.""" + judgment = root_cause_trace.judge_root_cause_via_work_graph( + conn, task_id, target_folder_path, bug_description, context, timeout=timeout, + ) category = judgment["root_cause_category"] route = _ROUTE_FOR_CATEGORY.get(category, "full_replan") diff --git a/src/root_cause_trace.py b/src/root_cause_trace.py index 0c09434..72f616d 100644 --- a/src/root_cause_trace.py +++ b/src/root_cause_trace.py @@ -14,11 +14,27 @@ perspectives 04-P names (正确性视角/需求完整性视角/Skill 系统性 concrete question each perspective asks is a prompt-engineering concern for the caller (see real_validator_perspective below for the real-call version), not this module's. + +**F (2026-08-06, docs/todo/verifier-fan-out-jury-design.md)**: judge_root_cause +above is the closure-based implementation -- each perspective is a Python +function call, sequentially, in the SAME process/context. Kept as-is (not +replaced): it's the cheap path for scenarios that don't need worktree/Work +Graph overhead, and its already-proven >=2/3 rule is now shared (via +_tally_perspectives) with judge_root_cause_via_work_graph below, the real +Work Graph fan-out version -- each perspective becomes an independent Work +Node (own claude -p call, own clean context, own execution_event log +entries), scheduled by G's topological_layers()/run_graph() same as any +other graph_manifest. The fan-in vote itself stays plain Python (a +zero_token aggregation per the design-dynamic-work-graph Skill's own +edge_type distinction -- counting doesn't need an Agent call), not a +fourth Work Node. """ +import json +import uuid from collections import Counter -from src import bug_record +from src import bug_record, work_graph_orchestrate class InconclusiveEscalateError(Exception): @@ -31,6 +47,19 @@ class InconclusiveEscalateError(Exception): super().__init__("perspectives did not reach 2/3 agreement: " + repr(judgments)) +def _tally_perspectives(judgments): + """>=2/3 agreement rule (04-P: "至少 2/3 视角一致才定案") -- shared by both + judge_root_cause (closures) and judge_root_cause_via_work_graph (Work + Graph fan-out), so the rule is one piece of code, not two that could + drift apart. `judgments`: list of {"category", "detail"}.""" + counts = Counter(j["category"] for j in judgments) + winner, count = counts.most_common(1)[0] + if count * 2 < len(judgments): # strictly "at least 2/3", not just plurality + raise InconclusiveEscalateError(judgments) + detail = " | ".join(j["detail"] for j in judgments if j["category"] == winner) + return {"root_cause_category": winner, "root_cause_detail": detail, "judgments": judgments} + + def judge_root_cause(perspective_fns): """Runs all perspective_fns, requires >=2/3 to agree on the same category (04-P guard: "至少 2/3 视角一致才定案"). Returns @@ -38,12 +67,7 @@ def judge_root_cause(perspective_fns): raises InconclusiveEscalateError otherwise. """ judgments = [fn() for fn in perspective_fns] - counts = Counter(j["category"] for j in judgments) - winner, count = counts.most_common(1)[0] - if count * 2 < len(judgments): # strictly "at least 2/3", not just plurality - raise InconclusiveEscalateError(judgments) - detail = " | ".join(j["detail"] for j in judgments if j["category"] == winner) - return {"root_cause_category": winner, "root_cause_detail": detail, "judgments": judgments} + return _tally_perspectives(judgments) def trace_root_cause_test_failed(conn, task_id, description, perspective_fns, suggested_route, related_artifact_ids=None): @@ -126,3 +150,86 @@ def real_validator_perspective(perspective_name, bug_description, context, targe return {"category": category, "detail": detail or response} return perspective_fn + + +_PERSPECTIVE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]} +_PERSPECTIVE_OUTPUT_SCHEMA = {"type": "object", "required": ["category", "detail"]} + + +def _perspective_prompt(perspective_name, bug_description, context): + framing = _PERSPECTIVE_FRAMINGS[perspective_name] + return ( + "{framing}\n\nBug: {bug_description}\n\nContext:\n{context}\n\n" + "Reply with your judgment: category must be exactly one of 'planning', 'coder', or " + "'skill_defect'; detail is one sentence citing the specific SDD section or code " + "location that supports it." + ).format(framing=framing, bug_description=bug_description, context=context) + + +def _node_id_for_perspective(name): + return "perspective-" + name + + +def _build_jury_graph_manifest(perspective_names): + """FR-1: a reusable fan-out graph_manifest shape -- N mutually + independent perspective nodes, generic over any set of perspective + names (not hardcoded to root-cause's three). No fan-in node: see this + module's own docstring / design doc §1 for why the vote stays plain + Python (zero_token aggregation, not worth an Agent call).""" + return { + "nodes": [{"node_id": _node_id_for_perspective(name), "depends_on": []} for name in perspective_names], + "edges": [], + } + + +def _jury_prompt_builder(perspective_prompts): + """FR-2: perspective content is parameterized -- this scheduling code + has no idea what the perspectives actually ask, it just runs whatever + prompt each node was given.""" + def build_prompt(node, outputs): + name = node["node_id"][len("perspective-"):] + return perspective_prompts[name] + return build_prompt + + +def judge_root_cause_via_work_graph( + conn, task_id, target_folder_path, bug_description, context, + org_node="validator", skill_id="qa-task-validation", timeout=300, +): + """Work Graph fan-out version of judge_root_cause (FR-5's real landing + scenario) -- each perspective is an independent Work Node (own + `claude -p` call with real schema-constrained structured output, own + clean context -- work_graph_orchestrate._make_run_fn's dynamic- + workflow-script mechanism, validated for real 2026-08-01, see that + module's docstring) instead of a Python closure called in a loop, so + unlike real_validator_perspective there's no 'CATEGORY:'/'DETAIL:' + text parsing needed -- the node's output_payload already IS + {"category", "detail"}. Same return shape as judge_root_cause, and + raises InconclusiveEscalateError on <2/3 agreement via the same shared + _tally_perspectives call -- callers don't need to know which + implementation they're getting. + + No plan_work_graph() call: this graph's shape (N independent nodes) is + fixed and known ahead of time, same reasoning e2e_pipeline.py's + implement_code()/validate_implementation_with_own_graph() already use + for skipping an LLM planning step of their own.""" + perspective_prompts = { + name: _perspective_prompt(name, bug_description, context) for name in _PERSPECTIVE_FRAMINGS + } + manifest = _build_jury_graph_manifest(list(perspective_prompts)) + + run_id = str(uuid.uuid4()) + conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES (?, ?, ?, ?, 'running')", + (run_id, task_id, org_node, json.dumps(manifest)), + ) + conn.commit() + + outputs = work_graph_orchestrate.run_graph( + conn, run_id, org_node, skill_id, target_folder_path, manifest, + _jury_prompt_builder(perspective_prompts), _PERSPECTIVE_INPUT_SCHEMA, _PERSPECTIVE_OUTPUT_SCHEMA, + timeout=timeout, + ) + judgments = [outputs[_node_id_for_perspective(name)] for name in perspective_prompts] + return _tally_perspectives(judgments) diff --git a/src/scheduler.py b/src/scheduler.py index ffc88b6..b6fe870 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -193,7 +193,17 @@ def run_goal_to_completion(db_path, target_folder_path, goal_text): task always lands on a real terminal-ish status with a real error detail, never a caller-crashes-and-the-task-is-just-stuck outcome. """ - conn = schema.connect(db_path) + # check_same_thread=False (2026-08-06, F): this conn can reach + # work_graph_orchestrate.run_graph() via finish_task_lifecycle -> + # org_level_loop.run_fix_loop -> root_cause_trace. + # judge_root_cause_via_work_graph, whose 3-perspective layer runs on + # its OWN ThreadPoolExecutor worker threads (G), not this function's + # thread -- a plain schema.connect() here would raise "SQLite objects + # created in a thread can only be used in that same thread" the first + # time a real multi-node concurrent layer actually runs (previously + # never exercised in production: every graph_manifest before F was + # either single-node or a strictly serial chain). + conn = schema.connect(db_path, check_same_thread=False) conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS)) try: project_id = conn.execute( diff --git a/src/task_execution.py b/src/task_execution.py index f12d6c5..6795689 100644 --- a/src/task_execution.py +++ b/src/task_execution.py @@ -157,7 +157,13 @@ def start_task_execution_in_background(db_path, task_id, target_folder_path, goa started one reaches the same real terminal state. Returns the Thread (already started) so callers/tests can `.join()` it if needed.""" def run(): - conn = schema.connect(db_path) + # check_same_thread=False (2026-08-06, F): same reasoning as + # scheduler.run_goal_to_completion's own conn -- this thread's conn + # can reach work_graph_orchestrate.run_graph()'s multi-node + # concurrent layers (G) via finish_task_lifecycle -> run_fix_loop -> + # judge_root_cause_via_work_graph, whose worker threads aren't this + # one. + conn = schema.connect(db_path, check_same_thread=False) try: result = run_or_resume_task_lifecycle(conn, task_id, target_folder_path, goal_text) scheduler.finish_task_lifecycle(conn, result, target_folder_path, goal_text) diff --git a/src/test_org_level_loop.py b/src/test_org_level_loop.py index 9b2c4aa..2eb4833 100644 --- a/src/test_org_level_loop.py +++ b/src/test_org_level_loop.py @@ -3,23 +3,35 @@ fix -> re-verify -> done), mocked at the subprocess boundary (CLAUDE.md Testing constraint). Real coverage is in src/test_org_level_loop_e2e.py (RUN_REAL_AGENT_E2E=1). +**F (2026-08-06)**: root-cause judgment now runs through +root_cause_trace.judge_root_cause_via_work_graph (a real Work Graph fan-out, +see docs/todo/verifier-fan-out-jury-design.md), so the 3 perspective calls +are mocked at work_graph_execute.write_workflow_script/execute_work_graph +(same boundary as test_work_graph_orchestrate.py), not at +org_node_trigger.trigger_org_node -- that mock is now only used for the +fixer/verify calls, which are unchanged. + Run: python3 -m unittest src.test_org_level_loop -v """ import unittest from unittest.mock import patch -from src import org_level_loop, schema +from src import org_level_loop, root_cause_trace, schema, work_graph_execute + +def _make_perspective_execute(perspective_category): + """Fakes work_graph_execute.execute_work_graph for the 3 perspective + Work Nodes -- all agree on perspective_category, real schema-shaped + output (no more 'CATEGORY:'/'DETAIL:' text parsing).""" + def fake_execute(cwd, name, args, output_schema, timeout=300): + return {"category": perspective_category, "detail": "because reasons"} + return fake_execute -def _make_trigger(perspective_category, fix_ok=True): - """Fakes org_node_trigger.trigger_org_node for the whole loop: 3 - perspective calls (all agreeing on perspective_category), one fixer - call, one verify call.""" +def _make_trigger(fix_ok=True): + """Fakes org_node_trigger.trigger_org_node for the fixer/verify calls only.""" def fake(org_node, prompt, target_folder_path, timeout=300): - if "CATEGORY:" in prompt: - return "CATEGORY: {}\nDETAIL: because reasons".format(perspective_category) if org_node in ("coder", "planner"): return "Fixed it." if org_node == "validator": @@ -31,7 +43,9 @@ def _make_trigger(perspective_category, fix_ok=True): class RunFixLoopTestBase(unittest.TestCase): def setUp(self): - self.conn = schema.connect(":memory:") + # check_same_thread=False: the 3 root-cause perspective nodes form + # one concurrent layer under G, run via ThreadPoolExecutor. + self.conn = schema.connect(":memory:", check_same_thread=False) schema.create_schema(self.conn) self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " @@ -46,13 +60,21 @@ class RunFixLoopTestBase(unittest.TestCase): def tearDown(self): self.conn.close() + def _run_fix_loop(self, perspective_category, fix_ok=True): + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object( + work_graph_execute, "execute_work_graph", + side_effect=_make_perspective_execute(perspective_category), + ), \ + patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=fix_ok)): + return org_level_loop.run_fix_loop( + self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it" + ) + class CoderCategoryFastLaneTest(RunFixLoopTestBase): def test_coder_category_routes_fast_lane_and_reaches_done(self): - with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("coder", fix_ok=True)): - result = org_level_loop.run_fix_loop( - self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it" - ) + result = self._run_fix_loop("coder", fix_ok=True) self.assertEqual(result["root_cause_category"], "coder") self.assertEqual(result["route"], "fast_lane") self.assertTrue(result["verified"]) @@ -69,10 +91,7 @@ class CoderCategoryFastLaneTest(RunFixLoopTestBase): class PlanningCategoryFullReplanTest(RunFixLoopTestBase): def test_planning_category_routes_full_replan(self): - with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("planning", fix_ok=True)): - result = org_level_loop.run_fix_loop( - self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it" - ) + result = self._run_fix_loop("planning", fix_ok=True) self.assertEqual(result["route"], "full_replan") self.assertEqual(result["final_status"], "done") assigned = self.conn.execute("SELECT assigned_org_node FROM task WHERE task_id = 'task-1'").fetchone()[0] @@ -83,15 +102,36 @@ class PlanningCategoryFullReplanTest(RunFixLoopTestBase): class VerificationFailsTest(RunFixLoopTestBase): def test_fix_not_verified_leaves_task_out_of_done(self): - with patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger("coder", fix_ok=False)): - result = org_level_loop.run_fix_loop( - self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it" - ) + result = self._run_fix_loop("coder", fix_ok=False) self.assertFalse(result["verified"]) self.assertNotEqual(result["final_status"], "done") status = self.conn.execute("SELECT status FROM task WHERE task_id = 'task-1'").fetchone()[0] self.assertNotEqual(status, "done") +class InconclusiveDisagreementTest(RunFixLoopTestBase): + """Real Work Graph fan-out still surfaces InconclusiveEscalateError the + same way the closure-based version did -- run_fix_loop adds no fallback + of its own (04-P: disagreement escalates to a human, it isn't silently + routed anywhere).""" + + def test_three_way_split_raises_and_creates_no_bug_record(self): + categories = iter(["planning", "coder", "skill_defect"]) + + def fake_execute(cwd, name, args, output_schema, timeout=300): + return {"category": next(categories), "detail": "d"} + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute), \ + patch("src.org_node_trigger.trigger_org_node", side_effect=_make_trigger(fix_ok=True)): + with self.assertRaises(root_cause_trace.InconclusiveEscalateError): + org_level_loop.run_fix_loop( + self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", "fix it", "verify it" + ) + + count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0] + self.assertEqual(count, 0) + + if __name__ == "__main__": unittest.main() diff --git a/src/test_root_cause_trace.py b/src/test_root_cause_trace.py index f5ed0d9..6b5e16b 100644 --- a/src/test_root_cause_trace.py +++ b/src/test_root_cause_trace.py @@ -6,9 +6,15 @@ coverage is in src/test_root_cause_trace_e2e.py (RUN_REAL_AGENT_E2E=1). Run: python3 -m unittest src.test_root_cause_trace -v """ +import json +import os +import tempfile +import threading +import time import unittest +from unittest.mock import patch -from src import bug_record, root_cause_trace, schema +from src import bug_record, root_cause_trace, schema, work_graph_execute def _fixed(category, detail="detail"): @@ -128,5 +134,180 @@ class RealValidatorPerspectiveParsingTest(unittest.TestCase): perspective_fn() +# F (2026-08-06): Work Graph fan-out version -- each perspective becomes a +# real Work Node (mocked at the claude subprocess boundary, same as +# test_work_graph_orchestrate.py) instead of a Python closure. + +class JudgeRootCauseViaWorkGraphTestBase(unittest.TestCase): + def setUp(self): + self.home_tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.home_tmp.cleanup) + self.old_graphlemon_home = os.environ.get("GRAPHLEMON_HOME") + os.environ["GRAPHLEMON_HOME"] = self.home_tmp.name + self.addCleanup(self._restore_graphlemon_home) + + # check_same_thread=False: the three perspective nodes form one + # concurrent layer under G, run via ThreadPoolExecutor. + self.conn = schema.connect(":memory:", check_same_thread=False) + schema.create_schema(self.conn) + self.conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 'validator')" + ) + self.conn.commit() + + def _restore_graphlemon_home(self): + if self.old_graphlemon_home is None: + os.environ.pop("GRAPHLEMON_HOME", None) + else: + os.environ["GRAPHLEMON_HOME"] = self.old_graphlemon_home + + def tearDown(self): + self.conn.close() + + def _perspective_for_script(self, prompt_by_script, script_name): + content = prompt_by_script[script_name] + for name, framing in root_cause_trace._PERSPECTIVE_FRAMINGS.items(): + if framing in content: + return name + raise AssertionError("could not identify perspective for script " + script_name) + + +class RealWorkNodesPerPerspectiveTest(JudgeRootCauseViaWorkGraphTestBase): + """Acceptance criterion 1: three independent Work Node executions (real + work_node_execution rows, real execution_event logs), not three closure + calls inside one function.""" + + def test_three_perspectives_are_three_real_work_node_rows(self): + prompt_by_script = {} + + def fake_write_script(cwd, script_name, script_content): + prompt_by_script[script_name] = script_content + + def fake_execute(cwd, name, args, output_schema, timeout=600): + perspective = self._perspective_for_script(prompt_by_script, name) + return {"category": "coder", "detail": "agreed via " + perspective} + + with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + result = root_cause_trace.judge_root_cause_via_work_graph( + self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", + ) + + self.assertEqual(result["root_cause_category"], "coder") + self.assertEqual(len(result["judgments"]), 3) + + rows = self.conn.execute( + "SELECT node_id, status, output_payload FROM work_node_execution WHERE run_id = (" + " SELECT run_id FROM work_graph_run WHERE task_id = 'task-1'" + ")" + ).fetchall() + self.assertEqual( + {r[0] for r in rows}, + {"perspective-correctness", "perspective-requirement_completeness", "perspective-skill_defect"}, + ) + for node_id, status, output_payload in rows: + self.assertEqual(status, "done") + self.assertEqual(json.loads(output_payload)["category"], "coder") + + +class JudgmentRuleEquivalenceTest(JudgeRootCauseViaWorkGraphTestBase): + """Acceptance criterion 2: same >=2/3 rule, same results, as the + closure-based judge_root_cause on equivalent inputs -- both call the + same shared _tally_perspectives.""" + + def _run_with_categories(self, categories_by_perspective): + prompt_by_script = {} + + def fake_write_script(cwd, script_name, script_content): + prompt_by_script[script_name] = script_content + + def fake_execute(cwd, name, args, output_schema, timeout=600): + perspective = self._perspective_for_script(prompt_by_script, name) + return {"category": categories_by_perspective[perspective], "detail": "d-" + perspective} + + with patch.object(work_graph_execute, "write_workflow_script", side_effect=fake_write_script), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + return root_cause_trace.judge_root_cause_via_work_graph( + self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", + ) + + def test_two_of_three_agreement_wins(self): + result = self._run_with_categories( + {"correctness": "planning", "requirement_completeness": "planning", "skill_defect": "coder"} + ) + self.assertEqual(result["root_cause_category"], "planning") + + closure_result = root_cause_trace.judge_root_cause( + [_fixed("planning"), _fixed("planning"), _fixed("coder")] + ) + self.assertEqual(result["root_cause_category"], closure_result["root_cause_category"]) + + def test_three_way_split_is_inconclusive(self): + with self.assertRaises(root_cause_trace.InconclusiveEscalateError): + self._run_with_categories( + {"correctness": "planning", "requirement_completeness": "coder", "skill_defect": "skill_defect"} + ) + with self.assertRaises(root_cause_trace.InconclusiveEscalateError): + root_cause_trace.judge_root_cause([_fixed("planning"), _fixed("coder"), _fixed("skill_defect")]) + + +class IndependentCleanContextTest(JudgeRootCauseViaWorkGraphTestBase): + """Acceptance criterion 3: each perspective node's input_payload only + carries what it needs to judge -- not other perspectives' reasoning.""" + + def test_input_payload_has_no_cross_perspective_content(self): + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object( + work_graph_execute, "execute_work_graph", + return_value={"category": "coder", "detail": "d"}, + ): + root_cause_trace.judge_root_cause_via_work_graph( + self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", + ) + + rows = self.conn.execute("SELECT node_id, input_payload FROM work_node_execution").fetchall() + self.assertEqual(len(rows), 3) + for node_id, input_payload in rows: + payload = json.loads(input_payload) + self.assertEqual(payload, {"node_id": node_id}) + + +class ConcurrentPerspectiveExecutionTest(JudgeRootCauseViaWorkGraphTestBase): + """G is already in place -- three mutually-independent perspective + nodes land in one topological layer, so they run concurrently, not + queued one after another.""" + + def test_three_perspective_calls_overlap_in_wall_clock_time(self): + events = [] + events_lock = threading.Lock() + + def fake_execute(cwd, name, args, output_schema, timeout=600): + start = time.monotonic() + time.sleep(0.05) + end = time.monotonic() + with events_lock: + events.append((start, end)) + return {"category": "coder", "detail": "d"} + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_execute): + root_cause_trace.judge_root_cause_via_work_graph( + self.conn, "task-1", "/tmp/proj-1", "bug desc", "context", + ) + + self.assertEqual(len(events), 3) + # at least two of the three calls must have overlapped + overlaps = sum( + 1 for i in range(len(events)) for j in range(i + 1, len(events)) + if events[i][0] < events[j][1] and events[j][0] < events[i][1] + ) + self.assertGreater(overlaps, 0) + + if __name__ == "__main__": unittest.main() diff --git a/src/test_scheduler.py b/src/test_scheduler.py index fa7e639..d6fa2bf 100644 --- a/src/test_scheduler.py +++ b/src/test_scheduler.py @@ -377,10 +377,10 @@ class UnroutedExceptionStillBlocksTaskTest(RoutableExceptionTestBase): def _fix_loop_trigger(org_node, prompt, target_folder_path, timeout=300): """org_node_trigger.trigger_org_node fake for org_level_loop.run_fix_loop's - 3 perspective calls (all agree 'coder') + fixer + verify -- verify always - fails, so route_org_loop keeps incrementing loop_round on retries.""" - if "CATEGORY:" in prompt: - return "CATEGORY: coder\nDETAIL: because reasons" + fixer + verify calls -- verify always fails, so route_org_loop keeps + incrementing loop_round on retries. Root-cause judgment itself (F, + 2026-08-06) no longer goes through trigger_org_node -- see + _fix_loop_perspective_execute below for that boundary.""" if org_node == "coder": return "Fixed it." if org_node == "validator": @@ -388,6 +388,12 @@ def _fix_loop_trigger(org_node, prompt, target_folder_path, timeout=300): raise AssertionError("unexpected call: " + org_node) +def _fix_loop_perspective_execute(cwd, name, args, output_schema, timeout=300): + """work_graph_execute.execute_work_graph fake for run_fix_loop's 3 + root-cause perspective Work Nodes (F) -- all agree 'coder'.""" + return {"category": "coder", "detail": "because reasons"} + + class SchedulerContinuesPastBlockedTaskTest(unittest.TestCase): """REQ-05 failure / REQ-09 boundary, against the REAL scheduler (U-032) -- test_org_loop_route.py's own test for this literally says "no scheduler @@ -430,7 +436,9 @@ class SchedulerContinuesPastBlockedTaskTest(unittest.TestCase): goals = [("/tmp/proj-1", "goal-ok-1"), ("/tmp/proj-1", "goal-doomed"), ("/tmp/proj-1", "goal-ok-2")] with patch.object(scheduler.e2e_pipeline, "run_task_lifecycle", side_effect=fake_run_task_lifecycle), \ patch.object(scheduler.e2e_pipeline, "_read_artifact_content", return_value="benchmark content"), \ - patch("src.org_node_trigger.trigger_org_node", side_effect=_fix_loop_trigger): + patch("src.org_node_trigger.trigger_org_node", side_effect=_fix_loop_trigger), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=_fix_loop_perspective_execute): results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=3) by_goal = {r["goal_text"]: r for r in results} -- libgit2 0.21.4