Add a per-call spend threshold (claude --max-budget-usd) and report call cost

doc/architecture/SPEC/kb_service.md

@@ -301,3 +301,13 @@ diff 时保留原摘录（证据不会因此变得更"好过"）。只扩展带
 
 ## 2026-09-30 模型回复解析
 `models.parse_json_object` 取裸 JSON 对象或最后一个 ```json 围栏块，用 `json.loads(strict=False)`：字符串里的原始控制字符（多行值里的换行、制表符，例如图示）按数据接受，不再把结构正确的回答判为 `ModelUnavailable`。测试：`test_kb_intake_gate.py`。
+
+## 2026-09-30 单次调用花费阈值（kb init 预算硬上限的一块积木）
+`ModelGateway.call_json(..., max_budget_usd=x)`：x 是**停止阈值**，不是硬上限 —— 花费达到 x 后不再发起新请求，但越过 x 的
+那个请求照样计费。硬上限由调用方负责：调用前预留 x + 单个请求的最坏情况。transport 的 `stops_at_spend` 为 False 时在发出调用
+**之前**抛 `ModelUnavailable("<provider> cannot stop a call at a spend threshold")`（fail-closed，绝不用 `max_tokens` 近似）；
+x 必须为正。因阈值停下的调用（`stop_reason="max_budget"`）→ `ModelUnavailable`，并作为失败记录（不会成为训练样本）。
+`ModelReply.cost_usd` 取自 `usage["cost_usd"]`（未知为 None），trace 记录同时写入 `cost_usd` 与 `max_budget_usd`。
+不传上限的现有调用方行为不变：该参数根本不会发给 transport。
+`runtime.trace_recorder` 把阈值与花费持久化到 `model_call` 记录的 `result.max_budget_usd` / `result.cost_usd`
+（未设或未知为 null），包括 transport 抛错的失败记录。

doc/architecture/SPEC/providers/base.md

@@ -1,6 +1,6 @@
 # providers/base.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~161 · provider 层契约 + 子进程环境白名单 · refactor-status: ok`
 
@@ -53,3 +53,10 @@ stdlib + `..scopes.ToolScope`。它是一个叶子契约模块。
 
 ## 2026-09-28
 `complete()` 接受 `effort`（推理强度）：codex 以 `-c model_reasoning_effort="<effort>"` 生效并校验取值；其他 transport 接受并忽略（模型 id 已决定推理预算）。
+
+## 2026-09-30 单次调用花费阈值（不是硬上限）
+`complete()` 接受 `max_budget_usd`，类属性 `stops_at_spend`（默认 False）声明 CLI 是否会在本次调用花费达到阈值后**不再发起**
+新的 API 请求。越过阈值的那一个请求仍会完成并计费，所以超出量最多一个请求 —— 这是阈值，不是硬上限；
+需要硬上限的调用方要预留"阈值 + 单个请求的最坏情况"。不支持的 transport 接受并忽略该参数，调用方必须先检查
+`stops_at_spend`（知识服务的 gateway 会拒绝）。不得出现声称能硬性限制单次调用花费的属性。
+`max_tokens` 对 CLI transport 不是花费上限。transport 报告的本次花费放在 `Reply.usage["cost_usd"]`（未知时不出现）。

doc/architecture/SPEC/providers/claude_code.md

@@ -1,6 +1,6 @@
 # providers/claude_code.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~205 · harness transport（Claude 订阅） · refactor-status: ok`
 
@@ -43,3 +43,10 @@ stdlib + `.base` + `..agent_loop.AgentOutcome` + `..llm` 的类型。
 
 ## 2026-09-28
 `complete()` 接受并忽略 `effort`（与 base 契约一致）。
+
+## 2026-09-30 单次调用花费阈值
+`stops_at_spend = True`：`complete(max_budget_usd=x)` 把 `--max-budget-usd x` 传给 CLI（仅 print 模式有效，`_run` 正是 print 模式），
+x 必须为正。CLI 在每个 API 请求完成**之后**检查预算：它阻止下一个请求，不能截断当前请求，所以花费可超出 x 一个请求。CLI 因预算停下（`subtype` 含 budget）时返回 `stop_reason="max_budget"` 且无文本 —— 被截断的回答永远不可用。
+`total_cost_usd` 以 `usage["cost_usd"]` 返回。
+阈值参数由 `budget_arg()` 精确生成：浮点数的最短往返十进制、普通（非科学）记数法；非正、非有限值 → `ValueError`。
+固定精度格式会移动边界（`.4f` 把 0.00001 变成 "0.0000" 即没有阈值，把 0.12345 变成 "0.1235" 即更高的阈值），所以不允许。

doc/architecture/SPEC/providers/codex.md

@@ -1,6 +1,6 @@
 # providers/codex.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~197 · harness transport（ChatGPT 订阅） · refactor-status: ok`
 
@@ -42,3 +42,6 @@ OS 级控制可用的后端；把两者混同，会模糊掉"当前实际生效
 
 ## 2026-09-28
 `complete()` 接受 `effort`（推理强度）：codex 以 `-c model_reasoning_effort="<effort>"` 生效并校验取值；其他 transport 接受并忽略（模型 id 已决定推理预算）。
+
+## 2026-09-30
+`complete()` 接受并忽略 `max_budget_usd`；`stops_at_spend` 为 False（CLI 无花费阈值）。

doc/architecture/SPEC/providers/cursor.md

@@ -1,6 +1,6 @@
 # providers/cursor.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~207 · harness transport（Cursor 订阅） · refactor-status: ok`
 
@@ -48,3 +48,6 @@ scope 约束，native 调用只被**记录**。
 
 ## 2026-09-28
 `complete()` 接受并忽略 `effort`（与 base 契约一致）。
+
+## 2026-09-30
+`complete()` 接受并忽略 `max_budget_usd`；`stops_at_spend` 为 False（CLI 无花费阈值）。

doc/architecture/SPEC/providers/deepseek.md

@@ -1,6 +1,6 @@
 # providers/deepseek.py —— 规范
 
-<!-- verified-against: 2026-09-28 -->
+<!-- verified-against: 2026-09-30 -->
 
 `LOC ~502 · harness transport（dsh，API-keyed） · refactor-status: oversized`
 
@@ -59,3 +59,6 @@ stdlib + `.base` + `.registry` + `..agent_loop` + `..llm` 的类型 + dsh SDK（
 
 ## 2026-09-28
 `complete()` 接受并忽略 `effort`（与 base 契约一致）。
+
+## 2026-09-30
+`complete()` 接受并忽略 `max_budget_usd`；`stops_at_spend` 为 False。

doc/architecture/SPEC/providers/zcode.md

@@ -49,3 +49,5 @@ stdlib + `.base` + `..agent_loop.AgentOutcome` + `..llm` 的类型与 `ModelMism
 ## 测试
 `test_provider_zcode.py`（离线，事件形状取自 zcode 0.16.9 实跑）；2026-09-30 用真实 zcode 做过
 一次端到端冒烟（工具桥 `read_file` 被调用，`bridge_trace.jsonl` 有记录）。
+
+`complete()` 接受并忽略 `max_budget_usd`；`stops_at_spend` 为 False（CLI 无花费阈值，`ModelGateway` 请求阈值时在派发前拒绝）。

src/infermatrix_copilot/kb_service/models.py

@@ -2,8 +2,8 @@
 
 The generator drafts knowledge changes; the judge (a different model family)
 grades them. Both are pinned by (provider, model, reasoning effort) and
-recorded on every call. An unavailable backend, a timeout or an unparseable
-reply raises ``ModelUnavailable``; the caller leaves the work queued. It never
+recorded on every call. An unavailable backend, a timeout, a call stopped
+at its spend cap or an unparseable reply raises ``ModelUnavailable``; the caller leaves the work queued. It never
 switches to a weaker model: a verdict must always say which model made it.
 """
 
@@ -61,6 +61,7 @@ class ModelReply:
     served_model: str
     usage: dict
     seconds: float
+    cost_usd: float | None = None  # the transport's reported spend (None: unknown)
 
 
 def parse_json_object(text: str) -> dict:
@@ -104,36 +105,57 @@ def _transport(self, provider: str):
         return transport
 
     def call_json(self, role: ModelRole, *, system: str, prompt: str,
-                  validate: Callable[[dict], None] | None = None) -> ModelReply:
+                  validate: Callable[[dict], None] | None = None,
+                  max_budget_usd: float | None = None) -> ModelReply:
+        """``max_budget_usd`` is a per-call STOP THRESHOLD, not a hard cap:
+        the transport starts no further API request once the call's spend
+        reaches it, but the request that crosses it is billed in full. A
+        caller enforcing a hard ceiling must reserve the threshold plus one
+        request's worst case before calling. Requesting a threshold from a
+        transport that cannot stop at one (``stops_at_spend`` False) is
+        refused before dispatch; it is never approximated with ``max_tokens``."""
         transport = self._transport(role.provider)
+        cap: dict = {}
+        if max_budget_usd is not None:
+            if not max_budget_usd > 0:
+                raise ModelUnavailable(f"{role.label()}: max_budget_usd must be positive")
+            if not getattr(transport, "stops_at_spend", False):
+                raise ModelUnavailable(f"{role.provider} cannot stop a call at a spend threshold")
+            cap = {"max_budget_usd": max_budget_usd}
         started = time.time()
         identity = {"role": role.name, "requested": role.label(), "provider": role.provider,
                     "model": role.model, "effort": role.effort}
         try:
             reply = transport.complete(
                 system=system, messages=[{"role": "user", "content": prompt}],
-                model=role.model, effort=role.effort, role=role.name)
+                model=role.model, effort=role.effort, role=role.name, **cap)
         except Exception as exc:  # the transport's own failure modes
             if self._recorder is not None:
                 self._recorder({**identity, "served_model": "", "stop_reason": "", "usage": {},
+                                "cost_usd": None, "max_budget_usd": max_budget_usd,
                                 "seconds": round(time.time() - started, 3), "system": system,
                                 "prompt": prompt, "reply": "", "error": str(exc)[:2000]})
             raise ModelUnavailable(f"{role.label()} failed: {exc}") from exc
         seconds = time.time() - started
         text = "".join(getattr(block, "text", "") or "" for block in getattr(reply, "blocks", []))
+        usage = dict(getattr(reply, "usage", {}) or {})
+        cost = usage.get("cost_usd")
+        cost_usd = float(cost) if isinstance(cost, (int, float)) and not isinstance(cost, bool) else None
         record = {
             **identity,
             "served_model": getattr(reply, "model", "") or "",
             "stop_reason": getattr(reply, "stop_reason", ""),
-            "usage": dict(getattr(reply, "usage", {}) or {}),
+            "usage": usage, "cost_usd": cost_usd, "max_budget_usd": max_budget_usd,
             "seconds": round(seconds, 3), "system": system, "prompt": prompt, "reply": text,
         }
         # the record carries the call's FINAL verdict: a truncated, empty,
         # unparseable or schema-failing reply is recorded as a failure, so it
         # never becomes a training example
         failure: ModelUnavailable | None = None
         data: dict = {}
-        if getattr(reply, "stop_reason", "") == "max_tokens" or not text.strip():
+        if getattr(reply, "stop_reason", "") == "max_budget":
+            failure = ModelUnavailable(f"{role.label()} stopped at its spend threshold ${max_budget_usd}")
+        elif getattr(reply, "stop_reason", "") == "max_tokens" or not text.strip():
             failure = ModelUnavailable(f"{role.label()} timed out or returned nothing")
         else:
             try:
@@ -149,4 +171,4 @@ def call_json(self, role: ModelRole, *, system: str, prompt: str,
             self._recorder({**record, "error": str(failure) if failure else ""})
         if failure is not None:
             raise failure
-        return ModelReply(role, data, text, record["served_model"], record["usage"], seconds)
+        return ModelReply(role, data, text, record["served_model"], usage, seconds, cost_usd)

src/infermatrix_copilot/kb_service/runtime.py

@@ -128,7 +128,12 @@ def record(entry: dict) -> None:
                 outputs={"reply": entry["reply"]},
                 model={k: entry.get(k, "") for k in ("role", "provider", "model", "effort", "served_model")},
                 usage=entry.get("usage") or {}, seconds=entry.get("seconds"),
-                result={"stop_reason": entry.get("stop_reason", "")}, error=entry.get("error", ""))
+                # spend: the requested stop threshold and the reported cost
+                # (None when unset/unknown), on failures too
+                result={"stop_reason": entry.get("stop_reason", ""),
+                        "max_budget_usd": entry.get("max_budget_usd"),
+                        "cost_usd": entry.get("cost_usd")},
+                error=entry.get("error", ""))
         except Exception:  # noqa: BLE001 - tracing never breaks a model call
             pass
 

src/infermatrix_copilot/providers/base.py

@@ -140,18 +140,35 @@ def auth_gap(self) -> str | None:
         return None
 
     # -- contract ------------------------------------------------------------
+    # True only where the CLI itself stops a `complete(max_budget_usd=...)`
+    # call from STARTING another API request once its spend reaches the
+    # threshold. The request that crosses it still completes and is billed,
+    # so this bounds the overshoot to one request; it is not a hard ceiling.
+    # `max_tokens` is no spend bound at all for CLI transports.
+    stops_at_spend: bool = False
+
     def run_session(self, req: AgentSessionRequest):
         """Run one delegated agent step; returns `agent_loop.AgentOutcome`."""
         raise NotImplementedError
 
     def complete(self, *, system: str, messages: list[dict],
                  model: str = "", max_tokens: int | None = None,
-                 role: str = "", effort: str = ""):
+                 role: str = "", effort: str = "",
+                 max_budget_usd: float | None = None):
         """One-shot tool-less completion; returns a normalized `llm.Reply`.
 
         `effort` is a reasoning-effort pin (`low`/`medium`/`high`/`xhigh`) for
         transports whose CLI takes one (codex); the others accept and ignore
-        it, because their model id already fixes the reasoning budget."""
+        it, because their model id already fixes the reasoning budget.
+
+        `max_budget_usd` is a stop threshold where the CLI honours one
+        (`stops_at_spend`): no further API request starts once the call's
+        spend reaches it, but the request that crosses it is still billed. A
+        caller that needs a hard ceiling reserves the threshold PLUS one
+        request's worst case. Transports without it accept and ignore the
+        argument, so callers must check `stops_at_spend` first (the knowledge
+        service's gateway refuses them). The call's cost, when the transport
+        reports one, is `Reply.usage["cost_usd"]`."""
         raise NotImplementedError
 
 

src/infermatrix_copilot/providers/claude_code.py

@@ -22,8 +22,10 @@
 from __future__ import annotations
 
 import json
+import math
 import subprocess
 import sys
+from decimal import Decimal
 from pathlib import Path
 
 from ..agent_loop import AgentOutcome
@@ -43,15 +45,31 @@
 _BRIDGE_SERVER = "infermatrix-tools"
 
 
+def budget_arg(value: float) -> str:
+    """The threshold as the CLI argument, EXACTLY: the float's shortest
+    round-trip decimal in plain (never scientific) notation. Fixed-precision
+    formatting would move the boundary: `.4f` turned 0.00001 into "0.0000"
+    (no threshold at all) and 0.12345 into "0.1235" (a higher one)."""
+    if isinstance(value, bool) or not isinstance(value, (int, float)) \
+            or not math.isfinite(value) or not value > 0:
+        raise ValueError(f"max_budget_usd must be a positive finite number, got {value!r}")
+    return format(Decimal(repr(float(value))), "f")
+
+
 class ClaudeCodeTransport(HarnessTransport):
     """claude CLI (headless) as a Strict backend."""
 
     spec = PROVIDERS["claude-code"]
+    # `--max-budget-usd` (print mode, which _run uses) is checked after each
+    # API request completes: it stops the next one, it cannot cut the current
+    # one short. A threshold, never a hard cap.
+    stops_at_spend = True
 
     # -- process plumbing ----------------------------------------------------
     def _run(self, prompt_text: str, *, system: str, cwd: str,
              timeout_s: float, max_turns: int, model: str = "",
-             mcp_config: Path | None = None) -> tuple[dict, bool]:
+             mcp_config: Path | None = None,
+             max_budget_usd: float | None = None) -> tuple[dict, bool]:
         """One CLI invocation → (parsed result object, timed_out). Prompt on
         stdin (argv has a 128KiB per-arg limit; evidence packs exceed it)."""
         cmd = [self.require_cli(), "-p", "--output-format", "json",
@@ -63,6 +81,8 @@ def _run(self, prompt_text: str, *, system: str, cwd: str,
         selected = model or self.settings.strict_backend_model
         if selected:
             cmd += ["--model", selected]
+        if max_budget_usd is not None:
+            cmd += ["--max-budget-usd", budget_arg(max_budget_usd)]
         if mcp_config is not None:
             cmd += ["--mcp-config", str(mcp_config), "--strict-mcp-config",
                     "--allowedTools", f"mcp__{_BRIDGE_SERVER}"]
@@ -177,29 +197,42 @@ def run_session(self, req: AgentSessionRequest) -> AgentOutcome:
 
     def complete(self, *, system: str, messages: list[dict],
                  model: str = "", max_tokens: int | None = None,
-                 role: str = "", effort: str = "") -> Reply:
+                 role: str = "", effort: str = "",
+                 max_budget_usd: float | None = None) -> Reply:
         """Tool-less one-shot: built-ins denied, no MCP config, two turns
         (one to think, the cap as a backstop). Runs in the run-less scratch
         of the process cwd — with every tool denied there is nothing to
-        contain."""
+        contain.
+
+        `max_budget_usd` is passed as `--max-budget-usd`: once the call's
+        spend reaches it the CLI starts no further API request (the request
+        that crossed it is already billed, so spend can exceed the threshold
+        by one request). A call the CLI ended on its budget (a `subtype`
+        naming the budget) comes back with `stop_reason "max_budget"` and no
+        text. The CLI's `total_cost_usd` is reported as
+        `usage["cost_usd"]` (absent when the CLI gave none)."""
         import tempfile
 
         with tempfile.TemporaryDirectory(prefix="imc-claude-oneshot-") as td:
             data, timed_out = self._run(
                 flatten_messages("", messages), system=system, cwd=td,
                 timeout_s=self.settings.strict_backend_timeout_s,
-                max_turns=2, model=model)
+                max_turns=2, model=model, max_budget_usd=max_budget_usd)
         usage = self._usage(data)
-        text = str(data.get("result") or "")
+        over_budget = "budget" in str(data.get("subtype") or "")
+        text = "" if over_budget else str(data.get("result") or "")
+        extra = {} if usage.cost_usd is None else {"cost_usd": usage.cost_usd}
         return Reply(
             blocks=[Block(type="text", text=text)] if text else [],
-            stop_reason="max_tokens" if timed_out else "end_turn",
+            stop_reason=("max_tokens" if timed_out
+                         else "max_budget" if over_budget else "end_turn"),
             usage={"input_tokens": usage.input_tokens,
                    "output_tokens": usage.output_tokens,
                    "cache_read_input_tokens": int(
                        (data.get("usage") or {}).get(
                            "cache_read_input_tokens") or 0),
                    "cache_creation_input_tokens": int(
                        (data.get("usage") or {}).get(
-                           "cache_creation_input_tokens") or 0)},
+                           "cache_creation_input_tokens") or 0),
+                   **extra},
             model=usage.served_model)

src/infermatrix_copilot/providers/codex.py

@@ -182,7 +182,8 @@ def run_session(self, req: AgentSessionRequest) -> AgentOutcome:
 
     def complete(self, *, system: str, messages: list[dict],
                  model: str = "", max_tokens: int | None = None,
-                 role: str = "", effort: str = "") -> Reply:
+                 role: str = "", effort: str = "",
+                 max_budget_usd: float | None = None) -> Reply:
         """Tool-less one-shot in an empty scratch cwd (read-only sandbox +
         nothing to read = contained)."""
         import tempfile