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 脚本。 状态: 设计完成，待实施。

---

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 调用的家）

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，或提到更底层的共享位置）

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 改成：

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（主执行路径）：

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/todo/org-graph-unification-phase1-review-gate-design.md（Phase 1）不依赖这次改动，可以并行/先后独立推进。但 docs/todo/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，不是必须但值得。