feat(engine): 产物 Gate 流水线 P0+P1 + run 修改清单 + Ollama 实测修复

- engine/pipeline.py: 受限 gate DSL 求值器(AST 白名单)、artifact store
  多上游 consumes、运行时契约校验、write-ahead 状态、LLM-judge gate
  (default_judge 可注入, 本地 ollama 默认)、judge 作 term 进成本折叠;
  含两轮 codex 评审修订(路径穿越/retry resume/attempt 清空等, 设计文档 §15)
- engine/sandbox.py: build_change_manifest() 每 run 落 manifest.json
  (目录扫描+best-effort git diff); create_run_workspace 加 run_dir 参;
  审计⑧: 先文件系统探测 .git 再起 git 子进程(非 git 目录 58ms→1.5ms)
- primitives.py: _call_llm 剥 'ollama/' 前缀修 404; 本地模型计费 $0
- openai_compat_provider.py: chat_typed usage 累积修复
- tests: test_pipeline.py(39, 含 live ollama skip)、test_run_workspace.py
  manifest 用例(25)、test_ollama.py(成本单测+live skip)
- agentexample/pipelines/: research-demo.yml(本地实跑 COMPLETED) + research-v1.yml
- docs: PRODUCT_GATE_DESIGN.md(15 节)、OLLAMA_VERIFICATION.md

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

agentexample/pipelines/qa-gated-demo.yml

@@ -0,0 +1,31 @@
+# 真案例：给一个【现有 agent】套 gate。
+# answer-generator 是 qaagent67 里已存在的问答 skill（type: simple, ollama）。
+# 这里不改它，只把它当成流水线的一个 stage，在它之后加一道 LLM-judge 验收：
+# 答案必须基于参考文档、直接回答问题，否则 gate 拒绝、流水线 halt。
+#
+# 运行：
+#   from agentpaas.engine.pipeline import run_pipeline
+#   run_pipeline("agentexample/pipelines/qa-gated-demo.yml",
+#                "问题：...\n\n参考文档：...", agent_dir, base_dir="agentexample")
+pipeline:
+  id: qa-gated-demo
+  name: QA agent + 验收 gate
+  defaults:
+    on_fail: halt
+
+  stages:
+    - id: answer
+      name: 生成回答（复用现有 answer-generator skill）
+      agent: {ref: qaagent67/skills/answer-generator.yml}   # ← 现有 agent，原样引用
+      model_override: qwen2.5:7b                              # ← 本地模型覆盖（原配 32b）
+      produces:
+        - {name: answer.md, type: md, path: "final/answer.md", required: true}
+      gate:
+        kind: llm_judge
+        judge:
+          model: {provider: ollama, name: qwen2.5:7b}
+          rubric: >
+            The answer must be grounded in the provided reference documents and
+            directly answer the user's question. Reject if it is a refusal,
+            off-topic, or makes claims not supported by the documents.
+          threshold: 0.6

agentexample/pipelines/research-demo.yml

@@ -0,0 +1,45 @@
+# 科研流水线 Demo — 可端到端运行（plan → literature），全部本地 ollama。
+# rule gate + llm_judge gate 各一，验证产物 gate 逐级验收闭环。
+# 运行：run_pipeline("agentexample/pipelines/research-demo.yml", "<研究方向>", agent_dir)
+pipeline:
+  id: research-demo
+  name: 科研流水线 Demo (plan→lit)
+  defaults:
+    on_fail: halt
+
+  stages:
+    - id: plan
+      name: 研究计划
+      agent:
+        type: simple
+        model: {provider: ollama, name: qwen2.5:7b, temperature: 0.2, maxTokens: 600}
+        systemPrompt: |
+          You are a research planner. Given a topic, write a concise research plan
+          (<= 200 words) that EXPLICITLY states a research hypothesis using the word
+          "hypothesis". Plain prose only.
+      produces:
+        - {name: plan.md, type: md, path: "final/plan.md", required: true}
+      gate:
+        kind: rule
+        rule: 'regex("plan.md", "(?i)hypothesis")'
+
+    - id: lit
+      name: 文献综述
+      consumes: [plan]
+      agent:
+        type: simple
+        model: {provider: ollama, name: qwen2.5:7b, temperature: 0.3, maxTokens: 700}
+        systemPrompt: |
+          You are a literature reviewer. Based on the given research plan (provided
+          under UPSTREAM ARTIFACTS), write a short related-work survey (<= 250 words)
+          discussing prior approaches relevant to the plan.
+      produces:
+        - {name: survey.md, type: md, path: "final/survey.md", required: true}
+      gate:
+        kind: llm_judge
+        judge:
+          model: {provider: ollama, name: qwen2.5:7b}
+          rubric: >
+            The text is a literature / related-work survey that discusses prior
+            approaches or references relevant to a research plan — not off-topic text.
+          threshold: 0.5

agentexample/pipelines/research-v1.yml

@@ -0,0 +1,67 @@
+# 科研五阶段产物 Gate 流水线（模板）— 对应 docs/PRODUCT_GATE_DESIGN.md §8.1。
+# plan → literature → simulation → analysis → paper，逐级验收。
+# 注：plan/literature 已可端到端跑（见 research-demo.yml）；simulation/analysis/paper
+# 需要带工具的 react agent（WriteFile / 代码执行），属 v1.1 工作，这里作为模板占位。
+pipeline:
+  id: research-v1
+  name: 科研五阶段流水线
+  defaults:
+    on_fail: retry
+    retry: 2
+
+  stages:
+    - id: plan
+      name: 研究计划
+      agent: {ref: agents/planner.yml}
+      produces:
+        - {name: plan.md, type: md, path: "final/plan.md", required: true}
+      gate:
+        kind: rule
+        rule: 'has("plan.md") and regex("plan.md", "(?i)hypothesis")'
+
+    - id: literature
+      name: 文献综述
+      consumes: [plan]
+      agent: {ref: agents/lit-mapper.yml}
+      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: 仿真实验
+      consumes: [plan, literature]
+      agent: {ref: agents/simulator.yml}    # react + 代码执行工具（v1.1）
+      produces:
+        - {name: sim_results, type: dir, path: "results/sim/*", required: true}
+      gate:
+        kind: rule
+        rule: 'count() >= 1'
+      on_fail: halt
+
+    - id: analysis
+      name: 结果分析
+      consumes: [simulation]
+      agent: {ref: agents/analyst.yml}
+      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: 论文成稿
+      consumes: [plan, literature, analysis]
+      agent: {ref: agents/writer.yml}
+      produces:
+        - {name: paper.md, type: md, path: "final/paper.md", required: true}
+      gate:
+        kind: llm_judge
+        judge:
+          rubric: "结构完整(摘要/方法/结果/讨论)、引用与 refs.json 一致、无明显逻辑断裂"
+          threshold: 0.8

agentpaas/src/agentpaas/engine/pipeline.py

@@ -0,0 +1,1029 @@
+"""
+engine.pipeline — 产物 Gate 逐级验收流水线（P0+P1 骨架）.
+
+设计文档：docs/PRODUCT_GATE_DESIGN.md
+本模块实现 P0（数据类 + YAML 解析 + 静态校验）和 P1（顺序执行 + 每阶段
+workspace/manifest/verdict + artifact store 多上游 consumes + 运行时契约校验
++ write-ahead 状态）。
+
+codex 评审（§15）的修复已内建：
+  P1-A 受限 gate DSL 求值器（非 _safe_eval，固定函数集，禁属性/方法链）
+  P1-B write-ahead 状态（attempt 前/gate 前/verdict 后各写一次，原子 rename）
+  P1-C 多上游 consumes 经共享 artifact store（每阶段从上游 workspace 读 declared 产物）
+  P2-E 运行时 ArtifactSpec 契约校验（gate 前解析、required 缺失即 fail、记 hash 进 verdict）
+  P3-F PipelineRunner 自己拥有 stage 目录（create_run_workspace(run_dir=...)）
+
+未实现（留 P2/P3，已用 NotImplementedError / TODO 标出）：
+  - LLM-judge gate（须建成 term 图里的真 agent，§15 P2-D）
+  - human gate 的完整 resume 语义（§6.3）
+  - 崩溃-孤儿 attempt 的确定性恢复（§15 P1-B 的完整版）
+"""
+from __future__ import annotations
+
+import ast
+import glob as _glob
+import hashlib
+import json
+import os
+import re
+import time
+from dataclasses import dataclass, field, asdict
+from enum import Enum
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+from .sandbox import create_run_workspace, save_run_artifacts, build_change_manifest
+
+
+# ============================================================
+# Exceptions
+# ============================================================
+
+class PipelineError(Exception):
+    """流水线配置/执行错误。"""
+
+
+class GateError(Exception):
+    """gate 求值错误（含不安全表达式）。"""
+
+
+# ============================================================
+# Enums
+# ============================================================
+
+class StageStatus(str, Enum):
+    PENDING = "pending"
+    RUNNING = "running"
+    GATING = "gating"
+    PASSED = "passed"
+    FAILED = "failed"
+    SKIPPED = "skipped"
+    AWAITING_HUMAN = "awaiting_human"
+
+
+class PipelineStatus(str, Enum):
+    RUNNING = "running"
+    COMPLETED = "completed"
+    HALTED = "halted"
+    AWAITING_HUMAN = "awaiting_human"
+
+
+class OnFail(str, Enum):
+    RETRY = "retry"        # 用 retry 次数重试后仍失败 → halt
+    HALT = "halt"
+    ESCALATE = "escalate"  # → AWAITING_HUMAN
+    SKIP = "skip"
+
+
+# ============================================================
+# Domain model
+# ============================================================
+
+@dataclass
+class ArtifactSpec:
+    name: str
+    type: str = "file"          # file | dir | json | md | ...
+    path: str = ""              # glob，相对 stage workspace
+    required: bool = True
+    cardinality: str = "one"    # one | many
+
+
+@dataclass
+class Gate:
+    kind: str = "rule"          # rule | llm_judge | human
+    rule: str = "True"          # kind=rule 时的 DSL 表达式
+    judge: dict = field(default_factory=dict)   # kind=llm_judge
+    prompt: str = ""            # kind=human
+
+
+@dataclass
+class Stage:
+    id: str
+    name: str = ""
+    agent: dict = field(default_factory=dict)        # {ref: path} 或内联配置
+    model_override: str = ""                          # 覆盖被引用 agent 的模型（如本地 ollama 跑云端 agent）
+    consumes: List[str] = field(default_factory=list)
+    produces: List[ArtifactSpec] = field(default_factory=list)
+    gate: Gate = field(default_factory=Gate)
+    on_fail: OnFail = OnFail.RETRY
+    retry: int = 0
+
+
+@dataclass
+class Pipeline:
+    id: str
+    name: str = ""
+    stages: List[Stage] = field(default_factory=list)
+
+
+@dataclass
+class ResolvedArtifact:
+    name: str
+    type: str
+    path: str
+    exists: bool
+    size_bytes: int = 0
+    sha256: Optional[str] = None
+
+
+@dataclass
+class Verdict:
+    stage_id: str
+    passed: bool
+    kind: str
+    attempt: int
+    reasons: List[str] = field(default_factory=list)
+    score: Optional[float] = None
+    artifacts: List[dict] = field(default_factory=list)   # resolved artifact metadata
+    timestamp: str = ""
+
+
+@dataclass
+class StageResult:
+    stage_id: str
+    status: StageStatus
+    attempts: int
+    workspace_path: str = ""
+    verdict: Optional[Verdict] = None
+    error: str = ""
+
+
+@dataclass
+class PipelineResult:
+    pipeline_id: str
+    status: PipelineStatus
+    workspace_path: str
+    stages: List[StageResult] = field(default_factory=list)
+
+
+# ============================================================
+# P0: 配置加载 + 静态校验
+# ============================================================
+
+def load_pipeline(cfg: Any) -> Pipeline:
+    """从 dict 或 YAML 文件路径加载 Pipeline。"""
+    if isinstance(cfg, str):
+        import yaml
+        with open(cfg, encoding="utf-8") as f:
+            data = yaml.safe_load(f)
+    else:
+        data = cfg
+    if not isinstance(data, dict):
+        raise PipelineError("pipeline config must be a mapping")
+
+    pdata = data.get("pipeline", data)
+    stages: List[Stage] = []
+    for sd in pdata.get("stages", []):
+        produces = [
+            ArtifactSpec(
+                name=a["name"],
+                type=a.get("type", "file"),
+                path=a.get("path", ""),
+                required=a.get("required", True),
+                cardinality=a.get("cardinality", "one"),
+            )
+            for a in sd.get("produces", [])
+        ]
+        gd = sd.get("gate", {}) or {}
+        gate = Gate(
+            kind=gd.get("kind", "rule"),
+            rule=gd.get("rule", "True"),
+            judge=gd.get("judge", {}) or {},
+            prompt=gd.get("prompt", ""),
+        )
+        defaults = pdata.get("defaults", {}) or {}
+        on_fail_raw = sd.get("on_fail", defaults.get("on_fail", "retry"))
+        stages.append(Stage(
+            id=sd["id"],
+            name=sd.get("name", sd["id"]),
+            agent=sd.get("agent", {}) or {},
+            model_override=sd.get("model_override", ""),
+            consumes=list(sd.get("consumes", [])),
+            produces=produces,
+            gate=gate,
+            on_fail=OnFail(on_fail_raw),
+            retry=int(sd.get("retry", defaults.get("retry", 0))),
+        ))
+    return Pipeline(id=pdata.get("id", "pipeline"), name=pdata.get("name", ""), stages=stages)
+
+
+def validate_pipeline(p: Pipeline) -> List[str]:
+    """静态校验（§15 P2-E 的静态部分）。返回错误列表（空 = 合法）。"""
+    errs: List[str] = []
+    seen_ids: set = set()
+    for i, st in enumerate(p.stages):
+        if not st.id:
+            errs.append(f"stage[{i}] missing id")
+            continue
+        if st.id in seen_ids:
+            errs.append(f"duplicate stage id: {st.id}")
+        # consumes 必须指向先前已声明的阶段（线性，禁前向引用）
+        for c in st.consumes:
+            if c not in seen_ids:
+                errs.append(f"stage '{st.id}' consumes unknown/forward stage '{c}'")
+        # produced 名字阶段内唯一
+        names = [a.name for a in st.produces]
+        if len(names) != len(set(names)):
+            errs.append(f"stage '{st.id}' has duplicate produced artifact names")
+        if st.gate.kind not in ("rule", "llm_judge", "human"):
+            errs.append(f"stage '{st.id}' invalid gate kind: {st.gate.kind}")
+        # §15 P2-8: 静态校验规则 gate（语法 + AST 白名单），别等执行后才报错
+        if st.gate.kind == "rule":
+            try:
+                _validate_gate_expr(ast.parse(st.gate.rule, mode="eval"))
+            except SyntaxError as e:
+                errs.append(f"stage '{st.id}' gate syntax error: {e}")
+            except GateError as e:
+                errs.append(f"stage '{st.id}' gate not allowed: {e}")
+        seen_ids.add(st.id)
+    return errs
+
+
+# ============================================================
+# P1-A: 受限 gate DSL 求值器（不复用 _safe_eval）
+# ============================================================
+
+# 仅这些节点类型允许出现在 gate 表达式中（禁 Attribute/Subscript/Lambda/推导式/import）
+_GATE_ALLOWED_NODES = (
+    ast.Expression, ast.BoolOp, ast.And, ast.Or,
+    ast.UnaryOp, ast.Not,
+    ast.Compare, ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.Gt, ast.GtE,
+    ast.Call, ast.Name, ast.Load, ast.Constant,
+)
+_GATE_FUNCS = frozenset({"has", "count", "contains", "regex", "json_len", "json_get", "size"})
+_GATE_READ_MAX_BYTES = 1_000_000
+_GATE_REGEX_MAX_PATTERN = 200       # §15 P2-7: cap regex pattern length
+_GATE_REGEX_READ_MAX = 65_536       # §15 P2-7: smaller window for regex search
+
+
+class GateContext:
+    """gate DSL 的固定函数集，作用于本阶段已解析的产物。无任意属性/方法暴露。"""
+
+    def __init__(self, resolved: Dict[str, ResolvedArtifact], max_bytes: int = _GATE_READ_MAX_BYTES):
+        self._a = resolved
+        self._max = max_bytes
+
+    def _read(self, name: str) -> str:
+        ra = self._a.get(name)
+        if not ra or not ra.exists or not os.path.isfile(ra.path):
+            return ""
+        try:
+            with open(ra.path, encoding="utf-8", errors="replace") as f:
+                return f.read(self._max)
+        except OSError:
+            return ""
+
+    def has(self, name: str) -> bool:
+        ra = self._a.get(name)
+        return bool(ra and ra.exists)
+
+    def count(self) -> int:
+        return sum(1 for ra in self._a.values() if ra.exists)
+
+    def contains(self, name: str, sub: str) -> bool:
+        return str(sub) in self._read(name)
+
+    def regex(self, name: str, pattern: str) -> bool:
+        # ReDoS 缓解（§15 P2-7）：限制 pattern 长度 + 缩小搜索窗口。真正的时间界需
+        # RE2/超时引擎，留 v1.1；此处用保守上限降低风险。
+        if len(str(pattern)) > _GATE_REGEX_MAX_PATTERN:
+            return False
+        try:
+            return re.search(pattern, self._read(name)[:_GATE_REGEX_READ_MAX]) is not None
+        except re.error:
+            return False
+
+    def size(self, name: str) -> int:
+        ra = self._a.get(name)
+        return ra.size_bytes if (ra and ra.exists) else 0
+
+    def json_len(self, name: str) -> int:
+        try:
+            return len(json.loads(self._read(name)))
+        except Exception:
+            return 0
+
+    def json_get(self, name: str, path: str) -> Any:
+        try:
+            d = json.loads(self._read(name))
+        except Exception:
+            return None
+        for part in str(path).split("."):
+            if isinstance(d, dict) and part in d:
+                d = d[part]
+            else:
+                return None
+        return d
+
+
+def _validate_gate_expr(tree: ast.AST) -> None:
+    for n in ast.walk(tree):
+        if isinstance(n, ast.Call):
+            if not isinstance(n.func, ast.Name) or n.func.id not in _GATE_FUNCS:
+                raise GateError(f"gate: only {sorted(_GATE_FUNCS)} calls allowed")
+            if n.keywords:
+                raise GateError("gate: keyword args not allowed")
+        elif isinstance(n, ast.Name):
+            if n.id not in _GATE_FUNCS:
+                raise GateError(f"gate: name not allowed: {n.id!r}")
+        elif not isinstance(n, _GATE_ALLOWED_NODES):
+            raise GateError(f"gate: node not allowed: {type(n).__name__}")
+
+
+def eval_rule_gate(expr: str, resolved: Dict[str, ResolvedArtifact]) -> Tuple[bool, List[str]]:
+    """求值规则型 gate。返回 (passed, reasons)。"""
+    try:
+        tree = ast.parse(expr, mode="eval")
+    except SyntaxError as e:
+        raise GateError(f"gate syntax error: {e}") from e
+    _validate_gate_expr(tree)
+    gctx = GateContext(resolved)
+    ns = {fn: getattr(gctx, fn) for fn in _GATE_FUNCS}
+    try:
+        result = eval(compile(tree, "<gate>", "eval"), {"__builtins__": {}}, ns)  # noqa: S307 - AST-whitelisted
+    except Exception as e:
+        raise GateError(f"gate eval error: {e}") from e
+    passed = bool(result)
+    return passed, [f"rule {'passed' if passed else 'failed'}: {expr}"]
+
+
+# ============================================================
+# P2-E: 运行时 ArtifactSpec 契约校验
+# ============================================================
+
+def _sha256(path: str) -> Optional[str]:
+    try:
+        h = hashlib.sha256()
+        with open(path, "rb") as f:
+            for chunk in iter(lambda: f.read(65536), b""):
+                h.update(chunk)
+        return h.hexdigest()
+    except OSError:
+        return None
+
+
+def _is_safe_relpath(p: str) -> bool:
+    """拒绝空/绝对路径与含 `..` 的路径（§15 P1：防 artifact path 逃逸 stage_ws）。"""
+    if not p or os.path.isabs(p):
+        return False
+    return ".." not in os.path.normpath(p).split(os.sep)
+
+
+def _under(path: str, root_real: str) -> bool:
+    try:
+        return os.path.commonpath([os.path.realpath(path), root_real]) == root_real
+    except ValueError:
+        return False
+
+
+def _type_ok(path: str, t: str) -> bool:
+    """运行时类型校验（§15 P2-E）。"""
+    if t == "dir":
+        return os.path.isdir(path)
+    if t == "json":
+        if not os.path.isfile(path):
+            return False
+        try:
+            with open(path, encoding="utf-8") as f:
+                json.load(f)
+            return True
+        except Exception:
+            return False
+    return os.path.isfile(path)   # file / md / 其它当作文件
+
+
+def resolve_artifacts(stage: Stage, stage_ws: str) -> Tuple[Dict[str, ResolvedArtifact], List[str]]:
+    """对 stage workspace 解析每个 ArtifactSpec（§15 P1 路径安全 + P2-E 类型/基数校验）。
+
+    返回 (resolved, problems)。problems 非空表示 required 产物缺失/非法/基数不符，
+    上层据此在 gate 前 fail 该阶段。
+    """
+    resolved: Dict[str, ResolvedArtifact] = {}
+    problems: List[str] = []
+    sw_real = os.path.realpath(stage_ws)
+
+    for spec in stage.produces:
+        def _absent(reason: str) -> None:
+            resolved[spec.name] = ResolvedArtifact(spec.name, spec.type, "", False)
+            if spec.required:
+                problems.append(f"{spec.name}: {reason}")
+
+        # P1 路径安全：拒绝绝对路径 / ..
+        if not _is_safe_relpath(spec.path):
+            _absent(f"unsafe or empty path {spec.path!r}")
+            continue
+
+        raw = _glob.glob(os.path.join(stage_ws, spec.path))
+        # 强制命中项落在 stage_ws 内（防 glob/symlink 逃逸）
+        in_ws = sorted(m for m in raw if _under(m, sw_real))
+        # P2-E 类型校验
+        valid = [m for m in in_ws if _type_ok(m, spec.type)]
+
+        if not valid:
+            _absent(f"no valid {spec.type} match for {spec.path!r}")
+            continue
+        if spec.cardinality == "one" and len(valid) > 1:
+            # 声明 one 却命中多个 = 契约漂移
+            path = valid[0]
+            is_file = os.path.isfile(path)
+            resolved[spec.name] = ResolvedArtifact(
+                spec.name, spec.type, path, True,
+                os.path.getsize(path) if is_file else 0,
+                _sha256(path) if is_file else None,
+            )
+            if spec.required:
+                problems.append(f"{spec.name}: cardinality=one but {len(valid)} matches")
+            continue
+
+        # 正常解析（many 取第一个作主路径；count() 仍按 spec 计）
+        path = valid[0]
+        is_file = os.path.isfile(path)
+        resolved[spec.name] = ResolvedArtifact(
+            spec.name, spec.type, path, True,
+            os.path.getsize(path) if is_file else 0,
+            _sha256(path) if is_file else None,
+        )
+
+    return resolved, problems
+
+
+# ============================================================
+# Stage executor（可注入，便于测试）
+# ============================================================
+
+@dataclass
+class StageExecContext:
+    stage: Stage
+    stage_ws: str
+    input_text: str
+    consumed: Dict[str, Dict[str, ResolvedArtifact]]   # 上游 stage_id -> 其 resolved 产物
+    base_dir: str = ""
+
+
+@dataclass
+class StageExecResult:
+    output: str = ""
+    status: str = "completed"
+    error: str = ""
+    trace: list = field(default_factory=list)
+    usage: Optional[dict] = None
+
+
+StageExecutor = Callable[[StageExecContext], StageExecResult]
+
+
+def default_stage_executor(ec: StageExecContext) -> StageExecResult:
+    """默认执行体：把 stage.agent 配置编译成 term 在进程内跑。
+
+    P1 骨架版：in-process（Level 0）。接 Sandbox 隔离级别留 P2。
+    """
+    agent = ec.stage.agent or {}
+    try:
+        from lambdagent.fromconfig import from_config
+        from lambdagent.core import Context
+
+        overrides = {"workspace_path": ec.stage_ws}
+        if ec.stage.model_override:        # 覆盖被引用 agent 的模型（本地 ollama 跑云端 agent）
+            overrides["model"] = ec.stage.model_override
+        if "ref" in agent:
+            ref = agent["ref"]
+            cfg_path = ref if os.path.isabs(ref) else os.path.join(ec.base_dir, ref)
+            term = from_config(cfg_path, **overrides)
+        else:
+            # 内联配置：写临时 yml 再编译
+            import tempfile, yaml
+            with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as f:
+                yaml.dump(agent, f, allow_unicode=True)
+                tmp = f.name
+            try:
+                term = from_config(tmp, **overrides)
+            finally:
+                try:
+                    os.unlink(tmp)
+                except OSError:
+                    pass
+
+        # 把上游 consumes 的产物内容拼进输入，实现真正的跨阶段数据流（P1-C）
+        agent_input = ec.input_text
+        up_parts: List[str] = []
+        for cid, arts in (ec.consumed or {}).items():
+            for nm, ra in arts.items():
+                if ra.exists and os.path.isfile(ra.path):
+                    try:
+                        with open(ra.path, encoding="utf-8", errors="replace") as uf:
+                            up_parts.append(f"[{cid}/{nm}]\n{uf.read(4000)}")
+                    except OSError:
+                        pass
+        if up_parts:
+            agent_input = f"{ec.input_text}\n\nUPSTREAM ARTIFACTS:\n" + "\n\n".join(up_parts)
+
+        ctx = Context(workspace_path=ec.stage_ws)
+        result = term.apply(agent_input, ctx)
+        out = str(result)
+        # 产物桥：简单/内容型 agent 只返回文本、不写文件。若它没产出声明的主产物，
+        # 就把文本写进 produces[0]，让 simple agent 也能在流水线里产出 artifact。
+        _write_primary_artifact(ec.stage, ec.stage_ws, out)
+        return StageExecResult(output=out, status="completed",
+                               trace=list(getattr(ctx, "trace", [])))
+    except Exception as e:  # noqa: BLE001
+        return StageExecResult(status="failed", error=str(e))
+
+
+def _write_primary_artifact(stage: "Stage", stage_ws: str, text: str) -> None:
+    """若 stage 声明了产物且 agent 没自己写出来，把 agent 文本写进 produces[0]。
+
+    仅对非 glob 的安全相对路径生效（glob 如 results/*  无法当作写入目标）。
+    """
+    if not stage.produces:
+        return
+    spec = stage.produces[0]
+    if not _is_safe_relpath(spec.path) or any(c in spec.path for c in "*?["):
+        return
+    if _glob.glob(os.path.join(stage_ws, spec.path)):
+        return   # agent 已自己产出（如 react 用 WriteFile）
+    target = os.path.join(stage_ws, spec.path)
+    os.makedirs(os.path.dirname(target) or stage_ws, exist_ok=True)
+    try:
+        with open(target, "w", encoding="utf-8") as f:
+            f.write(text)
+    except OSError:
+        pass
+
+
+# ============================================================
+# LLM-judge gate（语义验收，可用本地 ollama）
+# ============================================================
+
+@dataclass
+class JudgeRequest:
+    rubric: str
+    threshold: float
+    model: dict                  # {provider, name}
+    artifacts_text: str
+    stage_id: str = ""
+
+
+@dataclass
+class JudgeResult:
+    score: float
+    passed: bool
+    reasons: List[str] = field(default_factory=list)
+    raw: str = ""
+    usage: Optional[dict] = None
+
+
+JudgeFn = Callable[[JudgeRequest], JudgeResult]
+
+_JUDGE_SYS = (
+    "You are a strict acceptance reviewer. Score how well the ARTIFACTS satisfy "
+    "the RUBRIC, from 0.0 (fails) to 1.0 (fully satisfies). Respond with ONLY a "
+    'JSON object, no prose: {"score": <float 0..1>, "reasons": ["...", "..."]}.'
+)
+
+
+def _parse_judge_response(text: str) -> Tuple[float, List[str]]:
+    """从（可能不干净的）LLM 输出里稳健解析 score + reasons。"""
+    score: Optional[float] = None
+    reasons: List[str] = []
+    m = re.search(r"\{.*\}", text, re.DOTALL)
+    if m:
+        try:
+            d = json.loads(m.group(0))
+            if isinstance(d, dict):
+                if "score" in d:
+                    score = float(d["score"])
+                r = d.get("reasons")
+                if isinstance(r, list):
+                    reasons = [str(x) for x in r]
+                elif isinstance(r, str):
+                    reasons = [r]
+        except Exception:
+            pass
+    if score is None:   # fallback：抓一个 0..1 的数
+        m2 = re.search(r"(\d\.\d+|[01](?:\.0+)?)", text)
+        if m2:
+            try:
+                score = float(m2.group(1))
+            except ValueError:
+                score = 0.0
+    score = max(0.0, min(1.0, score if score is not None else 0.0))
+    return score, reasons
+
+
+def default_judge(req: JudgeRequest) -> JudgeResult:
+    """默认 LLM-judge：调 provider（默认 ollama 本地）给产物打分。
+
+    §15 P2-D 注记：当前 judge 是 runner 内的独立 provider 调用，其 token 成本
+    暂未进 stage 的 cost.json（须把 judge 建成 term 图里的真 agent 才计入，留 v1.1）。
+    本函数返回 usage，已写进 verdict.reasons 供审计。
+    """
+    from lambdagent.providers import create_provider, ChatMessage
+
+    model = req.model or {}
+    provider = model.get("provider", "ollama")
+    kwargs: dict = {"timeout": 120}
+    if model.get("name"):
+        kwargs["model"] = model["name"]
+    p = create_provider(provider, **kwargs)
+
+    user = f"RUBRIC:\n{req.rubric}\n\nARTIFACTS:\n{req.artifacts_text}"
+    resp = p.chat_typed(
+        [ChatMessage(role="system", content=_JUDGE_SYS),
+         ChatMessage(role="user", content=user)],
+        temperature=0.0, max_tokens=512,
+    )
+    score, reasons = _parse_judge_response(resp.text)
+    usage = {"provider": provider,
+             "input_tokens": getattr(resp, "input_tokens", 0),
+             "output_tokens": getattr(resp, "output_tokens", 0)}
+    return JudgeResult(score=score, passed=score >= req.threshold,
+                       reasons=reasons or [resp.text[:200]], raw=resp.text, usage=usage)
+
+
+# ── ②：judge 作为 term 图里的真 agent（成本可被 estimate_cost 计入，直连 Paper34）──
+
+def _judge_model_str(model_cfg: dict) -> str:
+    """{provider, name} → Lam 的 'provider/name' model 字符串。"""
+    provider = (model_cfg or {}).get("provider", "ollama")
+    name = (model_cfg or {}).get("name") or ("qwen2.5:7b" if provider == "ollama" else "")
+    return f"{provider}/{name}" if name else provider
+
+
+def build_judge_term(judge_cfg: dict):
+    """把 LLM-judge 构造成一个 lambdagent `Lam` term（§14.2/§15.2 P2-D）。
+
+    这样 judge 进入 term 图，`estimate_cost(term)` 能把它的 token/money 计入流水线
+    成本预测——正是 Paper34 cost-soundness 的实证落点。
+    """
+    from lambdagent.primitives import Lam
+    rubric = (judge_cfg or {}).get("rubric", "")
+    prompt = _JUDGE_SYS + "\n\nRUBRIC:\n" + rubric
+    return Lam(name="gate-judge", prompt=prompt,
+               model=_judge_model_str((judge_cfg or {}).get("model", {})),
+               temperature=0.0, max_tokens=512)
+
+
+def term_judge(req: JudgeRequest) -> JudgeResult:
+    """term 版 judge：把 judge 当 Lam term 跑（在 term 图内），并报告其预测成本。
+
+    与 default_judge 等价的语义，但 judge 是真 term —— 用于成本可统计的流水线。
+    """
+    from lambdagent.core import Context
+    term = build_judge_term({"rubric": req.rubric, "model": req.model})
+    text = str(term.apply(req.artifacts_text, Context()))
+    score, reasons = _parse_judge_response(text)
+    usage = None
+    try:
+        from lambdagent.cost_grade import estimate_cost
+        g = estimate_cost(term)
+        usage = {"provider": (req.model or {}).get("provider", "ollama"),
+                 "predicted_tokens": g.tokens, "predicted_money": round(g.money, 6),
+                 "in_term_graph": True}
+    except Exception:
+        pass
+    return JudgeResult(score=score, passed=score >= req.threshold,
+                       reasons=reasons or [text[:200]], raw=text, usage=usage)
+
+
+def _compile_stage_agent(st: Stage, base_dir: str = ""):
+    """编译 stage 的 agent 配置成 term（仅用于成本估计，不运行）。失败/无配置返回 None。"""
+    agent = st.agent or {}
+    ov = {"model": st.model_override} if st.model_override else {}
+    try:
+        from lambdagent.fromconfig import from_config
+        if "ref" in agent:
+            ref = agent["ref"]
+            cfg_path = ref if os.path.isabs(ref) else os.path.join(base_dir, ref)
+            return from_config(cfg_path, **ov)
+        if agent:
+            import tempfile, yaml
+            with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False, encoding="utf-8") as f:
+                yaml.dump(agent, f, allow_unicode=True)
+                tmp = f.name
+            try:
+                return from_config(tmp, **ov)
+            finally:
+                try:
+                    os.unlink(tmp)
+                except OSError:
+                    pass
+    except Exception:
+        return None
+    return None
+
+
+def estimate_pipeline_cost(pipeline: Pipeline, base_dir: str = "") -> Tuple[Any, List[tuple]]:
+    """运行前预测整条流水线的成本上界（§14.2）。
+
+    每阶段成本 = estimate_cost(stage_agent) ·(serial) estimate_cost(judge_term，若 llm_judge)，
+    再按 retry 用 grade_guard 放大；阶段间用 grade_serial 折叠。
+    返回 (total_CostGrade, [(stage_id, CostGrade), ...])。这是 Paper34 "actual ≤ predicted"
+    实证的预测侧。
+    """
+    from lambdagent.cost_grade import estimate_cost, grade_serial, grade_guard, CostGrade
+    total = CostGrade()
+    breakdown: List[tuple] = []
+    for st in pipeline.stages:
+        agent_term = _compile_stage_agent(st, base_dir)
+        sc = estimate_cost(agent_term) if agent_term is not None else CostGrade()
+        if st.gate.kind == "llm_judge":
+            try:
+                sc = grade_serial(sc, estimate_cost(build_judge_term(st.gate.judge or {})))
+            except Exception:
+                pass
+        if st.on_fail == OnFail.RETRY and st.retry > 0:
+            sc = grade_guard(sc, st.retry)
+        breakdown.append((st.id, sc))
+        total = grade_serial(total, sc)
+    return total, breakdown
+
+
+# ============================================================
+# P1-B: write-ahead 状态
+# ============================================================
+
+def _write_state_atomic(pipeline_ws: str, state: dict) -> None:
+    """原子写 pipeline_state.json（tmp + os.replace），write-ahead 用。"""
+    path = os.path.join(pipeline_ws, "pipeline_state.json")
+    tmp = path + ".tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(state, f, ensure_ascii=False, indent=2)
+    os.replace(tmp, path)
+
+
+def _read_state(pipeline_ws: str) -> dict:
+    path = os.path.join(pipeline_ws, "pipeline_state.json")
+    if not os.path.isfile(path):
+        return {}
+    try:
+        with open(path, encoding="utf-8") as f:
+            return json.load(f)
+    except Exception:
+        return {}
+
+
+# ============================================================
+# PipelineRunner
+# ============================================================
+
+class PipelineRunner:
+    def __init__(self, pipeline: Pipeline, pipeline_ws: str, input_text: str,
+                 stage_executor: StageExecutor, base_dir: str = "",
+                 resume_state: Optional[dict] = None,
+                 judge: Optional[JudgeFn] = None):
+        self.p = pipeline
+        self.ws = pipeline_ws
+        self.input_text = input_text
+        self.exec = stage_executor
+        self.judge = judge or default_judge
+        self.base_dir = base_dir
+        # artifact store：stage_id -> resolved 产物（P1-C 多上游来源）
+        self.store: Dict[str, Dict[str, ResolvedArtifact]] = {}
+        self.state: dict = resume_state or {"pipeline_id": pipeline.id, "stages": {}}
+
+    def _stage_dir(self, stage_id: str) -> str:
+        return os.path.join(self.ws, stage_id)
+
+    def _persist(self, status: PipelineStatus) -> None:
+        self.state["status"] = status.value
+        _write_state_atomic(self.ws, self.state)
+
+    def _set_stage_state(self, stage_id: str, **kw) -> None:
+        s = self.state["stages"].setdefault(stage_id, {})
+        s.update(kw)
+
+    def run(self) -> PipelineResult:
+        results: List[StageResult] = []
+        self._persist(PipelineStatus.RUNNING)
+
+        for st in self.p.stages:
+            prior = self.state["stages"].get(st.id, {})
+            if prior.get("status") in (StageStatus.PASSED.value, StageStatus.SKIPPED.value):
+                # resume：跳过已通过/跳过的阶段，但要把其产物补进 store。
+                # §15 P1-2：用持久化的 workspace_path（可能是重试 attempt 的 _aN 目录），
+                # 不能想当然用 base stage_dir，否则下游消费到错目录。
+                sd = prior.get("workspace_path") or self._stage_dir(st.id)
+                if os.path.isdir(sd):
+                    self.store[st.id], _ = resolve_artifacts(st, sd)
+                results.append(StageResult(st.id, StageStatus(prior["status"]),
+                                           attempts=prior.get("attempt_no", 0),
+                                           workspace_path=sd))
+                continue
+
+            sr = self._run_stage(st)
+            results.append(sr)
+
+            if sr.status == StageStatus.PASSED:
+                continue
+            if sr.status == StageStatus.SKIPPED:
+                continue
+            if sr.status == StageStatus.AWAITING_HUMAN:
+                self._persist(PipelineStatus.AWAITING_HUMAN)
+                return PipelineResult(self.p.id, PipelineStatus.AWAITING_HUMAN, self.ws, results)
+            # FAILED → halt
+            self._persist(PipelineStatus.HALTED)
+            return PipelineResult(self.p.id, PipelineStatus.HALTED, self.ws, results)
+
+        self._persist(PipelineStatus.COMPLETED)
+        return PipelineResult(self.p.id, PipelineStatus.COMPLETED, self.ws, results)
+
+    def _consumed_view(self, st: Stage) -> Dict[str, Dict[str, ResolvedArtifact]]:
+        return {cid: self.store.get(cid, {}) for cid in st.consumes}
+
+    def _run_stage(self, st: Stage) -> StageResult:
+        stage_ws = self._stage_dir(st.id)
+        # §15 P1-4：retry 预算只在 on_fail==RETRY 时给。halt/escalate/skip = 立即处置
+        #（与 design §6.1 "on_fail: halt → 直接 halt" 一致）。
+        attempts = (1 + max(0, st.retry)) if st.on_fail == OnFail.RETRY else 1
+        last_err = ""
+        last_ws = stage_ws
+        last_verdict: Optional[Verdict] = None
+
+        for attempt in range(1, attempts + 1):
+            # write-ahead：attempt 开始前
+            self._set_stage_state(st.id, status=StageStatus.RUNNING.value,
+                                   attempt_no=attempt, attempt_id=f"{st.id}#{attempt}",
+                                   started_at=time.strftime("%Y-%m-%dT%H:%M:%S"))
+            self._persist(PipelineStatus.RUNNING)
+
+            # 每个 attempt 一个干净的 stage workspace（PipelineRunner 拥有，§15 P3-F）。
+            # §15 P1-3：复用前先清空，避免旧 attempt 的残留产物虚假满足 has()。
+            attempt_dir = stage_ws if attempt == 1 else f"{stage_ws}_a{attempt}"
+            if os.path.isdir(attempt_dir):
+                import shutil as _sh
+                _sh.rmtree(attempt_dir, ignore_errors=True)
+            ws = create_run_workspace(
+                agent_dir="", run_id=f"{st.id}_{attempt}",
+                input_text=self.input_text, config=(st.agent or {}),
+                run_dir=attempt_dir,
+            )
+            last_ws = ws
+
+            ec = StageExecContext(stage=st, stage_ws=ws, input_text=self.input_text,
+                                  consumed=self._consumed_view(st), base_dir=self.base_dir)
+            res = self.exec(ec)
+            save_run_artifacts(ws, res.output, res.trace, 0,
+                               error=res.error, status=res.status, usage=res.usage)
+            build_change_manifest(ws)
+
+            if res.status != "completed":
+                last_err = res.error
+                continue   # 执行失败 → 重试
+
+            # write-ahead：gating 前
+            self._set_stage_state(st.id, status=StageStatus.GATING.value)
+            self._persist(PipelineStatus.RUNNING)
+
+            # P2-E 运行时契约校验（类型/基数/路径安全在 resolve_artifacts 内）
+            resolved, problems = resolve_artifacts(st, ws)
+            if problems:
+                last_err = f"artifact contract problems: {problems}"
+                last_verdict = Verdict(
+                    stage_id=st.id, passed=False, kind=st.gate.kind, attempt=attempt,
+                    reasons=[last_err], artifacts=[asdict(r) for r in resolved.values()],
+                    timestamp=time.strftime("%Y-%m-%dT%H:%M:%S"))
+                self._write_verdict(ws, st, last_verdict)
+                continue   # 契约不符 → 不判 gate，直接重试
+
+            # gate
+            passed, reasons, score = self._eval_gate(st, resolved)
+            verdict = Verdict(stage_id=st.id, passed=passed, kind=st.gate.kind,
+                              attempt=attempt, reasons=reasons, score=score,
+                              artifacts=[asdict(r) for r in resolved.values()],
+                              timestamp=time.strftime("%Y-%m-%dT%H:%M:%S"))
+            self._write_verdict(ws, st, verdict)
+            last_verdict = verdict
+
+            if passed:
+                self.store[st.id] = resolved
+                # §15 P1-2：持久化通过那次 attempt 的 workspace_path，供 resume 取数
+                self._set_stage_state(st.id, status=StageStatus.PASSED.value,
+                                      workspace_path=ws,
+                                      verdict_path=os.path.join(ws, "verdict.json"),
+                                      completed_at=time.strftime("%Y-%m-%dT%H:%M:%S"))
+                self._persist(PipelineStatus.RUNNING)
+                return StageResult(st.id, StageStatus.PASSED, attempt, ws, verdict)
+            last_err = "; ".join(reasons)
+
+        # 重试耗尽 → 按 on_fail 终态处置
+        return self._apply_on_fail(st, last_ws, attempts, last_err, last_verdict)
+
+    @staticmethod
+    def _collect_artifacts_text(resolved: Dict[str, ResolvedArtifact], max_each: int = 4000) -> str:
+        parts: List[str] = []
+        for name, ra in resolved.items():
+            if ra.exists and os.path.isfile(ra.path):
+                try:
+                    with open(ra.path, encoding="utf-8", errors="replace") as f:
+                        parts.append(f"### {name}\n{f.read(max_each)}")
+                except OSError:
+                    pass
+        return "\n\n".join(parts) if parts else "(no readable artifacts)"
+
+    def _eval_gate(self, st: Stage, resolved: Dict[str, ResolvedArtifact]) -> Tuple[bool, List[str], Optional[float]]:
+        if st.gate.kind == "rule":
+            passed, reasons = eval_rule_gate(st.gate.rule, resolved)
+            return passed, reasons, None
+        if st.gate.kind == "llm_judge":
+            jcfg = st.gate.judge or {}
+            threshold = float(jcfg.get("threshold", 0.7))
+            req = JudgeRequest(
+                rubric=jcfg.get("rubric", ""),
+                threshold=threshold,
+                model=jcfg.get("model", {}) or {},
+                artifacts_text=self._collect_artifacts_text(resolved),
+                stage_id=st.id,
+            )
+            jr = self.judge(req)
+            reasons = [f"llm_judge score={jr.score:.2f} threshold={threshold:.2f}"]
+            reasons += [str(r) for r in jr.reasons]
+            if jr.usage:
+                reasons.append(f"judge_usage={jr.usage}")
+            return (jr.score >= threshold), reasons, jr.score
+        if st.gate.kind == "human":
+            # TODO P3：human gate 走 YIELD + resume(approved/rejected) 语义（§6.3）。
+            raise NotImplementedError("human gate not implemented in P1 skeleton (see §6.3)")
+        raise GateError(f"unknown gate kind: {st.gate.kind}")
+
+    def _apply_on_fail(self, st: Stage, stage_ws: str, attempts: int, err: str,
+                       verdict: Optional[Verdict] = None) -> StageResult:
+        # §15 P1-5：终态转换立即原子持久化（别等 caller），关闭崩溃窗口。
+        ts = time.strftime("%Y-%m-%dT%H:%M:%S")
+        if st.on_fail == OnFail.SKIP:
+            self._set_stage_state(st.id, status=StageStatus.SKIPPED.value,
+                                  workspace_path=stage_ws, error=err, completed_at=ts)
+            _write_state_atomic(self.ws, self.state)
+            return StageResult(st.id, StageStatus.SKIPPED, attempts, stage_ws, verdict, error=err)
+        if st.on_fail == OnFail.ESCALATE:
+            self._set_stage_state(st.id, status=StageStatus.AWAITING_HUMAN.value,
+                                  workspace_path=stage_ws, error=err, completed_at=ts)
+            _write_state_atomic(self.ws, self.state)
+            return StageResult(st.id, StageStatus.AWAITING_HUMAN, attempts, stage_ws, verdict, error=err)
+        # RETRY（已重试完）/ HALT → FAILED 终态
+        self._set_stage_state(st.id, status=StageStatus.FAILED.value,
+                              workspace_path=stage_ws, error=err, completed_at=ts)
+        _write_state_atomic(self.ws, self.state)
+        return StageResult(st.id, StageStatus.FAILED, attempts, stage_ws, verdict, error=err)
+
+    @staticmethod
+    def _write_verdict(ws: str, st: Stage, verdict: Verdict) -> None:
+        try:
+            with open(os.path.join(ws, "verdict.json"), "w", encoding="utf-8") as f:
+                json.dump(asdict(verdict), f, ensure_ascii=False, indent=2)
+        except OSError:
+            pass
+
+
+# ============================================================
+# Public API
+# ============================================================
+
+def _create_pipeline_workspace(agent_dir: str, p: Pipeline, input_text: str) -> str:
+    ts = time.strftime("%Y%m%d_%H%M%S")
+    ws = os.path.join(agent_dir, "workspace", f"pipeline_{ts}")
+    if os.path.exists(ws):
+        ws = f"{ws}_{p.id}"
+    os.makedirs(ws, exist_ok=True)
+    with open(os.path.join(ws, "input.json"), "w", encoding="utf-8") as f:
+        json.dump({"pipeline_id": p.id, "input": input_text, "timestamp": ts},
+                  f, ensure_ascii=False, indent=2)
+    return os.path.abspath(ws)
+
+
+def run_pipeline(pipeline_cfg: Any, input_text: str, agent_dir: str, *,
+                 base_dir: str = "", stage_executor: Optional[StageExecutor] = None,
+                 judge: Optional[JudgeFn] = None) -> PipelineResult:
+    """加载、校验并运行一条流水线。"""
+    p = load_pipeline(pipeline_cfg)
+    errs = validate_pipeline(p)
+    if errs:
+        raise PipelineError("invalid pipeline: " + "; ".join(errs))
+    ws = _create_pipeline_workspace(agent_dir, p, input_text)
+    runner = PipelineRunner(p, ws, input_text, stage_executor or default_stage_executor,
+                            base_dir, judge=judge)
+    return runner.run()
+
+
+def resume_pipeline(workspace_path: str, pipeline_cfg: Any, input_text: str = "", *,
+                    base_dir: str = "", stage_executor: Optional[StageExecutor] = None,
+                    judge: Optional[JudgeFn] = None) -> PipelineResult:
+    """从已有 pipeline workspace 续跑（跳过已通过阶段）。
+
+    P1 版：依赖 write-ahead 的 pipeline_state.json 跳过 PASSED/SKIPPED 阶段。
+    崩溃-孤儿 attempt 的完整确定性恢复留 P3（§15 P1-B）。
+    """
+    if not os.path.isdir(workspace_path):
+        raise PipelineError(f"workspace not found: {workspace_path}")
+    p = load_pipeline(pipeline_cfg)
+    errs = validate_pipeline(p)
+    if errs:
+        raise PipelineError("invalid pipeline: " + "; ".join(errs))
+    if not input_text:
+        ij = os.path.join(workspace_path, "input.json")
+        if os.path.isfile(ij):
+            with open(ij, encoding="utf-8") as f:
+                input_text = json.load(f).get("input", "")
+    state = _read_state(workspace_path)
+    runner = PipelineRunner(p, workspace_path, input_text,
+                            stage_executor or default_stage_executor, base_dir,
+                            resume_state=state, judge=judge)
+    return runner.run()

agentpaas/src/agentpaas/engine/sandbox.py

@@ -12,8 +12,10 @@ Run Workspace:
 """
 from __future__ import annotations
 import asyncio
+import hashlib
 import json
 import os
+import subprocess
 import time
 import traceback
 from dataclasses import dataclass, field
@@ -51,6 +53,7 @@ def create_run_workspace(
     input_text: str,
     config: dict,
     bare: bool = False,
+    run_dir: str = "",
 ) -> str:
     """
     为一次 run 创建持久化工作目录。
@@ -72,6 +75,10 @@ def create_run_workspace(
         input_text: 用户输入
         config: agent YAML 配置字典
         bare: 若 True，不追加 workspace/ 子目录，直接在 agent_dir 下建 run_* 目录
+        run_dir: 若提供，直接用作 workspace 绝对/相对路径（不做时间戳命名）。
+                 供 PipelineRunner 把每个 stage 目录放到 pipeline 目录下用
+                 （绕开时间戳发现，见 PRODUCT_GATE_DESIGN §15 P3-F）。
+                 优先级高于 bare/agent_dir。
 
     Returns:
         workspace_path: 创建的 run 目录绝对路径
@@ -79,13 +86,15 @@ def create_run_workspace(
     import yaml
 
     timestamp = time.strftime("%Y%m%d_%H%M%S")
-    if bare:
+    if run_dir:
+        workspace = run_dir
+    elif bare:
         workspace = os.path.join(agent_dir, f"run_{timestamp}")
     else:
         workspace = os.path.join(agent_dir, "workspace", f"run_{timestamp}")
 
-    # 如果同一秒内有多次 run（罕见），追加 run_id 后缀避免冲突
-    if os.path.exists(workspace):
+    # 如果同一秒内有多次 run（罕见），追加 run_id 后缀避免冲突（run_dir 显式路径不冲突处理）
+    if not run_dir and os.path.exists(workspace):
         workspace = f"{workspace}_{run_id[-6:]}"
 
     os.makedirs(workspace, exist_ok=True)
@@ -125,9 +134,16 @@ def _compute_cost_usd(usage: dict) -> float:
       qwen-plus  Input: ¥0.0008/1K → ~$0.11/M  Output: ¥0.002/1K → ~$0.28/M
       qwen-turbo Input: ¥0.0003/1K → ~$0.04/M  Output: ¥0.0006/1K → ~$0.08/M
       (Uses qwen-max rates as conservative default; CNY/USD ≈ 7.2)
+
+    ── Ollama (local / self-hosted) ─────────────────────────────────────
+      No per-token API cost → always $0.00.
     """
     provider = usage.get("provider", "claude")
 
+    # Local / self-hosted models incur no per-token API cost.
+    if provider in ("ollama",):
+        return 0.0
+
     if provider in ("dashscope", "qwen"):
         # Conservative: use qwen-max rates regardless of actual model tier.
         # $5.56/M input, $16.67/M output (¥0.04 and ¥0.12 per 1K ÷ 7.2)
@@ -145,6 +161,152 @@ def _compute_cost_usd(usage: dict) -> float:
     )
 
 
+# ============================================================
+# Change Manifest（修改清单）
+# ============================================================
+
+# Framework-written files at the run-dir root — NOT agent-produced changes
+_MANIFEST_EXCLUDE = {
+    "input.json", "config.yml", "output.json",
+    "trace.json", "cost.json", "manifest.json",
+}
+_MANIFEST_HASH_MAX_BYTES = 100 * 1024 * 1024   # don't hash files larger than this
+_GIT_DIFF_MAX_CHARS = 200_000                   # cap diff text stored in manifest
+
+
+def _sha256_file(path: str) -> str:
+    h = hashlib.sha256()
+    with open(path, "rb") as fh:
+        for chunk in iter(lambda: fh.read(65536), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def _git_diff_for(path: str) -> dict:
+    """Best-effort git capture for a run.
+
+    If *path* lives inside a git work tree, record modifications to *tracked*
+    files (this complements the workspace file scan, which already covers newly
+    created artifacts).  Honours the original "wire get_diff() into the run
+    artifacts" intent for the worktree-isolation case.  Returns
+    ``{"available": False}`` whenever git or a repo is absent — never raises.
+    """
+    # 改进⑧ (AUDIT_2026-06-11): 先用纯文件系统探测向上找 .git，目录树里
+    # 没有就直接返回，省掉每次 ~50ms 的 git 子进程启动开销（多 stage
+    # pipeline 下每 stage 一次）。找到了再交给 git rev-parse 权威确认
+    # （.git 文件型 worktree、裸仓库等边角仍由 git 判定）。
+    probe = os.path.realpath(path)
+    while True:
+        if os.path.exists(os.path.join(probe, ".git")):
+            break
+        parent = os.path.dirname(probe)
+        if parent == probe:  # 到根了，没有仓库
+            return {"available": False}
+        probe = parent
+
+    try:
+        root = subprocess.run(
+            ["git", "-C", path, "rev-parse", "--show-toplevel"],
+            capture_output=True, text=True, timeout=10,
+        )
+        if root.returncode != 0 or not root.stdout.strip():
+            return {"available": False}
+        repo_root = root.stdout.strip()
+        status = subprocess.run(
+            ["git", "-C", path, "status", "--porcelain"],
+            capture_output=True, text=True, timeout=10,
+        )
+        changed = [ln for ln in status.stdout.splitlines() if ln.strip()]
+        diff = subprocess.run(
+            ["git", "-C", path, "diff", "HEAD"],
+            capture_output=True, text=True, timeout=15,
+        )
+        diff_text = diff.stdout
+        truncated = len(diff_text) > _GIT_DIFF_MAX_CHARS
+        if truncated:
+            diff_text = diff_text[:_GIT_DIFF_MAX_CHARS]
+        return {
+            "available": True,
+            "repo_root": repo_root,
+            "changed_files": changed,
+            "changed_count": len(changed),
+            "diff": diff_text,
+            "diff_truncated": truncated,
+        }
+    except Exception:
+        return {"available": False}
+
+
+def build_change_manifest(workspace_path: str) -> dict:
+    """Scan a run workspace and write ``manifest.json`` — the per-run 修改清单.
+
+    Records every file the agent produced under the run directory (excluding the
+    framework's own metadata files in ``_MANIFEST_EXCLUDE``) with relative path,
+    size, sha256 and mtime, plus a best-effort git diff of tracked-file edits.
+
+    Because each run gets a fresh workspace, everything under it *is* this run's
+    output — making the scan an authoritative evidence trail.  Returns the
+    manifest dict (also written to ``{workspace_path}/manifest.json``); returns
+    ``{}`` for an invalid path.  Never raises on per-file errors.
+    """
+    if not workspace_path or not os.path.isdir(workspace_path):
+        return {}
+
+    run_id = ""
+    input_json = os.path.join(workspace_path, "input.json")
+    if os.path.isfile(input_json):
+        try:
+            with open(input_json, encoding="utf-8") as fh:
+                run_id = json.load(fh).get("run_id", "")
+        except Exception:
+            run_id = ""
+
+    files: list[dict] = []
+    total_bytes = 0
+    for dirpath, _dirs, filenames in os.walk(workspace_path):
+        for name in filenames:
+            abspath = os.path.join(dirpath, name)
+            rel = os.path.relpath(abspath, workspace_path)
+            if rel in _MANIFEST_EXCLUDE:        # framework metadata, not a change
+                continue
+            try:
+                st = os.stat(abspath)
+            except OSError:
+                continue
+            total_bytes += st.st_size
+            if st.st_size <= _MANIFEST_HASH_MAX_BYTES:
+                try:
+                    digest = _sha256_file(abspath)
+                except OSError:
+                    digest = None
+            else:
+                digest = None                    # too large to hash
+            files.append({
+                "path": rel.replace(os.sep, "/"),
+                "size_bytes": st.st_size,
+                "sha256": digest,
+                "modified_at": time.strftime(
+                    "%Y-%m-%dT%H:%M:%S", time.localtime(st.st_mtime)
+                ),
+            })
+
+    files.sort(key=lambda e: e["path"])
+
+    manifest = {
+        "run_id": run_id,
+        "generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
+        "file_count": len(files),
+        "total_bytes": total_bytes,
+        "files": files,
+        "git": _git_diff_for(workspace_path),
+    }
+
+    with open(os.path.join(workspace_path, "manifest.json"), "w", encoding="utf-8") as fh:
+        json.dump(manifest, fh, ensure_ascii=False, indent=2)
+
+    return manifest
+
+
 def save_run_artifacts(
     workspace_path: str,
     result: str,
@@ -203,7 +365,10 @@ def save_run_artifacts(
         cache_read   = usage.get("cache_read_input_tokens", 0)
         cache_create = usage.get("cache_creation_input_tokens", 0)
         cost_usd     = _compute_cost_usd(usage)
-        source       = "claude_jsonl"
+        _prov        = usage.get("provider", "claude")
+        # Accurate source label: Claude reads JSONL transcripts; OpenAI-compatible
+        # providers (ollama/qwen/...) report usage from the API response.
+        source       = "claude_jsonl" if _prov in ("claude", "claude-code") else f"{_prov}_api"
     else:
         # Fallback: sum ctx.trace tokens_used (usually 0 for claude-code sessions)
         real_input   = input_tokens
@@ -230,6 +395,12 @@ def save_run_artifacts(
             "cost_usd": round(cost_usd, 6),
         }, f, ensure_ascii=False, indent=2)
 
+    # 保存修改清单（change manifest）— never let manifest failure break a run
+    try:
+        build_change_manifest(workspace_path)
+    except Exception:
+        pass
+
 
 # ============================================================
 # Sandbox 执行器
@@ -440,6 +611,12 @@ class Sandbox:
                 os.unlink(script_path)
             except OSError:
                 pass
+            # 保存修改清单（subprocess 路径不走 save_run_artifacts）
+            if workspace_path:
+                try:
+                    build_change_manifest(workspace_path)
+                except Exception:
+                    pass
             if not workspace_path:
                 for p in (config_path, input_path):
                     try:

docs/OLLAMA_VERIFICATION.md

@@ -0,0 +1,51 @@
+# Ollama（本地 / 自托管）实测记录
+
+> 日期：2026-06-09 · 环境：macOS，Ollama @ `127.0.0.1:11434`，模型 `qwen2.5:7b` / `qwen2.5-coder:32b`
+
+把 claim ④ 的"本地 Ollama 灵活切换"从**纸面声明**变为**实测通过**。坐实"自托管"这条结构性护城河（大厂 SaaS 无法提供气隙部署）。
+
+## 实测结论
+
+| 层级 | 结果 |
+|---|---|
+| Provider `chat()` | ✅ 正常，usage 累积 |
+| Provider `chat_typed()` | ✅ 正常，返回 token/model/finish_reason |
+| `get_usage()` 会话累计 | ✅ 正常（修复后，见下） |
+| 全栈 simple agent（YAML→compile→Sandbox→workspace） | ✅ 完成，产物三件套齐全 |
+| 全栈 react agent（含 terminate 工具） | ✅ 完成，token 捕获正确（232 tokens） |
+| 本地模型计费 | ✅ **$0**（修复后，见下） |
+
+## 实测中发现并修复的 3 个真问题
+
+1. **本地模型被按 Claude 定价计费**（`agentpaas/engine/sandbox.py::_compute_cost_usd`）
+   - 修复：`provider == "ollama"` → 直接返回 `0.0`。本地模型无 per-token API 成本。
+
+2. **`chat_typed()` 不累积 session usage**（`lambdagent/.../openai_compat_provider.py`）
+   - 只有 `chat()` 累积，`chat_typed()` 返回了 usage 却没加进 `_usage_input/_usage_output`，导致用 typed 接口时 `get_usage()` 永远返回 0。
+   - 修复：`chat_typed()` 内同步累积，与 `chat()` 对齐。
+
+3. **cost.json 的 `source` 标签对非 Claude provider 误导**
+   - 原来只要有 usage 就写死 `"claude_jsonl"`，即使是 ollama/qwen。
+   - 修复：按 provider 反映真实来源——Claude=`claude_jsonl`，其余=`{provider}_api`。
+
+## 已知遗留（非 ollama 专属，跨 provider 既有限制）
+
+- **simple agent token 计数为 0**：simple 类型没有 react 的 `_think_ref`，Sandbox 读不到 provider 已累积的 usage → `source: trace_fallback, tokens: 0`。对 ollama 无害（成本本就 $0），但对计费型 provider 会低估 token。属既有架构限制，待单独处理（让 simple 路径也能回读 provider.get_usage()）。
+
+## 可重复测试
+
+`tests/test_ollama.py`：
+- 成本单测（**不需** ollama 运行）：local=$0、Claude/qwen 定价回归。
+- live 测试（**ollama 可达才跑，否则自动 skip**）：chat 往返、usage 累积、全栈 $0 run。
+- 复现前置：`ollama pull qwen2.5:7b && ollama serve`。
+
+## 配置用法
+
+```yaml
+model:
+  provider: ollama
+  name: qwen2.5:7b      # 或 qwen2.5-coder:32b 等本地模型
+  temperature: 0.0
+  maxTokens: 256
+```
+默认 base_url `http://127.0.0.1:11434/v1`，可用 `base_url` 覆盖远程 ollama。

docs/PRODUCT_GATE_DESIGN.md

@@ -0,0 +1,440 @@
+# 产物 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。

lambdagent/src/lambdagent/primitives.py

@@ -159,8 +159,12 @@ class Lam(Term):
             # Ollama local model via HTTP
             import json, urllib.request
             url = "http://localhost:11434/api/chat"
+            # Strip the "ollama/" routing prefix before sending (mirrors the
+            # dashscope branch); ollama only knows the bare model name, so
+            # sending "ollama/qwen2.5:7b" returns HTTP 404 Not Found.
+            model_name = self.model.replace("ollama/", "")
             body = json.dumps({
-                "model": self.model,
+                "model": model_name,
                 "messages": [
                     {"role": "system", "content": self.prompt},
                     {"role": "user", "content": input_text},

lambdagent/src/lambdagent/providers/openai_compat_provider.py

@@ -157,6 +157,10 @@ class OpenAICompatProvider(LLMProvider):
                 data = json.loads(resp.read())
                 choice = data["choices"][0]
                 usage = data.get("usage", {})
+                # Accumulate session usage so get_usage() is consistent whether
+                # the caller used chat() or chat_typed() (cost tracking parity).
+                self._usage_input  += usage.get("prompt_tokens", 0)
+                self._usage_output += usage.get("completion_tokens", 0)
                 return ChatResponse(
                     text=(choice.get("message", {}).get("content") or "").strip(),
                     input_tokens=usage.get("prompt_tokens", 0),

tests/test_ollama.py

@@ -0,0 +1,114 @@
+"""
+Tests for Ollama (local / self-hosted) provider support.
+
+Two groups:
+  1. Cost accounting — pure unit tests, no Ollama needed. Verify local models
+     are billed at $0 and the cost source label reflects the real provider.
+  2. Live provider — skipped automatically unless an Ollama server is reachable
+     at http://127.0.0.1:11434. Run a local model first, e.g.:
+         ollama pull qwen2.5:7b && ollama serve
+"""
+
+import urllib.request
+
+import pytest
+
+from agentpaas.engine.sandbox import _compute_cost_usd
+
+
+# ============================================================
+# 1. Cost accounting (no Ollama required)
+# ============================================================
+
+class TestOllamaCost:
+
+    def test_ollama_costs_zero(self):
+        """Local Ollama models incur no per-token API cost."""
+        usage = {"input_tokens": 10_000, "output_tokens": 5_000, "provider": "ollama"}
+        assert _compute_cost_usd(usage) == 0.0
+
+    def test_claude_still_billed(self):
+        """Regression: non-local providers keep their pricing."""
+        usage = {"input_tokens": 1_000_000, "output_tokens": 0, "provider": "claude"}
+        assert _compute_cost_usd(usage) == pytest.approx(3.00)
+
+    def test_qwen_still_billed(self):
+        usage = {"input_tokens": 1_000_000, "output_tokens": 0, "provider": "qwen"}
+        assert _compute_cost_usd(usage) == pytest.approx(5.56)
+
+
+# ============================================================
+# 2. Live provider (skipped unless Ollama is up)
+# ============================================================
+
+_OLLAMA_URL = "http://127.0.0.1:11434/api/tags"
+
+
+def _ollama_models():
+    """Return the list of locally available Ollama model names, or [] if down."""
+    try:
+        with urllib.request.urlopen(_OLLAMA_URL, timeout=3) as resp:
+            import json
+            data = json.loads(resp.read())
+        return [m["name"] for m in data.get("models", [])]
+    except Exception:
+        return []
+
+
+_MODELS = _ollama_models()
+_HAS_OLLAMA = bool(_MODELS)
+# Prefer a small model if present, else use whatever is available.
+_MODEL = next((m for m in _MODELS if "7b" in m or "3b" in m), _MODELS[0] if _MODELS else "qwen2.5:7b")
+
+ollama_required = pytest.mark.skipif(
+    not _HAS_OLLAMA, reason="No Ollama server reachable at 127.0.0.1:11434"
+)
+
+
+@ollama_required
+class TestOllamaLive:
+
+    def test_chat_roundtrip(self):
+        from lambdagent.providers import create_provider
+        p = create_provider("ollama", model=_MODEL, timeout=120)
+        assert p.provider_name == "ollama"
+        assert p.base_url.startswith("http://127.0.0.1:11434")
+        out = p.chat([{"role": "user", "content": "Reply with exactly: OLLAMA_OK"}])
+        assert "OLLAMA_OK" in out
+
+    def test_usage_accumulates(self):
+        from lambdagent.providers import create_provider, ChatMessage
+        p = create_provider("ollama", model=_MODEL, timeout=120)
+        r = p.chat_typed([ChatMessage(role="user", content="Say hi.")], max_tokens=16)
+        assert r.text
+        assert r.input_tokens > 0
+        usage = p.get_usage()
+        assert usage["provider"] == "ollama"
+        assert usage["input_tokens"] > 0
+
+    def test_full_stack_run_costs_zero(self):
+        """A simple agent run on Ollama produces artifacts and bills $0."""
+        import asyncio, tempfile, os, json, shutil
+        from agentpaas.engine.sandbox import Sandbox
+
+        config = {
+            "name": "ollama-smoke", "type": "simple",
+            "systemPrompt": "You are concise.",
+            "model": {"provider": "ollama", "name": _MODEL,
+                      "temperature": 0.0, "maxTokens": 64},
+        }
+        agent_dir = tempfile.mkdtemp(prefix="ollama_test_")
+        try:
+            sb = Sandbox(level=0)
+            r = asyncio.run(sb.execute(
+                config, "Reply with exactly: FULLSTACK_OK",
+                timeout=120, agent_dir=agent_dir, run_id="run_olltest",
+            ))
+            assert r.status == "completed", r.error
+            ws = r.workspace_path
+            for fn in ("output.json", "cost.json", "manifest.json"):
+                assert os.path.isfile(os.path.join(ws, fn))
+            cost = json.load(open(os.path.join(ws, "cost.json")))
+            assert cost["cost_usd"] == 0.0
+        finally:
+            shutil.rmtree(agent_dir, ignore_errors=True)

tests/test_pipeline.py

@@ -0,0 +1,552 @@
+"""
+Tests for engine.pipeline — 产物 Gate 逐级验收流水线 (P0+P1 骨架).
+
+覆盖：
+  1. 静态校验（forward-ref consumes、duplicate id）
+  2. 受限 gate DSL 求值器（功能 + 安全：拒绝属性/import/裸名）
+  3. resolve_artifacts 运行时契约校验（required 缺失）
+  4. run_pipeline 端到端（fake executor）：通过 / 多上游 consumes / gate 失败 retry→halt / skip
+  5. 每阶段 verdict + manifest + write-ahead state
+  6. resume_pipeline 跳过已通过阶段
+"""
+import json
+import os
+import shutil
+import tempfile
+
+import pytest
+
+from agentpaas.engine.pipeline import (
+    load_pipeline, validate_pipeline, eval_rule_gate, resolve_artifacts,
+    run_pipeline, resume_pipeline, _parse_judge_response,
+    estimate_pipeline_cost, build_judge_term, term_judge, JudgeRequest,
+    Stage, ArtifactSpec, ResolvedArtifact, GateError, PipelineError,
+    StageStatus, PipelineStatus, StageExecResult, JudgeResult,
+)
+
+_REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+_DEMO_PRESET = os.path.join(_REPO, "agentexample", "pipelines", "research-demo.yml")
+
+
+def _ollama_up() -> bool:
+    import urllib.request
+    try:
+        urllib.request.urlopen("http://127.0.0.1:11434/api/tags", timeout=2)
+        return True
+    except Exception:
+        return False
+
+
+@pytest.fixture
+def agent_dir():
+    d = tempfile.mkdtemp(prefix="test_pipeline_")
+    yield d
+    shutil.rmtree(d, ignore_errors=True)
+
+
+# ============================================================
+# helpers — fake executors that write artifacts (no LLM)
+# ============================================================
+
+def _writer(files: dict):
+    """Return a stage executor that writes {relpath: content} into the stage ws."""
+    def _exec(ec):
+        for rel, content in files.items():
+            p = os.path.join(ec.stage_ws, rel)
+            os.makedirs(os.path.dirname(p), exist_ok=True)
+            with open(p, "w", encoding="utf-8") as f:
+                f.write(content)
+        return StageExecResult(output="ok", status="completed")
+    return _exec
+
+
+# ============================================================
+# 1. Static validation
+# ============================================================
+
+class TestStaticValidation:
+
+    def test_valid_pipeline(self):
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "produces": [{"name": "x.md", "path": "final/x.md"}],
+             "gate": {"kind": "rule", "rule": 'has("x.md")'}},
+            {"id": "b", "consumes": ["a"], "gate": {"kind": "rule", "rule": "True"}},
+        ]}}
+        p = load_pipeline(cfg)
+        assert validate_pipeline(p) == []
+
+    def test_forward_consumes_rejected(self):
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "consumes": ["b"]},   # b not yet declared
+            {"id": "b"},
+        ]}}
+        errs = validate_pipeline(load_pipeline(cfg))
+        assert any("forward" in e for e in errs)
+
+    def test_duplicate_stage_id_rejected(self):
+        cfg = {"pipeline": {"id": "p", "stages": [{"id": "a"}, {"id": "a"}]}}
+        errs = validate_pipeline(load_pipeline(cfg))
+        assert any("duplicate" in e for e in errs)
+
+    def test_run_rejects_invalid(self, agent_dir):
+        cfg = {"pipeline": {"id": "p", "stages": [{"id": "a", "consumes": ["z"]}]}}
+        with pytest.raises(PipelineError):
+            run_pipeline(cfg, "in", agent_dir, stage_executor=_writer({}))
+
+    def test_unsafe_rule_caught_statically(self, agent_dir):
+        """§15 P2-8: a banned gate expression fails validation BEFORE the executor runs."""
+        ran = {"n": 0}
+
+        def exec_spy(ec):
+            ran["n"] += 1
+            return StageExecResult(output="ok", status="completed")
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "gate": {"kind": "rule", "rule": 'open("/etc/passwd")'}},
+        ]}}
+        errs = validate_pipeline(load_pipeline(cfg))
+        assert any("not allowed" in e for e in errs)
+        with pytest.raises(PipelineError):
+            run_pipeline(cfg, "in", agent_dir, stage_executor=exec_spy)
+        assert ran["n"] == 0   # executor never ran
+
+
+# ============================================================
+# 2. Bounded gate DSL evaluator
+# ============================================================
+
+class TestGateEvaluator:
+
+    def _resolved(self, tmp, name="plan.md", content="contains hypothesis here"):
+        p = os.path.join(tmp, name)
+        with open(p, "w") as f:
+            f.write(content)
+        return {name: ResolvedArtifact(name=name, type="md", path=p, exists=True,
+                                       size_bytes=len(content))}
+
+    def test_has_and_contains_pass(self, agent_dir):
+        r = self._resolved(agent_dir)
+        passed, reasons = eval_rule_gate('has("plan.md") and contains("plan.md", "hypothesis")', r)
+        assert passed is True
+
+    def test_contains_fail(self, agent_dir):
+        r = self._resolved(agent_dir, content="no keyword")
+        passed, _ = eval_rule_gate('contains("plan.md", "hypothesis")', r)
+        assert passed is False
+
+    def test_json_len(self, agent_dir):
+        p = os.path.join(agent_dir, "refs.json")
+        with open(p, "w") as f:
+            json.dump(list(range(12)), f)
+        r = {"refs.json": ResolvedArtifact("refs.json", "json", p, True, os.path.getsize(p))}
+        assert eval_rule_gate('json_len("refs.json") >= 10', r)[0] is True
+        assert eval_rule_gate('json_len("refs.json") >= 100', r)[0] is False
+
+    def test_count(self, agent_dir):
+        r = self._resolved(agent_dir)
+        assert eval_rule_gate("count() >= 1", r)[0] is True
+
+    # --- safety: dangerous expressions must raise GateError, not execute ---
+
+    @pytest.mark.parametrize("expr", [
+        '__import__("os").system("echo pwned")',
+        'open("/etc/passwd")',
+        'has.__class__',
+        'count().__class__.__bases__',
+        '[].append(1)',
+        'lambda: 1',
+    ])
+    def test_unsafe_expr_rejected(self, expr):
+        with pytest.raises(GateError):
+            eval_rule_gate(expr, {})
+
+
+# ============================================================
+# 3. Runtime artifact contract validation
+# ============================================================
+
+class TestResolveArtifacts:
+
+    def test_missing_required(self, agent_dir):
+        st = Stage(id="a", produces=[ArtifactSpec(name="x.md", path="final/x.md", required=True)])
+        resolved, problems = resolve_artifacts(st, agent_dir)
+        assert len(problems) == 1 and problems[0].startswith("x.md:")
+        assert resolved["x.md"].exists is False
+
+    def test_resolved_with_hash(self, agent_dir):
+        os.makedirs(os.path.join(agent_dir, "final"))
+        with open(os.path.join(agent_dir, "final", "x.md"), "w") as f:
+            f.write("data")
+        st = Stage(id="a", produces=[ArtifactSpec(name="x.md", path="final/x.md")])
+        resolved, missing = resolve_artifacts(st, agent_dir)
+        assert missing == []
+        assert resolved["x.md"].exists is True
+        assert resolved["x.md"].sha256 is not None
+
+    def test_path_breakout_rejected(self, agent_dir):
+        """§15 P1-1: absolute / .. paths must not escape the stage workspace."""
+        for bad in ("/etc/passwd", "../../../../etc/passwd", "final/../../secret"):
+            st = Stage(id="a", produces=[ArtifactSpec(name="x", path=bad, required=True)])
+            resolved, problems = resolve_artifacts(st, agent_dir)
+            assert problems, f"unsafe path not rejected: {bad}"
+            assert resolved["x"].exists is False
+
+    def test_invalid_json_type_fails(self, agent_dir):
+        """§15 P2-E: type=json that doesn't parse must NOT count as a valid artifact."""
+        os.makedirs(os.path.join(agent_dir, "results"))
+        with open(os.path.join(agent_dir, "results", "refs.json"), "w") as f:
+            f.write("not valid json {{{")
+        st = Stage(id="a", produces=[ArtifactSpec(name="refs.json", type="json",
+                                                  path="results/refs.json", required=True)])
+        resolved, problems = resolve_artifacts(st, agent_dir)
+        assert problems            # invalid json → contract problem
+        assert resolved["refs.json"].exists is False
+
+
+# ============================================================
+# 4-5. End-to-end run
+# ============================================================
+
+class TestRunPipeline:
+
+    def _two_stage_cfg(self):
+        return {"pipeline": {"id": "research", "defaults": {"retry": 1}, "stages": [
+            {"id": "plan", "produces": [{"name": "plan.md", "path": "final/plan.md"}],
+             "gate": {"kind": "rule", "rule": 'has("plan.md") and contains("plan.md", "hypothesis")'}},
+            {"id": "lit", "consumes": ["plan"],
+             "produces": [{"name": "refs.json", "path": "results/refs.json", "type": "json"}],
+             "gate": {"kind": "rule", "rule": 'json_len("refs.json") >= 3'}},
+        ]}}
+
+    def test_completes_and_persists(self, agent_dir):
+        execs = {
+            "plan": _writer({"final/plan.md": "our hypothesis is X"}),
+            "lit": _writer({"results/refs.json": json.dumps([1, 2, 3, 4])}),
+        }
+        result = run_pipeline(self._two_stage_cfg(), "topic", agent_dir,
+                              stage_executor=lambda ec: execs[ec.stage.id](ec))
+        assert result.status == PipelineStatus.COMPLETED
+        assert [s.status for s in result.stages] == [StageStatus.PASSED, StageStatus.PASSED]
+        # verdict + manifest + state persisted
+        plan_ws = os.path.join(result.workspace_path, "plan")
+        assert os.path.isfile(os.path.join(plan_ws, "verdict.json"))
+        assert os.path.isfile(os.path.join(plan_ws, "manifest.json"))
+        assert os.path.isfile(os.path.join(result.workspace_path, "pipeline_state.json"))
+
+    def test_multi_upstream_consumes(self, agent_dir):
+        """stage 3 consumes two non-adjacent upstream stages via the artifact store."""
+        seen = {}
+
+        def s3_exec(ec):
+            # read upstream artifacts through the store-backed consumed view
+            seen["consumed_keys"] = sorted(ec.consumed.keys())
+            seen["plan_path"] = ec.consumed["plan"]["plan.md"].path
+            with open(os.path.join(ec.stage_ws, "final", "paper.md"), "w") as f:
+                f.write("done")
+            return StageExecResult(output="ok", status="completed")
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "plan", "produces": [{"name": "plan.md", "path": "final/plan.md"}],
+             "gate": {"kind": "rule", "rule": 'has("plan.md")'}},
+            {"id": "lit", "consumes": ["plan"], "produces": [{"name": "s.md", "path": "final/s.md"}],
+             "gate": {"kind": "rule", "rule": 'has("s.md")'}},
+            {"id": "paper", "consumes": ["plan", "lit"],
+             "produces": [{"name": "paper.md", "path": "final/paper.md"}],
+             "gate": {"kind": "rule", "rule": 'has("paper.md")'}},
+        ]}}
+        execs = {
+            "plan": _writer({"final/plan.md": "h"}),
+            "lit": _writer({"final/s.md": "survey"}),
+            "paper": s3_exec,
+        }
+        result = run_pipeline(cfg, "x", agent_dir, stage_executor=lambda ec: execs[ec.stage.id](ec))
+        assert result.status == PipelineStatus.COMPLETED
+        assert seen["consumed_keys"] == ["lit", "plan"]   # both non-adjacent upstreams visible
+        assert seen["plan_path"].endswith("plan.md")
+
+    def test_retry_then_halts(self, agent_dir):
+        """on_fail=retry + retry=2 → 1+2 attempts, then FAILED/HALTED."""
+        calls = {"n": 0}
+
+        def failing(ec):
+            calls["n"] += 1
+            return StageExecResult(output="", status="completed")  # never writes artifact
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "retry": 2, "on_fail": "retry",
+             "produces": [{"name": "x.md", "path": "final/x.md", "required": True}],
+             "gate": {"kind": "rule", "rule": 'has("x.md")'}},
+        ]}}
+        result = run_pipeline(cfg, "x", agent_dir, stage_executor=failing)
+        assert result.status == PipelineStatus.HALTED
+        assert result.stages[0].status == StageStatus.FAILED
+        assert calls["n"] == 3   # 1 + retry(2)
+
+    def test_halt_does_not_retry(self, agent_dir):
+        """§15 P1-4: on_fail=halt ignores retry budget — single attempt, immediate halt."""
+        calls = {"n": 0}
+
+        def failing(ec):
+            calls["n"] += 1
+            return StageExecResult(output="", status="completed")
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "retry": 2, "on_fail": "halt",   # retry=2 must be ignored
+             "produces": [{"name": "x.md", "path": "final/x.md", "required": True}],
+             "gate": {"kind": "rule", "rule": 'has("x.md")'}},
+        ]}}
+        result = run_pipeline(cfg, "x", agent_dir, stage_executor=failing)
+        assert result.status == PipelineStatus.HALTED
+        assert calls["n"] == 1   # halt = no retry
+
+    def test_skip_on_fail_continues(self, agent_dir):
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "on_fail": "skip",
+             "produces": [{"name": "x.md", "path": "final/x.md", "required": True}],
+             "gate": {"kind": "rule", "rule": 'has("x.md")'}},
+            {"id": "b", "produces": [{"name": "y.md", "path": "final/y.md"}],
+             "gate": {"kind": "rule", "rule": 'has("y.md")'}},
+        ]}}
+        execs = {"a": _writer({}), "b": _writer({"final/y.md": "ok"})}
+        result = run_pipeline(cfg, "x", agent_dir, stage_executor=lambda ec: execs[ec.stage.id](ec))
+        assert result.status == PipelineStatus.COMPLETED
+        assert result.stages[0].status == StageStatus.SKIPPED
+        assert result.stages[1].status == StageStatus.PASSED
+
+
+# ============================================================
+# 6. Resume
+# ============================================================
+
+class TestResume:
+
+    def test_resume_skips_passed(self, agent_dir):
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "produces": [{"name": "x.md", "path": "final/x.md"}],
+             "gate": {"kind": "rule", "rule": 'has("x.md")'}},
+            {"id": "b", "produces": [{"name": "y.md", "path": "final/y.md"}],
+             "gate": {"kind": "rule", "rule": 'has("y.md")'}},
+        ]}}
+
+        # First run: stage a passes, stage b halts (writes nothing)
+        run_counts = {"a": 0, "b": 0}
+
+        def exec1(ec):
+            run_counts[ec.stage.id] += 1
+            if ec.stage.id == "a":
+                with open(os.path.join(ec.stage_ws, "final", "x.md"), "w") as f:
+                    f.write("ok")
+            return StageExecResult(output="ok", status="completed")
+
+        cfg["pipeline"]["stages"][1]["on_fail"] = "halt"
+        r1 = run_pipeline(cfg, "x", agent_dir, stage_executor=exec1)
+        assert r1.status == PipelineStatus.HALTED
+        assert run_counts == {"a": 1, "b": 1}
+
+        # Resume: stage a should be skipped (already passed); only b re-runs
+        def exec2(ec):
+            run_counts[ec.stage.id] += 1
+            with open(os.path.join(ec.stage_ws, "final", "y.md"), "w") as f:
+                f.write("ok")
+            return StageExecResult(output="ok", status="completed")
+
+        r2 = resume_pipeline(r1.workspace_path, cfg, stage_executor=exec2)
+        assert r2.status == PipelineStatus.COMPLETED
+        assert run_counts["a"] == 1   # NOT re-run
+        assert run_counts["b"] == 2   # re-run once on resume
+
+    def test_resume_after_retry_pass_uses_correct_dir(self, agent_dir):
+        """§15 P1-2: stage passes on attempt 2 (_a2 dir); downstream after resume must
+        consume that attempt's artifact, not the empty base stage dir."""
+        # stage a fails attempt 1, passes attempt 2; stage b halts first run, then passes on resume.
+        a_calls = {"n": 0}
+
+        def a_exec(ec):
+            a_calls["n"] += 1
+            if a_calls["n"] >= 2:   # succeed only on the 2nd attempt
+                with open(os.path.join(ec.stage_ws, "final", "x.md"), "w") as f:
+                    f.write("content-from-attempt-2")
+            return StageExecResult(output="ok", status="completed")
+
+        consumed_path = {}
+
+        def b_exec(ec):
+            # capture the path a's artifact resolves to through the store
+            ra = ec.consumed.get("a", {}).get("x.md")
+            consumed_path["p"] = ra.path if ra else None
+            return StageExecResult(output="", status="completed")  # fail → halt first time
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "retry": 2, "on_fail": "retry",
+             "produces": [{"name": "x.md", "path": "final/x.md", "required": True}],
+             "gate": {"kind": "rule", "rule": 'has("x.md")'}},
+            {"id": "b", "consumes": ["a"], "on_fail": "halt",
+             "produces": [{"name": "y.md", "path": "final/y.md", "required": True}],
+             "gate": {"kind": "rule", "rule": 'has("y.md")'}},
+        ]}}
+
+        r1 = run_pipeline(cfg, "x", agent_dir,
+                          stage_executor=lambda ec: (a_exec if ec.stage.id == "a" else b_exec)(ec))
+        assert r1.status == PipelineStatus.HALTED
+        # a passed on attempt 2 → its artifact lives in the _a2 dir
+        assert consumed_path["p"] and consumed_path["p"].endswith("x.md")
+        assert "_a2" in consumed_path["p"]
+
+        # resume: a skipped (uses persisted _a2 workspace_path); b passes
+        def b_exec2(ec):
+            ra = ec.consumed.get("a", {}).get("x.md")
+            consumed_path["resume"] = ra.path if ra else None
+            with open(os.path.join(ec.stage_ws, "final", "y.md"), "w") as f:
+                f.write("ok")
+            return StageExecResult(output="ok", status="completed")
+
+        r2 = resume_pipeline(r1.workspace_path, cfg,
+                             stage_executor=lambda ec: b_exec2(ec))
+        assert r2.status == PipelineStatus.COMPLETED
+        # after resume, a's artifact still resolves from the correct (_a2) dir
+        assert consumed_path["resume"] and "_a2" in consumed_path["resume"]
+
+
+# ============================================================
+# 7. LLM-judge gate
+# ============================================================
+
+class TestLLMJudgeGate:
+
+    @pytest.mark.parametrize("text,expected", [
+        ('{"score": 0.9, "reasons": ["good"]}', 0.9),
+        ('Here is my verdict: {"score":0.4,"reasons":["meh"]} thanks', 0.4),
+        ('score: 0.75 overall', 0.75),
+        ('1.0', 1.0),
+        ('total garbage, no number at all', 0.0),
+        ('{"score": 5}', 1.0),     # clamped to [0,1]
+    ])
+    def test_parse_judge_response(self, text, expected):
+        score, _ = _parse_judge_response(text)
+        assert abs(score - expected) < 1e-6
+
+    def test_judge_pass_injected(self, agent_dir):
+        captured = {}
+
+        def fake_judge(req):
+            captured["rubric"] = req.rubric
+            captured["text"] = req.artifacts_text
+            return JudgeResult(score=0.85, passed=True, reasons=["covers everything"],
+                               usage={"provider": "fake", "input_tokens": 10, "output_tokens": 5})
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "produces": [{"name": "essay.md", "path": "final/essay.md"}],
+             "gate": {"kind": "llm_judge", "judge": {"rubric": "states a hypothesis", "threshold": 0.7}}},
+        ]}}
+        r = run_pipeline(cfg, "x", agent_dir,
+                         stage_executor=_writer({"final/essay.md": "our hypothesis is X"}),
+                         judge=fake_judge)
+        assert r.status == PipelineStatus.COMPLETED
+        v = r.stages[0].verdict
+        assert v.passed is True and v.score == 0.85
+        assert any("score=0.85" in s for s in v.reasons)
+        assert any("judge_usage" in s for s in v.reasons)
+        assert "hypothesis" in captured["text"]   # judge actually saw the artifact content
+
+    def test_judge_fail_halts(self, agent_dir):
+        def low_judge(req):
+            return JudgeResult(score=0.30, passed=False, reasons=["too thin"])
+
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "on_fail": "halt",
+             "produces": [{"name": "essay.md", "path": "final/essay.md"}],
+             "gate": {"kind": "llm_judge", "judge": {"rubric": "deep analysis", "threshold": 0.7}}},
+        ]}}
+        r = run_pipeline(cfg, "x", agent_dir,
+                         stage_executor=_writer({"final/essay.md": "thin"}),
+                         judge=low_judge)
+        assert r.status == PipelineStatus.HALTED
+        assert r.stages[0].status == StageStatus.FAILED
+        assert r.stages[0].verdict.score == 0.30
+
+    @pytest.mark.skipif(not _ollama_up(), reason="ollama not reachable on :11434")
+    def test_judge_live_ollama(self, agent_dir):
+        """Real local-ollama judge — backs the self-hosted '逐级验收' selling point."""
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a", "on_fail": "halt",
+             "produces": [{"name": "essay.md", "path": "final/essay.md"}],
+             "gate": {"kind": "llm_judge", "judge": {
+                 "model": {"provider": "ollama", "name": "qwen2.5:7b"},
+                 "rubric": "The text must clearly state a research hypothesis.",
+                 "threshold": 0.5}}},
+        ]}}
+        good = ("Our central hypothesis is that cost-graded effect types can predict "
+                "an agent's resource consumption before execution. We test it on 835 configs.")
+        r = run_pipeline(cfg, "topic", agent_dir,
+                         stage_executor=_writer({"final/essay.md": good}))
+        v = r.stages[0].verdict
+        assert v is not None and v.score is not None
+        assert 0.0 <= v.score <= 1.0
+        assert r.status in (PipelineStatus.COMPLETED, PipelineStatus.HALTED)
+
+
+# ============================================================
+# 8. Cost estimation (②: judge as term, Paper34 prediction side)
+# ============================================================
+
+class TestCostEstimate:
+
+    def test_estimate_pipeline_cost_includes_judge(self):
+        """estimate_pipeline_cost folds agent + judge-term cost; llm_judge stage > 0."""
+        cfg = {"pipeline": {"id": "p", "stages": [
+            {"id": "a",
+             "agent": {"type": "simple",
+                       "model": {"provider": "ollama", "name": "qwen2.5:7b"},
+                       "systemPrompt": "x"},
+             "produces": [{"name": "a.md", "path": "final/a.md"}],
+             "gate": {"kind": "llm_judge",
+                      "judge": {"model": {"provider": "ollama", "name": "qwen2.5:7b"},
+                                "rubric": "r", "threshold": 0.5}}},
+        ]}}
+        total, breakdown = estimate_pipeline_cost(load_pipeline(cfg))
+        assert len(breakdown) == 1
+        assert breakdown[0][0] == "a"
+        assert total.tokens > 0            # judge term contributes a token upper bound
+        assert 0 < total.probability <= 1
+
+    def test_build_judge_term_is_a_term(self):
+        from lambdagent.cost_grade import estimate_cost
+        term = build_judge_term({"rubric": "r", "model": {"provider": "ollama", "name": "qwen2.5:7b"}})
+        g = estimate_cost(term)            # judge is in the term graph → cost computable
+        assert g.tokens > 0
+
+
+# ============================================================
+# 9. Live ollama: term judge + research demo end-to-end
+# ============================================================
+
+class TestLiveOllama:
+
+    @pytest.mark.skipif(not _ollama_up(), reason="ollama not reachable on :11434")
+    def test_term_judge_live(self, agent_dir):
+        req = JudgeRequest(
+            rubric="The text must clearly state a research hypothesis.",
+            threshold=0.5, model={"provider": "ollama", "name": "qwen2.5:7b"},
+            artifacts_text="### x\nOur hypothesis is that graded cost predicts agent resource use.")
+        jr = term_judge(req)
+        assert 0.0 <= jr.score <= 1.0
+        assert jr.usage and jr.usage.get("in_term_graph") is True
+
+    @pytest.mark.skipif(not _ollama_up(), reason="ollama not reachable on :11434")
+    def test_research_demo_end_to_end(self, agent_dir):
+        """① Real plan→lit run on local ollama via the research-demo preset.
+
+        Asserts ORCHESTRATION invariants (terminal status, stage order, state +
+        verdict persisted), not the small model's exact output — a live local-LLM
+        e2e must not flake on model nondeterminism or transient ollama hiccups.
+        """
+        assert os.path.isfile(_DEMO_PRESET)
+        r = run_pipeline(_DEMO_PRESET, "形式化资源语义在 AI agent 调度中的应用", agent_dir)
+        # orchestration reached a terminal state and ran in declared order
+        assert r.status in (PipelineStatus.COMPLETED, PipelineStatus.HALTED)
+        assert [s.stage_id for s in r.stages][:1] == ["plan"]
+        assert os.path.isfile(os.path.join(r.workspace_path, "pipeline_state.json"))
+        # when the plan stage passes, the artifact bridge must have produced plan.md
+        plan = r.stages[0]
+        if plan.status == StageStatus.PASSED:
+            assert os.path.isfile(os.path.join(plan.workspace_path, "final", "plan.md"))

tests/test_run_workspace.py

@@ -19,6 +19,7 @@ import pytest
 from agentpaas.engine.sandbox import (
     create_run_workspace,
     save_run_artifacts,
+    build_change_manifest,
     ExecutionResult,
 )
 from lambdagent.core import Context, TraceEntry
@@ -193,3 +194,80 @@ class TestNoWorkspaceWithoutAgentDir:
     def test_execution_result_default_empty(self):
         r = ExecutionResult()
         assert r.workspace_path == ""
+
+
+# ============================================================
+# 6. Change Manifest (修改清单)
+# ============================================================
+
+class TestChangeManifest:
+
+    def test_save_run_artifacts_writes_manifest(self, agent_dir):
+        """save_run_artifacts auto-generates manifest.json"""
+        ws = create_run_workspace(agent_dir, "run_m1", "input", {})
+        save_run_artifacts(ws, "result", [], 100)
+        assert os.path.isfile(os.path.join(ws, "manifest.json"))
+
+    def test_manifest_lists_agent_produced_files(self, agent_dir):
+        """Files the agent wrote under the run dir appear in the manifest"""
+        ws = create_run_workspace(agent_dir, "run_m2", "input", {})
+        with open(os.path.join(ws, "code", "main.py"), "w") as f:
+            f.write("print('hi')\n")
+        with open(os.path.join(ws, "results", "data.csv"), "w") as f:
+            f.write("a,b\n1,2\n")
+
+        manifest = build_change_manifest(ws)
+        paths = {e["path"] for e in manifest["files"]}
+        assert "code/main.py" in paths
+        assert "results/data.csv" in paths
+        assert manifest["file_count"] == 2
+        assert manifest["total_bytes"] > 0
+
+    def test_manifest_excludes_framework_metadata(self, agent_dir):
+        """input/config/output/trace/cost/manifest themselves are not counted"""
+        ws = create_run_workspace(agent_dir, "run_m3", "input", {})
+        save_run_artifacts(ws, "result", [], 100)  # writes output/trace/cost + manifest
+        with open(os.path.join(ws, "manifest.json")) as f:
+            manifest = json.load(f)
+        paths = {e["path"] for e in manifest["files"]}
+        assert paths.isdisjoint({
+            "input.json", "config.yml", "output.json",
+            "trace.json", "cost.json", "manifest.json",
+        })
+
+    def test_manifest_records_sha256(self, agent_dir):
+        """Each produced file carries a sha256 digest for the evidence chain"""
+        ws = create_run_workspace(agent_dir, "run_m4", "input", {})
+        with open(os.path.join(ws, "final", "report.txt"), "w") as f:
+            f.write("hello world")
+        manifest = build_change_manifest(ws)
+        entry = next(e for e in manifest["files"] if e["path"] == "final/report.txt")
+        # sha256("hello world")
+        assert entry["sha256"] == (
+            "b94d27b9934d3e08a52e52d7da7dabfac484efe37a5380ee9088f7ace2efcde9"
+        )
+
+    def test_manifest_carries_run_id(self, agent_dir):
+        """run_id from input.json is propagated into the manifest"""
+        ws = create_run_workspace(agent_dir, "run_m5", "input", {})
+        manifest = build_change_manifest(ws)
+        assert manifest["run_id"] == "run_m5"
+
+    def test_manifest_git_field_present(self, agent_dir):
+        """git section is always present (available False outside a repo)"""
+        ws = create_run_workspace(agent_dir, "run_m6", "input", {})
+        manifest = build_change_manifest(ws)
+        assert "git" in manifest
+        assert "available" in manifest["git"]
+
+    def test_manifest_noop_on_invalid_path(self):
+        """Invalid path returns {} and does not raise"""
+        assert build_change_manifest("") == {}
+        assert build_change_manifest("/nonexistent/xyz") == {}
+
+    def test_empty_run_yields_empty_manifest(self, agent_dir):
+        """A run that produced nothing still yields a valid (empty) manifest"""
+        ws = create_run_workspace(agent_dir, "run_m7", "input", {})
+        manifest = build_change_manifest(ws)  # only framework metadata present
+        assert manifest["file_count"] == 0
+        assert manifest["files"] == []