diff --git a/docs/development-plan.md b/docs/development-plan.md index 74430b5..e7c9e73 100644 --- a/docs/development-plan.md +++ b/docs/development-plan.md @@ -237,6 +237,7 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。 - **U-024.5（2026-07-30 新增）[Value Path]** REQ-12 完整验收（事件多样性联调） - 依赖：U-024, U-015, U-016, U-018, U-019, U-020, U-021（需要 PH2 大部分 Unit 已上线，事件类型覆盖足够丰富才能真正验证 boundary/failure 场景——例如 usage-limit 等待、护栏拦截、Skill 回滚各自产生的 `execution_event` 都要真实出现过） - Done：REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准，U-024 本身只解决"链路通不通" + - **实现补充（2026-08-01）——U-031 回归审计中发现的真实缺口**：核对本 Unit 是否真的 Done 时，重新对照 06 节 REQ-12 的 boundary 用例原文——"重连后能补齐断线期间遗漏的关键状态（通过重新拉取 task/work_graph_run 当前快照兜底）"——grep 了 `frontend/app.js` 里所有 `onStatusChange` 用法，发现只有 `v_dashboard` 做了一点反应（重连时把断线提示 banner 隐藏），而且连它自己也没有真的重新拉取数据；`v_task_board`/`v_task_detail`（含内嵌的 `v_graph_visualizer`）的 `onStatusChange` 全是空函数 `() => {}`。也就是说这条 boundary 用例此前从未被真正实现过，U-024/U-024.5 之前的"REQ-12 完整通过"是不准确的。修复：`mountDashboard`/`mountTaskBoard`/`mountTaskDetail` 三处的 `connectEventSocket` 都加了 `wasDisconnected` 标志，在状态从 `disconnected` 转回 `connected` 时调用各自页面已有的刷新函数（`refreshTasksAndOrgGraph`+`refreshBudget` / `refresh` / `refresh`）重新拉取当前快照。`v_artifact_browser` 的 `onStatusChange` 保持空——REQ-12 命名的验收范围只到 `v_dashboard`/`v_graph_visualizer`/`v_task_detail`，artifact 浏览器不在其中，不属于这次修复范围。浏览器扩展本次会话不可用，验证方式：`node --check` 过语法 + 全量 `unittest` 回归（361 tests）；WS 重连本身的行为逻辑此前已被 `src/test_backend_server.py`/Node 原生 WebSocket 端到端验证过（U-024/U-027），这次只是前端消费侧新增的刷新调用，未改动服务端重连协议本身。 - 覆盖：REQ-12（完整） - **U-025 [Value Path]** `v_dashboard` + `v_task_board` + `v_task_detail` @@ -288,6 +289,15 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。 - **U-031 [Risk Gate]** REQ-01~14 全量自动化回归套件 - 依赖：PH1/PH2/PH3 全部 Unit + - Done：06 节全部 14 条 REQ 的全部用例通过一次完整回归（原文档缺 Done/覆盖行，2026-08-01 补上） + - **实现补充（2026-08-01）——发现并修复一个结构性集成缺口，这是本 Unit 目前为止最重要的产出**：逐条核对 REQ-01~14 验收用例对应的测试覆盖时，发现 REQ-02/REQ-03 验收锚点明确列出的控制器覆盖（`c_work_graph_plan`/`c_work_graph_execute`/`c_work_node_run`）——也就是"Org Graph 触发 → Work Graph 规划 → Work Graph 执行 → 每个 Work Node 独立运行"这套两层架构——虽然在 U-011/U-012/U-013/U-020/U-021 各自都建好、也各自做过完整单元测试（U-013/U-021 甚至做过真实 git worktree 验证），但 `plan_work_graph()`/`execute_work_graph()`/`run_work_node()` 这三个函数彼此之间、以及和唯一真正跑通验证过的闭环（U-014 的 `e2e_pipeline.py`，snake-game 那次真实生成）之间，从未被真正接起来过——`e2e_pipeline.py` 走的是完全独立、更简单的路径（对 Planner/Coder/Validator 各打一次 `org_node_trigger.trigger_org_node()` 直连调用），完全绕开了这套两层架构。**这个决策已跟用户确认（AskUserQuestion），选择立即补上集成而不是把现状订正进文档了事**，因为 REQ-06（断点恢复）的验收要求"重启后不重跑已完成节点"，只有 `work_node_run.py` 的按节点 Python 层持久化状态能做到，这和"整个 Work Graph 包在一次不透明的 `claude -p` 调用里"（`execute_work_graph()` 原本设想的用法）根本互斥，所以这个缺口不是"锦上添花"，是 REQ-06 能不能算真正满足的前提。 + - **设计**：新增 `src/work_graph_orchestrate.py`，每个 Work Node 的真实执行走**它自己单独的一次动态工作流调用**（`write_workflow_script` + `execute_work_graph`），不是把整张图塞进一次调用——这是刻意的取舍，牺牲了 03-C §3.5 原本"多节点编排摊薄到一次调用里，编排本身不消耗 token"的效率立意，换来 REQ-06 要求的按节点级持久化/断点恢复。`topological_order()` 对 `graph_manifest` 做拓扑排序，同时读 `depends_on` 和 `edges` 并合并（原因见下一条发现）。`build_prompt(node, outputs)` 在拓扑序上逐节点调用，`outputs` 里已经有该节点全部上游依赖的真实产出内容，天然支持"prd 的 prompt 需要嵌入 brd 的真实内容"这种链式依赖，也天然支持断点恢复（`run_work_node` 对已 `done` 节点直接返回持久化结果，恢复执行时上游内容照样能取到）。 + - **`work_node_run.py` 补了一个真实 bug**：原来只捕获 `_UsageLimitSignal`，任何其他异常（比如新集成会遇到的 `ClassifierBlockedError`）会直接从 `run_work_node()` 冒泡出去，`work_node_execution.status` 永远停在 `running`，没有任何"为什么卡住了"的记录。改成任何非 usage-limit 异常都先把节点标记 `blocked` 再重新抛出。 + - **真实调用验证时连续发现两个集成 bug，都已修复**：① 第一次真实跑 `plan_work_graph()` 时，模型返回的 `graph_manifest` 每个节点用的是 `id` 而不是文档要求的 `node_id`，而且完全没填任何节点的 `depends_on`（依赖关系只体现在 `edges` 里）——`plan_work_graph()` 本身从来没做过 schema 校验（`--json-schema` 配 `--agent` 不生效，05 §5.4.1.1 早就记录过），这是真实模型输出会偏离文档契约的活例子。新增 `work_graph_plan._normalize_graph_manifest()`，在 parse 完、持久化前统一把 `id`→`node_id`、从 `edges` 反推缺失的 `depends_on`——这样不只是我这个新集成受益，`v_task_detail` 的 to-do 树、`v_graph_visualizer` 的 DAG 分层以后遇到真实（而不是测试里手写整齐的）manifest 时也不会因为字段名对不上而显示错误。② 修好第一个 bug 后，第二次真实跑又炸了：`work_graph_orchestrate.run_graph()` 调 `run_work_node()` 时忘了传 `repo_path`，`cwd` 解析成 `None`，写工作流脚本直接崩——两次真实调用暴露的都是"单元测试全绿但没人真正把这几个函数接起来跑过"的典型症状，修完重跑，真实 4 篇文档（BRD→PRD→SDD→Benchmark）全部生成成功、血缘链正确、内容长度合理。 + - **`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，不需要动。 + - 覆盖：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`）+ 长时间运行稳定性压测 - 依赖：U-031 diff --git a/docs/sdd/05-实施约束与依赖.md b/docs/sdd/05-实施约束与依赖.md index 220f409..55dcc06 100644 --- a/docs/sdd/05-实施约束与依赖.md +++ b/docs/sdd/05-实施约束与依赖.md @@ -61,6 +61,13 @@ - **应对方式**：参照 `create_server()` 已有的 `db_path` 参数模式，加了 `skill_repo_path` 参数：默认值是 `skill_query._REPO_ROOT`（graphlemon 真实仓库根，生产环境下这就是唯一正确的行为——Skill 变更本来就应该落到这里），但可以在构造 server 时覆盖；测试统一传一个 `tempfile` 建的临时仓库。**这个参数只在 Python 层（进程启动/测试构造）可设，绝不通过 HTTP 请求体/query string 暴露给客户端**——否则等于开了一个"客户端指定任意目录、服务端替它跑 git commit"的口子，属于路径注入/任意写入范畴。 - 这不是"发现"的约束（不是实测跑出来的意外行为），是设计阶段就该考虑到、写代码时主动做的防护，记录下来是为了以后改这块代码的人不要图省事把 `skill_repo_path` 从请求里读出来。 +## 5.4.1.5 动态工作流内部 `agent()` 的 `agentType`+`schema` 组合已实测可用（发现，2026-08-01，U-031） + +- 背景：U-031 补 Work Graph 三模块集成（`work_graph_orchestrate.py`）时，每个 Work Node 的真实执行都要用一份一次性的动态工作流脚本，脚本内唯一的一次 `agent()` 调用需要同时做到两件事——① 加载目标 org_node（Planner/Coder/Validator）迁移后的人设（`agentType: ''`，对应 `.claude/agents/.md`），② 拿到结构化输出（`schema: {...}`）。这和 §5.4.1.1 记录的"顶层 `--agent` + `--json-schema` 不能同时生效"表面上是同一类组合，但机制完全不同：§5.4.1.1 是 CLI 顶层参数 `claude --agent ... --json-schema ...`；这里是动态工作流脚本**内部** JS DSL 的 `agent(prompt, {agentType, schema})` 调用，二者互不影响，不能假设前者的限制也适用于后者——必须单独验证，不能凭直觉套用。 +- **实测确认（两次独立验证）**：① 先用一个最小 spike（仅一个一节点工作流，唯一动作是 `agent({agentType:'planner', schema:{type:'object',...}})`）验证——`structured_output` 正确按 schema 返回，且额外追问角色身份时模型明确自称"Planning Agent (PL)"，证明迁移后的 SOUL.md 人设确实通过 `agentType` 正确加载（不是退化成某个通用 agent 后蒙对了格式）；② 后续 `e2e_pipeline.plan_documents()` 完整四节点（BRD→PRD→SDD→Benchmark）真实跑通，再次在真实多节点场景下确认这个组合稳定可用。 +- **实际用法**：`work_graph_orchestrate._render_node_script()` 渲染的脚本固定形如 `const result = await agent(prompt, { agentType: , schema: })`，`execute_work_graph()` 直接把这个一次性脚本文件写到 `.claude/workflows/` 后调用 `claude -p 'Run /...'` 执行，不复用同一个脚本名跑第二个节点（每个 `node_exec_id` 生成自己独立的脚本文件名，避免并发/重试之间互相覆盖）。 +- 同 §5.4.1.1/§5.4.1.2：这条结论是实测发现的，官方文档没有专门说明动态工作流 `agent()` 调用里 `agentType`+`schema` 的组合行为，不排除后续 Claude Code 版本变化后需要重新验证。 + ## 5.4.2 Claude Code plan 额度等待重试参数（架构文档 7.5 节决策，2026-07-29） - 退避轮询间隔（5→10→20→30 分钟，上限 30 分钟）、超时兜底阈值（session 8 小时、weekly 9 天）是 MVP 阶段的硬编码默认值，不在管理后台开放用户配置，和 `log_retention_config` 的"保留天数可配、清理频率不可配"是同一个 MVP 简化原则 diff --git a/docs/sdd/unit_matrix.json b/docs/sdd/unit_matrix.json index 92253d2..6828815 100644 --- a/docs/sdd/unit_matrix.json +++ b/docs/sdd/unit_matrix.json @@ -39,7 +39,7 @@ {"id": "U-022", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_project_setup + c_pick_folder 全平台（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-007", "U-008"], "coverage": {"model": [], "view": ["v_project_setup"], "controller": ["c_pick_folder", "c_project_create"], "process": [], "requirement": ["REQ-11"]}, "done": "REQ-11 里 v_project_setup/c_pick_folder/c_project_create 部分通过；c_pick_folder 补齐 Windows/Linux；c_project_create 首次接上 HTTP 路由；REQ-11 完整 happy（含建 task）随 U-023 累计满足"}, {"id": "U-023", "phase": "PH3", "sub_phase": "PH3-S1", "name": "v_goal_composer + v_bug_feedback（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-022"], "coverage": {"model": [], "view": ["v_goal_composer", "v_bug_feedback"], "controller": ["c_task_create", "c_bug_report_create"], "process": [], "requirement": []}, "done": "能通过 UI 创建 task 和人工 bug_record；补上 c_bug_report_create 的 project_id 必填缺口（03-C input_contract 遗漏）；related_task_id 简化为文本输入，等 U-025 的 c_task_list 落地后换真下拉"}, {"id": "U-024", "phase": "PH3", "sub_phase": "PH3-S2", "name": "c_ws_subscribe + execution_event 实时推送（最小连通版，见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-012", "U-013"], "coverage": {"model": ["execution_event"], "view": [], "controller": ["c_ws_subscribe"], "process": [], "requirement": ["REQ-12"]}, "done": "推送链路本身连通，≤2 秒延迟达成；REQ-12 happy 用例通过（boundary/failure 留给 U-024.5）；手写 RFC 6455 握手+帧编解码（无第三方 websockets 依赖），轮询式推送（无单一写路径可插桩），Node 原生 WebSocket 客户端验证过真实互操作性"}, - {"id": "U-024.5", "phase": "PH3", "sub_phase": "PH3-S2", "name": "REQ-12 完整验收：事件多样性联调（2026-07-30 新增）", "classification": "Value Path", "depends_on": ["U-024", "U-015", "U-016", "U-018", "U-019", "U-020", "U-021"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-12"]}, "done": "REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准"}, + {"id": "U-024.5", "phase": "PH3", "sub_phase": "PH3-S2", "name": "REQ-12 完整验收：事件多样性联调（见 development-plan.md 2026-08-01 实现补充）", "classification": "Value Path", "depends_on": ["U-024", "U-015", "U-016", "U-018", "U-019", "U-020", "U-021"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-12"]}, "done": "REQ-12 全部 happy/boundary/failure 用例通过；PH3 Phase 验收里的 REQ-12 以本 Unit 为准。boundary 用例（重连补齐断线期间遗漏状态）此前从未真正实现，仅 v_dashboard 有 UI banner 切换、无数据重拉——2026-08-01 于 U-031 回归审计中发现并补上：v_dashboard/v_task_board/v_task_detail（含内嵌的 v_graph_visualizer）的 connectEventSocket onStatusChange 均已在 disconnected→connected 转换时重新拉取当前快照；v_artifact_browser 不在 REQ-12 命名范围内，未改动"}, {"id": "U-025", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_dashboard + v_task_board + v_task_detail（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_dashboard", "v_task_board", "v_task_detail"], "controller": ["c_task_list", "c_task_get", "c_org_loop_route"], "process": [], "requirement": []}, "done": "三个页面可正常展示实时数据；补上此前无 Unit 认领的 c_task_list/c_task_get HTTP 路由，c_org_loop_route 首次接上 HTTP；v_graph_visualizer/v_log_viewer 子模块占位等 U-026/U-027"}, {"id": "U-026", "phase": "PH3", "sub_phase": "PH3-S2", "name": "v_graph_visualizer（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": [], "view": ["v_graph_visualizer"], "controller": [], "process": [], "requirement": []}, "done": "Work Graph DAG 分层视图含 waiting_for_quota 信息态；Org Graph 层复用 v_task_detail 已有的 org-strip 面板，不重复渲染"}, {"id": "U-027", "phase": "PH3", "sub_phase": "PH3-S3", "name": "execution_log_file + 日志查看器（见 development-plan.md 2026-07-31 实现补充）", "classification": "Value Path", "depends_on": ["U-024"], "coverage": {"model": ["execution_log_file", "log_retention_config"], "view": ["v_log_viewer"], "controller": ["c_execution_log_get", "c_ws_log_tail"], "process": ["p_log_retention_cleanup"], "requirement": ["REQ-13"]}, "done": "REQ-13 happy/boundary/failure 通过；补上此前从未存在的 append_log 写路径；顺带修了 U-024 就存在的 server_close() 并发 segfault 隐患（conn_lock 未覆盖）"}, @@ -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 全量自动化回归", "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 的全部用例通过一次完整回归"}, + {"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-032", "phase": "PH4", "sub_phase": "PH4-S2", "name": "多 Goal 并发 + 长时间运行稳定性压测", "classification": "Risk Gate", "depends_on": ["U-031"], "coverage": {"model": ["org_concurrency_config"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "并发数调高后多个独立 Goal 可同时稳定运行"}, {"id": "U-033", "phase": "PH4", "sub_phase": "PH4-S3", "name": "A-2 假设收口", "classification": "Risk Gate", "depends_on": ["U-004"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": [], "assumption": ["A-2"]}, "done": "A-2 状态确认为 confirmed 或记录实际偏差"}, diff --git a/frontend/app.js b/frontend/app.js index 533411e..8e900b7 100644 --- a/frontend/app.js +++ b/frontend/app.js @@ -548,6 +548,7 @@ function mountDashboard() { refreshBudget(); eventTimeline.innerHTML = '
暂无事件
'; + let wasDisconnected = false; const socket = connectEventSocket(projectId, null, { onEvent: (event) => { prependEvent(event); @@ -556,6 +557,16 @@ function mountDashboard() { }, onStatusChange: (status) => { wsBanner.style.display = status === "disconnected" ? "block" : "none"; + if (status === "disconnected") { + wasDisconnected = true; + } else if (status === "connected" && wasDisconnected) { + // REQ-12 boundary: reconnect doesn't replay every missed event, but + // must at least catch up on current state via a fresh snapshot + // fetch (03-C §3.14's own documented reconnect story). + wasDisconnected = false; + refreshTasksAndOrgGraph(); + refreshBudget(); + } }, }); registerRouteCleanup(() => socket.close()); @@ -637,11 +648,20 @@ function mountTaskBoard() { orgNodeEl.addEventListener("change", refresh); refresh(); + let wasDisconnected = false; const socket = connectEventSocket(projectId, null, { onEvent: (event) => { if (event.event_type === "task_status_changed") refresh(); }, - onStatusChange: () => {}, + onStatusChange: (status) => { + if (status === "disconnected") { + wasDisconnected = true; + } else if (status === "connected" && wasDisconnected) { + // REQ-12 boundary: catch up via a fresh snapshot on reconnect. + wasDisconnected = false; + refresh(); + } + }, }); registerRouteCleanup(() => socket.close()); } @@ -969,9 +989,18 @@ function mountTaskDetail(taskId) { refresh(); + let wasDisconnected = false; const socket = connectEventSocket(projectId, taskId, { onEvent: () => refresh(), - onStatusChange: () => {}, + onStatusChange: (status) => { + if (status === "disconnected") { + wasDisconnected = true; + } else if (status === "connected" && wasDisconnected) { + // REQ-12 boundary: catch up via a fresh snapshot on reconnect. + wasDisconnected = false; + refresh(); + } + }, }); registerRouteCleanup(() => { socket.close(); diff --git a/src/e2e_pipeline.py b/src/e2e_pipeline.py index 3269f35..75fd73f 100644 --- a/src/e2e_pipeline.py +++ b/src/e2e_pipeline.py @@ -5,51 +5,82 @@ benchmark/tests and produces a root-cause-attributed verdict). Criteria 4-7 (Org-level Loop, breakpoint recovery, ...) are explicitly out of this Unit's scope -- PH2 (U-017/U-021). +**Rewired 2026-08-01 (U-031 regression audit)**: `plan_documents` used to +call `org_node_trigger.trigger_org_node()` directly, bypassing +`work_graph_plan.plan_work_graph()`/`work_graph_execute.execute_work_graph()` +/`work_node_run.run_work_node()` entirely -- those three modules were each +fully unit-tested (and even real-API-validated) in isolation since PH1/PH2, +but nothing in production ever called any of them; this was the one real, +validated closed loop, and it used a completely separate, simpler mechanism. +REQ-02's own coverage line names `c_work_graph_plan`/`c_work_graph_execute` +explicitly, so that gap is now closed: `plan_documents` calls +`plan_work_graph()` for a real graph_manifest, then +`work_graph_orchestrate.run_graph()` executes each node for real (see that +module's docstring for why each node gets its own one-node dynamic-workflow +script rather than one big multi-node script -- REQ-06 breakpoint recovery +needs it). `implement_code` (Coder) goes through the same per-node +mechanism with a synthetic single-node manifest, since REQ-03's own +coverage only names `c_work_node_run`, not graph planning -- there is no +real "plan a Coder module graph" LLM capability built (a much bigger, +separate undertaking; planning docs and planning code modules are +different problems), so a one-node graph is the honest minimum that makes +`c_work_node_run` genuinely load-bearing for Coder too. +`validate_implementation` (Validator) is unchanged -- REQ-04's own +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. Content generation is -separate, plain org_node_trigger.trigger_org_node calls (no status -side-effects) so each document/step's real text is available to write as an -artifact -- trigger_org_node_full's own contract (a status dict, no text) -isn't changed to accommodate this. +phase with a lightweight "confirm ready" prompt. """ +import json import os import subprocess +import uuid -from src import artifact_write, org_node_lifecycle, org_node_trigger, task_create +from src import ( + artifact_write, org_node_lifecycle, org_node_trigger, task_create, + work_graph_orchestrate, work_graph_plan, +) _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"]} + _DOC_PROMPTS = { "brd": ( "Write a BRD (Business Requirement Document) for this goal, following your own " "SOUL.md methodology: {goal_text}\n" - "Output ONLY the BRD content in Markdown -- no preamble, no meta-commentary about " - "what you're about to do." + "Reply with the document content (Markdown) via the required schema -- no preamble, " + "no meta-commentary about what you're about to do." ), "prd": ( "Here is the approved BRD for goal '{goal_text}':\n\n{brd_content}\n\n" "Write the PRD (Product Requirement Document) based on this BRD. " - "Output ONLY the PRD content in Markdown." + "Reply with the document content (Markdown) via the required schema." ), "sdd": ( "Here is the approved PRD for goal '{goal_text}':\n\n{prd_content}\n\n" "Write the SDD (System Design Document) based on this PRD, including a Development " - "Unit breakdown. Output ONLY the SDD content in Markdown." + "Unit breakdown. Reply with the document content (Markdown) via the required schema." ), "benchmark": ( "Here is the approved SDD for goal '{goal_text}':\n\n{sdd_content}\n\n" "Write the Benchmark/acceptance-test document based on this SDD: concrete test " "scenarios (happy/boundary/failure) with observable pass criteria. " - "Output ONLY the Benchmark content in Markdown." + "Reply with the document content (Markdown) via the required schema." ), } +_CODE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]} +_CODE_OUTPUT_SCHEMA = {"type": "object", "required": ["response"]} + _CODER_PROMPT = ( "Implement this goal per the approved SDD below. Write real code and real tests, run the " "tests yourself, and only stop once they pass (or once you've exhausted reasonable attempts " - "and can explain why not). Goal: {goal_text}\n\nSDD:\n\n{sdd_content}" + "and can explain why not). Reply via the required schema with a `response` field describing " + "what you did and the real test commands/output. Goal: {goal_text}\n\nSDD:\n\n{sdd_content}" ) _VALIDATOR_PROMPT = ( @@ -61,42 +92,86 @@ _VALIDATOR_PROMPT = ( ) -def plan_documents(conn, task_id, run_id, goal_text, target_folder_path, timeout=300): - """Architecture §11 criterion 1: real Planner calls producing BRD->PRD->SDD - ->Benchmark, each written via c_artifact_write with correct lineage. - Returns {"brd": artifact_id, "prd": ..., "sdd": ..., "benchmark": ...}. +def _planner_prompt_builder(goal_text, node_artifact_types): + """node_artifact_types: dict[node_id] -> artifact_type, read off the + real graph_manifest before running it (build_prompt only ever sees one + node/outputs-so-far pair at a time, not the whole manifest).""" + def build_prompt(node, outputs): + node_id = node["node_id"] + artifact_type = node_artifact_types.get(node_id, node_id) + content_kwargs = {} + for dep_id in (node.get("depends_on") or []): + dep_output = outputs.get(dep_id) + if dep_output is not None: + dep_type = node_artifact_types.get(dep_id, dep_id) + content_kwargs[dep_type + "_content"] = dep_output["content"] + return _DOC_PROMPTS[artifact_type].format(goal_text=goal_text, **content_kwargs) + return build_prompt + + +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 + work_graph_orchestrate.run_graph) producing BRD->PRD->SDD->Benchmark, + each written via c_artifact_write with correct lineage. + Returns {"run_id": ..., "artifact_ids": {artifact_type: artifact_id}}. """ + plan_result = work_graph_plan.plan_work_graph( + conn, "planner", task_id, goal_text, target_folder_path, timeout=timeout + ) + run_id = plan_result["run_id"] + graph_manifest = plan_result["graph_manifest"] + node_artifact_types = {n["node_id"]: n.get("artifact_type", n["node_id"]) for n in graph_manifest["nodes"]} + + 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, + ) + artifact_ids = {} - contents = {} - prev_type = None - for artifact_type in ("brd", "prd", "sdd", "benchmark"): - prompt = _DOC_PROMPTS[artifact_type].format( - goal_text=goal_text, **{k + "_content": v for k, v in contents.items()} - ) - content = org_node_trigger.trigger_org_node("planner", prompt, target_folder_path, timeout=timeout) - deps = [artifact_ids[prev_type]] if prev_type else [] + for node in work_graph_orchestrate.topological_order(graph_manifest): + node_id = node["node_id"] + artifact_type = node_artifact_types[node_id] + deps = [ + artifact_ids[node_artifact_types[dep_id]] + for dep_id in (node.get("depends_on") or []) + if node_artifact_types.get(dep_id) in artifact_ids + ] result = artifact_write.write_artifact( conn, target_folder_path, artifact_type, task_id, run_id, - ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), content, + ".graphlemon/runs/{}/{}.md".format(task_id, artifact_type), outputs[node_id]["content"], depends_on_artifact_ids=deps, ) artifact_ids[artifact_type] = result["artifact_id"] - contents[artifact_type] = content - prev_type = artifact_type - return artifact_ids + return {"run_id": run_id, "artifact_ids": artifact_ids} -def implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path, timeout=1200): - """Architecture §11 criterion 2: real Coder call, then registers whatever - the Coder actually committed as code_diff (the diff itself) and - test_report (the Coder's own response text, which per its migrated SOUL.md - is required to include real test commands + real output) artifacts. +def implement_code(conn, task_id, sdd_content, goal_text, target_folder_path, timeout=1200): + """Architecture §11 criterion 2: a real Coder call via c_work_node_run + (work_graph_orchestrate.run_graph with a synthetic single-node + manifest -- see module docstring for why there's no real Coder graph + planner), then registers whatever the Coder actually committed as + code_diff (the diff itself) and test_report (the Coder's own response + text, which per its migrated SOUL.md is required to include real test + commands + real output) artifacts. """ before_sha = _current_commit(target_folder_path) - response = org_node_trigger.trigger_org_node( - "coder", _CODER_PROMPT.format(goal_text=goal_text, sdd_content=sdd_content), - target_folder_path, timeout=timeout, + run_id = str(uuid.uuid4()) + manifest = {"nodes": [{"node_id": "implement", "depends_on": []}], "edges": []} + conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES (?, ?, 'coder', ?, 'running')", + (run_id, task_id, json.dumps(manifest)), + ) + conn.commit() + + outputs = work_graph_orchestrate.run_graph( + conn, run_id, "coder", "coder-implementation", target_folder_path, manifest, + lambda node, outputs: _CODER_PROMPT.format(goal_text=goal_text, sdd_content=sdd_content), + _CODE_INPUT_SCHEMA, _CODE_OUTPUT_SCHEMA, timeout=timeout, ) + response = outputs["implement"]["response"] + diff = subprocess.run( ["git", "diff", before_sha, "HEAD"], cwd=target_folder_path, capture_output=True, text=True ).stdout @@ -115,7 +190,11 @@ def implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_ def validate_implementation(task_id, benchmark_content, goal_text, target_folder_path, timeout=1200): """Architecture §11 criterion 3: real Validator call, parses the 'VERDICT: PASS|FAIL|INCONCLUSIVE' line this Unit's prompt asks for. - Returns {"verdict", "raw_response"}. + Returns {"verdict", "raw_response"}. Unchanged by the 2026-08-01 + rewiring -- REQ-04's own coverage never named the Work Graph + controllers (c_bug_record_create/p_validator_root_cause_trace only), + so the simpler flat trigger_org_node call is already faithful to what + this REQ actually requires. """ response = org_node_trigger.trigger_org_node( "validator", @@ -145,30 +224,19 @@ def run_end_to_end(conn, target_folder_path, goal_text): 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] + 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) - # trigger_org_node_full itself doesn't create a work_graph_run row (that's - # U-011's plan_work_graph, a separate call this pipeline doesn't invoke, - # since REQ-02's graph_manifest and this pipeline's actual document - # content are orthogonal concerns) -- artifact.produced_by_run_id is a - # real FK, so a row has to exist; this one just records that the e2e - # pipeline (not a real planned Work Graph) produced these artifacts - run_id = "e2e-" + task_id - conn.execute( - "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES (?, ?, 'planner', '{}', 'running')", - (run_id, task_id), - ) - conn.commit() - - artifact_ids = plan_documents(conn, task_id, run_id, goal_text, target_folder_path) + plan_result = plan_documents(conn, task_id, goal_text, target_folder_path) + artifact_ids = plan_result["artifact_ids"] sdd_content = _read_artifact_content(conn, target_folder_path, artifact_ids["sdd"]) benchmark_content = _read_artifact_content(conn, target_folder_path, artifact_ids["benchmark"]) org_node_lifecycle.trigger_org_node_full(conn, "coder", task_id, target_folder_path, _READY_PROMPT) - code_artifact_ids = implement_code(conn, task_id, run_id, sdd_content, goal_text, target_folder_path) + code_artifact_ids = implement_code(conn, task_id, sdd_content, goal_text, target_folder_path) org_node_lifecycle.trigger_org_node_full(conn, "validator", task_id, target_folder_path, _READY_PROMPT) verdict = validate_implementation(task_id, benchmark_content, goal_text, target_folder_path) diff --git a/src/test_e2e_pipeline.py b/src/test_e2e_pipeline.py index 89f281f..f86bb03 100644 --- a/src/test_e2e_pipeline.py +++ b/src/test_e2e_pipeline.py @@ -6,6 +6,19 @@ correctly, not what a real Planner/Coder/Validator would actually produce. Real coverage (as much of it as is safe/affordable to run) is in src/test_e2e_pipeline_e2e.py (RUN_REAL_AGENT_E2E=1). +**Rewired 2026-08-01 (U-031 regression audit)**: plan_documents/ +implement_code now go through work_graph_plan.plan_work_graph + +work_graph_orchestrate.run_graph (which calls work_graph_execute. +execute_work_graph per node) instead of a flat org_node_trigger. +trigger_org_node call -- see e2e_pipeline.py's module docstring for why. +plan_work_graph is mocked at its own subprocess boundary (same pattern as +test_work_graph_plan.py) so its real graph_manifest-parsing logic and its +real work_graph_run INSERT both still run for real; execute_work_graph is +mocked directly (one level higher than its own subprocess boundary) since +this file's job is testing e2e_pipeline's wiring, not re-proving +execute_work_graph's own mechanics (covered in test_work_graph_execute.py) +or work_graph_orchestrate's own mechanics (test_work_graph_orchestrate.py). + Run: python3 -m unittest src.test_e2e_pipeline -v """ @@ -15,27 +28,33 @@ import tempfile import unittest from unittest.mock import patch -from src import e2e_pipeline, project_scaffold, schema +from src import e2e_pipeline, org_node_trigger, project_scaffold, schema, work_graph_execute _FAKE_DOCS = {"brd": "# BRD\nfake", "prd": "# PRD\nfake", "sdd": "# SDD\nfake", "benchmark": "# Benchmark\nfake"} +_FAKE_GRAPH_MANIFEST = { + "nodes": [ + {"node_id": "brd", "artifact_type": "brd", "depends_on": []}, + {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]}, + {"node_id": "sdd", "artifact_type": "sdd", "depends_on": ["prd"]}, + {"node_id": "benchmark", "artifact_type": "benchmark", "depends_on": ["sdd"]}, + ], + "edges": [ + {"from": "brd", "to": "prd", "edge_type": "needs_judgment"}, + {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"}, + {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"}, + ], +} + + +def _fake_plan_subprocess_run(*args, **kwargs): + payload = {"is_error": False, "result": "```json\n" + json.dumps(_FAKE_GRAPH_MANIFEST) + "\n```"} + return subprocess.CompletedProcess(args=["claude"], returncode=0, stdout=json.dumps(payload), stderr="") -def _fake_trigger(org_node, prompt, target_folder_path, timeout=300): - if prompt == e2e_pipeline._READY_PROMPT: - return "Ready." - if "BRD (Business" in prompt: - return _FAKE_DOCS["brd"] - if "PRD (Product" in prompt: - return _FAKE_DOCS["prd"] - if "SDD (System" in prompt: - return _FAKE_DOCS["sdd"] - if "Benchmark/acceptance-test" in prompt: - return _FAKE_DOCS["benchmark"] - if prompt.startswith("Implement this goal"): - return "Implemented. Ran `pytest`: 3 passed." - if prompt.startswith("Independently verify"): - return "VERDICT: PASS\nAll benchmark scenarios reproduced and passing." - raise AssertionError("unexpected prompt: " + prompt[:80]) + +def _fake_ready_trigger(org_node, prompt, target_folder_path, timeout=300): + assert prompt == e2e_pipeline._READY_PROMPT + return "Ready." class PipelineTestBase(unittest.TestCase): @@ -58,17 +77,18 @@ class PlanDocumentsTest(PipelineTestBase): "VALUES ('task-1', (SELECT project_id FROM project), '实现贪吃蛇游戏', 'goal', " "'planning_brd_prd_sdd', 'planner')" ) - self.conn.execute( - "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')" - ) self.conn.commit() - with patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger): - artifact_ids = e2e_pipeline.plan_documents( - self.conn, "task-1", "run-1", "实现贪吃蛇游戏", self.tmp.name - ) + with patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = [ + {"content": _FAKE_DOCS["brd"]}, {"content": _FAKE_DOCS["prd"]}, + {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, + ] + plan_result = e2e_pipeline.plan_documents(self.conn, "task-1", "实现贪吃蛇游戏", self.tmp.name) + artifact_ids = plan_result["artifact_ids"] self.assertEqual(set(artifact_ids), {"brd", "prd", "sdd", "benchmark"}) prd_deps = json.loads( self.conn.execute( @@ -83,11 +103,25 @@ class PlanDocumentsTest(PipelineTestBase): ) self.assertEqual(sdd_deps, [artifact_ids["prd"]]) + # plan_work_graph's own real work_graph_run INSERT actually ran (not bypassed) + run_row = self.conn.execute( + "SELECT org_node, status FROM work_graph_run WHERE run_id = ?", (plan_result["run_id"],) + ).fetchone() + self.assertEqual(run_row, ("planner", "planning")) + class RunEndToEndTest(PipelineTestBase): def test_full_pipeline_wiring(self): - with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_trigger), \ - patch("src.org_node_trigger.trigger_org_node", side_effect=_fake_trigger): + with patch("src.org_node_lifecycle.trigger_org_node", side_effect=_fake_ready_trigger), \ + patch("src.work_graph_plan.subprocess.run", side_effect=_fake_plan_subprocess_run), \ + patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute, \ + 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"]}, + {"content": _FAKE_DOCS["sdd"]}, {"content": _FAKE_DOCS["benchmark"]}, + {"response": "Implemented. Ran `pytest`: 3 passed."}, + ] result = e2e_pipeline.run_end_to_end(self.conn, self.tmp.name, "实现贪吃蛇游戏") self.assertEqual(result["verdict"]["verdict"], "PASS") @@ -104,6 +138,14 @@ class RunEndToEndTest(PipelineTestBase): ).fetchone()[0] self.assertEqual(all_artifacts, 6) # brd, prd, sdd, benchmark, code_diff, test_report + # planner and coder each got their own work_graph_run (not sharing one) + run_org_nodes = { + row[0] for row in self.conn.execute( + "SELECT org_node FROM work_graph_run WHERE task_id = ?", (result["task_id"],) + ).fetchall() + } + self.assertEqual(run_org_nodes, {"planner", "coder"}) + class ValidateImplementationTest(unittest.TestCase): def test_parses_pass_verdict(self): diff --git a/src/test_e2e_pipeline_e2e.py b/src/test_e2e_pipeline_e2e.py index 1aebb3e..8ed3e67 100644 --- a/src/test_e2e_pipeline_e2e.py +++ b/src/test_e2e_pipeline_e2e.py @@ -29,14 +29,12 @@ class RealPlanDocumentsTest(unittest.TestCase): agent_profile.deploy_agent_definitions(tmp) project_id = conn.execute("SELECT project_id FROM project").fetchone()[0] task_id = task_create.create_task(conn, project_id, "实现贪吃蛇游戏")["task_id"] - conn.execute( - "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " - "VALUES ('run-1', ?, 'planner', '{}', 'running')", - (task_id,), - ) - conn.commit() - artifact_ids = e2e_pipeline.plan_documents(conn, task_id, "run-1", "实现贪吃蛇游戏", tmp) + # plan_documents (2026-08-01 rewiring) creates its own real + # work_graph_run via plan_work_graph() -- no manual seed row + # needed anymore. + plan_result = e2e_pipeline.plan_documents(conn, task_id, "实现贪吃蛇游戏", tmp) + artifact_ids = plan_result["artifact_ids"] for artifact_type, artifact_id in artifact_ids.items(): row = conn.execute( diff --git a/src/test_work_graph_orchestrate.py b/src/test_work_graph_orchestrate.py new file mode 100644 index 0000000..0c86eab --- /dev/null +++ b/src/test_work_graph_orchestrate.py @@ -0,0 +1,212 @@ +"""U-031 tests: work_graph_orchestrate.py -- the graph_manifest -> per-node +execution bridge closed during the REQ-01~14 regression audit. Real sqlite, +run_fn mocked at the claude subprocess boundary (execute_work_graph is +tested for real elsewhere, in test_work_graph_execute.py / the 2026-08-01 +spike -- this module's own job is orchestration, not re-proving that). + +Run: python3 -m unittest src.test_work_graph_orchestrate -v +""" + +import unittest +from unittest.mock import patch + +from src import schema, work_graph_execute, work_graph_orchestrate as wgo + + +class TopologicalOrderTest(unittest.TestCase): + def test_linear_chain_via_depends_on(self): + manifest = { + "nodes": [ + {"node_id": "brd", "depends_on": []}, + {"node_id": "prd", "depends_on": ["brd"]}, + {"node_id": "sdd", "depends_on": ["prd"]}, + ], + "edges": [], + } + self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["brd", "prd", "sdd"]) + + def test_linear_chain_via_edges_only(self): + # real LLM output might populate edges but not each node's own + # depends_on -- must still order correctly + manifest = { + "nodes": [{"node_id": "a"}, {"node_id": "b"}, {"node_id": "c"}], + "edges": [{"from": "a", "to": "b"}, {"from": "b", "to": "c"}], + } + self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b", "c"]) + + def test_diamond_fan_out_fan_in(self): + manifest = { + "nodes": [ + {"node_id": "n1", "depends_on": []}, + {"node_id": "n2", "depends_on": ["n1"]}, + {"node_id": "n3", "depends_on": ["n1"]}, + {"node_id": "n4", "depends_on": ["n2", "n3"]}, + ], + "edges": [], + } + order = [n["node_id"] for n in wgo.topological_order(manifest)] + self.assertEqual(order[0], "n1") + self.assertEqual(order[-1], "n4") + self.assertEqual(set(order[1:3]), {"n2", "n3"}) + + def test_cycle_raises(self): + manifest = { + "nodes": [{"node_id": "x", "depends_on": ["y"]}, {"node_id": "y", "depends_on": ["x"]}], + "edges": [], + } + with self.assertRaises(ValueError): + wgo.topological_order(manifest) + + def test_independent_nodes_ordered_deterministically(self): + manifest = {"nodes": [{"node_id": "b"}, {"node_id": "a"}], "edges": []} + self.assertEqual([n["node_id"] for n in wgo.topological_order(manifest)], ["a", "b"]) + + +class RunGraphTestBase(unittest.TestCase): + def setUp(self): + self.conn = schema.connect(":memory:") + schema.create_schema(self.conn) + self.conn.execute( + "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " + "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" + ) + self.conn.execute( + "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " + "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'coding', 'planner')" + ) + self.conn.execute( + "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) " + "VALUES ('run-1', 'task-1', 'planner', '{}', 'running')" + ) + self.conn.commit() + + def tearDown(self): + self.conn.close() + + +_OUT_SCHEMA = {"type": "object", "required": ["content"]} +_IN_SCHEMA = {"type": "object", "required": ["node_id"]} + + +class RunGraphHappyPathTest(RunGraphTestBase): + def test_runs_nodes_in_order_and_returns_outputs(self): + manifest = { + "nodes": [ + {"node_id": "brd", "artifact_type": "brd", "depends_on": []}, + {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]}, + ], + "edges": [], + } + calls = [] + + def build_prompt(node, outputs): + calls.append((node["node_id"], dict(outputs))) + return "prompt for " + node["node_id"] + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = [{"content": "brd text"}, {"content": "prd text"}] + outputs = wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + build_prompt, _IN_SCHEMA, _OUT_SCHEMA, + ) + + self.assertEqual(outputs, {"brd": {"content": "brd text"}, "prd": {"content": "prd text"}}) + self.assertEqual(calls[0], ("brd", {})) + self.assertEqual(calls[1], ("prd", {"brd": {"content": "brd text"}})) + + def test_creates_work_node_execution_rows_and_placeholder_skill(self): + manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", return_value={"content": "x"}): + wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, + ) + row = self.conn.execute( + "SELECT status, subagent_type, skill_ref FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'brd'" + ).fetchone() + self.assertEqual(row, ("done", "planner", "planner-doc-writing")) + skill_row = self.conn.execute("SELECT owner_org_node FROM skill WHERE skill_id = 'planner-doc-writing'").fetchone() + self.assertEqual(skill_row[0], "planner") + + def test_resumes_without_rerunning_done_nodes(self): + manifest = { + "nodes": [ + {"node_id": "brd", "depends_on": []}, + {"node_id": "prd", "depends_on": ["brd"]}, + ], + "edges": [], + } + calls = [] + + def build_prompt(node, outputs): + return "p-" + node["node_id"] + + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute: + mock_execute.side_effect = lambda cwd, name, args, schema, timeout=600: calls.append(name) or {"content": "c-" + name} + wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + build_prompt, _IN_SCHEMA, _OUT_SCHEMA, + ) + self.assertEqual(len(calls), 2) + + # simulate a restart: run_graph called again for the same run_id + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph") as mock_execute2: + mock_execute2.side_effect = AssertionError("should not be called -- both nodes are already done") + outputs = wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + build_prompt, _IN_SCHEMA, _OUT_SCHEMA, + ) + self.assertEqual(set(outputs.keys()), {"brd", "prd"}) + + +class RunGraphUsageLimitTest(RunGraphTestBase): + def test_usage_limit_reached_translates_to_quota_wait(self): + manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} + attempts = {"n": 0} + + def fake_execute(cwd, name, args, output_schema, timeout=600): + attempts["n"] += 1 + if attempts["n"] == 1: + raise work_graph_execute.UsageLimitReachedError("hit your session limit", "session") + return {"content": "recovered"} + + sleeps = [] + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object(work_graph_execute, "execute_work_graph", side_effect=fake_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, + sleep=lambda s: sleeps.append(s), now=lambda: 0.0, + ) + self.assertEqual(outputs["brd"], {"content": "recovered"}) + self.assertEqual(sleeps, [5 * 60]) + + +class RunGraphClassifierBlockedTest(RunGraphTestBase): + def test_classifier_blocked_marks_node_blocked_and_propagates(self): + manifest = {"nodes": [{"node_id": "brd", "depends_on": []}], "edges": []} + with patch.object(work_graph_execute, "write_workflow_script"), \ + patch.object( + work_graph_execute, "execute_work_graph", + side_effect=work_graph_execute.ClassifierBlockedError("blocked by classifier"), + ): + with self.assertRaises(work_graph_execute.ClassifierBlockedError): + wgo.run_graph( + self.conn, "run-1", "planner", "planner-doc-writing", "/tmp/proj-1", manifest, + lambda node, outputs: "p", _IN_SCHEMA, _OUT_SCHEMA, + ) + node_exec_id = self.conn.execute( + "SELECT node_exec_id FROM work_node_execution WHERE run_id = 'run-1' AND node_id = 'brd'" + ).fetchone()[0] + status = self.conn.execute( + "SELECT status FROM work_node_execution WHERE node_exec_id = ?", (node_exec_id,) + ).fetchone()[0] + self.assertEqual(status, "blocked") + + +if __name__ == "__main__": + unittest.main() diff --git a/src/test_work_graph_plan.py b/src/test_work_graph_plan.py index e75b465..562f358 100644 --- a/src/test_work_graph_plan.py +++ b/src/test_work_graph_plan.py @@ -72,6 +72,63 @@ class ExtractGraphManifestTest(unittest.TestCase): self.assertIsNone(work_graph_plan._extract_graph_manifest("sorry, I can't do that")) +class NormalizeGraphManifestTest(unittest.TestCase): + """U-031 regression: a real Planner call (2026-08-01) came back using + `id` instead of the documented `node_id`, and never populated any + node's own `depends_on` -- dependencies existed only as `edges`. This + is a real, observed deviation, not a hypothetical one; the fixture + below is the actual shape captured from that call (fields trimmed to + what normalization touches).""" + + def test_already_well_formed_manifest_untouched(self): + manifest = { + "nodes": [ + {"node_id": "brd", "artifact_type": "brd", "depends_on": []}, + {"node_id": "prd", "artifact_type": "prd", "depends_on": ["brd"]}, + ], + "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}], + } + result = work_graph_plan._normalize_graph_manifest(manifest) + self.assertEqual(result["nodes"][1]["depends_on"], ["brd"]) + + def test_id_key_normalized_to_node_id(self): + manifest = { + "nodes": [{"id": "brd", "artifact_type": "brd"}, {"id": "prd", "artifact_type": "prd"}], + "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}], + } + result = work_graph_plan._normalize_graph_manifest(manifest) + self.assertEqual([n["node_id"] for n in result["nodes"]], ["brd", "prd"]) + + def test_depends_on_derived_from_edges_when_absent(self): + manifest = { + "nodes": [{"id": "brd"}, {"id": "prd"}, {"id": "sdd"}, {"id": "benchmark"}], + "edges": [ + {"from": "brd", "to": "prd", "edge_type": "needs_judgment"}, + {"from": "prd", "to": "sdd", "edge_type": "needs_judgment"}, + {"from": "sdd", "to": "benchmark", "edge_type": "needs_judgment"}, + ], + } + result = work_graph_plan._normalize_graph_manifest(manifest) + by_id = {n["node_id"]: n for n in result["nodes"]} + self.assertEqual(by_id["brd"]["depends_on"], []) + self.assertEqual(by_id["prd"]["depends_on"], ["brd"]) + self.assertEqual(by_id["sdd"]["depends_on"], ["prd"]) + self.assertEqual(by_id["benchmark"]["depends_on"], ["sdd"]) + + def test_does_not_duplicate_dependency_already_present(self): + manifest = { + "nodes": [{"node_id": "brd", "depends_on": []}, {"node_id": "prd", "depends_on": ["brd"]}], + "edges": [{"from": "brd", "to": "prd", "edge_type": "needs_judgment"}], + } + result = work_graph_plan._normalize_graph_manifest(manifest) + self.assertEqual(result["nodes"][1]["depends_on"], ["brd"]) + + def test_edge_referencing_unknown_node_id_ignored(self): + manifest = {"nodes": [{"id": "brd"}], "edges": [{"from": "brd", "to": "no-such-node"}]} + result = work_graph_plan._normalize_graph_manifest(manifest) + self.assertEqual(result["nodes"][0]["depends_on"], []) + + class HappyPathTest(PlanWorkGraphTestBase): def test_returns_run_id_and_graph_manifest(self): with patch("src.work_graph_plan.subprocess.run") as run: diff --git a/src/test_work_node_run.py b/src/test_work_node_run.py index ca32e0c..54dcaf6 100644 --- a/src/test_work_node_run.py +++ b/src/test_work_node_run.py @@ -109,6 +109,24 @@ class HappyPathTest(WorkNodeRunTestBase): final_status = self.conn.execute("SELECT status FROM execution_log_file").fetchone()[0] self.assertEqual(final_status, "closed") + def test_unretryable_run_fn_failure_marks_node_blocked_not_stuck_running(self): + # any exception from run_fn other than a recognized usage-limit + # signal (e.g. CLASSIFIER_BLOCKED surfacing from + # work_graph_execute.execute_work_graph, 03-C §3.5 point 6) must + # not leave the row stuck at status='running' with no record of why + class SomeUnretryableError(Exception): + pass + + def boom(cwd, model=None): + raise SomeUnretryableError("classifier blocked or similar") + + with self.assertRaises(SomeUnretryableError): + work_node_run.run_work_node(self.conn, "node-1", {"goal": "x"}, _IN_SCHEMA, _OUT_SCHEMA, run_fn=boom) + status = self.conn.execute( + "SELECT status FROM work_node_execution WHERE node_exec_id = 'node-1'" + ).fetchone()[0] + self.assertEqual(status, "blocked") + class RetryOnSchemaFailureTest(WorkNodeRunTestBase): def test_retries_then_succeeds(self): diff --git a/src/work_graph_orchestrate.py b/src/work_graph_orchestrate.py new file mode 100644 index 0000000..ce373e9 --- /dev/null +++ b/src/work_graph_orchestrate.py @@ -0,0 +1,182 @@ +"""Bridges `plan_work_graph()`'s `graph_manifest` to real per-node execution +via `execute_work_graph()` + `run_work_node()` -- the missing integration +between three previously isolated, individually-tested modules (U-011/ +U-012/U-013), closed during U-031's regression audit (2026-08-01): each of +the three had its own unit tests (and even real-API validation) but none of +them had ever actually been called by production code, and the one +genuinely-validated closed loop (`e2e_pipeline.py`, U-014) used a much +simpler, separate mechanism (flat `org_node_trigger.trigger_org_node()` +calls) that bypassed all three. + +Design: each node's actual work happens via its OWN one-node dynamic +workflow script (`write_workflow_script` + `execute_work_graph`), not one +big multi-node script for the whole graph. This is a deliberate trade-off, +not the "amortize orchestration cost across many nodes in one `claude -p` +call" pattern 03-C §3.5 originally motivated `execute_work_graph` with -- +REQ-06's breakpoint-recovery requirement (resume after a process restart +without re-running already-`done` nodes) is only achievable with per-node, +Python-level persisted state (`work_node_run.py`'s whole design), which is +fundamentally incompatible with wrapping the WHOLE graph in one opaque +`claude` subprocess call Python has no visibility into mid-flight. + +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. +""" + +import json +import time +import uuid + +from src import work_graph_execute, work_node_run + + +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 + output isn't schema-validated (it can't be, `--json-schema` doesn't + work with `--agent`, see work_graph_plan.py), so a real LLM response + might populate one and not the other; trusting only one would silently + mis-order nodes rather than raise. Raises ValueError on a cycle or a + dependency referencing an unknown node_id. + """ + nodes = {n["node_id"]: n for n in graph_manifest["nodes"]} + deps = {node_id: set(n.get("depends_on") or []) for node_id, n in nodes.items()} + for edge in graph_manifest.get("edges", []): + to_id, from_id = edge.get("to"), edge.get("from") + if to_id in deps and from_id in nodes: + deps[to_id].add(from_id) + + in_degree = {node_id: len(d) for node_id, d in deps.items()} + dependents = {node_id: [] for node_id in nodes} + for node_id, d in deps.items(): + for dep in d: + if dep in dependents: + dependents[dep].append(node_id) + + ready = sorted(node_id for node_id, degree in in_degree.items() if degree == 0) + ordered = [] + while ready: + node_id = ready.pop(0) + ordered.append(nodes[node_id]) + for dependent in sorted(dependents[node_id]): + in_degree[dependent] -= 1 + if in_degree[dependent] == 0: + ready.append(dependent) + ready.sort() + if len(ordered) != len(nodes): + raise ValueError( + "graph_manifest has a dependency cycle or a depends_on/edge references an unknown node_id" + ) + return ordered + + +def ensure_placeholder_skill(conn, skill_id, owner_org_node): + """No Skill content has actually been migrated from the Hermes profiles + yet (skill_update.py's own documented gap) -- work_node_execution. + skill_ref is a required FK, so a real row has to exist regardless. + Idempotent.""" + exists = conn.execute("SELECT 1 FROM skill WHERE skill_id = ?", (skill_id,)).fetchone() + if exists is None: + conn.execute( + "INSERT INTO skill (skill_id, owner_org_node, file_path, current_git_commit_sha, source, status) " + "VALUES (?, ?, ?, 'unknown', 'new', 'active')", + (skill_id, owner_org_node, "skills/" + skill_id + "/"), + ) + conn.commit() + + +def _ensure_node_row(conn, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, input_payload): + existing = conn.execute( + "SELECT node_exec_id FROM work_node_execution WHERE run_id = ? AND node_id = ?", (run_id, node_id) + ).fetchone() + if existing: + return existing[0] + node_exec_id = str(uuid.uuid4()) + conn.execute( + "INSERT INTO work_node_execution (node_exec_id, run_id, node_id, subagent_type, skill_ref, " + "input_schema_ref, output_schema_ref, input_payload, status) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending')", + (node_exec_id, run_id, node_id, subagent_type, skill_ref, input_schema_ref, output_schema_ref, json.dumps(input_payload)), + ) + conn.commit() + return node_exec_id + + +def _render_node_script(script_name, org_node, prompt, output_schema, model): + model_opt = ", model: " + json.dumps(model) if model else "" + return ( + "export const meta = {{ name: {name}, description: 'graphlemon Work Node ({org_node})' }}\n\n" + "const result = await agent({prompt}, {{ agentType: {org_node_json}, schema: {schema}{model_opt} }})\n\n" + "return result\n" + ).format( + name=json.dumps(script_name), org_node=org_node, org_node_json=json.dumps(org_node), + prompt=json.dumps(prompt), schema=json.dumps(output_schema), model_opt=model_opt, + ) + + +def _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout): + script_name = "graphlemon-node-" + node_exec_id.replace("-", "") + + def run_fn(cwd, model=None): + script_content = _render_node_script(script_name, org_node, prompt, output_schema, model) + work_graph_execute.write_workflow_script(cwd, script_name, script_content) + try: + return work_graph_execute.execute_work_graph(cwd, script_name, {}, output_schema, timeout=timeout) + except work_graph_execute.UsageLimitReachedError as e: + work_node_run.raise_usage_limit_signal(e.quota_limit_type) + # ClassifierBlockedError / generic WorkGraphExecuteError propagate + # as-is -- run_work_node's own `except Exception` clause marks the + # node 'blocked' and re-raises, no special handling needed here. + + return run_fn + + +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, +): + """Runs every node in `graph_manifest` (topologically ordered) through + `run_work_node()`, each backed by a real one-node dynamic-workflow + `agent()` call using `org_node`'s migrated persona. + + `build_prompt(node, outputs)` is called right before running that + node -- `outputs` contains every already-completed node's + output_payload so far (topological order guarantees a node's + dependencies are always already present), so prompts needing upstream + content (e.g. "here is the approved BRD...") can be built just-in-time + instead of requiring the whole graph's prompts precomputed before + anything runs. This also makes resume-after-restart work correctly: + `run_work_node` short-circuits already-`done` nodes and returns their + persisted output, which lands in `outputs` the same as a freshly-run + one would. + + `needs_worktree=True` (03-C §3.6: "并行写代码的节点必须先申请 + git_worktree_ref") isolates each node's actual file changes in its own + git worktree, merged back only once the node succeeds -- callers + writing code (Coder) should pass this; callers only producing document + content (Planner) don't need it. + + Returns {node_id: output_payload}. + """ + ensure_placeholder_skill(conn, skill_id, org_node) + ordered = topological_order(graph_manifest) + outputs = {} + for node in ordered: + node_id = node["node_id"] + node_exec_id = _ensure_node_row( + conn, run_id, node_id, org_node, skill_id, + "work_graph_node_input.schema.json", "work_graph_node_output.schema.json", + {"node_id": node_id}, + ) + prompt = build_prompt(node, outputs) + run_fn = _make_run_fn(target_folder_path, node_exec_id, org_node, prompt, output_schema, timeout) + output = work_node_run.run_work_node( + conn, node_exec_id, {"node_id": node_id}, input_schema, output_schema, run_fn, + needs_worktree=needs_worktree, repo_path=target_folder_path, + sleep=sleep, now=now, + ) + outputs[node_id] = output + return outputs diff --git a/src/work_graph_plan.py b/src/work_graph_plan.py index f981fba..f150f73 100644 --- a/src/work_graph_plan.py +++ b/src/work_graph_plan.py @@ -102,6 +102,35 @@ def _extract_graph_manifest(text): return None +def _normalize_graph_manifest(manifest): + """Real LLM output has been observed (2026-08-01, U-031 integration + validation) to use `id` instead of the documented `node_id` key, and to + express dependencies purely via `edges` (from/to) without ever + populating each node's own `depends_on` array -- even though this + schema requires both. Normalizing here, right after parsing and before + persisting, means every downstream consumer (work_graph_orchestrate.py's + topological sort, v_task_detail's to-do tree, v_graph_visualizer's DAG + layering) gets a manifest that actually matches the documented shape, + instead of each one needing its own defensive "id or node_id, + depends_on or edges" fallback logic repeated independently. Mutates and + returns `manifest`. + """ + nodes = manifest.get("nodes", []) + for node in nodes: + if "node_id" not in node and "id" in node: + node["node_id"] = node["id"] + node.setdefault("depends_on", []) + + node_ids = {n.get("node_id") for n in nodes} + for edge in manifest.get("edges", []): + from_id, to_id = edge.get("from"), edge.get("to") + if from_id in node_ids and to_id in node_ids: + for node in nodes: + if node.get("node_id") == to_id and from_id not in node["depends_on"]: + node["depends_on"].append(from_id) + return manifest + + def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, timeout=300): """Returns {"run_id", "graph_manifest"}. Creates one `work_graph_run` row (status='planning') -- moving it to 'running'/executing it is out of scope.""" @@ -134,6 +163,7 @@ def plan_work_graph(conn, org_node, task_id, goal_text, target_folder_path, time graph_manifest = _extract_graph_manifest(payload.get("result", "")) if not graph_manifest: raise PlanningFailedError("could not extract a graph_manifest for task {}: {}".format(task_id, payload)) + graph_manifest = _normalize_graph_manifest(graph_manifest) run_id = str(uuid.uuid4()) conn.execute( diff --git a/src/work_node_run.py b/src/work_node_run.py index bc0ecd4..1400781 100644 --- a/src/work_node_run.py +++ b/src/work_node_run.py @@ -291,6 +291,22 @@ def run_work_node( backoff_index = min(int(elapsed // (30 * 60)), len(_BACKOFF_SCHEDULE_MINUTES) - 1) sleep(_BACKOFF_SCHEDULE_MINUTES[backoff_index] * 60) continue + except Exception: + # Anything else run_fn raises (e.g. CLASSIFIER_BLOCKED from + # work_graph_execute.execute_work_graph, 03-C §3.5 point 6) + # is an unretryable node-level failure -- mark 'blocked' + # before re-raising so the row doesn't sit stuck at + # 'running' forever with no record of why. This is + # deliberately broad (not just ClassifierBlockedError): + # REQ-06's "系统应明确报错并转 blocked_pending_human，而不是 + # 静默丢失进度" applies to any node-level catastrophic + # failure work_node_run.py doesn't have specific retry + # logic for, not just the one named error type. + conn.execute( + "UPDATE work_node_execution SET status = 'blocked' WHERE node_exec_id = ?", (node_exec_id,) + ) + conn.commit() + raise if _validate_schema_lite(output_payload, output_schema): conn.execute( @@ -347,6 +363,15 @@ def raise_if_usage_limited(text): raise _UsageLimitSignal(quota_type) +def raise_usage_limit_signal(quota_limit_type): + """Same signal as raise_if_usage_limited, for callers that already know + the quota type structurally (e.g. work_graph_orchestrate.py wrapping + work_graph_execute.UsageLimitReachedError, which already parsed this + from the same detection regex at its own layer) rather than needing to + regex-match free text a second time.""" + raise _UsageLimitSignal(quota_limit_type) + + def resume_work_graph_run(conn, run_id): """REQ-06 (04-P §4.2 p_work_graph_execution, breakpoint recovery): after a process restart, determine which nodes in `run_id` still need