# Work Graph 节点执行去黑盒化——需求+设计文档（合并，方向已定）

**日期**: 2026-08-08
**性质**: 需求+设计合并文档——用户已经明确拍板方向（自研，不复用 Claude Code 原生 Dynamic Workflow），且调研发现代码库里已经有一个生产验证过的现成替代方案（`plan_work_graph()`），不需要走"逐条开放问题讨论"流程，直接给出可实施设计。对应 `docs/graphlemon_architecture.md` §10.2（2026-08-08 决策修订）。
**触发场景**：用户追问"之前设计说用 Claude Code 自带 dynamic workflow，现在要去掉、自己实现，这个描述改了吗？为了可控，不再黑盒"——调研发现 `docs/graphlemon_architecture.md` §10.1"方案 A"（已确认：Work Graph 执行层复用原生 Dynamic Workflow）从未被正式撤回，但图编排层早就自研了（`work_graph_orchestrate.py`，REQ-06 断点续跑驱动）；**单节点执行层还没有**——三处调用点仍在用 `write_workflow_script`/`execute_work_graph()` 生成/执行一次性 Dynamic Workflow `.js` 脚本。
**状态**: ✅ 已实施（2026-08-13）。三处调用点（`_check_prd_consistency`/`_llm_judgment_verify_fn`/`_make_run_fn`）全部迁移到新增的 `org_node_trigger.trigger_org_node_structured()`；实施过程中额外发现并顺带迁移了文档撰写之后新落地的两处调用点（`_run_reflection_perspective`、`_fix_skill_for_node`，Phase 2a Reflection-review/Fix-Skill 引入）。`extract_json_payload` 泛化落在 `work_graph_plan.py`；`UsageLimitReachedError` 的两份重复定义未强行合并——`org_node_trigger.py` 新增一份规范版本（`.quota_limit_type` 属性齐全），`org_node_lifecycle.py`/`work_graph_execute.py` 各自的旧版本原样保留（前者服务不同调用路径，后者随宿主模块一起冻结）。FR-4 的真实验证（`RUN_REAL_AGENT_E2E=1`，`src/test_org_node_trigger_e2e.py`）确认人设加载、`--model` 覆盖、JSON 围栏提取三点全部符合预期。`write_workflow_script`/`execute_work_graph`/`execute_work_graph_with_usage_limit_retry` 确认无生产调用方，文件保留、docstring 标注历史遗留；`CLAUDE.md` 对应描述已同步更新。`discover -s src`（797 tests）+ `discover -s supervisor`（127 tests）全绿。

---

## 1. 背景：为什么单节点层至今还留着原生 Workflow

`work_graph_orchestrate.py` 模块docstring 记录的原因：`claude --agent <org_node> -p ... --json-schema <schema>` 这个 CLI 顶层组合会**静默丢弃** `structured_output`（05 §5.4.1.1 实测发现，人设加载正常、schema 校验失效）。Workflow DSL 内部的 `agent(prompt, {agentType, schema})` 组合是当时（2026-08-01，U-031）唯一验证过能同时做到"加载 org_node 人设"+"拿到 schema 校验过的结构化输出"的机制（05 §5.4.1.5）。

**这个理由已经不成立**：`work_graph_plan.py::plan_work_graph()`（`c_work_graph_plan`）面对的是完全相同的"`--agent` 身份 + 结构化输出"需求，但它从一开始就没有用 Workflow——直接 `subprocess.run(["claude", "--agent", org_node, "-p", prompt, "--output-format", "json", ...])`，从自由文本 `result` 字段里用 ```` ```json ```` 围栏 + 正则提取解析（`_extract_graph_manifest`，05 §5.4.1.1 自己推荐的应对方式），且早已在生产环境真实跑通（Planner 四节点文档链、这次新设计的 Phase 1 复核调用也是这个模式）。"必须用 Workflow 才能两者兼得"这个前提是错的，Workflow 只是**也**能做到，不是唯一路径。

### 1.1 当前仍依赖 `write_workflow_script`/`execute_work_graph` 的三个调用点

| # | 位置 | 用途 | `agentType` |
|---|---|---|---|
| 1 | `work_graph_orchestrate.py::_make_run_fn`（`_render_node_script`，L361-370） | 每个 Work Node 真正的内容生成调用——**主执行路径，影响面最大** | 该节点所属的 `org_node` |
| 2 | `work_graph_orchestrate.py::_llm_judgment_verify_fn`（L420-467） | `judgment.kind="llm"` 节点判断（Part A，2026-08-08 新增，Phase 2 会大量增加这类调用） | 该节点所属的 `org_node` |
| 3 | `e2e_pipeline.py::_check_prd_consistency`（L309-350） | PRD 内部矛盾检测（REQ-02 边界情况） | 固定 `'planner'` |

三处都是"渲染一段一次性 `.js` 脚本 → `write_workflow_script` 落盘 → `execute_work_graph()` 跑 `claude -p 'Run /<name>...'`"的同一个模式，且都已经有独立于 Workflow 的 self-target worktree 隔离逻辑包裹（`memory: project` 写入需要隔离，跟用不用 Workflow 无关，迁移时**不动**这部分）。

---

## 2. 目标（FR）

| # | 需求描述 |
|---|---|
| FR-1 | 新增一个通用的、`output_schema`-无关的辅助函数 `trigger_org_node_structured()`，用"flat `claude --agent` 调用 + JSON 围栏提取"取代"渲染 Workflow 脚本 + `claude -p 'Run /<name>...'`"，供三个调用点共用。 |
| FR-2 | 把 `work_graph_plan.py::_extract_graph_manifest` 里"整段当 JSON 解析、失败则逐个 ```` ```json ```` 围栏尝试"的算法抽成一个通用 helper，`_extract_graph_manifest` 和 `trigger_org_node_structured` 都调用它，不重复实现。 |
| FR-3 | 三个调用点依次迁移到 `trigger_org_node_structured()`，删除对 `write_workflow_script`/`execute_work_graph` 的调用。 |
| FR-4 | 保留 `work_node_run.py`（U-020）按节点降级模型重试的能力——`trigger_org_node_structured()` 必须支持等价的模型覆盖参数，且**必须真实验证**（不能假设）`claude --agent <name> -p ... --model <model>` 这个 CLI 组合行为符合预期（05 §5.4.1.5 的教训：DSL 内部选项和 CLI 顶层参数是两条不同路径，不能凭直觉套用其中一条的结论）。 |
| FR-5 | 保留 `execute_work_graph_with_usage_limit_retry()` 的 5→10→20→30 分钟退避重试能力，迁移到包裹 `trigger_org_node_structured()` 外层，退避算法本身不变，只换内层调用。 |
| FR-6 | 迁移完成后，`write_workflow_script`/`execute_work_graph`/`execute_work_graph_with_usage_limit_retry` 三个函数不再被生产代码调用——**不删除** `work_graph_execute.py` 文件本身（先确认没有其他调用方，尤其是 `work_graph_fork.py` 时间旅行调试功能是否也依赖它），同步更新 `CLAUDE.md` 里 `.claude/workflows/` 一节的描述，说明这是历史遗留机制。 |

---

## 3. 非目标

1. **不改变 Work Graph 的图拓扑/编排逻辑**——`work_graph_orchestrate.py` 的 `topological_layers()`/`run_graph()`/loop_back 机制本来就是纯 Python 自研的，不依赖 Workflow，不受这次改动影响。
2. **不改变 `org_node_trigger.trigger_org_node()` 本身**——新函数是它的姊妹函数（多了 schema 提取），不是替换它；`_READY_PROMPT` 握手调用、Validator 最终 verdict 调用等现有的 flat 调用完全不变。
3. **不处理 worktree 隔离/`memory: project` 写入隔离逻辑**——这部分独立于"用 Workflow 还是用 flat 调用"，迁移时原样保留包裹逻辑，只换内层实现。
4. **不在这次改动里处理 `docs/graphlemon_architecture.md` 第 10 节表格第 14 条本身**——那是课程技法的对照记录，不是决策，不需要改；改的只是 §10.1/§10.2 的决策结论（已在本文档触发前完成）。

---

## 4. 详细设计

### 4.1 新函数：放在 `org_node_trigger.py`（flat 调用的家）

```python
class StructuredOutputExtractionError(Exception):
    """trigger_org_node_structured 在 result 文本里找不到匹配 output_schema
    必需字段的 JSON payload。"""


_JSON_FENCE_INSTRUCTION = (
    "Your final answer MUST be a single JSON object wrapped in exactly one "
    "```json fenced code block, matching this schema: {schema}. Do not "
    "describe what you did or are about to do -- the fenced block's content "
    "IS the answer, not a summary of it."
)


def trigger_org_node_structured(org_node, prompt, target_folder_path, output_schema, model=None, timeout=300):
    """Like trigger_org_node(), but also extracts a JSON payload matching
    output_schema's top-level required keys from the free-text response --
    05 §5.4.1.1's documented workaround (never pass --json-schema alongside
    --agent; ask for a single ```json fenced block in the prompt instead,
    same convention plan_work_graph()'s _PROMPT_TEMPLATE/_GENERIC_PROMPT_
    TEMPLATE already use). `_JSON_FENCE_INSTRUCTION` is appended to `prompt`
    automatically -- callers don't repeat this boilerplate themselves.
    `model`: real CLI --model override (FR-4 -- verify empirically before
    relying on this in production, see this module's test file for the
    real-call verification this needs).
    """
    args = ["claude", "--agent", org_node, "-p", prompt + "\n\n" + _JSON_FENCE_INSTRUCTION.format(schema=json.dumps(output_schema))]
    if model:
        args += ["--model", model]
    args += ["--output-format", "json", "--permission-mode", "auto"]
    result = subprocess.run(
        args, cwd=target_folder_path, stdin=subprocess.DEVNULL,
        capture_output=True, text=True, timeout=timeout,
    )
    if result.returncode != 0:
        raise OrgNodeTriggerError(
            "trigger_org_node_structured failed for {}\nstderr: {}\nstdout (truncated): {}".format(
                org_node, result.stderr, result.stdout[:4000]
            )
        )
    payload = json.loads(result.stdout)
    if _looks_like_usage_limit_response(payload):   # reuses org_node_lifecycle's existing phrase-match logic
        raise UsageLimitReachedError(...)
    extracted = extract_json_payload(payload.get("result", ""), required_keys=output_schema.get("required", []))
    if extracted is None:
        raise StructuredOutputExtractionError(
            "could not extract a payload matching {} from org_node={}'s response".format(
                output_schema.get("required"), org_node
            )
        )
    return extracted
```

（`UsageLimitReachedError`/`_looks_like_usage_limit` 目前分别定义在 `work_graph_execute.py`/`org_node_lifecycle.py` 两处、逻辑不完全相同——这次顺带看要不要统一成一个共享 helper，具体放哪个模块留给实施阶段判断，不在本设计强行拍板，避免引入不必要的耦合决策。）

### 4.2 `extract_json_payload` 泛化（新增在 `work_graph_plan.py`，或提到更底层的共享位置）

```python
def extract_json_payload(text, required_keys=()):
    """整段先当 JSON 解析，失败则逐个 ```json 围栏尝试；只接受包含全部
    required_keys 的 dict（或包一层同名 wrapper key，模型有时会加这层）。
    `_extract_graph_manifest(text)` 等价于
    `extract_json_payload(text, required_keys=["nodes", "edges"])` 外加它自己
    的 "graph_manifest" wrapper-key 特殊处理——保留 `_extract_graph_manifest`
    作为薄封装，不强行让所有调用方都用同一个 wrapper-key 名字。
    """
    ...
```

`_extract_graph_manifest` 改成：

```python
def _extract_graph_manifest(text):
    result = extract_json_payload(text, required_keys=["nodes", "edges"])
    if result is not None:
        return result
    # existing "graph_manifest" wrapper-key fallback, unchanged
    ...
```

### 4.3 三个调用点的迁移

**`_make_run_fn`（主执行路径）**：

```python
def run_fn(cwd, model=None):
    _log_event(conn, node_exec_id, "node_start", org_node=org_node, model=model)
    try:
        output = org_node_trigger.trigger_org_node_structured(
            org_node, prompt, cwd, output_schema, model=model, timeout=timeout,
        )
    except org_node_trigger.UsageLimitReachedError as e:
        _log_event(conn, node_exec_id, "quota_limit", quota_limit_type=e.quota_limit_type)
        work_node_run.raise_usage_limit_signal(e.quota_limit_type)
    except Exception as e:
        _log_event(conn, node_exec_id, "node_failed", error=str(e))
        raise
    else:
        _log_event(conn, node_exec_id, "node_completed", output=output)
        return output
```

`_render_node_script`/对 `work_graph_execute` 的 import 一并删除。

**`_llm_judgment_verify_fn`**：`try` 块内 `work_graph_execute.write_workflow_script(...)` + `execute_work_graph(...)` 两行替换成一行 `result = org_node_trigger.trigger_org_node_structured(org_node, prompt, cwd, _JUDGMENT_SCHEMA, timeout=timeout)`；worktree 包裹逻辑（create/ensure_committed/merge/remove）完全不动。

**`_check_prd_consistency`**：同样把 `write_workflow_script`+`execute_work_graph` 两行替换成 `org_node_trigger.trigger_org_node_structured('planner', prompt, cwd, _PRD_CONSISTENCY_SCHEMA, timeout=timeout)`；worktree 逻辑不动。

### 4.4 usage-limit 退避重试（FR-5）

`work_graph_execute.execute_work_graph_with_usage_limit_retry` 的退避骨架（5→10→20→30 分钟，`sleep`/`now` 参数注入以便测试）整体保留，唯一改动是内层从 `execute_work_graph(...)` 换成 `org_node_trigger.trigger_org_node_structured(...)`——这个包裹函数本身可以留在 `work_graph_execute.py`（模块名不用改，职责变成"给 flat 调用包一层 usage-limit 退避"，不是"驱动 Dynamic Workflow"），或者搬到 `org_node_trigger.py` 跟 `trigger_org_node_structured` 放在一起——搬迁与否留给实施阶段判断，不影响外部调用方看到的行为。

### 4.5 `.claude/workflows/` 目录的后续状态

三个调用点全部迁移完成后，生产代码不再往 `.claude/workflows/` 写新文件（`work_graph_fork.py` 的时间旅行调试功能如果也依赖 Workflow 脚本重放机制，需要在实施前单独确认——这是这次调研没有覆盖到的一个盲点，**实施第一步必须先查**，如果它依赖，这次迁移范围要么把它也迁掉、要么明确排除并记录原因）。已经生成的历史 `.js` 文件保留在版本库里（执行痕迹，不回溯删除），`CLAUDE.md` 关于这一节的描述在迁移完成后需要同步更新为"历史机制，2026-08-08 后不再新增"。

---

## 5. 测试方案

- `src/test_org_node_trigger.py`（新文件或扩展现有测试）：`trigger_org_node_structured()` 的 mock 化测试（构造各种 `result` 文本形态：整段 JSON、单个 ```json 围栏、多个围栏取第一个匹配的、找不到任何匹配抛 `StructuredOutputExtractionError`）；usage-limit 响应识别。
- `src/test_work_graph_plan.py`：`extract_json_payload` 泛化后，`_extract_graph_manifest` 的既有测试必须原样全绿（回归测试，不能因为泛化引入行为变化）。
- `src/test_work_graph_orchestrate.py`：`_make_run_fn`/`_llm_judgment_verify_fn` 迁移后的 mock 测试，从 mock `work_graph_execute.write_workflow_script`/`execute_work_graph` 改成 mock `org_node_trigger.trigger_org_node_structured`——**这是这次改动里回归面最大的一处**，`work_graph_orchestrate.py` 自己的现有测试大概率有几十处需要同步改 mock 目标，实施时预留足够时间。
- `src/test_e2e_pipeline.py`：`_check_prd_consistency` 同上。
- **`RUN_REAL_AGENT_E2E=1` 真实验证（FR-4 的强制要求，不是可选项）**：至少跑一次真实 `--agent <org_node> -p ... --model <downgrade_model>` 组合，确认①人设正确加载（跟 U-031 当初验证 DSL 路径时做的"追问角色身份，确认自称迁移后的人设"同一手法）、②`--model` 覆盖真的生效、③JSON 围栏提取在真实模型输出上稳定工作（不是只在 mock 数据上测得过）。这条不能用 mock 替代，因为要验证的正是"从一个已知能用的路径换到另一个之前从没在这三个调用点上真实跑过的路径"。

---

## 6. 实施顺序

1. **先查 `work_graph_fork.py` 是否依赖 Workflow 脚本重放**——决定这次迁移范围是否要覆盖它（§4.5 提到的盲点）。
2. `extract_json_payload` 泛化 + `_extract_graph_manifest` 回归测试（独立、低风险，先做）。
3. `trigger_org_node_structured()` 新增 + mock 测试。
4. 迁移影响面最小的调用点：`e2e_pipeline.py::_check_prd_consistency`（一个函数，独立）。
5. 迁移 `_llm_judgment_verify_fn`（次小，Phase 2 会大量新增这类调用，越早切过去越好，减少 Phase 2 在两套机制上重复踩坑）。
6. 迁移 `_make_run_fn`（影响面最大，放最后，前两步已经验证过新机制的可靠性）。
7. `RUN_REAL_AGENT_E2E=1` 真实验证（FR-4 的模型覆盖 + 三个调用点各跑一次真实场景）。
8. 全量回归 `discover -s src` + `discover -s supervisor`。
9. `write_workflow_script`/`execute_work_graph`/`execute_work_graph_with_usage_limit_retry` 确认无生产调用方后，标记为历史遗留（不删除文件，加 docstring 说明），更新 `CLAUDE.md` 对应描述。

---

## 7. 跟 Org-Graph-Unification 系列文档的关系

正交但有先后关系：这次改动只影响**单节点执行机制**，不影响 Org Graph 流程形状或自规划范围，`docs/done/org-graph-unification-phase1-review-gate-design.md`（Phase 1）不依赖这次改动，可以并行/先后独立推进。但 `docs/done/org-graph-unification-and-universal-self-planning-requirements.md` 的 **Phase 2**（FR-4，节点级验证 skill）会大量新增 `judgment.kind="llm"` 判断调用——如果这次去黑盒化工作能在 Phase 2 动工前完成，Phase 2 就不需要在两套机制（Workflow 和新的 flat 调用）之间选一套来复用，直接基于 `trigger_org_node_structured()` 设计，减少后续返工。建议顺序：这份文档 → Phase 2，不是必须但值得。