feat: P3 双残留完成 — pipeline human-gate resume + simple agent token 透传

1. human-gate YIELD + resume（设计 §6.3 完成版）:
   - _run_stage 在 gate 求值前拦截 kind=human: 产物契约校验通过后
     写 PENDING_HUMAN verdict、stage 标 awaiting_human(含 gate_prompt/
     workspace_path)、原子落盘后干净返回 — 不再 NotImplementedError
   - 新增 resolve_human_gate(cfg, ws, approved|rejected, note=):
     approved → 该阶段 PASSED(human verdict 覆盖占位)从断点续跑,
     已通过阶段不重算(产物从各自 workspace 读回 store);
     rejected → 按 on_fail: retry=清状态重做后再次等审 / skip=跳过
     继续 / halt|escalate=HALTED。续跑复用 resume_pipeline(崩溃续跑
     同一路径)
   - 价值: 「导师把关」可插进自动化科研流水线中间, 打回只重跑一个
     阶段, 长流水线不再因人工环节整条重来
   - 测试 +6: 停在门口/批准续跑/打回三种 on_fail/重做再审/异常输入,
     executor 计数证明断点前后零重算

2. simple agent token 透传(修「单轮智能体 token/成本恒 0」):
   - agents.py _find_usage_source: 原只认 react 的 _think_ref —
     现沿包装链查找(term 自身 ConversationLam → Guard/Memory 的
     .agent 递归), 单轮路径 usage 真实入库
   - AnthropicProvider 补 usage 累积 + get_usage/reset_session
     (与 openai_compat 同形, 全 provider 口径一致)
   - 测试 +2; live 验证: ollama simple run 报 97/17 tokens(原恒 0)

全量回归: tests/ 290 + lambdagent 566 passed

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

agentpaas/src/agentpaas/api/v1/agents.py

@@ -1929,15 +1929,27 @@ def _execute_agent(config: dict, input_text: str, on_step=None,
         result = term.apply(input_text, ctx)
         duration_ms = int((time.time() - t0_exec) * 1000)
 
-        # ── Accurate token counts from ClaudeCode JSONL transcript ──
-        # _compile_react attaches _think_ref (ConversationLam) to the Loop
-        # term; calling get_usage() reads ~/.claude/projects/*/{session_id}.jsonl
-        # and sums all assistant-turn usage for the actual token numbers.
-        think_ref = getattr(term, "_think_ref", None)
+        # ── Accurate token counts ──
+        # react: _compile_react 把 _think_ref(ConversationLam) 挂在 Loop 上。
+        # simple (P3 token 透传修复): term 本身就是 ConversationLam，或被
+        # Guard/Memory 包装一层 —— 此前只认 _think_ref，单轮智能体的
+        # usage 恒为 None，运行记录 token/成本一律记 0。现在沿包装链
+        # 找到第一个带 get_usage 的节点。
+        def _find_usage_source(t, depth=0):
+            if t is None or depth > 5:
+                return None
+            ref = getattr(t, "_think_ref", None)
+            if ref is not None and callable(getattr(ref, "get_usage", None)):
+                return ref
+            if callable(getattr(t, "get_usage", None)):
+                return t
+            return _find_usage_source(getattr(t, "agent", None), depth + 1)
+
         usage = None
-        if think_ref is not None and callable(getattr(think_ref, "get_usage", None)):
+        usage_src = _find_usage_source(term)
+        if usage_src is not None:
             try:
-                usage = think_ref.get_usage()
+                usage = usage_src.get_usage()
             except Exception as _gu_err:
                 logger.warning("get_usage failed: %s", _gu_err)
 

agentpaas/src/agentpaas/engine/pipeline.py

@@ -13,9 +13,13 @@ codex 评审（§15）的修复已内建：
   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）
+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
@@ -884,6 +888,24 @@ class PipelineRunner:
                 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,
@@ -940,8 +962,9 @@ class PipelineRunner:
                 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)")
+            # 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,
@@ -1003,6 +1026,89 @@ def run_pipeline(pipeline_cfg: Any, input_text: str, agent_dir: str, *,
     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:

lambdagent/src/lambdagent/providers/anthropic_provider.py

@@ -21,6 +21,10 @@ class AnthropicProvider(LLMProvider):
         super().__init__(config)
         self._client = None
         self._api_key = api_key  # L02: Allow explicit api_key param
+        # P3 token 透传: 与 OpenAICompatProvider 对齐的会话级 usage 累积，
+        # 让 get_usage() 在 simple/react 两类路径下都有真实数字
+        self._usage_input = 0
+        self._usage_output = 0
 
     def _get_client(self):
         if self._client is None:
@@ -58,6 +62,10 @@ class AnthropicProvider(LLMProvider):
                 system=system_prompt,
                 messages=chat_messages,
             )
+            usage = getattr(response, "usage", None)
+            if usage is not None:
+                self._usage_input += getattr(usage, "input_tokens", 0) or 0
+                self._usage_output += getattr(usage, "output_tokens", 0) or 0
             return response.content[0].text.strip()
         except Exception as e:
             raise ProviderError(f"Anthropic API error: {e}", "anthropic", retryable="overloaded" in str(e).lower())
@@ -88,6 +96,8 @@ class AnthropicProvider(LLMProvider):
                 temperature=temperature,
                 max_tokens=max_tokens,
             )
+            self._usage_input += response.usage.input_tokens or 0
+            self._usage_output += response.usage.output_tokens or 0
             return ChatResponse(
                 text=response.content[0].text,
                 input_tokens=response.usage.input_tokens,
@@ -107,6 +117,19 @@ class AnthropicProvider(LLMProvider):
             None, lambda: self.chat_typed(messages, model, temperature, max_tokens)
         )
 
+    def get_usage(self) -> dict:
+        """会话级累计 usage（与 OpenAICompatProvider 同形）。"""
+        return {
+            "input_tokens": self._usage_input,
+            "output_tokens": self._usage_output,
+            "cache_creation_input_tokens": 0,
+            "cache_read_input_tokens": 0,
+        }
+
+    def reset_session(self) -> None:
+        self._usage_input = 0
+        self._usage_output = 0
+
     @property
     def provider_name(self) -> str:
         return "anthropic"

tests/test_pipeline.py

@@ -550,3 +550,120 @@ class TestLiveOllama:
         plan = r.stages[0]
         if plan.status == StageStatus.PASSED:
             assert os.path.isfile(os.path.join(plan.workspace_path, "final", "plan.md"))
+
+
+# ============================================================
+# Human gate: YIELD + resolve_human_gate（§6.3，P3 完成版）
+# ============================================================
+
+from agentpaas.engine.pipeline import resolve_human_gate
+
+
+class TestHumanGate:
+
+    def _cfg(self, on_fail="halt"):
+        """s1 规则门 → s2 人工门 → s3 规则门（验证断点前后都正确）。"""
+        return {"pipeline": {"id": "hg", "stages": [
+            {"id": "s1",
+             "produces": [{"name": "plan", "type": "md", "path": "plan.md"}],
+             "gate": {"kind": "rule", "rule": 'has("plan")'}},
+            {"id": "s2", "consumes": ["s1"],
+             "produces": [{"name": "draft", "type": "md", "path": "draft.md"}],
+             "gate": {"kind": "human", "prompt": "请导师确认草稿方向"},
+             "on_fail": on_fail, "retry": 1},
+            {"id": "s3", "consumes": ["s2"],
+             "produces": [{"name": "final", "type": "md", "path": "final.md"}],
+             "gate": {"kind": "rule", "rule": 'has("final")'}},
+        ]}}
+
+    def _execs(self, counter):
+        def make(files):
+            def _exec(ec):
+                counter[ec.stage.id] = counter.get(ec.stage.id, 0) + 1
+                for rel, content in files.items():
+                    p = os.path.join(ec.stage_ws, rel)
+                    os.makedirs(os.path.dirname(p) or ec.stage_ws, exist_ok=True)
+                    with open(p, "w", encoding="utf-8") as f:
+                        f.write(content)
+                return StageExecResult(output="ok", status="completed")
+            return _exec
+        table = {"s1": make({"plan.md": "the plan"}),
+                 "s2": make({"draft.md": "the draft"}),
+                 "s3": make({"final.md": "the final"})}
+        return lambda ec: table[ec.stage.id](ec)
+
+    def test_yields_at_human_gate(self, agent_dir):
+        n = {}
+        r = run_pipeline(self._cfg(), "topic", agent_dir, stage_executor=self._execs(n))
+        assert r.status == PipelineStatus.AWAITING_HUMAN
+        assert n == {"s1": 1, "s2": 1}  # s3 没跑
+        # state 落盘正确，verdict 是 pending 占位
+        state = json.load(open(os.path.join(r.workspace_path, "pipeline_state.json")))
+        s2 = state["stages"]["s2"]
+        assert s2["status"] == "awaiting_human"
+        assert s2["gate_prompt"] == "请导师确认草稿方向"
+        v = json.load(open(os.path.join(s2["workspace_path"], "verdict.json")))
+        assert v["passed"] is False and "PENDING_HUMAN" in v["reasons"][0]
+
+    def test_approved_resumes_to_completion(self, agent_dir):
+        n = {}
+        ex = self._execs(n)
+        r = run_pipeline(self._cfg(), "topic", agent_dir, stage_executor=ex)
+        r2 = resolve_human_gate(self._cfg(), r.workspace_path, "approved",
+                                note="方向没问题", stage_executor=ex)
+        assert r2.status == PipelineStatus.COMPLETED
+        # s1/s2 没重算，只补跑了 s3
+        assert n == {"s1": 1, "s2": 1, "s3": 1}
+        # human verdict 覆盖了 pending 占位
+        state = json.load(open(os.path.join(r.workspace_path, "pipeline_state.json")))
+        s2 = state["stages"]["s2"]
+        assert s2["status"] == "passed"
+        v = json.load(open(s2["verdict_path"]))
+        assert v["passed"] is True and "方向没问题" in v["reasons"][0]
+
+    def test_rejected_halt(self, agent_dir):
+        n = {}
+        ex = self._execs(n)
+        r = run_pipeline(self._cfg(on_fail="halt"), "topic", agent_dir, stage_executor=ex)
+        r2 = resolve_human_gate(self._cfg(on_fail="halt"), r.workspace_path,
+                                "rejected", note="重写", stage_executor=ex)
+        assert r2.status == PipelineStatus.HALTED
+        assert n == {"s1": 1, "s2": 1}  # 不重跑
+        state = json.load(open(os.path.join(r.workspace_path, "pipeline_state.json")))
+        assert state["stages"]["s2"]["status"] == "failed"
+
+    def test_rejected_retry_redoes_stage_then_awaits_again(self, agent_dir):
+        n = {}
+        ex = self._execs(n)
+        r = run_pipeline(self._cfg(on_fail="retry"), "topic", agent_dir, stage_executor=ex)
+        r2 = resolve_human_gate(self._cfg(on_fail="retry"), r.workspace_path,
+                                "rejected", note="再来一版", stage_executor=ex)
+        # 重做后再次停在人工门
+        assert r2.status == PipelineStatus.AWAITING_HUMAN
+        assert n == {"s1": 1, "s2": 2}  # s2 重跑一次, s1 不重算
+        # 再批准 → 跑完
+        r3 = resolve_human_gate(self._cfg(on_fail="retry"), r.workspace_path,
+                                "approved", stage_executor=ex)
+        assert r3.status == PipelineStatus.COMPLETED
+        assert n == {"s1": 1, "s2": 2, "s3": 1}
+
+    def test_rejected_skip_continues(self, agent_dir):
+        n = {}
+        ex = self._execs(n)
+        r = run_pipeline(self._cfg(on_fail="skip"), "topic", agent_dir, stage_executor=ex)
+        r2 = resolve_human_gate(self._cfg(on_fail="skip"), r.workspace_path,
+                                "rejected", stage_executor=ex)
+        assert r2.status == PipelineStatus.COMPLETED
+        state = json.load(open(os.path.join(r.workspace_path, "pipeline_state.json")))
+        assert state["stages"]["s2"]["status"] == "skipped"
+        assert n == {"s1": 1, "s2": 1, "s3": 1}
+
+    def test_bad_decision_and_no_awaiting(self, agent_dir):
+        n = {}
+        ex = self._execs(n)
+        r = run_pipeline(self._cfg(), "topic", agent_dir, stage_executor=ex)
+        with pytest.raises(PipelineError):
+            resolve_human_gate(self._cfg(), r.workspace_path, "maybe")
+        resolve_human_gate(self._cfg(), r.workspace_path, "approved", stage_executor=ex)
+        with pytest.raises(PipelineError):  # 已无 awaiting 阶段
+            resolve_human_gate(self._cfg(), r.workspace_path, "approved", stage_executor=ex)

tests/test_workspace_provision.py

@@ -183,3 +183,60 @@ def test_switch_model_unknown_agent_404(api_client):
     r = client.put("/api/v1/agents/ag_nonexistent/model", headers=_auth(key),
                    json={"provider": "ollama", "name": "x"})
     assert r.status_code == 404
+
+
+# ── simple agent token 透传（P3）────────────────────────────────────────────
+
+def test_simple_agent_usage_source_found():
+    """_find_usage_source 沿包装链找到 get_usage：裸 ConversationLam、
+    Guard 包装、Memory(Guard(...)) 双层包装都要命中。"""
+    from lambdagent.conversation import ConversationLam
+    from lambdagent.extensions import Guard, Memory
+
+    class FakeProvider:
+        default_model = "qwen-max"
+        model_name = "qwen-max"
+        context_window = 100000
+        def chat(self, messages, **kw):
+            return "ok"
+        def get_usage(self):
+            return {"input_tokens": 123, "output_tokens": 45,
+                    "cache_creation_input_tokens": 0,
+                    "cache_read_input_tokens": 0}
+
+    lam = ConversationLam(name="t", provider=FakeProvider(), system_prompt="s")
+
+    # 复用 _execute_agent 内的同款查找逻辑（提取为本测试的等价实现以
+    # 锁定语义：_think_ref 优先 → 自身 get_usage → .agent 递归）
+    def find(t, depth=0):
+        if t is None or depth > 5:
+            return None
+        ref = getattr(t, "_think_ref", None)
+        if ref is not None and callable(getattr(ref, "get_usage", None)):
+            return ref
+        if callable(getattr(t, "get_usage", None)):
+            return t
+        return find(getattr(t, "agent", None), depth + 1)
+
+    assert find(lam) is lam
+    guarded = Guard(lam, validator=lambda x: True)
+    assert find(guarded) is lam
+    wrapped = Memory(Guard(lam, validator=lambda x: True))
+    assert find(wrapped) is lam
+    assert find(wrapped).get_usage()["input_tokens"] == 123
+
+
+def test_anthropic_provider_accumulates_usage():
+    """AnthropicProvider 现在有 get_usage/reset_session（与 openai_compat 同形）。"""
+    from lambdagent.providers.anthropic_provider import AnthropicProvider
+    from lambdagent.providers.base import ProviderConfig
+
+    p = AnthropicProvider(ProviderConfig(model="claude-sonnet-4-6"))
+    u = p.get_usage()
+    assert u == {"input_tokens": 0, "output_tokens": 0,
+                 "cache_creation_input_tokens": 0,
+                 "cache_read_input_tokens": 0}
+    p._usage_input += 10; p._usage_output += 5
+    assert p.get_usage()["input_tokens"] == 10
+    p.reset_session()
+    assert p.get_usage()["output_tokens"] == 0