"""
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=...)）

P3 已完成：
  - human gate YIELD + resume 语义（§6.3）：_run_stage 在 gate 前拦截
    kind=human → 写 pending verdict、stage 标 awaiting_human、干净返回；
    审批走 resolve_human_gate(approved/rejected)，approved 续跑、
    rejected 按 on_fail 处置（retry 重做再审 / skip / halt）。

未实现（留后续）：
  - 崩溃-孤儿 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，直接重试

            # human gate（§6.3 YIELD 语义）：产物已合格（契约校验通过），
            # 在此处停下等人 —— 写 pending verdict、stage 标 awaiting_human
            # 并干净返回；审批走 resume_pipeline(approved/rejected)。
            if st.gate.kind == "human":
                pending = Verdict(
                    stage_id=st.id, passed=False, kind="human", attempt=attempt,
                    reasons=[f"PENDING_HUMAN: {st.gate.prompt or '等待人工审批'}"],
                    artifacts=[asdict(r) for r in resolved.values()],
                    timestamp=time.strftime("%Y-%m-%dT%H:%M:%S"))
                self._write_verdict(ws, st, pending)
                self._set_stage_state(
                    st.id, status=StageStatus.AWAITING_HUMAN.value,
                    workspace_path=ws, attempt_no=attempt,
                    gate_prompt=st.gate.prompt or "",
                    verdict_path=os.path.join(ws, "verdict.json"))
                _write_state_atomic(self.ws, self.state)
                return StageResult(st.id, StageStatus.AWAITING_HUMAN, attempt, ws, pending)

            # 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":
            # human gate 不在这里评估 —— _run_stage 在 gate 之前拦截并
            # YIELD（awaiting_human）；审批经 resume_pipeline 写 verdict。
            raise GateError("human gate must be handled by _run_stage YIELD path")
        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 resolve_human_gate(pipeline_cfg: Any, pipeline_ws: str, decision: str, *,
                       note: str = "", base_dir: str = "",
                       stage_executor: Optional[StageExecutor] = None,
                       judge: Optional[JudgeFn] = None) -> PipelineResult:
    """对 awaiting_human 的阶段做出审批决定并从断点续跑（§6.3，P3 完成版）。

    Args:
        pipeline_cfg: 与 run_pipeline 相同的流水线配置（yml 路径或 dict）。
        pipeline_ws:  run_pipeline 返回的 workspace_path（含 pipeline_state.json）。
        decision:     "approved" | "rejected"。
        note:         审批意见，写进 human verdict 的 reasons。

    语义（设计 §6.3）：
    - approved → 该阶段 PASSED（写 human verdict），从下一阶段续跑；
      已 PASSED/SKIPPED 的阶段不重算（产物从各自 workspace 读回）。
    - rejected → 写拒绝 verdict，按该阶段 on_fail 处置：
        retry    → 清该阶段状态重跑（重做后会再次停在 human gate 等审）；
        skip     → 标 SKIPPED 后继续后续阶段；
        halt / escalate → 流水线 HALTED（escalate 的"升级给人"已经发生过）。

    续跑部分复用 resume_pipeline（崩溃续跑同一条路径）。
    """
    if decision not in ("approved", "rejected"):
        raise PipelineError(f"decision must be approved|rejected, got {decision!r}")

    p = load_pipeline(pipeline_cfg)
    errs = validate_pipeline(p)
    if errs:
        raise PipelineError("invalid pipeline: " + "; ".join(errs))

    state = _read_state(pipeline_ws)
    if not state:
        raise PipelineError(f"no pipeline_state.json under {pipeline_ws}")

    awaiting = [(sid, s) for sid, s in (state.get("stages") or {}).items()
                if s.get("status") == StageStatus.AWAITING_HUMAN.value]
    if not awaiting:
        raise PipelineError("no stage is awaiting human approval")
    stage_id, sstate = awaiting[0]
    stage = next((s for s in p.stages if s.id == stage_id), None)
    if stage is None:
        raise PipelineError(f"awaiting stage {stage_id!r} not in pipeline config")

    ts = time.strftime("%Y-%m-%dT%H:%M:%S")
    stage_ws = sstate.get("workspace_path") or os.path.join(pipeline_ws, stage_id)
    resolved, _ = resolve_artifacts(stage, stage_ws)

    # 写 human verdict（覆盖 pending 占位）
    verdict = Verdict(
        stage_id=stage_id, passed=(decision == "approved"), kind="human",
        attempt=int(sstate.get("attempt_no", 1)),
        reasons=[f"human {decision}" + (f": {note}" if note else "")],
        artifacts=[asdict(r) for r in resolved.values()], timestamp=ts)
    try:
        with open(os.path.join(stage_ws, "verdict.json"), "w", encoding="utf-8") as f:
            json.dump(asdict(verdict), f, ensure_ascii=False, indent=2)
    except OSError:
        pass

    if decision == "approved":
        sstate.update(status=StageStatus.PASSED.value, completed_at=ts,
                      verdict_path=os.path.join(stage_ws, "verdict.json"))
    else:
        if stage.on_fail in (OnFail.HALT, OnFail.ESCALATE):
            sstate.update(status=StageStatus.FAILED.value,
                          error=f"human rejected: {note}", completed_at=ts)
            state["status"] = PipelineStatus.HALTED.value
            _write_state_atomic(pipeline_ws, state)
            return PipelineResult(p.id, PipelineStatus.HALTED, pipeline_ws, [
                StageResult(stage_id, StageStatus.FAILED,
                            int(sstate.get("attempt_no", 1)), stage_ws, verdict,
                            error=f"human rejected: {note}")])
        if stage.on_fail == OnFail.SKIP:
            sstate.update(status=StageStatus.SKIPPED.value, completed_at=ts,
                          error=f"human rejected: {note}")
        else:  # RETRY：清状态 → 重入 runner 重跑该阶段（重做后再次等审）
            state["stages"].pop(stage_id, None)
    _write_state_atomic(pipeline_ws, state)

    return resume_pipeline(pipeline_ws, pipeline_cfg, base_dir=base_dir,
                           stage_executor=stage_executor, judge=judge)


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()
