# 产物 Gate（逐级验收流水线）— 需求与设计文档

> 状态：设计草稿（2026-06-09）
> 范围：MVP（v1）
> 关联：`docs/research/AGENTOS_POSITIONING.md`（竞争壁垒 claim ③）、`docs/OLLAMA_VERIFICATION.md`、`docs/theory.md`（Guard/Compose 形式化）

---

## 0. 一句话

> 把一次科研任务拆成**有序阶段**（计划→文献→仿真→分析→论文），每个阶段产出**可审计的产物**，并由一个**验收谓词（gate）**判定通过与否——只有通过才进入下一阶段。失败则按策略重试 / 暂停 / 人工介入。

这是把竞争分析里"完全 aspirational、0 代码"的 claim ③ 落到地的最小可用实现。差异化来源：**通用编码 agent 不做垂直科研工作流 + 逐级验收带审计证据链**。

---

## 1. 背景与动机

- 现状：lambdagent 有单 agent / react / 多 agent 编排，但**没有跨阶段的"产物验收"概念**。一次复杂科研任务要么一个大 prompt 跑完（不可控、不可审计），要么人工串多个 agent（无验收、无证据链）。
- 缺口：竞争分析（AGENTOS_POSITIONING §competitive）确认"产物 gate 逐级验收"在商业计划书里被当壁垒宣传，但**运行时零实现**。这是最大的信誉风险，也是最大的差异化机会。
- 形式化契合：阶段验收 ≈ `Guard(stage_agent, predicate, retry)`；阶段串联 ≈ `Compose(g1, g2, ...)`。产物 gate **本就能用现有演算表达**——这同时坐实 AgentOS"形式化内核"叙事。

---

## 2. 目标与非目标

### 2.1 目标（v1）
- G1：用 YAML 声明一条**可配置**的阶段流水线（不硬编码五阶段）。
- G2：顺序执行各阶段，每阶段产物落到**独立的、可审计的 workspace**（复用 run workspace + manifest.json）。
- G3：每阶段一个 **gate（验收谓词）**，支持两类：规则型（确定性 safe-eval）+ LLM-judge 型（语义评分）。
- G4：gate 失败的处置策略：retry(N) / halt / escalate（人工）。
- G5：流水线**可恢复**（中断后从上次未通过的阶段续跑）。
- G6：内置一个"科研五阶段"模板（plan/literature/simulation/analysis/paper）作为开箱即用 preset。
- G7：Python API 可调用；每次流水线运行产出完整审计记录（各阶段产物 + manifest + gate verdict + 总 trace）。

### 2.2 非目标（v1 不做，留 v1.1+）
- DAG / 并行阶段（v1 仅线性）。
- 阶段回滚 / 分支重跑。
- REST API + WebUI 可视化（v1 仅 Python API + CLI）。
- 成本预算 gate（与 cost_grade 集成，留 v1.1）。
- 完整 human-in-the-loop 审批 UI（v1 的 escalate 仅暂停 + 留接口）。

---

## 3. 用户故事

- US1（研究者）：我提交一个课题，系统先产出研究计划；计划没通过验收（缺少可验证假设）就自动重写，通过后才去做文献综述。
- US2（研究者）：流水线跑到"仿真"阶段失败暂停，我查看该阶段 workspace 的产物和 gate 失败原因，修正配置后从该阶段续跑——前面阶段不重跑。
- US3（审稿/合规）：任务结束后我能拿到一条完整证据链：每阶段产出了哪些文件（含 sha256）、gate 怎么判的、判定理由是什么。
- US4（平台方）：我能复用这套流水线编排任意领域的"分阶段+验收"任务，不止科研（如尽调、合同审查）。

---

## 4. 核心概念（领域模型）

```
Pipeline                         一条流水线
├── id / name / version
├── input                        初始输入
├── stages: [Stage, ...]         有序阶段
└── defaults                     全局默认（on_fail、gate 类型等）

Stage                            单个阶段
├── id / name
├── agent                        本阶段执行体（lambdagent 配置引用 或 内联 YAML）
├── consumes: [artifact_ref]     消费的上游产物
├── produces: [ArtifactSpec]     声明应产出的产物（name/type/path-glob/required）
├── gate: Gate                   验收谓词
├── on_fail: OnFailPolicy        retry(N) | halt | escalate | skip
└── budget?                      可选：本阶段成本/步数上限（v1.1）

Gate                             验收谓词（pass/fail）
├── kind: rule | llm_judge | human
├── rule: <safe-eval 表达式>      kind=rule 时
├── judge: {rubric, threshold, model}   kind=llm_judge 时
└── prompt?                      kind=human 时给人看的说明

Verdict                          验收结果（持久化）
├── stage_id / passed: bool
├── score?: float                llm_judge 给的分
├── reasons: [str]
├── kind / attempt / timestamp
└── evidence_ref                 指向该阶段 manifest.json

ArtifactSpec                     产物声明
├── name / type(file|dir|json|md|...)
├── path: <glob，相对阶段 workspace>
└── required: bool
```

---

## 5. 架构与落点

| 层 | 职责 | 复用 |
|---|---|---|
| **`agentpaas/engine/pipeline.py`（新）** | Pipeline/Stage/Gate 编排、状态机、持久化、恢复 | run workspace、manifest、Sandbox |
| **专用 gate 求值器（新，非 `_safe_eval`）** | 规则型 gate 的受限 DSL 求值（仅 has/count/contains/regex/json_len，无属性访问/方法链） | pipeline.py 内（见 §7.1、§15 P1-A） |
| `lambdagent` compiler `from_config` | 把每阶段 agent 配置编译成可执行 term | compiler.py |
| `lambdagent` Guard | 控制流对应物（retry 语义参考，非直接复用 validator） | extensions.py:146 `Guard` |
| `agentpaas/engine/sandbox.py` | 阶段执行 + 每阶段 workspace + manifest | 已实现（含 build_change_manifest）；需加 `workspace_root` 显式参数（§15 P3-F） |

**关键设计决定**：编排器（pipeline.py）放在 **agentpaas 服务层**，因为它依赖 run workspace（agentpaas）且是产品级特性。**gate 求值不复用 `_safe_eval`**——`_safe_eval` 只允许单名 `x` + 白名单 builtins，无法表达 `outputs.has(...)`，放开属性访问又会能力逃逸（§15 P1-A）。改为 pipeline.py 内一个**受限 gate DSL 求值器**（固定函数集，无任意属性/方法）。

**形式化对应（设计锚点，非严格等价 — 见 §15 P1-C/P2-D）**：一条 pipeline 的**控制流**对应
`Compose(Guard(s1.agent, gate1, r1), Guard(s2.agent, gate2, r2), …)`；但**数据流不是 Compose 的单值穿引**——多上游 `consumes` 经**共享 artifact store（CEK 的 store σ / `Memory`，即 `state` effect）**读取。严格说法是"Compose（控制）+ store 读写（数据）"，而非 plain Compose。v1 用专用 `PipelineRunner` 实现，保持这条对应关系，为将来"pipeline 即 term、可进 CEK 机"留路。

---

## 6. 执行模型（状态机）

### 6.1 阶段状态
```
PENDING ─run─▶ RUNNING ─done─▶ GATING ─pass─▶ PASSED
                                  │
                                  └─fail─▶ FAILED
                                            ├ on_fail=retry & attempt<N ─▶ RUNNING
                                            ├ on_fail=halt              ─▶ (pipeline HALTED)
                                            ├ on_fail=escalate          ─▶ AWAITING_HUMAN
                                            └ on_fail=skip              ─▶ SKIPPED ─▶ 下一阶段
```

### 6.2 流水线状态
```
RUNNING ─▶ COMPLETED            （所有 required 阶段 PASSED/SKIPPED）
        ─▶ HALTED               （某阶段 halt）
        ─▶ AWAITING_HUMAN       （某阶段 escalate；等待外部 resume 信号）
```

### 6.3 恢复（resumability，crash-safe — 见 §15 P1-B）
- **write-ahead**：每次 attempt 开始前、gate 求值前各写一次 `pipeline_state.json`，记 `stage_status / attempt_no / attempt_id / workspace_path / started_at / completed_at / verdict_path`。原子写（temp 文件 + rename）+ pipeline 锁防并发。
- **孤儿处理**：resume 时遇到状态停在 `RUNNING/GATING` 的 attempt（崩在半途），确定性处理——检查 `output.json/manifest.json/verdict.json` 是否完整：完整则继续 gating，不完整则判该 attempt 失败并**恰好消耗一次** retry 预算（不重复、不跳过）。
- 续跑跳过 `PASSED/SKIPPED` 阶段，从首个非通过阶段重入；上游产物从其 workspace（artifact store）读取，不重算。
- **人工 resume 语义**：`approved` → 该阶段 `PASSED`（写一条 human verdict）；`rejected` → `FAILED`，再走 `on_fail`。

---

## 7. 验收谓词（gate）机制

### 7.1 规则型（kind=rule，确定性，默认）
- **不复用 `_safe_eval`**（§15 P1-A）：它只允许单名 `x`，表达不了产物访问；放开属性访问会能力逃逸。改用一个**受限 gate DSL 求值器**（pipeline.py 内）。
- **固定函数集**（无任意属性/方法链/对象实例，硬性 size/time 上限，路径按 ArtifactSpec 白名单）：
  - `has(name)` → 该 required/declared 产物是否存在
  - `count()` → 本阶段产物文件数
  - `contains(name, sub)` / `regex(name, pattern)` → 文本匹配（读取上限 `max_bytes`）
  - `json_len(name)` / `json_get(name, path)` → JSON 产物访问
  - `size(name)` → 字节数
- 求值器实现：`ast.parse(expr, mode="eval")` → **节点白名单**（仅 BoolOp/UnaryOp(Not)/Compare/Call-到-固定函数/Constant；**禁** Attribute/Subscript/Lambda/推导式/import）→ 在仅含上述函数、`__builtins__={}` 的命名空间内 eval。
- effect：规则型 gate 是 `pure`。
- 例：`has("plan.md") and contains("plan.md", "hypothesis")`
- 例：`json_len("refs.json") >= 10`
- 例：`count() >= 1`

### 7.2 LLM-judge 型（kind=llm_judge，语义）— ✅ 已实现（v1）
- 给定 rubric + 产物内容 → LLM 打分 → 返回 `{score, reasons}`；`score ≥ threshold` 即通过。
- 实现：`pipeline.py` 的 `default_judge(JudgeRequest) -> JudgeResult`，可注入（`run_pipeline(judge=...)`，测试用 fake judge）。
- `model: {provider, name}`，默认 `ollama`（本地，省钱/隐私）。**已用本地 ollama qwen2.5:7b 实测**：好产物 0.80 通过、跑题 0.00 不通过。
- 稳健解析：`_parse_judge_response` 先抓 JSON `{score,reasons}`，失败回退抓 0..1 数字，clamp 到 [0,1]。
- score/reasons/judge_usage 写进 `verdict.json` 供审计。
- effect：`llm(m)`（effectful gate，对齐 §14.4 G3）。⚠️ **judge token 成本暂未进 stage cost.json**——须把 judge 建成 term 图里的真 agent（`stage≫judge≫parser`）才计入，留 v1.1（§15.2 P2-D）。当前 usage 已记入 verdict.reasons。

### 7.3 人工型（kind=human，v1 仅最小支持）
- 进入 `AWAITING_HUMAN`，把产物 + prompt 暴露给外部；收到 resume(approved/rejected) 信号后继续。
- v1 只提供 Python API 的 resume 钩子，不做 UI。

---

## 8. 数据 / 配置 schema

### 8.1 流水线 YAML（示例：科研五阶段 preset）
```yaml
pipeline:
  id: research-v1
  name: 科研五阶段流水线
  defaults:
    on_fail: retry
    retry: 2
  stages:
    - id: plan
      name: 研究计划
      agent: { ref: agents/planner.yml }      # 或内联 type/model/systemPrompt
      produces:
        - { name: plan.md, type: md, path: "final/plan.md", required: true }
      gate:
        kind: rule
        rule: 'has("plan.md") and contains("plan.md", "hypothesis")'

    - id: literature
      name: 文献综述
      agent: { ref: agents/lit-mapper.yml }
      consumes: [plan]
      produces:
        - { name: refs.json,    type: json, path: "results/refs.json", required: true }
        - { name: survey.md,    type: md,   path: "final/survey.md",   required: true }
      gate:
        kind: rule
        rule: 'has("refs.json") and json_len("refs.json") >= 10'

    - id: simulation
      name: 仿真实验
      agent: { ref: agents/simulator.yml }
      consumes: [plan, literature]
      produces:
        - { name: sim_results, type: dir, path: "results/sim/*", required: true }
      gate:
        kind: rule
        rule: 'count() >= 1'
      on_fail: halt        # 仿真失败不自动重试，直接暂停人工查

    - id: analysis
      name: 结果分析
      agent: { ref: agents/analyst.yml }
      consumes: [simulation]
      produces:
        - { name: analysis.md, type: md, path: "final/analysis.md", required: true }
      gate:
        kind: llm_judge
        judge:
          model: { provider: ollama, name: qwen2.5:7b }
          rubric: "分析是否覆盖了所有仿真指标、是否有统计显著性讨论、结论是否有数据支撑"
          threshold: 0.7

    - id: paper
      name: 论文成稿
      agent: { ref: agents/writer.yml }
      consumes: [plan, literature, analysis]
      produces:
        - { name: paper.md, type: md, path: "final/paper.md", required: true }
      gate:
        kind: llm_judge
        judge:
          rubric: "结构完整(摘要/方法/结果/讨论)、引用与 refs.json 一致、无明显逻辑断裂"
          threshold: 0.8
```

### 8.2 持久化目录布局
```
{agent_dir}/workspace/pipeline_{YYYYMMDD_HHMMSS}/
├── pipeline.yml              ← 流水线配置快照
├── pipeline_state.json       ← 全局状态（可恢复）
├── input.json
├── plan/                     ← 各阶段 = 一个 run workspace
│   ├── code/ results/ final/
│   ├── manifest.json         ← 该阶段产物清单（已实现）
│   ├── trace.json / cost.json
│   └── verdict.json          ← 该阶段验收结果
├── literature/  ...
├── simulation/  ...
├── analysis/    ...
└── paper/       ...
```

### 8.3 verdict.json
```json
{
  "stage_id": "plan",
  "passed": true,
  "score": null,
  "kind": "rule",
  "attempt": 1,
  "reasons": ["rule passed: outputs.has('plan.md') and ..."],
  "evidence_ref": "plan/manifest.json",
  "timestamp": "2026-06-09T..."
}
```

---

## 9. API 表面（v1）

```python
from agentpaas.engine.pipeline import Pipeline, run_pipeline, resume_pipeline

# 加载并运行
result = run_pipeline(
    pipeline_cfg="pipelines/research-v1.yml",
    input_text="研究方向：...",
    agent_dir="/data/agents/researcher",
)
# result: PipelineResult(status, stages=[StageResult...], workspace_path, final_artifacts)

# 恢复（从上次未通过阶段续跑）
result = resume_pipeline(workspace_path="/data/.../pipeline_20260609_...")

# 人工 gate 放行
resume_pipeline(workspace_path=..., human_decision={"stage_id": "simulation", "approved": True})
```

CLI（薄封装）：`agentpaas pipeline run research-v1.yml --input "..."` / `... resume <workspace>`。

---

## 10. MVP 范围裁剪（明确边界）

| 能力 | v1 | v1.1+ |
|---|---|---|
| 线性顺序阶段 | ✅ | |
| 规则型 gate（safe-eval） | ✅ | |
| LLM-judge gate | ✅ | |
| 人工 gate（暂停+API resume） | ✅ 最小 | UI 审批流 |
| on_fail: retry / halt / escalate / skip | ✅ | |
| 每阶段 workspace + manifest + verdict | ✅ | |
| 恢复续跑 | ✅ | |
| 科研五阶段 preset | ✅ | 更多领域模板 |
| DAG / 并行阶段 | ❌ | ✅ |
| 回滚 / 分支重跑 | ❌ | ✅ |
| 成本预算 gate（cost_grade 集成） | ❌ | ✅ |
| REST API + WebUI | ❌ | ✅ |
| 降解为 Compose-of-Guard term（进 CEK） | ❌（保持语义一致） | ✅ |

---

## 11. 风险与对策

- **R1 gate 表达力不足**：规则型只能查存在/数量/正则，复杂语义判不了。
  对策：LLM-judge 兜底；rule context 提供 `read`/`read_json` 等够用的访问器。
- **R2 LLM-judge 不稳定**（同一产物判定漂移）：
  对策：threshold + 固定 temperature=0；judge 失败也走 on_fail（可重试取多数）；记录 reasons 可审计。
- **R3 阶段间耦合/产物契约漂移**：上游改了产物路径，下游 consumes 找不到。
  对策：ArtifactSpec 显式声明 + 启动时静态校验（缺失 required 产物声明即报错）。
- **R4 长流水线半途失败成本高**：
  对策：恢复续跑（G5）+ 每阶段产物永不删除（沿用 run workspace 策略）。
- **R5 与现有 Guard 语义重复造轮子**：
  对策：gate 复用 `_safe_eval`，PipelineRunner 仅做编排/持久化；保持与 Compose-of-Guard 等价。

---

## 12. 实施计划（建议）

| 阶段 | 产出 |
|---|---|
| P0 骨架 | `pipeline.py`：Pipeline/Stage/Gate 数据类 + YAML 解析 + ArtifactSpec 静态校验 |
| P1 执行 | 顺序执行 + 每阶段 workspace（`PipelineRunner` 拥有 stage 目录，§15 P3-F）/manifest/verdict 落盘 + **artifact store**（多上游 consumes，§15 P1-C）+ **运行时 ArtifactSpec 契约校验**（§15 P2-E） |
| P2 gate | 规则型（**受限 gate DSL 求值器**，非 `_safe_eval`，§15 P1-A）✅ → **LLM-judge ✅ 已实现**（可注入 `default_judge`，默认本地 ollama，已实测）→ 人工最小版（TODO） |
| P3 状态机 | on_fail 策略 + **write-ahead** pipeline_state.json + 孤儿处理 + resume/续跑（§15 P1-B） |
| P4 preset+测试 | 科研五阶段 preset + 端到端测试（含 ollama judge 跑通）+ CLI |
| P5 文档 | 用法文档 + 把 claim ③ 从 aspirational 更新为 shipped |

每阶段都应有可跑的测试（参考 test_run_workspace.py / test_ollama.py 的 skip-if-unavailable 模式）。

---

## 13. 验收标准（这份设计实现后算"做到了 claim ③"）

- [x] **能用 YAML 定义并跑通科研流水线（plan→literature 两阶段端到端）**。✅ `agentexample/pipelines/research-demo.yml` 本地 ollama 实跑 COMPLETED：plan 规则 gate 过、lit judge 0.80 过。五阶段模板见 `research-v1.yml`（sim/analysis/paper 需带工具 react agent，v1.1）。
- [ ] 某阶段 gate 失败时按 on_fail 正确处置（retry 能复跑、halt 能暂停）。
- [ ] 中断后 resume 能跳过已通过阶段续跑。
- [ ] 每阶段产出 manifest + verdict，构成可审计证据链。
- [x] **LLM-judge gate 能用本地 ollama 跑**（坐实"自托管+逐级验收"组合卖点）。✅ 已实测：qwen2.5:7b 好产物 0.80 通过 / 跑题 0.00 拒。
- [x] 有自动化测试覆盖状态机的通过/失败/恢复路径。✅ test_pipeline.py 35 个（含 live ollama judge，skip-if-unreachable）。

---

## 14. 形式化对齐（与 LambdaAgent 理论的一致性）

> 结论：产物 gate 与 LambdaAgent 形式化（Papers I/II/III）**高度自洽**——线性流水线 v1 完全在现有构造（`Guard` + `Compose` + store σ + graded cost）的表达力之内。其中三项理论结果直接产生产品红利（运行前成本预测、可化简、可恢复）。需要的理论补丁仅 4 处，v1 只触及其中 2 处小扩展。

### 14.1 可直接落到现有构造的部分

| gate 概念 | 形式化构造 | 出处 | 契合度 |
|---|---|---|---|
| 单个验收阶段 | `Guard(stage_agent, predicate, retry=N)` | Paper I 构造 10 | 字面相同 |
| 阶段串联（线性） | `Compose(g1, g2, …)` | Paper I 构造 3 | 字面相同 |
| 阶段间产物传递 | CEK 机 store σ 读写 / `Memory` | Paper II ⟨C,E,K,**σ**,c⟩ / Paper I 构造 11 | 契合 |
| 中断恢复（resume） | CEK 状态序列化；`pipeline_state.json` = 序列化机器状态 | Paper II CEK + YIELD | 优雅契合 |
| 人工 gate（AWAITING_HUMAN） | YIELD 挂起 + Human Oracle | Paper IV（Oracle Duality） | 契合（依赖 Paper IV） |
| 流水线必然终止 | Bounded Termination，retry 有界 | Paper II Thm 5.4 | 契合 |

**对应式（非严格等价 — 见 §15 P1-C）**：线性流水线的**控制流**对应
`Compose(Guard(s1.agent, gate1, r1), Guard(s2.agent, gate2, r2), …)`；**数据流**经共享 store σ（多上游 `consumes`），不是 Compose 单值穿引。故严格说法 = "Compose（控制）+ store 读写（`state` effect，数据）"。v1 用专用 `PipelineRunner` 实现，保持这条对应，为"pipeline 即 term、可进 CEK"留路。

### 14.2 最强契合：graded cost 给"运行前成本预测"

Paper III Def 12 的成本组合规则给出 gate 成本的**目标对应**（非"已严格覆盖" — 见 §15 P2-D）：

- **Serial(Compose)** → `(p₁·p₂, t₁+t₂, l₁+l₂, m₁+m₂)`
- **Guard(k)**，其中 **k = 1 + retries**（与 `Guard.apply` 的 `1+retry` 循环一致）→ `(1-(1-p)^k, k·t, k·l, k·m)`

因此流水线成本上界 = 各阶段 `grade_guard` 再 `grade_serial` 折叠，**原则上跑前可推断** "≤ $X、≤ T tokens、成功率 ≥ p"；`p_fatal`：gate 没过 = 非致命（重试），API 5xx = 致命。

✅ **已实现 `estimate_pipeline_cost(pipeline)`**（pipeline.py）：把 judge 建成真 `Lam` term（`build_judge_term`），用 `grade_serial(stage_agent, judge_term)` 再按 retry `grade_guard`、阶段间 `grade_serial` 折叠，给出运行前成本上界。实测 research-demo：lit 阶段 judge 计入 ≤800 tokens / ≤$0.0024 / p≥0.99。这是 Paper33/34 "actual ≤ predicted" 实证的预测侧。

⚠️ **残留差距**：`estimate_cost` 只认 `Lam`，**不认 simple agent 编出的 `ConversationLam`** → 上面 plan 阶段 agent 成本算成 0（只有 judge 那段被计）。要全保真，需给 `estimate_cost` 加 ConversationLam 分支（小改 cost_grade.py，留 v1.1）。规则型 gate 成本近 0（pure）可忽略。

### 14.3 代数律给"流水线变换"

- **Compose 结合律** `(s1≫s2)≫s3 = s1≫(s2≫s3)`：子流水线嵌套/分组不改变语义。
- **Guard 幂等** `Guard(Guard(f,P),P) = Guard(f,P,retry×2)`：叠 gate = 叠重试，可用于静态化简。

### 14.4 待扩展的 gap

| gap | 现状 | 需要的扩展 | v1 是否受影响 |
|---|---|---|---|
| **G1 多上游 consumes** | Compose 只串一条数据流；但**v1 preset 本身**就有 simulation 消费 [plan,lit]、paper 消费 [plan,lit,analysis]（§15 P1-C 纠正了"v1 不受影响"的错误） | 数据流改走**共享 artifact store**（每阶段从上游 workspace 读 declared 产物），即 `state` effect；成本/DAG 拓扑折叠留 v1.1 | ⚠️ **v1 受影响**：数据流必须经 store，已纳入 §5/§6.3 设计；控制流仍线性 |
| **G2 谓词作用域** | Paper 的 Guard 谓词 `P(r)` 只看返回值；gate 要看 workspace 产物（store） | Guard 谓词 `P(r)` → **`P(r, ctx)`**（值 + store/产物访问器） | ✅ v1 采纳（自然扩展，σ 本就在 CEK 内） |
| **G3 effectful gate** | Paper 隐含谓词 `pure`；LLM-judge gate 有 `llm` effect | T-Guard 允许 `ε_P ≠ pure`，validator 的 effect/cost 计入组合 | ✅ v1 采纳（cost 折叠多加一项） |
| **G4 人工 gate** | 需 Human Oracle 形式化 | Paper IV：Oracle Duality（LLM oracle ≃ human oracle） | Paper IV 未成文；v1 人工 gate 仅工程最小版 |

### 14.5 对 v1 实现的约束（由对齐导出）

1. **gate 谓词签名采用 `P(r, ctx)`**（见 §7.1），而非 Paper 原始的 `P(r)`——对齐 G2。`ctx` 暴露 store/产物访问器。`_compile_guard` 的 `validator_fn` 相应从 `validator_fn(x)` 扩展为 `validator_fn(x, ctx=None)`，向后兼容（旧 Guard 不传 ctx）。
2. **LLM-judge gate 视为 effectful validator**（effect = `llm(m)`）——对齐 G3；其 token/latency/money 计入该阶段 cost。
3. **v1 控制流线性、数据流经 store**——回避 DAG 成本拓扑，但 multi-upstream consumes 必须走 artifact store（§15 P1-C）。
4. **人工 gate 走 YIELD 语义**——为将来接 Paper IV Human Oracle 留接口。

---

## 15. 评审修订记录（codex 2026-06-09，实现前）

codex 评审在动手前找出 6 个问题，全部接受并已修订本文档。摘要如下：

| # | 严重度 | 问题 | 修订 |
|---|---|---|---|
| P1-A | 致命 | `_safe_eval` 表达不了 `outputs.has(...)`，放开属性访问会能力逃逸——"复用 `_safe_eval`"既不可行也不安全 | §5/§7.1：改为 pipeline.py 内**受限 gate DSL 求值器**（固定函数集 has/count/contains/regex/json_len，禁属性/方法链，size/time 上限，路径按 ArtifactSpec 白名单） |
| P1-B | 致命 | 只在"阶段结束"写状态，崩在半途会重复/跳过 retry、孤儿 workspace；retry 计数只在内存；人工 resume 语义未定义 | §6.3：**write-ahead**（attempt 前 + gate 前各写一次，原子 rename + 锁）；孤儿确定性处理（恰好消耗一次 retry）；人工 resume 语义明确（approved→PASSED / rejected→FAILED→on_fail） |
| P1-C | 致命 | 多上游 `consumes` 与"v1 plain Compose"矛盾——v1 preset 本身就需非相邻上游读取 | §5/§14.1/§14.4-G1：数据流改走**共享 artifact store**（`state` effect），对应式降级为"Compose（控制）+ store（数据）"，不再声称严格 plain-Compose 等价 |
| P2-D | 中 | §14 成本声称过强：`estimate_cost(Guard)` 不含 validator 成本；`k` 含糊 | §14.2：降级为"目标对应"；统一 **k = 1+retries**；LLM-judge 须建成 term 图里的真 agent（`stage≫judge≫parser`）其成本才计入 |
| P2-E | 中 | ArtifactSpec 静态校验太弱，只证名字存在，不防契约漂移 | §12-P1：加**运行时契约校验**（每 attempt 后、gate 前解析 spec、校验 type/cardinality、把路径+hash 记进 verdict、required 缺失即 fail）；静态校验另查 consumes 引用合法 |
| P3-F | 低 | workspace 布局与现有 `create_run_workspace`（时间戳 run 目录）不符，无 API 强制 stage 目录在 pipeline 下 | §5/§12-P1：`PipelineRunner` 自己拥有 stage 目录创建（或给 `create_run_workspace` 加 `workspace_root` 参数），不依赖时间戳发现 |

**结论**：三个 P1 在 v1 就必须处理（已纳入 §5/§6.3/§7.1）；P2/P3 纳入 P1/P2 实施阶段。原 §14 "高度自洽/严格等价"的措辞过乐观，已据实修正为"控制流对应 + 数据流经 store"。

### 15.2 实现审查（codex，P0+P1 骨架成稿后）

骨架 `pipeline.py` 写完后第二轮 codex review，又找出 8 个实现级问题，已修（除完整孤儿恢复）：

| # | 严重度 | 问题 | 状态 |
|---|---|---|---|
| P1-1 | 致命(安全) | `ArtifactSpec.path` 可用绝对路径/`..` 逃逸 stage_ws，gate 能读 `/etc/passwd` | ✅ 已修：`_is_safe_relpath` + glob 后 `commonpath` 强制在 ws 内 |
| P1-2 | 致命 | 阶段在重试 attempt 通过后，resume 从 base `stage_dir` 取数（错目录），下游消费陈旧产物 | ✅ 已修：持久化通过 attempt 的 `workspace_path`，resume 用它 |
| P1-3 | 致命 | `create_run_workspace(run_dir=)` 用 `exist_ok=True` 不清空，复用目录旧产物虚假满足 `has()` | ✅ 已修：attempt 目录复用前 `rmtree` 清空 |
| P1-4 | 致命 | `on_fail=halt` 仍跑 `1+retry` 次，与"直接 halt"矛盾 | ✅ 已修：retry 预算仅 `on_fail==RETRY` 时给；halt/escalate/skip = 单次 |
| P1-5 | 致命 | 终态失败只在内存设置、由 caller 后续持久化，留崩溃窗口 | ✅ 部分修：`_apply_on_fail` 立即原子持久化终态；**完整孤儿恢复仍 P3** |
| P2-6 | 中 | 契约校验未做 type/cardinality：json 不解析、dir 可指文件、one 命中多个静默忽略 | ✅ 已修：`_type_ok` 校验 + cardinality=one 多命中记问题 |
| P2-7 | 中 | `regex()` 无 timeout，ReDoS | ⚠️ 缓解：pattern 长度上限 200 + 搜索窗 64KB；真 timeout/RE2 留 v1.1 |
| P2-8 | 中 | 规则 gate 未静态校验，执行后才报错 | ✅ 已修：`validate_pipeline` 对 kind=rule 调 `ast.parse + _validate_gate_expr` |

codex 同时确认：AST 白名单 + `{"__builtins__":{}}` **无直接 Python 对象逃逸**（Attribute/Subscript/推导式/f-string/walrus/lambda/starred/方法调用全被拒）。

测试 `test_pipeline.py` 加了负向用例（路径穿越、非法 JSON、静态 gate 校验、halt 不重试、retry-pass resume 取对目录），共 26 个。**遗留 P3**：崩溃-孤儿 attempt 的确定性恢复、真正的 regex timeout。