PH4 closure audit: close 5 real REQ-01~14 gaps found before starting PH5

CLAUDE.md is explicit that PH5 (self-modification/auto-rollback) must not start until PH4 has genuinely closed -- not just "all unit tests green," but 06-验收与测试锚点.md's REQ-01~14 happy/boundary/failure cases actually covered. A REQ-by-REQ audit (the same discipline that caught U-031's integration gap) found 5 real gaps: - REQ-03 failure: work_node_execution.status jumped straight running->retrying, skipping 'reflecting' (declared in the schema enum, never once written by any code). - REQ-05 failure / REQ-09 boundary: the "scheduler keeps serving other queued tasks past one blocked_pending_human" scenario was never tested against a real scheduler -- the old test literally said "no scheduler exists yet to test against directly" (stale since U-032). - REQ-06 / REQ-07 / REQ-14 failure (the biggest gap, shared root cause): CorruptPersistedStateError, ClassifierBlockedError, and UsageLimitWaitTimeoutError were each correctly raised and each correctly marked their own work_node_execution 'blocked', but nothing ever routed that to task.status='blocked_pending_human' with the blocked_reason each REQ's acceptance text names. e2e_pipeline.run_end_to_end is split into run_task_lifecycle() so scheduler.py's run_goal_to_completion can create the task itself and catch these three specific exceptions, following the same UPDATE+execution_event pattern budget_guardrail.check_budget() already established for REQ-10. - REQ-02 boundary: PRD internal-contradiction check (verifying->reflecting) had no implementation anywhere -- artifact_write always marked validation_status='passed' unconditionally. Added an optional verify_fn hook to work_graph_orchestrate.run_graph() (one regeneration attempt, then ConsistencyCheckFailedError) and a real consistency-check agent() call scoped to the PRD node only. The first real run of this new call caught a genuine bug: the model returned "consistent": "false" as a JSON string, not a boolean, which `if ok:` silently treated as truthy -- fixed with explicit _coerce_bool(). - REQ-13 failure: minor test-assertion weakness (WS 404 rejection test didn't check the JSON error body). 384 tests green. REQ-01~14 now have real, exercised coverage rather than a "tests are green" assumption -- PH4 can be treated as genuinely closed. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

PH4 closure audit: close 5 real REQ-01~14 gaps found before starting PH5
CLAUDE.md is explicit that PH5 (self-modification/auto-rollback) must not start until PH4 has genuinely closed -- not just "all unit tests green," but 06-验收与测试锚点.md's REQ-01~14 happy/boundary/failure cases actually covered. A REQ-by-REQ audit (the same discipline that caught U-031's integration gap) found 5 real gaps: - REQ-03 failure: work_node_execution.status jumped straight running->retrying, skipping 'reflecting' (declared in the schema enum, never once written by any code). - REQ-05 failure / REQ-09 boundary: the "scheduler keeps serving other queued tasks past one blocked_pending_human" scenario was never tested against a real scheduler -- the old test literally said "no scheduler exists yet to test against directly" (stale since U-032). - REQ-06 / REQ-07 / REQ-14 failure (the biggest gap, shared root cause): CorruptPersistedStateError, ClassifierBlockedError, and UsageLimitWaitTimeoutError were each correctly raised and each correctly marked their own work_node_execution 'blocked', but nothing ever routed that to task.status='blocked_pending_human' with the blocked_reason each REQ's acceptance text names. e2e_pipeline.run_end_to_end is split into run_task_lifecycle() so scheduler.py's run_goal_to_completion can create the task itself and catch these three specific exceptions, following the same UPDATE+execution_event pattern budget_guardrail.check_budget() already established for REQ-10. - REQ-02 boundary: PRD internal-contradiction check (verifying->reflecting) had no implementation anywhere -- artifact_write always marked validation_status='passed' unconditionally. Added an optional verify_fn hook to work_graph_orchestrate.run_graph() (one regeneration attempt, then ConsistencyCheckFailedError) and a real consistency-check agent() call scoped to the PRD node only. The first real run of this new call caught a genuine bug: the model returned "consistent": "false" as a JSON string, not a boolean, which `if ok:` silently treated as truthy -- fixed with explicit _coerce_bool(). - REQ-13 failure: minor test-assertion weakness (WS 404 rejection test didn't check the JSON error body). 384 tests green. REQ-01~14 now have real, exercised coverage rather than a "tests are green" assumption -- PH4 can be treated as genuinely closed. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent 69c3ea62
Showing 12 changed files with 636 additions and 26 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/e2e_pipeline.py
src/scheduler.py
src/test_backend_server.py
src/test_e2e_pipeline.py
src/test_e2e_pipeline_e2e.py
src/test_scheduler.py
src/test_work_graph_orchestrate.py
src/test_work_node_run.py
src/work_graph_orchestrate.py
src/work_node_run.py
@@ -297,6 +297,13 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
     - **`agentType` + `schema` 能否在动态工作流的 `agent()` 调用里正确共存，是这次集成最大的未知数，做了两次独立真实验证**：先用一个最小 spike（一个只有一次 `agent({agentType:'planner', schema:{...}})` 调用的一节点工作流）验证——不仅 `structured_output` 正确返回，追问角色身份时模型明确自称"Planning Agent (PL)"，说明迁移后的 SOUL.md 人设真的通过 `agentType` 正确加载了（不是走到某个通用 agent）；后续完整 `plan_documents()` 真实跑通又在真实的四节点场景下再次确认。记录在 05 §5.4.1.5。
     - **Coder 阶段的范围决策**：REQ-03 的验收锚点只写了 `c_work_node_run`，没写 `c_work_graph_plan`/`c_work_graph_execute`——因为"规划一个真正的多模块代码图"是另一个完全不同的 LLM 能力，这次没有人要求建，也没有 SDD 规定怎么建，属于超出这次集成范围的新能力，不是"补上就完事"的小活。`implement_code()` 因此只用一个合成的单节点 manifest 走 `work_graph_orchestrate.run_graph()`，让 `c_work_node_run` 对 Coder 也真正生效，但没有尝试造一个"Coder 图规划器"。`needs_worktree` 参数留在 `run_graph()` 里但暂时两处调用都没启用——worktree 隔离要求 `.claude/agents/<org_node>.md` 已经**提交**到分支上（`git worktree add` 只看已提交内容，`deploy_agent_definitions()` 目前只写工作区文件不 commit），这个"部署后要不要顺手提交"的问题这次没有一并解决，留作后续发现，不是遗漏是有意不在这次范围内展开。真实验证也只覆盖了 Planner 全流程（Coder/Validator 走同一套机制，机制本身已经被 Planner 的真实跑通证明，没有再花真钱跑一次完整 Coder 会话，跟 `test_e2e_pipeline_e2e.py` 一贯的成本节制原则一致）。
     - **Validator 阶段未改动**：REQ-04 的覆盖从来没提过 Work Graph 控制器（只有 `c_bug_record_create`/`p_validator_root_cause_trace`），`validate_implementation()` 原来的直连调用已经如实对应它自己的 REQ，不需要动。
+  - **实现补充（2026-08-02）——PH5 开工前对 U-031 自己的 Done 标准（"06 节全部 14 条 REQ 的全部用例通过一次完整回归"）做逐条复核，而不是只信任"全部单测通过"**：CLAUDE.md 明确写着 PH5 依赖 PH4 真正收口，逐条核对 06-验收与测试锚点.md 每条 REQ 的 happy/boundary/failure 用例（不含 PH5 自己的 REQ-15）后，找到并修复了 5 个真实缺口（其中一个是本 Unit 自己就该覆盖但漏掉的更大缺口）：
+    1. **REQ-03 failure**：`work_node_execution.status` 一直是直接从 `running` 跳到 `retrying`，`reflecting` 这个状态值虽然在 01-M 的枚举里，但从没被任何代码真正写过。`work_node_run.py` 的 schema 校验失败分支现在先写一次 `reflecting` 再决定重试/放弃，新增 `test_schema_failure_passes_through_reflecting_before_retrying`（用 sqlite `trace_callback` 侦测真实执行过的 SQL，不是靠时序猜测）。
+    2. **REQ-05 failure / REQ-09 boundary**（"多个任务同时有一个 blocked_pending_human、其余 queued 时，调度器应继续调度 queued 任务"）：`test_org_loop_route.py` 里对应这条的测试自己写着"no scheduler exists yet to test against directly"——U-032 建好调度器后这句注释已经过期，但一直没人回头补一个真正对着调度器跑的测试。新增 `test_scheduler.SchedulerContinuesPastBlockedTaskTest`：真实跑 `run_scheduler_once`，一个 Goal 的任务预置 `loop_round=3`，其真实触发 `org_loop_route.LoopLimitReachedError`（真实的 `blocked_pending_human` 写入，不是 mock 出来的），其余 Goal 正常跑完，验证调度器没有被卡住。
+    3. **REQ-06 / REQ-07 / REQ-14 三条 failure 用例共享的同一个真实缺口，是这次审计里最重要的发现**：`CorruptPersistedStateError`（REQ-06）、`ClassifierBlockedError`（REQ-07）、`UsageLimitWaitTimeoutError`（REQ-14）都被正确地抛出，也都正确地把各自的 `work_node_execution.status` 标成了 `blocked`（后者是 U-031 自己加的 catch-all），但从来没有任何代码把这个后果真正传导到 `task.status='blocked_pending_human'`——`work_node_run.py` 自己的 docstring 甚至早就写明"callers should route it"，但没有一个调用方真的这么做。这正是 U-032 的调度器该管的事：`e2e_pipeline.py` 拆出 `run_task_lifecycle()`（`run_end_to_end()` 变成"建 task + 调它"的两行包装，签名/行为不变），`scheduler.py` 新增 `_block_task()`（照抄 `budget_guardrail.check_budget()` 已有的 UPDATE + `task_status_changed` 事件写法，不是发明新模式），`run_goal_to_completion()` 自己建 task 后捕获这三种异常并正确路由；其余异常仍然原样冒泡，不吞掉。三个新测试类（`CorruptPersistedStateRoutingTest`/`ClassifierBlockedRoutingTest`/`UsageLimitWaitTimeoutRoutingTest`）逐一验证。
+    4. **REQ-02 boundary**（PRD 出现自相矛盾的场景假设应被拦截，触发 `reflecting` 而非直接放行）：这条此前完全没有实现——`artifact_write.write_artifact()` 一直无条件把 `validation_status` 写成 `'passed'`，`artifact_write.py` 自己的 docstring 也如实记录了这个缺口。因为这是全新的 LLM 判断能力而不是接线修复，就此询问用户是否现在补上，用户选择"现在就建"。新增 `work_graph_orchestrate.run_graph()` 的可选 `verify_fn` 钩子（每个节点跑完、下游节点开始前调用，命中失败时把 `work_graph_run.status` 写成 `reflecting`、对同一节点重新生成一次并再次校验，仍失败则写 `blocked` 并抛 `ConsistencyCheckFailedError`——`work_graph_run.status` 在此之前从未被任何代码真正写过，一直停在初始的 `'planning'`，这次是它第一次被真正用起来，但只做了 `verifying`/`reflecting`/`blocked` 三个和这条用例直接相关的状态，没有借机把 `running`/`merging` 那一整套图级状态机也补全——04-P 没具体描述过那部分，属于超出这次范围的新设计）；`e2e_pipeline._check_prd_consistency()` 是新增的一次真实 `agent(agentType:'planner', schema:...)` 调用，只挂在 PRD 节点上（`_prd_consistency_verify_fn`，其余节点直接放行不调用）。真实验证：`test_e2e_pipeline_e2e.RealPrdConsistencyCheckTest` 用一段刻意自相矛盾的撞墙规则（"必须立即结束" vs "可以穿墙不结束"）+一段真实一致的 PRD 各跑一次真实调用。**第一次真实跑就抓到一个真 bug**：模型把 `consistent` 字段返回成 JSON 字符串 `"false"` 而不是布尔值 `false`（即使 schema 里已经写了 `"consistent": {"type": "boolean"}`——05 §5.4.1.1 早就说过不能盲目信任 schema 真的会被强制），Python 里 `if "false":` 是真值，直接把"发现矛盾"误判成"通过"，是全场景里最隐蔽的一类 bug（mock 测试永远返回真的 Python bool，只有真实调用能暴露）。新增 `_coerce_bool()` 做防御性转换（bool 原样、字符串按大小写不敏感比较 "true"、其余走 `bool()`），修复后重跑真实测试通过；同时补了两个 mock 层回归测试锁死这个具体行为（`test_string_false_from_a_real_call_is_not_treated_as_truthy`/`test_string_true_from_a_real_call_is_treated_as_consistent`）。
+    5. **REQ-13 failure**：一个次要的测试质量问题（不是功能缺口）——`test_ws_nonexistent_node_rejected_before_upgrade` 只检查了 HTTP 状态行是 404，没检查响应体里的 `NODE_EXEC_NOT_FOUND`（生产代码本身是对的）。顺手加了 `_read_http_response_with_json_body` 辅助函数，把这条和同类的 `PROJECT_NOT_FOUND`（REQ-12 failure）测试都补上了 body 断言。
+    - 全部 5 处修复后，`python3 -m unittest discover -s src` 380 个测试全绿（含新增的）；REQ-01~14 现在逐条都有真实覆盖，PH4 可以视为真正收口，PH5 排期不再依赖对这份回归的信任假设。
   - 覆盖：01-M `work_graph_run`/`work_node_execution`（真正被生产代码写入，不再只是测试 fixture）；03-C `c_work_graph_plan`/`c_work_graph_execute`/`c_work_node_run`（首次真正接通）；REQ-01~14 全部
 - **U-032 [Risk Gate]** 多 Goal 并发（`org_concurrency_config`）+ 长时间运行稳定性压测
@@ -47,7 +47,7 @@
     {"id": "U-029", "phase": "PH3", "sub_phase": "PH3-S3", "name": "v_skill_library + v_skill_audit_log（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-018"], "coverage": {"model": [], "view": ["v_skill_library", "v_skill_audit_log"], "controller": ["c_skill_diff", "c_skill_list", "c_skill_get", "c_skill_rollback"], "process": [], "requirement": []}, "done": "Skill 库与变更审计日志可查看；c_skill_rollback 首次接上 HTTP（skill_repo_path 可注入，测试绝不碰真实仓库）"},
     {"id": "U-030", "phase": "PH3", "sub_phase": "PH3-S4", "name": "并发/预算/日志保留配置页（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-019", "U-021"], "coverage": {"model": ["org_concurrency_config"], "view": ["v_concurrency_settings", "v_budget_settings", "v_log_retention_settings"], "controller": ["c_concurrency_config_update", "c_budget_config_update", "c_log_retention_config_update"], "process": [], "requirement": []}, "done": "三个配置页可用且改动生效；PH3 至此全部完成"},
-    {"id": "U-031", "phase": "PH4", "sub_phase": "PH4-S1", "name": "REQ-01~14 全量自动化回归（见 development-plan.md 2026-08-01 实现补充）", "classification": "Risk Gate", "depends_on": ["U-014", "U-017", "U-020", "U-021", "U-022", "U-023", "U-024.5", "U-025", "U-026", "U-027", "U-028", "U-029", "U-030"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-01", "REQ-02", "REQ-03", "REQ-04", "REQ-05", "REQ-06", "REQ-07", "REQ-08", "REQ-09", "REQ-10", "REQ-11", "REQ-12", "REQ-13", "REQ-14"]}, "done": "06 节全部 14 条 REQ 的全部用例通过一次完整回归；回归审计中发现 c_work_graph_plan/c_work_graph_execute/c_work_node_run 三个模块彼此隔离测试通过但从未被生产代码真正串联（唯一验证过的闭环 e2e_pipeline.py 绕开了三者，直接用 org_node_trigger），已按用户决策现场补上集成：新增 src/work_graph_orchestrate.py（拓扑排序 + 每节点一份独立动态工作流脚本调用 run_work_node），e2e_pipeline.py 的 plan_documents/implement_code 改为真调用 plan_work_graph→run_graph；真实 API 验证过程中发现并修复两个真实 bug（graph_manifest 的 id/node_id 字段名及 depends_on 缺失见 work_graph_plan._normalize_graph_manifest；run_work_node 缺 repo_path 导致 cwd=None）"},
+    {"id": "U-031", "phase": "PH4", "sub_phase": "PH4-S1", "name": "REQ-01~14 全量自动化回归（见 development-plan.md 2026-08-01/2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-014", "U-017", "U-020", "U-021", "U-022", "U-023", "U-024.5", "U-025", "U-026", "U-027", "U-028", "U-029", "U-030"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-01", "REQ-02", "REQ-03", "REQ-04", "REQ-05", "REQ-06", "REQ-07", "REQ-08", "REQ-09", "REQ-10", "REQ-11", "REQ-12", "REQ-13", "REQ-14"]}, "done": "06 节全部 14 条 REQ 的全部用例通过一次完整回归；回归审计中发现 c_work_graph_plan/c_work_graph_execute/c_work_node_run 三个模块彼此隔离测试通过但从未被生产代码真正串联（唯一验证过的闭环 e2e_pipeline.py 绕开了三者，直接用 org_node_trigger），已按用户决策现场补上集成：新增 src/work_graph_orchestrate.py（拓扑排序 + 每节点一份独立动态工作流脚本调用 run_work_node），e2e_pipeline.py 的 plan_documents/implement_code 改为真调用 plan_work_graph→run_graph；真实 API 验证过程中发现并修复两个真实 bug（graph_manifest 的 id/node_id 字段名及 depends_on 缺失见 work_graph_plan._normalize_graph_manifest；run_work_node 缺 repo_path 导致 cwd=None）。**2026-08-02（PH5 开工前逐条复核）**：对照 06 节重新逐条核对后又发现并修复 5 处真实缺口——REQ-03 failure 缺失 'reflecting' 状态写入、REQ-05/09 的调度器场景此前从未真正测试过（U-032 建好后才补测）、REQ-06/07/14 三条 failure 共享的同一个缺口（CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError 均未路由到 task.status=blocked_pending_human）、REQ-02 boundary（PRD 一致性校验）此前完全未实现现已用新的 verify_fn 机制补上、REQ-13 failure 一处测试断言过弱；全部详见 development-plan.md 对应实现补充，380 测试全绿，此时才是 06 节 REQ-01~14 真正逐条有覆盖"},
     {"id": "U-032", "phase": "PH4", "sub_phase": "PH4-S2", "name": "多 Goal 并发 + 长时间运行稳定性压测（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-031"], "coverage": {"model": ["org_concurrency_config"], "view": [], "controller": [], "process": [], "requirement": ["REQ-09"]}, "done": "并发数调高后多个独立 Goal 可同时稳定运行；发现并补上一个此前从未存在的结构性缺口——04-P/01-M/REQ-09 boundary 反复假设存在的自动调度器实际从未被建过，新增 src/scheduler.py（ThreadPoolExecutor + 每线程独立 sqlite 连接，避免复刻 U-027 的共享连接锁陷阱）真正实现，并顺带补上 e2e_pipeline.run_end_to_end() 自己文档过的缺口（PASS/FAIL 后从不调用 close_task_done/run_fix_loop，卡在 validating 不动）"},
     {"id": "U-033", "phase": "PH4", "sub_phase": "PH4-S3", "name": "A-2 假设收口（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-004"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": [], "assumption": ["A-2"]}, "done": "A-2 状态确认为 confirmed 或记录实际偏差；最终结论：维持 U-004 记录的\"部分成立\"，PH1~PH4 积累的真实运行数据未暴露新偏差，偏差范围到 U-004 为止已完全摸清（Coder SOUL.md 三处实质性重写），不改判 confirmed，正式收口"},
@@ -31,6 +31,18 @@ coverage never named the Work Graph controllers at all.
 Status-transition bookkeeping (task.status, precondition guards) stays
 U-010's job (org_node_lifecycle.trigger_org_node_full) -- called once per
 phase with a lightweight "confirm ready" prompt.
+
+**PRD consistency check (2026-08-02, PH4-closure audit)**: REQ-02's
+boundary case (contradictory PRD scenario assumptions must be intercepted,
+not silently pass validation) had no implementation anywhere -- found while
+auditing REQ-01~14 for real coverage before starting PH5.
+`_check_prd_consistency` is one extra real `agent()` call (reusing
+`agentType='planner'` + a structured-output schema, same mechanism as every
+other node) that reviews the PRD's own content for genuine internal
+contradiction (not incompleteness/style) right after it's produced, wired
+in as `plan_documents`'s `verify_fn` -- see `work_graph_orchestrate.
+run_graph`'s docstring for what happens on a failed check (one regeneration
+attempt, then `ConsistencyCheckFailedError` if it's still contradictory).
 """
 import json
@@ -40,7 +52,7 @@ import uuid
 from src import (
     artifact_write, org_node_lifecycle, org_node_trigger, task_create,
-    work_graph_orchestrate, work_graph_plan,
+    work_graph_execute, work_graph_orchestrate, work_graph_plan,
 )
 _READY_PROMPT = "Confirm you're ready to begin. Take no other action yet."
@@ -48,6 +60,22 @@ _READY_PROMPT = &quot;Confirm you&#39;re ready to begin. Take no other action yet.&quot;
 _DOC_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
 _DOC_OUTPUT_SCHEMA = {"type": "object", "required": ["content"]}
+_PRD_CONSISTENCY_SCHEMA = {
+    "type": "object",
+    "required": ["consistent", "issues"],
+    "properties": {
+        "consistent": {"type": "boolean"},
+        "issues": {"type": "array", "items": {"type": "string"}},
+    },
+}
+_PRD_CONSISTENCY_PROMPT = (
+    "Review the following PRD for internally CONTRADICTORY scenario assumptions -- e.g. two "
+    "requirements that cannot both hold at once, or the same scenario described two incompatible "
+    "ways. Do NOT flag mere incompleteness, ambiguity, or style issues, only genuine "
+    "self-contradiction. Reply via the required schema: consistent=false only if you found a real "
+    "contradiction, and list each one in issues.\n\nPRD:\n\n{prd_content}"
+)
+
 _DOC_PROMPTS = {
     "brd": (
         "Write a BRD (Business Requirement Document) for this goal, following your own "
@@ -109,6 +137,48 @@ def _planner_prompt_builder(goal_text, node_artifact_types):
     return build_prompt
+def _coerce_bool(value):
+    """A real call (2026-08-02) returned `"consistent": "false"` as a JSON
+    STRING, not a boolean, even with `"type": "boolean"` in the schema --
+    `if "false":` is truthy in Python, so this silently inverted the check
+    until caught by a real test (mocks always return real Python bools, so
+    this never showed up until RUN_REAL_AGENT_E2E=1). Don't trust the
+    schema's declared type any more than 05 §5.4.1.1 already tells us to
+    trust --json-schema itself."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() == "true"
+    return bool(value)
+
+
+def _check_prd_consistency(target_folder_path, prd_content, timeout=180):
+    """REQ-02 boundary: one real agent() call reviewing the PRD's own
+    content for genuine internal contradiction. Returns (ok, issues)."""
+    script_name = "graphlemon-prd-consistency-" + uuid.uuid4().hex
+    prompt = _PRD_CONSISTENCY_PROMPT.format(prd_content=prd_content)
+    script_content = (
+        "export const meta = {{ name: {name}, description: 'graphlemon PRD consistency check' }}\n\n"
+        "const result = await agent({prompt}, {{ agentType: 'planner', schema: {schema} }})\n\n"
+        "return result\n"
+    ).format(name=json.dumps(script_name), prompt=json.dumps(prompt), schema=json.dumps(_PRD_CONSISTENCY_SCHEMA))
+    work_graph_execute.write_workflow_script(target_folder_path, script_name, script_content)
+    result = work_graph_execute.execute_work_graph(
+        target_folder_path, script_name, {}, _PRD_CONSISTENCY_SCHEMA, timeout=timeout
+    )
+    return _coerce_bool(result["consistent"]), result.get("issues", [])
+
+
+def _prd_consistency_verify_fn(node_artifact_types, target_folder_path):
+    """work_graph_orchestrate.run_graph's verify_fn -- only the PRD node
+    gets checked, every other node passes through untouched."""
+    def verify_fn(node, output):
+        if node_artifact_types.get(node["node_id"]) != "prd":
+            return True, None
+        return _check_prd_consistency(target_folder_path, output["content"])
+    return verify_fn
+
+
 def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
     """Architecture §11 criterion 1: a real Planner Work Graph
     (c_work_graph_plan -> per-node c_work_graph_execute/c_work_node_run via
@@ -126,6 +196,7 @@ def plan_documents(conn, task_id, goal_text, target_folder_path, timeout=300):
     outputs = work_graph_orchestrate.run_graph(
         conn, run_id, "planner", "planner-doc-writing", target_folder_path, graph_manifest,
         _planner_prompt_builder(goal_text, node_artifact_types), _DOC_INPUT_SCHEMA, _DOC_OUTPUT_SCHEMA,
+        verify_fn=_prd_consistency_verify_fn(node_artifact_types, target_folder_path),
     )
     artifact_ids = {}
@@ -219,16 +290,16 @@ def _current_commit(repo_path):
     ).stdout.strip()
-def run_end_to_end(conn, target_folder_path, goal_text):
-    """Full orchestration: create_task -> plan_documents -> implement_code ->
-    validate_implementation. Returns a dict with every intermediate result,
-    for inspection/debugging rather than hiding it behind a single verdict.
+def run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
+    """The body of run_end_to_end() for an ALREADY-CREATED task -- split out
+    2026-08-02 so scheduler.py can create the task itself first and still
+    have a task_id in scope if this raises partway through (needed to route
+    CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
+    to task.status='blocked_pending_human', which is scheduler.py's job, not
+    this module's -- see that module's docstring). Behavior/return shape is
+    unchanged from before the split; run_end_to_end() below is now a 2-line
+    wrapper over this.
     """
-    project_id = conn.execute(
-        "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
-    ).fetchone()[0]
-    task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
-
     org_node_lifecycle.trigger_org_node_full(conn, "planner", task_id, target_folder_path, _READY_PROMPT)
     plan_result = plan_documents(conn, task_id, goal_text, target_folder_path)
     artifact_ids = plan_result["artifact_ids"]
@@ -249,6 +320,18 @@ def run_end_to_end(conn, target_folder_path, goal_text):
     }
+def run_end_to_end(conn, target_folder_path, goal_text):
+    """Full orchestration: create_task -> plan_documents -> implement_code ->
+    validate_implementation. Returns a dict with every intermediate result,
+    for inspection/debugging rather than hiding it behind a single verdict.
+    """
+    project_id = conn.execute(
+        "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
+    ).fetchone()[0]
+    task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
+    return run_task_lifecycle(conn, task_id, target_folder_path, goal_text)
+
+
 def _read_artifact_content(conn, target_folder_path, artifact_id):
     file_path = conn.execute("SELECT file_path FROM artifact WHERE artifact_id = ?", (artifact_id,)).fetchone()[0]
     with open(os.path.join(target_folder_path, file_path), encoding="utf-8") as f:
@@ -26,15 +26,67 @@ replicating that discipline here would just serialize all DB work anyway and
 defeat the point of concurrency. Each connection sets a `busy_timeout` so
 concurrent writers on the shared file retry instead of raising "database is
 locked".
+
+**Second gap found and closed in the same pass (2026-08-02, PH4-closure
+audit before starting PH5)**: `work_node_run.CorruptPersistedStateError`
+(REQ-06 failure), `work_graph_execute.ClassifierBlockedError` (REQ-07
+failure), and `work_node_run.UsageLimitWaitTimeoutError` (REQ-14 failure)
+were each correctly raised and each correctly marked their own
+`work_node_execution.status='blocked'` (the last one via U-031's catch-all
+in `work_node_run.run_work_node`) -- but nothing ever caught them to apply
+the *task*-level consequence 06-验收与测试锚点.md's failure text for all
+three literally names: `task.status='blocked_pending_human'` with a specific
+`blocked_reason`. `work_node_run.py`'s own docstring already flagged this as
+a deliberate layering choice ("callers should route it to
+task.status='blocked_pending_human'") -- but no caller ever did. This is
+exactly the layer `budget_guardrail.check_budget()` already does the
+equivalent translation at for its own case (REQ-10), so `_block_task()`
+below follows that same established pattern (UPDATE + `task_status_changed`
+execution_event) rather than inventing a new one.
 """
+import json
+import uuid
 import concurrent.futures
-from src import e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config
+from src import (
+    e2e_pipeline, org_level_loop, org_loop_route, schema, settings_config,
+    task_create, work_graph_execute, work_graph_orchestrate, work_node_run,
+)
 _BUSY_TIMEOUT_MS = 30000
 _FIX_LOOP_TIMEOUT = 1200
+_BLOCKED_REASON_FOR_EXCEPTION = (
+    (work_node_run.CorruptPersistedStateError, "corrupt_persisted_state"),
+    (work_graph_execute.ClassifierBlockedError, "classifier_blocked"),
+    (work_node_run.UsageLimitWaitTimeoutError, "usage_limit_wait_timeout"),
+    (work_graph_orchestrate.ConsistencyCheckFailedError, "prd_consistency_check_failed"),
+)
+_ROUTABLE_EXCEPTIONS = tuple(exc for exc, _ in _BLOCKED_REASON_FOR_EXCEPTION)
+
+
+def _blocked_reason_for(exc):
+    for exc_type, reason in _BLOCKED_REASON_FOR_EXCEPTION:
+        if isinstance(exc, exc_type):
+            return reason
+    raise AssertionError("no blocked_reason registered for " + type(exc).__name__)
+
+
+def _block_task(conn, task_id, blocked_reason):
+    task_row = conn.execute("SELECT project_id, status FROM task WHERE task_id = ?", (task_id,)).fetchone()
+    conn.execute(
+        "UPDATE task SET status = 'blocked_pending_human', blocked_reason = ?, "
+        "updated_at = CURRENT_TIMESTAMP WHERE task_id = ?",
+        (blocked_reason, task_id),
+    )
+    conn.execute(
+        "INSERT INTO execution_event (event_id, project_id, task_id, event_type, payload) "
+        "VALUES (?, ?, ?, 'task_status_changed', ?)",
+        (str(uuid.uuid4()), task_row[0], task_id, json.dumps({"from_status": task_row[1], "to_status": "blocked_pending_human"})),
+    )
+    conn.commit()
+
 def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
     """PASS -> `close_task_done`; FAIL/INCONCLUSIVE -> `org_level_loop.
@@ -66,11 +118,30 @@ def finish_task_lifecycle(conn, run_result, target_folder_path, goal_text):
 def run_goal_to_completion(db_path, target_folder_path, goal_text):
     """One Goal, its own sqlite connection, start to a terminal task.status
     (done or blocked_pending_human). This is the unit of work each scheduler
-    worker thread runs."""
+    worker thread runs.
+
+    Creates the task itself (rather than delegating to e2e_pipeline.
+    run_end_to_end) so task_id is in scope even if e2e_pipeline.
+    run_task_lifecycle raises partway through -- needed to route
+    CorruptPersistedStateError/ClassifierBlockedError/UsageLimitWaitTimeoutError
+    to task.status='blocked_pending_human' (see module docstring's "Second
+    gap" note). Any OTHER exception still propagates uncaught, same as
+    before -- run_scheduler_once's own try/except is what keeps one Goal's
+    unexpected failure from stopping the rest.
+    """
     conn = schema.connect(db_path)
     conn.execute("PRAGMA busy_timeout = {}".format(_BUSY_TIMEOUT_MS))
     try:
-        result = e2e_pipeline.run_end_to_end(conn, target_folder_path, goal_text)
+        project_id = conn.execute(
+            "SELECT project_id FROM project WHERE target_folder_path = ?", (target_folder_path,)
+        ).fetchone()[0]
+        task_id = task_create.create_task(conn, project_id, goal_text)["task_id"]
+        try:
+            result = e2e_pipeline.run_task_lifecycle(conn, task_id, target_folder_path, goal_text)
+        except _ROUTABLE_EXCEPTIONS as e:
+            blocked_reason = _blocked_reason_for(e)
+            _block_task(conn, task_id, blocked_reason)
+            return {"task_id": task_id, "final_status": "blocked_pending_human", "blocked_reason": blocked_reason}
         return finish_task_lifecycle(conn, result, target_folder_path, goal_text)
     finally:
         conn.close()
@@ -536,6 +536,31 @@ def _read_http_response_head(sock):
     return buf.decode("ascii", errors="replace")
+def _read_http_response_with_json_body(sock):
+    """Like _read_http_response_head, but also reads and parses the JSON
+    body (via Content-Length) -- for pre-upgrade WS rejections, which DO
+    send a body (_send_json), unlike a 101 Switching Protocols response."""
+    buf = b""
+    while b"\r\n\r\n" not in buf:
+        chunk = sock.recv(4096)
+        if not chunk:
+            break
+        buf += chunk
+    head, _, rest = buf.partition(b"\r\n\r\n")
+    head_text = head.decode("ascii", errors="replace")
+    content_length = 0
+    for line in head_text.splitlines():
+        if line.lower().startswith("content-length:"):
+            content_length = int(line.split(":", 1)[1].strip())
+            break
+    while len(rest) < content_length:
+        chunk = sock.recv(4096)
+        if not chunk:
+            break
+        rest += chunk
+    return head_text, json.loads(rest.decode("utf-8"))
+
+
 def _read_server_frame_raw(sock):
     """Server->client frames are never masked (RFC 6455 §5.1) -- simpler
     decode than ws_subscribe.decode_client_frame, which expects masking.
@@ -608,9 +633,10 @@ class WsSubscribeRouteTest(unittest.TestCase):
         self.addCleanup(sock.close)
         key = base64.b64encode(b"0123456789012345").decode("ascii")
         sock.sendall(_ws_handshake_request("/api/ws/projects/no-such-project", key))
-        head = _read_http_response_head(sock)
+        head, body = _read_http_response_with_json_body(sock)
         self.assertIn("404", head.splitlines()[0])
         self.assertNotIn("101", head.splitlines()[0])
+        self.assertEqual(body["error"], "PROJECT_NOT_FOUND")
     def test_non_upgrade_request_rejected(self):
         sock = self._connect()
@@ -780,8 +806,9 @@ class ExecutionLogRouteTest(unittest.TestCase):
         self.addCleanup(sock.close)
         key = base64.b64encode(b"0123456789012345").decode("ascii")
         sock.sendall(_ws_handshake_request("/api/ws/work-nodes/no-such-node/log", key))
-        head = _read_http_response_head(sock)
+        head, body = _read_http_response_with_json_body(sock)
         self.assertIn("404", head.splitlines()[0])
+        self.assertEqual(body["error"], "NODE_EXEC_NOT_FOUND")
     def test_ws_pushes_appended_log_content(self):
         sock = self._connect()
@@ -84,6 +84,7 @@ class PlanDocumentsTest(PipelineTestBase):
              patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
             mock_execute.side_effect = [
                 {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
+                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                 {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
             ]
             plan_result = e2e_pipeline.plan_documents(self.conn, "task-1", "实现贪吃蛇游戏", self.tmp.name)
@@ -119,6 +120,7 @@ class RunEndToEndTest(PipelineTestBase):
              patch.object(org_node_trigger, "trigger_org_node", return_value="VERDICT: PASS\nAll scenarios pass."):
             mock_execute.side_effect = [
                 {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]},
+                {"consistent": True, "issues": []},  # PRD consistency check (REQ-02 boundary)
                 {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]},
                 {"response": "Implemented. Ran `pytest`: 3 passed."},
             ]
@@ -147,6 +149,59 @@ class RunEndToEndTest(PipelineTestBase):
         self.assertEqual(run_org_nodes, {"planner", "coder"})
+class PrdConsistencyVerifyFnTest(unittest.TestCase):
+    """REQ-02 boundary: only the PRD node gets the consistency check."""
+
+    def test_non_prd_node_passes_through_without_a_call(self):
+        node_artifact_types = {"brd": "brd"}
+        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
+        with patch.object(work_graph_execute, "execute_work_graph") as mock_execute:
+            ok, feedback = verify_fn({"node_id": "brd"}, {"content": "brd text"})
+        self.assertTrue(ok)
+        self.assertIsNone(feedback)
+        mock_execute.assert_not_called()
+
+    def test_string_false_from_a_real_call_is_not_treated_as_truthy(self):
+        # regression: a real call (2026-08-02) returned "consistent": "false"
+        # as a JSON string, not a boolean -- `if "false":` is truthy in
+        # Python, silently inverting the check until a real test caught it.
+        node_artifact_types = {"prd": "prd"}
+        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(
+                 work_graph_execute, "execute_work_graph",
+                 return_value={"consistent": "false", "issues": ["contradiction found"]},
+             ):
+            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
+        self.assertFalse(ok)
+
+    def test_string_true_from_a_real_call_is_treated_as_consistent(self):
+        node_artifact_types = {"prd": "prd"}
+        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(
+                 work_graph_execute, "execute_work_graph",
+                 return_value={"consistent": "true", "issues": []},
+             ):
+            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "consistent prd"})
+        self.assertTrue(ok)
+
+    def test_prd_node_triggers_a_real_check_call(self):
+        node_artifact_types = {"prd": "prd"}
+        verify_fn = e2e_pipeline._prd_consistency_verify_fn(node_artifact_types, "/tmp/proj-1")
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(
+                 work_graph_execute, "execute_work_graph",
+                 return_value={"consistent": False, "issues": ["contradiction found"]},
+             ) as mock_execute:
+            ok, feedback = verify_fn({"node_id": "prd"}, {"content": "contradictory prd"})
+        self.assertFalse(ok)
+        self.assertEqual(feedback, ["contradiction found"])
+        mock_execute.assert_called_once()
+        args = mock_execute.call_args[0]
+        self.assertEqual(args[0], "/tmp/proj-1")
+
+
 class ValidateImplementationTest(unittest.TestCase):
     def test_parses_pass_verdict(self):
         with patch("src.org_node_trigger.trigger_org_node", return_value="VERDICT: PASS\nlooks good"):
@@ -53,5 +53,45 @@ class RealPlanDocumentsTest(unittest.TestCase):
             self.assertEqual(prd_deps, [artifact_ids["brd"]])
+@skip_unless_env("RUN_REAL_AGENT_E2E")
+class RealPrdConsistencyCheckTest(unittest.TestCase):
+    """REQ-02 boundary, real coverage (2026-08-02): the new _check_prd_consistency
+    call is a genuinely new LLM-judgment capability, not a wiring change --
+    validated for real once each way (contradiction / no contradiction),
+    same discipline as every other new agent() call this project has added."""
+
+    def test_detects_a_real_deliberate_contradiction(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
+            agent_profile.deploy_agent_definitions(tmp)
+            contradictory_prd = (
+                "# PRD: Snake Game\n\n"
+                "## Wall Collision\n"
+                "Requirement 3.1: When the snake's head touches the board boundary, the game MUST "
+                "immediately end with a 'Game Over' screen.\n\n"
+                "Requirement 3.2: The snake must be able to pass through the board boundary and "
+                "wrap around to the opposite side, continuing play without ending the game.\n"
+            )
+            ok, issues = e2e_pipeline._check_prd_consistency(tmp, contradictory_prd)
+        self.assertFalse(ok)
+        self.assertTrue(len(issues) > 0)
+
+    def test_passes_a_real_consistent_prd(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            subprocess.run(["git", "init"], cwd=tmp, check=True, capture_output=True)
+            agent_profile.deploy_agent_definitions(tmp)
+            consistent_prd = (
+                "# PRD: Snake Game\n\n"
+                "## Wall Collision\n"
+                "Requirement 3.1: When the snake's head touches the board boundary, the game "
+                "immediately ends with a 'Game Over' screen showing the final score.\n\n"
+                "## Food\n"
+                "Requirement 4.1: Eating food grows the snake by one segment and spawns a new "
+                "food item at a random empty cell.\n"
+            )
+            ok, issues = e2e_pipeline._check_prd_consistency(tmp, consistent_prd)
+        self.assertTrue(ok, "false positive on a consistent PRD: {}".format(issues))
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -17,7 +17,7 @@ import time
 import unittest
 from unittest.mock import patch
-from src import org_level_loop, org_loop_route, schema, scheduler, settings_config
+from src import org_level_loop, org_loop_route, schema, scheduler, settings_config, work_graph_execute, work_node_run
 class FinishTaskLifecycleTestBase(unittest.TestCase):
@@ -178,13 +178,12 @@ class ConcurrentSqliteWritesStressTest(unittest.TestCase):
         self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))
     def test_many_concurrent_goals_write_to_the_same_db_file_without_locking_errors(self):
-        def fake_run_end_to_end(conn, target_folder_path, goal_text):
-            task_id = goal_text  # unique per goal, stands in for a real uuid
-            conn.execute(
-                "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) "
-                "VALUES (?, 'proj-1', ?, 'goal', 'validating', 'validator')",
-                (task_id, goal_text),
-            )
+        def fake_run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
+            # task_id already exists -- run_goal_to_completion creates it for
+            # real via task_create.create_task before calling this; stand in
+            # for org_node_lifecycle's real queued->...->validating march so
+            # finish_task_lifecycle's close_task_done (requires 'validating') works.
+            conn.execute("UPDATE task SET status = 'validating' WHERE task_id = ?", (task_id,))
             conn.commit()
             time.sleep(0.02)  # widen the contention window
             return {
@@ -193,7 +192,7 @@ class ConcurrentSqliteWritesStressTest(unittest.TestCase):
             }
         goals = [("/tmp/proj-1", "goal-{}".format(i)) for i in range(10)]
-        with patch.object(scheduler.e2e_pipeline, "run_end_to_end", side_effect=fake_run_end_to_end):
+        with patch.object(scheduler.e2e_pipeline, "run_task_lifecycle", side_effect=fake_run_task_lifecycle):
             results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=5)
         self.assertTrue(all("result" in r for r in results), results)
@@ -204,5 +203,158 @@ class ConcurrentSqliteWritesStressTest(unittest.TestCase):
         self.assertEqual(done_count, 10)
+class RoutableExceptionTestBase(unittest.TestCase):
+    """REQ-06/07/14 failure: work-node-level failures (CorruptPersistedState
+    /ClassifierBlocked/UsageLimitWaitTimeout) must actually reach
+    task.status='blocked_pending_human' with the matching blocked_reason --
+    found missing during the PH4-closure audit (2026-08-02): each exception
+    was correctly raised and correctly marked its own work_node_execution
+    'blocked', but nothing ever caught it to apply the task-level
+    consequence 06's own acceptance text names."""
+
+    def setUp(self):
+        self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        self.tmp.close()
+        self.db_path = self.tmp.name
+        conn = schema.connect(self.db_path)
+        schema.create_schema(conn)
+        conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        conn.commit()
+        conn.close()
+        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))
+
+    def _run_with_raising_lifecycle(self, exc):
+        def fake_run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
+            raise exc
+
+        with patch.object(scheduler.e2e_pipeline, "run_task_lifecycle", side_effect=fake_run_task_lifecycle):
+            result = scheduler.run_goal_to_completion(self.db_path, "/tmp/proj-1", "goal text")
+        return result
+
+    def _assert_task_blocked(self, task_id, blocked_reason):
+        conn = schema.connect(self.db_path)
+        row = conn.execute("SELECT status, blocked_reason FROM task WHERE task_id = ?", (task_id,)).fetchone()
+        event = conn.execute(
+            "SELECT event_type FROM execution_event WHERE task_id = ? ORDER BY rowid DESC LIMIT 1", (task_id,)
+        ).fetchone()
+        conn.close()
+        self.assertEqual(row, ("blocked_pending_human", blocked_reason))
+        self.assertEqual(event[0], "task_status_changed")
+
+
+class CorruptPersistedStateRoutingTest(RoutableExceptionTestBase):
+    def test_routes_to_blocked_pending_human(self):
+        result = self._run_with_raising_lifecycle(work_node_run.CorruptPersistedStateError("boom"))
+        self.assertEqual(result["final_status"], "blocked_pending_human")
+        self.assertEqual(result["blocked_reason"], "corrupt_persisted_state")
+        self._assert_task_blocked(result["task_id"], "corrupt_persisted_state")
+
+
+class ClassifierBlockedRoutingTest(RoutableExceptionTestBase):
+    def test_routes_to_blocked_pending_human(self):
+        result = self._run_with_raising_lifecycle(work_graph_execute.ClassifierBlockedError("boom"))
+        self.assertEqual(result["final_status"], "blocked_pending_human")
+        self.assertEqual(result["blocked_reason"], "classifier_blocked")
+        self._assert_task_blocked(result["task_id"], "classifier_blocked")
+
+
+class UsageLimitWaitTimeoutRoutingTest(RoutableExceptionTestBase):
+    def test_routes_to_blocked_pending_human(self):
+        result = self._run_with_raising_lifecycle(work_node_run.UsageLimitWaitTimeoutError("boom"))
+        self.assertEqual(result["final_status"], "blocked_pending_human")
+        self.assertEqual(result["blocked_reason"], "usage_limit_wait_timeout")
+        self._assert_task_blocked(result["task_id"], "usage_limit_wait_timeout")
+
+
+class ConsistencyCheckFailedRoutingTest(RoutableExceptionTestBase):
+    def test_routes_to_blocked_pending_human(self):
+        result = self._run_with_raising_lifecycle(
+            scheduler.work_graph_orchestrate.ConsistencyCheckFailedError("boom")
+        )
+        self.assertEqual(result["final_status"], "blocked_pending_human")
+        self.assertEqual(result["blocked_reason"], "prd_consistency_check_failed")
+        self._assert_task_blocked(result["task_id"], "prd_consistency_check_failed")
+
+
+class UnroutedExceptionStillPropagatesTest(RoutableExceptionTestBase):
+    def test_other_exceptions_are_not_swallowed(self):
+        with self.assertRaises(RuntimeError):
+            self._run_with_raising_lifecycle(RuntimeError("something else entirely"))
+
+
+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"
+    if org_node == "coder":
+        return "Fixed it."
+    if org_node == "validator":
+        return "VERDICT: FAIL\nstill broken"
+    raise AssertionError("unexpected call: " + org_node)
+
+
+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
+    exists yet to test against directly" (stale as of U-032). One Goal's
+    task is pre-seeded at loop_round=3 so its fix-loop attempt hits
+    org_loop_route.LoopLimitReachedError for real (route_org_loop's own real
+    blocked_pending_human write, not a mock) while the others complete via
+    close_task_done -- proving the scheduler doesn't stop serving the rest
+    of the batch."""
+
+    def setUp(self):
+        self.tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
+        self.tmp.close()
+        self.db_path = self.tmp.name
+        conn = schema.connect(self.db_path)
+        schema.create_schema(conn)
+        conn.execute(
+            "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) "
+            "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')"
+        )
+        conn.commit()
+        conn.close()
+        self.addCleanup(lambda: os.path.exists(self.db_path) and os.remove(self.db_path))
+
+    def test_one_task_hitting_loop_limit_does_not_block_the_others(self):
+        def fake_run_task_lifecycle(conn, task_id, target_folder_path, goal_text):
+            conn.execute("UPDATE task SET status = 'validating' WHERE task_id = ?", (task_id,))
+            conn.commit()
+            if goal_text == "goal-doomed":
+                conn.execute("UPDATE task SET loop_round = 3 WHERE task_id = ?", (task_id,))
+                conn.commit()
+                verdict = {"verdict": "FAIL", "raw_response": "VERDICT: FAIL\nstill broken"}
+            else:
+                verdict = {"verdict": "PASS", "raw_response": "VERDICT: PASS"}
+            return {
+                "task_id": task_id, "doc_artifact_ids": {"benchmark": "artifact-b"},
+                "code_artifact_ids": {}, "verdict": verdict,
+            }
+
+        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):
+            results = scheduler.run_scheduler_once(self.db_path, goals, max_workers=3)
+
+        by_goal = {r["goal_text"]: r for r in results}
+        self.assertEqual(by_goal["goal-ok-1"]["result"]["final_status"], "done")
+        self.assertEqual(by_goal["goal-ok-2"]["result"]["final_status"], "done")
+        self.assertIn("error", by_goal["goal-doomed"])
+
+        conn = schema.connect(self.db_path)
+        doomed_status = conn.execute(
+            "SELECT status, blocked_reason FROM task WHERE goal_text = 'goal-doomed'"
+        ).fetchone()
+        conn.close()
+        self.assertEqual(doomed_status, ("blocked_pending_human", "loop_limit_reached"))
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -208,5 +208,61 @@ class RunGraphClassifierBlockedTest(RunGraphTestBase):
         self.assertEqual(status, "blocked")
+class VerifyFnTest(RunGraphTestBase):
+    """REQ-02 boundary: verify_fn intercepts a bad node output before it
+    propagates downstream, via one regeneration attempt."""
+
+    def test_verify_fn_passing_immediately_does_not_regenerate(self):
+        manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []}
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "consistent prd"}) as mock_execute:
+            outputs = wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+                verify_fn=lambda node, output: (True, None),
+            )
+        self.assertEqual(outputs["prd"], {"content": "consistent prd"})
+        self.assertEqual(mock_execute.call_count, 1)
+        run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
+        self.assertEqual(run_status, "running")  # untouched -- verify_fn never failed
+
+    def test_verify_fn_failing_once_regenerates_and_succeeds(self):
+        manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []}
+        contents = iter(["contradictory prd", "fixed prd"])
+        checked = []
+
+        def verify_fn(node, output):
+            checked.append(output["content"])
+            return output["content"] == "fixed prd", ["scenario A contradicts scenario B"]
+
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", side_effect=lambda *a, **k: {"content": next(contents)}):
+            outputs = wgo.run_graph(
+                self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, verify_fn=verify_fn,
+            )
+        self.assertEqual(outputs["prd"], {"content": "fixed prd"})
+        self.assertEqual(checked, ["contradictory prd", "fixed prd"])
+        run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
+        self.assertEqual(run_status, "verifying")
+        reflect_row = self.conn.execute(
+            "SELECT status FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'prd-reflect'"
+        ).fetchone()
+        self.assertEqual(reflect_row[0], "done")
+
+    def test_verify_fn_failing_twice_raises_and_blocks_run(self):
+        manifest = {"nodes": [{"node_id": "prd", "depends_on": []}], "edges": []}
+        with patch.object(work_graph_execute, "write_workflow_script"), \
+             patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "still contradictory"}):
+            with self.assertRaises(wgo.ConsistencyCheckFailedError):
+                wgo.run_graph(
+                    self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest,
+                    lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA,
+                    verify_fn=lambda node, output: (False, ["still broken"]),
+                )
+        run_status = self.conn.execute("SELECT status FROM work_graph_run WHERE run_id = 'run-1'").fetchone()[0]
+        self.assertEqual(run_status, "blocked")
+
+
 if __name__ == "__main__":
     unittest.main()
@@ -147,6 +147,33 @@ class RetryOnSchemaFailureTest(WorkNodeRunTestBase):
         ).fetchone()
         self.assertEqual(row, ("done", 2))
+    def test_schema_failure_passes_through_reflecting_before_retrying(self):
+        # REQ-03 failure: "status 应为 reflecting->retrying" -- previously
+        # jumped straight to 'retrying', 'reflecting' was never written.
+        # execution_log_file/work_node_execution have no separate history
+        # table, so the only way to observe the transient write is sqlite's
+        # own trace callback (every SQL statement it actually executes).
+        executed_sql = []
+        self.conn.set_trace_callback(executed_sql.append)
+
+        attempts = {"n": 0}
+
+        def run_fn(cwd, model=None):
+            attempts["n"] += 1
+            if attempts["n"] == 1:
+                return {"wrong_key": True}
+            return {"value": "ok"}
+
+        try:
+            work_node_run.run_work_node(
+                self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=run_fn, max_retries=2
+            )
+        finally:
+            self.conn.set_trace_callback(None)
+
+        reflecting_writes = [sql for sql in executed_sql if "status = 'reflecting'" in sql]
+        self.assertEqual(len(reflecting_writes), 1)
+
     def test_exceeding_max_retries_blocks_the_node(self):
         with self.assertRaises(work_node_run.SchemaValidationFailedError):
             work_node_run.run_work_node(
@@ -23,6 +23,22 @@ Validated for real (2026-08-01): a one-node workflow script with a custom
 `agentType` + `schema` genuinely loads the target org_node's migrated
 persona (confirmed via a real call asking the agent to self-identify) and
 returns real `structured_output` -- see 05 §5.4.1.5.
+
+**`verify_fn` (2026-08-02, PH4-closure audit before starting PH5)**: closes
+REQ-02's boundary case (PRD internal-contradiction check must intercept
+before `verifying`, not silently pass) -- found genuinely missing
+(`artifact_write.write_artifact` always marks `validation_status='passed'`
+unconditionally, and no Unit ever built a consistency-judgment call).
+`work_graph_run.status` was ALSO never once written past its initial
+'planning' value anywhere in this codebase before this -- `verify_fn`'s
+'reflecting'/'verifying'/'blocked' writes below are the first real use of
+that enum. Scoped deliberately narrow: one optional per-node verify
+callback, one regeneration attempt on failure (matching 04-P's "回灌受影响
+的最小子图" -- for a single content-producing node, the minimal affected
+subgraph IS just itself), escalate to `work_graph_run.status='blocked'` +
+raise if the regenerated content still fails. Not a general multi-node
+re-planning mechanism -- that would be new, unscoped complexity 04-P itself
+doesn't describe either.
 """
 import json
@@ -32,6 +48,12 @@ import uuid
 from src import work_graph_execute, work_node_run
+class ConsistencyCheckFailedError(Exception):
+    """`verify_fn` rejected a node's output twice in a row (original +
+    one regeneration attempt) -- `work_graph_run.status` is already
+    'blocked' by the time this is raised."""
+
+
 def topological_order(graph_manifest):
     """Kahn's algorithm. Dependencies are read from BOTH `node['depends_on']`
     and `graph_manifest['edges']` and merged -- `plan_work_graph()`'s own
@@ -136,7 +158,7 @@ def _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_sche
 def run_graph(
     conn, run_id, org_node, skill_id, target_folder_path, graph_manifest,
     build_prompt, input_schema, output_schema, timeout=600, needs_worktree=False,
-    sleep=time.sleep, now=time.time,
+    verify_fn=None, sleep=time.sleep, now=time.time,
 ):
     """Runs every node in `graph_manifest` (topologically ordered) through
     `run_work_node()`, each backed by a real one-node dynamic-workflow
@@ -159,6 +181,17 @@ def run_graph(
     writing code (Coder) should pass this; callers only producing document
     content (Planner) don't need it.
+    `verify_fn(node, output) -> (ok: bool, feedback)`, if given, is called
+    right after each node completes, BEFORE moving on to nodes that depend
+    on it -- lets a caller intercept a bad output before it propagates
+    downstream (e.g. REQ-02's PRD-internal-contradiction check), not just
+    after the fact. On `ok=False`: `work_graph_run.status` -> 'reflecting',
+    the SAME node is regenerated once with `feedback` appended to its
+    prompt, and re-verified. If it fails again: `work_graph_run.status`
+    -> 'blocked' and `ConsistencyCheckFailedError` is raised. If it now
+    passes: `work_graph_run.status` -> 'verifying' and the regenerated
+    output replaces the original in `outputs`.
+
     Returns {node_id: output_payload}.
     """
     ensure_placeholder_skill(conn, skill_id, org_node)
@@ -178,5 +211,53 @@ def run_graph(
             needs_worktree=needs_worktree, repo_path=target_folder_path,
             sleep=sleep, now=now,
         )
+
+        if verify_fn is not None:
+            output = _verify_and_maybe_regenerate(
+                conn, run_id, org_node, skill_id, target_folder_path, node, node_id, prompt,
+                output, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
+            )
+
         outputs[node_id] = output
     return outputs
+
+
+def _verify_and_maybe_regenerate(
+    conn, run_id, org_node, skill_id, target_folder_path, node, node_id, prompt,
+    output, input_schema, output_schema, timeout, needs_worktree, verify_fn, sleep, now,
+):
+    ok, feedback = verify_fn(node, output)
+    if ok:
+        return output
+
+    conn.execute("UPDATE work_graph_run SET status = 'reflecting' WHERE run_id = ?", (run_id,))
+    conn.commit()
+
+    reflect_node_id = node_id + "-reflect"
+    reflect_node_exec_id = _ensure_node_row(
+        conn, run_id, reflect_node_id, org_node, skill_id,
+        "work_graph_node_input.schema.json", "work_graph_node_output.schema.json",
+        {"node_id": node_id, "feedback": feedback},
+    )
+    reflect_prompt = (
+        prompt + "\n\nA consistency reviewer flagged issues with your previous draft: "
+        + "; ".join(feedback or []) + "\nProduce a corrected version resolving these issues."
+    )
+    reflect_run_fn = _make_run_fn(target_folder_path, reflect_node_exec_id, org_node, reflect_prompt, output_schema, timeout)
+    regenerated = work_node_run.run_work_node(
+        conn, reflect_node_exec_id, {"node_id": node_id}, input_schema, output_schema, reflect_run_fn,
+        needs_worktree=needs_worktree, repo_path=target_folder_path,
+        sleep=sleep, now=now,
+    )
+
+    ok2, feedback2 = verify_fn(node, regenerated)
+    if not ok2:
+        conn.execute("UPDATE work_graph_run SET status = 'blocked' WHERE run_id = ?", (run_id,))
+        conn.commit()
+        raise ConsistencyCheckFailedError(
+            "node {} still failed verification after one regeneration attempt: {}".format(node_id, feedback2)
+        )
+
+    conn.execute("UPDATE work_graph_run SET status = 'verifying' WHERE run_id = ?", (run_id,))
+    conn.commit()
+    return regenerated
@@ -320,6 +320,17 @@ def run_work_node(
                     remove_worktree(repo_path, worktree_path)  # 03-C §3.6: "执行完成后清理 worktree"
                 return output_payload
+            # 04-P: verifying -> reflecting ("记录差距，决定重试路径") before the
+            # actual retry/give-up decision below -- REQ-03 failure's own
+            # acceptance text names this intermediate status explicitly;
+            # previously this jumped straight to 'retrying'/'blocked' and
+            # 'reflecting' was never written anywhere (found during the
+            # PH4-closure audit, 2026-08-02).
+            conn.execute(
+                "UPDATE work_node_execution SET status = 'reflecting' WHERE node_exec_id = ?", (node_exec_id,)
+            )
+            conn.commit()
+
             retry_count += 1
             if retry_count > max_retries:
                 conn.execute(