diff --git a/docs/development-plan.md b/docs/development-plan.md index 784933b..4e42857 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -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/.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`）+ 长时间运行稳定性压测 diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index b52f2a0..743644a 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -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，正式收口"}, diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 75fd73f..a444d29 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -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 = "Confirm you're ready to begin. Take no other action yet." _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: diff --git a/src/scheduler.py b/src/scheduler.py index a291351..c7fd3e2 100644 --- a/src/scheduler.py +++ b/src/scheduler.py @@ -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() diff --git a/src/test_backend_server.py b/src/test_backend_server.py index 3eefb37..7404c89 100644 --- a/src/test_backend_server.py +++ b/src/test_backend_server.py @@ -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() diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index f86bb03..ecea726 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -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"): diff --git a/src/test_e2e_pipeline_e2e.py b/src/test_e2e_pipeline_e2e.py index 8ed3e67..2d31556 100644 --- a/src/test_e2e_pipeline_e2e.py +++ b/src/test_e2e_pipeline_e2e.py @@ -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() diff --git a/src/test_scheduler.py b/src/test_scheduler.py index 1da7cb2..ed1c73e 100644 --- a/src/test_scheduler.py +++ b/src/test_scheduler.py @@ -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() diff --git a/src/test_work_graph_orchestrate.py b/src/test_work_graph_orchestrate.py index 0c86eab..c3e5f37 100644 --- a/src/test_work_graph_orchestrate.py +++ b/src/test_work_graph_orchestrate.py @@ -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() diff --git a/src/test_work_node_run.py b/src/test_work_node_run.py index 54dcaf6..46e9a35 100644 --- a/src/test_work_node_run.py +++ b/src/test_work_node_run.py @@ -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( diff --git a/src/work_graph_orchestrate.py b/src/work_graph_orchestrate.py index ce373e9..595db2b 100644 --- a/src/work_graph_orchestrate.py +++ b/src/work_graph_orchestrate.py @@ -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 diff --git a/src/work_node_run.py b/src/work_node_run.py index 1400781..d10723f 100644 --- a/src/work_node_run.py +++ b/src/work_node_run.py @@ -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(