Merge pull request #45 from openqj/real-cost

逐任务真实成本：上游用量事件 + 价目表计价

cloud/README.md

@@ -374,7 +374,7 @@ psycopg，写明“需安装 psycopg”），绝不静默回落到 SQLite，
 | `POST /tasks/{id}/heartbeat` | 该项目 `create_task` | 更新 `last_heartbeat = now`；状态为 `queued` 则改为 `running`；返回任务 |
 | `POST /admin/sweep` | 认证即可 | 调 `sweep_stale_tasks(db_path, threshold_seconds=300)`（body 可传 `threshold_seconds` 覆盖默认值）；把 `running` 且 `last_heartbeat`（无心跳则用 `updated_at`）早于 `now - threshold` 的任务改为 `interrupted` 并记审计 `sweep_interrupted`；返回 `{"swept": [ids]}` |
 | `POST /admin/backup` | maintainer 以上 | Postgres 时调 `backup_postgres`、SQLite 时调 `backup_db`（body 可传 `out_dir`，否则用 `CLOUD_BACKUP_DIR` 或默认 `cloud-backups`）；返回 `{backend, path, size_bytes, task_count}`，绝不返回数据库密码或备份文件内容 |
-| `POST /tasks/{id}/cost` | 该任务所属项目 `read` | 成本聚合：`{task_id, project_id, total_tokens, usd_micros, turns, fallback_count, actual_tokens, estimated_tokens}`；真数与估算分开；不存在 `404`，越权 `403` |
+| `GET /tasks/{id}/cost` | 该任务所属项目 `read` | 成本聚合：`{task_id, project_id, total_tokens, usd_micros, turns, fallback_count, actual_tokens, estimated_tokens, cached_input_tokens, unpriced_actual_tokens}`；真数与估算分开，无价 actual 行进 `unpriced_actual_tokens`；不存在 `404`，越权 `403` |
 | `POST /projects/{id}/memory` | 该项目 `create_task` | 写线程摘要 body `{thread_id?, goal, conclusion, files_changed?, open_issues?}`，返回 `201 {id}`，并记审计 `create_memory` |
 | `GET /projects/{id}/memory?top_k=5` | 该项目 `read` | 最近 active 摘要（默认 top_k=5）+ `prompt_context`（只含 top_k，不全量注入）；不存在 `404`，越权 `403` |
 | `POST /projects/{id}/memory/{mid}/supersede` | 该项目 `create_task` | 把摘要标废弃（deprecated，不是物理删除），并记审计 `supersede_memory`；摘要不属于该项目则 `404` |
@@ -697,9 +697,29 @@ interrupted 且终态任务 cancel 返回 400、心跳后 sweep 不扫刚心跳
   照样跑）。第二次的结果就是最终结果：成功则任务 completed，
   `actual_model` 写成 fallback 后的模型，成本记录 `fallback_count=1`；
   第二次也失败则任务 failed，`error_text` 同时写清主模型失败原因与
-  fallback 模型失败原因。边界：真实 token 成本仍是估算（host 协议
+  fallback 模型失败原因。成本来源与 fallback 无关（见下文“真实成本”）。
-  不暴露真实 token 数），fallback 不改变成本来源，只改变实际跑的
+- 真实成本（逐任务 actual）：host.py 在每回合结束时从 app-server
-  模型与 `actual_model`/`fallback_count` 的记录。
+  的 `thread/tokenUsage/updated` 事件里汇总上游真实用量（每个上游
+  补全发一次，`last` 是该次补全的用量，同一 turn 内累加；中继实测
+  五次 `last` 之和等于最后一次的 `total`，input/output/cached/
+  reasoning 四路都对得上）。host.py dispatch/run 打印的 JSON 带
+  `usage`，goal 循环在结论行后另打一行 `usage input=… cached=…
+  output=… total=… completions=…`；执行器解析后 worker 每任务记
+  一条 `source="actual"`（重试/fallback 各次尝试的用量都累加进
+  同一条，不按尝试次数重复记行）。没带回用量时（老版本 server、
+  回合未产生补全）回落字符估算记 `estimated`，两者在报表里分开。
+  金额 = 实测 token × 价目表：默认是 OpenAI 挂牌价（gpt-6.1-sol
+  $2.00/$0.10/$10.00、gpt-6-sol $2.00/$0.20/$10.00，单位美元/百万
+  token，依次为输入/缓存输入/输出；缓存是输入的子集，不另加；
+  推理 token 已含在输出里，不重复算），中继的实际账单可能不同，
+  可用 `CLOUD_MODEL_PRICES_JSON` 或 `CLOUD_MODEL_PRICES_FILE`
+  按模型覆盖；无价模型记 `usd_micros=None`，报表里进
+  `unpriced_actual_tokens` 而不是 0 美元。`GET /tasks/{id}/cost` 与
+  `GET /projects/{id}/cost` 的聚合新增 `cached_input_tokens` 与
+  `unpriced_actual_tokens`；按天行同样新增这两列。注意死胡同：
+  app-server 的 `account/usage/read` 在中继上不可用（要求 ChatGPT
+  账号登录，中继 API key 调它报 `codex account authentication
+  required`），账户级用量不是本功能的来源，不要重走这条路。
 - 容器清理兜底：`DockerExecutor.list_containers(prefix="codex-")`（
   `docker ps -a --format {{.Names}}`，失败返回 `[]`）与
   `cleanup_stale_containers()`（列出并逐个 kill+rm，单个失败只跳过，

cloud/cost.py

@@ -3,11 +3,30 @@
 Zero third-party dependencies; standard library only.
 
 Actual token counts come from the model/provider response when
-available; when they are not, token counts are estimates (see
+available: host.py collects the app-server's thread/tokenUsage/updated
-``estimate_tokens``) and are stored separately as ``source="estimated"``
+events during each turn (one per upstream completion, summed over the
-so estimates are never mixed into actual totals.
+turn window) and the worker records them with ``source="actual"``.
+When no usage events arrive (old server, turn produced no completion),
+the worker falls back to ``estimate_tokens`` with
+``source="estimated"``; the two are stored separately so estimates are
+never mixed into actual totals.
+
+USD is tokens times a price table, not the relay's bill: the defaults
+are OpenAI list prices (see ``_DEFAULT_MODEL_PRICES``), and the relay
+may charge differently. Override per model with
+``CLOUD_MODEL_PRICES_JSON`` or ``CLOUD_MODEL_PRICES_FILE``; models
+with no price record ``usd_micros=None`` and reports count their
+tokens in ``unpriced_actual_tokens`` instead of pricing them at zero.
+
+Dead end, documented so nobody retries it: the app-server's
+account/usage/read method requires ChatGPT account authentication and
+fails on a relay API key (2026-10-26 probe: ``codex account
+authentication required to read token usage``). Account-level usage
+is therefore not a source here; per-task actuals come from turn
+events only.
 """
 
+import json
 import os
 import sqlite3
 from cloud.db import connect as _db_connect
@@ -15,6 +34,101 @@ import time
 from datetime import datetime, timezone
 
 
+# OpenAI list prices, USD per 1M tokens, as (input, cached_input,
+# output). gpt-6.1-sol $2.00/$0.10/$10.00 and gpt-6-sol
+# $2.00/$0.20/$10.00, as reported at the 2026-09-30 launch.
+# Again: list prices, not the relay's bill; override below.
+_DEFAULT_MODEL_PRICES = {
+    "gpt-6.1-sol": (2.00, 0.10, 10.00),
+    "gpt-6-sol": (2.00, 0.20, 10.00),
+}
+
+
+def _coerce_rates(rates):
+    """Accept [input, cached, output] or {input, cached_input, output}."""
+    try:
+        if isinstance(rates, dict):
+            return (float(rates["input"]), float(rates["cached_input"]),
+                    float(rates["output"]))
+        return (float(rates[0]), float(rates[1]), float(rates[2]))
+    except (TypeError, ValueError, KeyError, IndexError):
+        return None
+
+
+def model_prices():
+    """Effective price table: defaults merged with env overrides.
+
+    ``CLOUD_MODEL_PRICES_JSON`` (a JSON object) and
+    ``CLOUD_MODEL_PRICES_FILE`` (path to a JSON file) both merge over
+    the defaults per model; unparsable content is ignored, never fatal.
+    """
+    prices = dict(_DEFAULT_MODEL_PRICES)
+    raw = os.environ.get("CLOUD_MODEL_PRICES_JSON") or ""
+    if raw:
+        try:
+            override = json.loads(raw)
+        except ValueError:
+            override = None
+        if isinstance(override, dict):
+            for model, rates in override.items():
+                coerced = _coerce_rates(rates)
+                if coerced is not None:
+                    prices[model] = coerced
+    path = os.environ.get("CLOUD_MODEL_PRICES_FILE") or ""
+    if path:
+        try:
+            with open(path) as f:
+                override = json.load(f)
+        except (OSError, ValueError):
+            override = None
+        if isinstance(override, dict):
+            for model, rates in override.items():
+                coerced = _coerce_rates(rates)
+                if coerced is not None:
+                    prices[model] = coerced
+    return prices
+
+
+def price_usd_micros(model, input_tokens, cached_input_tokens,
+                     output_tokens):
+    """USD micros for provider-measured tokens at the price table.
+
+    cached is a subset of input (prompt-cache reads): non-cached input
+    pays the input rate, cached pays the cached rate, output pays the
+    output rate. Reasoning tokens are already inside output_tokens
+    (provider total = input + output), never added twice. Returns None
+    when the model has no price entry (caller records usd as unknown).
+    """
+    rates = model_prices().get(model)
+    if rates is None:
+        return None
+    input_rate, cached_rate, output_rate = rates
+    total_in = max(0, int(input_tokens or 0))
+    cached = max(0, min(int(cached_input_tokens or 0), total_in))
+    non_cached = total_in - cached
+    usd = (non_cached * input_rate + cached * cached_rate
+           + max(0, int(output_tokens or 0)) * output_rate) / 1000000
+    return int(round(usd * 1000000))
+
+
+_USAGE_SUM_KEYS = ("input_tokens", "cached_input_tokens",
+                   "cache_write_input_tokens", "output_tokens",
+                   "reasoning_output_tokens", "total_tokens", "completions")
+
+
+def sum_usage(a, b):
+    """Add two usage dicts (host.py shape); either side may be None."""
+    if not a:
+        return dict(b) if b else None
+    if not b:
+        return dict(a)
+    out = dict(a)
+    for k in _USAGE_SUM_KEYS:
+        if isinstance(b.get(k), int):
+            out[k] = out.get(k, 0) + b[k]
+    return out
+
+
 def estimate_tokens(prompt_text, output_text) -> int:
     """Estimate total tokens as (chars of prompt + output) // 4."""
     prompt_text = prompt_text or ""
@@ -52,20 +166,14 @@ class CostStore:
 
     _SOURCES = ("actual", "estimated")
 
-    def __init__(self, db_path):
+    _COST_DDL = """CREATE TABLE IF NOT EXISTS cost_records (
-        self.db_path = db_path
-        if db_path != ":memory:":
-            parent = os.path.dirname(os.path.abspath(db_path))
-            os.makedirs(parent, exist_ok=True)
-        self._conn = _db_connect(db_path)
-        self._conn.execute(
-            """CREATE TABLE IF NOT EXISTS cost_records (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 task_id TEXT,
                 project_id TEXT,
                 model TEXT,
                 input_tokens INTEGER,
                 output_tokens INTEGER,
+                cached_input_tokens INTEGER DEFAULT 0,
                 total_tokens INTEGER,
                 usd_micros INTEGER,
                 source TEXT,
@@ -73,8 +181,36 @@ class CostStore:
                 day TEXT,
                 created_at REAL
             )"""
-        )
+
+    def _ensure_cached_column(self, conn):
+        """老库补 cached_input_tokens 列（SQLite 与 Postgres 两支：
+
+        PRAGMA table_info 经 db.py 适配层翻译，缺列时 ALTER 补列；
+        并发下两个进程同时补列时第二个 ALTER 报错，直接吞掉）。
+        """
+        try:
+            cols = {row[1] for row in conn.execute(
+                "PRAGMA table_info(cost_records)").fetchall()}
+        except Exception:
+            return
+        if "cached_input_tokens" not in cols:
+            try:
+                conn.execute(
+                    "ALTER TABLE cost_records"
+                    " ADD COLUMN cached_input_tokens INTEGER DEFAULT 0")
+                conn.commit()
+            except Exception:
+                pass
+
+    def __init__(self, db_path):
+        self.db_path = db_path
+        if db_path != ":memory:":
+            parent = os.path.dirname(os.path.abspath(db_path))
+            os.makedirs(parent, exist_ok=True)
+        self._conn = _db_connect(db_path)
+        self._conn.execute(self._COST_DDL)
         self._conn.commit()
+        self._ensure_cached_column(self._conn)
 
     def _new_conn(self):
         """每次写/读开一个新连接，用完即关。
@@ -92,32 +228,21 @@ class CostStore:
         if self.db_path == ":memory:":
             return self._conn
         conn = _db_connect(self.db_path)
-        conn.execute(
+        conn.execute(self._COST_DDL)
-            """CREATE TABLE IF NOT EXISTS cost_records (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                task_id TEXT,
-                project_id TEXT,
-                model TEXT,
-                input_tokens INTEGER,
-                output_tokens INTEGER,
-                total_tokens INTEGER,
-                usd_micros INTEGER,
-                source TEXT,
-                fallback_count INTEGER DEFAULT 0,
-                day TEXT,
-                created_at REAL
-            )"""
-        )
         conn.commit()
+        self._ensure_cached_column(conn)
         return conn
 
     def record(self, task_id, project_id, model, input_tokens=0,
-               output_tokens=0, usd_micros=None, source="actual",
+               output_tokens=0, cached_input_tokens=0, usd_micros=None,
-               fallback_count=0):
+               source="actual", fallback_count=0):
         """Insert one cost record; returns its row id.
 
         ``total_tokens`` is computed as input + output, and ``day`` is
-        the UTC date (YYYY-MM-DD) of ``created_at``.
+        the UTC date (YYYY-MM-DD) of ``created_at``. cached is a
+        subset of input (prompt-cache reads), stored separately so
+        reports can show why a task was cheap; it is never added to
+        total on top of input.
         """
         if source not in self._SOURCES:
             raise ValueError(
@@ -125,6 +250,7 @@ class CostStore:
             )
         input_tokens = int(input_tokens or 0)
         output_tokens = int(output_tokens or 0)
+        cached_input_tokens = int(cached_input_tokens or 0)
         total_tokens = input_tokens + output_tokens
         created_at = time.time()
         day = datetime.fromtimestamp(created_at, tz=timezone.utc).strftime("%Y-%m-%d")
@@ -133,12 +259,12 @@ class CostStore:
             cursor = conn.execute(
                 """INSERT INTO cost_records
                    (task_id, project_id, model, input_tokens, output_tokens,
-                    total_tokens, usd_micros, source, fallback_count, day,
+                    cached_input_tokens, total_tokens, usd_micros, source,
-                    created_at)
+                    fallback_count, day, created_at)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (task_id, project_id, model, input_tokens, output_tokens,
-                 total_tokens, usd_micros, source, int(fallback_count or 0),
+                 cached_input_tokens, total_tokens, usd_micros, source,
-                 day, created_at),
+                 int(fallback_count or 0), day, created_at),
             )
             conn.commit()
             return cursor.lastrowid
@@ -175,6 +301,10 @@ class CostStore:
                      COALESCE(SUM(CASE WHEN source = 'actual'
                                        THEN total_tokens ELSE 0 END), 0),
                      COALESCE(SUM(CASE WHEN source = 'estimated'
+                                       THEN total_tokens ELSE 0 END), 0),
+                     COALESCE(SUM(cached_input_tokens), 0),
+                     COALESCE(SUM(CASE WHEN source = 'actual'
+                                       AND usd_micros IS NULL
                                        THEN total_tokens ELSE 0 END), 0)
                    FROM cost_records WHERE %s""" % where,
                 params,
@@ -182,7 +312,8 @@ class CostStore:
         finally:
             if conn is not self._conn:
                 conn.close()
-        total_tokens, usd_micros, turns, fallback_count, actual, estimated = row
+        (total_tokens, usd_micros, turns, fallback_count, actual,
+         estimated, cached, unpriced) = row
         return {
             "total_tokens": int(total_tokens),
             "usd_micros": int(usd_micros),
@@ -190,6 +321,10 @@ class CostStore:
             "fallback_count": int(fallback_count),
             "actual_tokens": int(actual),
             "estimated_tokens": int(estimated),
+            "cached_input_tokens": int(cached),
+            # usd 求和只覆盖有价的行；无价 actual 行的 token 在此，
+            # 不会被静默算成 0 美元。
+            "unpriced_actual_tokens": int(unpriced),
         }
 
     def report_by_task(self, task_id) -> dict:

cloud/db.py

@@ -859,6 +859,7 @@ POSTGRES_SCHEMA = (
     "model TEXT, "
     "input_tokens BIGINT, "
     "output_tokens BIGINT, "
+    "cached_input_tokens BIGINT DEFAULT 0, "
     "total_tokens BIGINT, "
     "usd_micros BIGINT, "
     "source TEXT, "

cloud/sandbox.py

@@ -44,6 +44,41 @@ class Executor:
         raise NotImplementedError("该执行器不支持 goal 模式")
 
 
+def _clean_usage(raw):
+    """校验 host.py 带来的用量 dict：非 dict、非整数、或 input /
+    output / total 全为 0 时返回 None（调用方回落估算，绝不把
+    0 或脏数据当成真实用量记 actual）。"""
+    if not isinstance(raw, dict):
+        return None
+    keys = ("input_tokens", "cached_input_tokens",
+            "cache_write_input_tokens", "output_tokens",
+            "reasoning_output_tokens", "total_tokens", "completions")
+    try:
+        out = {k: int(raw.get(k) or 0) for k in keys}
+    except (TypeError, ValueError):
+        return None
+    if out["input_tokens"] <= 0 and out["output_tokens"] <= 0 \
+            and out["total_tokens"] <= 0:
+        return None
+    return out
+
+
+_USAGE_LINE_RE = re.compile(
+    r"^usage input=(\d+) cached=(\d+) cache_write=(\d+) output=(\d+)"
+    r" reasoning=(\d+) total=(\d+) completions=(\d+)\s*$", re.M)
+
+
+def _parse_goal_usage(stdout):
+    """从 host.py goal 的用量行解析用量 dict，无用量行返回 None。"""
+    m = _USAGE_LINE_RE.search(stdout or "")
+    if not m:
+        return None
+    keys = ("input_tokens", "cached_input_tokens",
+            "cache_write_input_tokens", "output_tokens",
+            "reasoning_output_tokens", "total_tokens", "completions")
+    return {k: int(v) for k, v in zip(keys, m.groups())}
+
+
 def _parse_goal_output(stdout):
     """解析 host.py goal 的 stdout，返回 (met, goal_error)。
 
@@ -155,18 +190,28 @@ class LocalExecutor(Executor):
         判断是否可 fallback），取不到 JSON 时按 completed 计。
         """
         returncode = getattr(result, "returncode", 1)
-        stdout = _truncate_stdout(getattr(result, "stdout", ""))
+        stdout_full = getattr(result, "stdout", "") or ""
+        stdout = _truncate_stdout(stdout_full)
         status = "failed"
+        usage = None
         if returncode == 0:
             status = "completed"
+            # JSON 解析必须用完整 stdout：截头后的前 2000 字在
+            # text 很长时根本不是合法 JSON，旧代码会把 host 的
+            # failed 误判成 completed；usage 在 JSON 末尾，截头
+            # 后同样丢。存库的 stdout 仍用截头版，不变。
             try:
-                parsed = json.loads(stdout)
+                parsed = json.loads(stdout_full)
             except (TypeError, ValueError):
                 parsed = None
             if isinstance(parsed, dict) and parsed.get("status"):
                 status = ("completed" if parsed["status"] == "completed"
                           else "failed")
                 host_status = parsed["status"]
+                # host.py dispatch/run 打印的 JSON 里带 usage 时
+                # 原样透传给 worker 记 actual；脏数据由
+                # _clean_usage 挡掉，不进 actual。
+                usage = _clean_usage(parsed.get("usage"))
             else:
                 host_status = None
         else:
@@ -182,6 +227,7 @@ class LocalExecutor(Executor):
             # stderr，worker 判断 dispatch 续跑时需要它；execute()
             # 的旧结果 dict 不含此字段，保持不变。
             "stderr": _truncate_stdout(getattr(result, "stderr", "")),
+            "usage": usage,
         }
 
     def _host_name(self, task):
@@ -309,6 +355,7 @@ class LocalExecutor(Executor):
                            "goal-unmet" if met is False else None),
             "goal_met": met,
             "goal_error": goal_error,
+            "usage": _parse_goal_usage(stdout_full),
         }
 
     def popen_task(self, task, kind="dispatch"):
@@ -852,6 +899,7 @@ class DockerExecutor(Executor):
             "container": self.container_name(task),
             "goal_met": met,
             "goal_error": goal_error,
+            "usage": _parse_goal_usage(stdout_full),
         }
 
     def network_gateway(self, network_name):
@@ -935,19 +983,25 @@ def _parse_host_result(task_id, fake):
     completed 外的状态均按 failed 计。
     """
     returncode = getattr(fake, "returncode", 1)
-    stdout = _truncate_stdout(getattr(fake, "stdout", ""))
+    stdout_full = getattr(fake, "stdout", "") or ""
+    stdout = _truncate_stdout(stdout_full)
     status = "failed"
     host_status = None
+    usage = None
     if returncode == 0:
         status = "completed"
+        # 与 _result_from_runner 同理：JSON 解析用完整 stdout，
+        # 否则长 text 会把 host 的 failed 误判成 completed，
+        # 也会丢掉 JSON 末尾的 usage。
         try:
-            parsed = json.loads(stdout)
+            parsed = json.loads(stdout_full)
         except (TypeError, ValueError):
             parsed = None
         if isinstance(parsed, dict) and parsed.get("status"):
             host_status = parsed["status"]
             status = ("completed" if host_status == "completed"
                       else "failed")
+            usage = _clean_usage(parsed.get("usage"))
     return {
         "executor": "docker",
         "task_id": task_id,
@@ -956,6 +1010,7 @@ def _parse_host_result(task_id, fake):
         "returncode": returncode,
         "host_status": host_status,
         "stderr": _truncate_stdout(getattr(fake, "stderr", "")),
+        "usage": usage,
     }
 
 

cloud/server.py

@@ -1092,14 +1092,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 row = conn.execute(
                     "SELECT day, COALESCE(SUM(total_tokens),0) AS t,"
                     " COALESCE(SUM(usd_micros),0) AS u, COUNT(*) AS c,"
-                    " COALESCE(SUM(fallback_count),0) AS f"
+                    " COALESCE(SUM(fallback_count),0) AS f,"
+                    " COALESCE(SUM(cached_input_tokens),0) AS cached,"
+                    " COALESCE(SUM(CASE WHEN source = 'actual'"
+                    " AND usd_micros IS NULL THEN total_tokens"
+                    " ELSE 0 END),0) AS unpriced"
                     " FROM cost_records WHERE project_id = ? AND day = ?"
                     " GROUP BY day",
                     (project_id, day)).fetchone()
                 if row:
                     daily = [{"day": row["day"], "total_tokens": row["t"],
                               "usd_micros": row["u"], "turns": row["c"],
-                              "fallback_count": row["f"]}]
+                              "fallback_count": row["f"],
+                              "cached_input_tokens": row["cached"],
+                              "unpriced_actual_tokens": row["unpriced"]}]
             elif days is not None:
                 try:
                     n = int(days)
@@ -1112,13 +1118,20 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
                 rows = conn.execute(
                     "SELECT day, COALESCE(SUM(total_tokens),0) AS t,"
                     " COALESCE(SUM(usd_micros),0) AS u, COUNT(*) AS c,"
-                    " COALESCE(SUM(fallback_count),0) AS f"
+                    " COALESCE(SUM(fallback_count),0) AS f,"
+                    " COALESCE(SUM(cached_input_tokens),0) AS cached,"
+                    " COALESCE(SUM(CASE WHEN source = 'actual'"
+                    " AND usd_micros IS NULL THEN total_tokens"
+                    " ELSE 0 END),0) AS unpriced"
                     " FROM cost_records WHERE project_id = ? AND day >= ?"
                     " GROUP BY day ORDER BY day",
                     (project_id, cutoff)).fetchall()
                 daily = [{"day": r["day"], "total_tokens": r["t"],
                           "usd_micros": r["u"], "turns": r["c"],
-                          "fallback_count": r["f"]} for r in rows]
+                          "fallback_count": r["f"],
+                          "cached_input_tokens": r["cached"],
+                          "unpriced_actual_tokens": r["unpriced"]}
+                         for r in rows]
         finally:
             try:
                 store._conn.close()

cloud/tests/test_models_cost.py

@@ -8,13 +8,15 @@ import os
 import sys
 import unittest
 from datetime import datetime, timezone
+from unittest import mock
 
 # cloud/ has no __init__.py by design (the coordinator creates it), so
 # make the project root importable regardless of the discover top level.
 sys.path.insert(0, os.path.dirname(os.path.dirname(
     os.path.dirname(os.path.abspath(__file__)))))
 
-from cloud.cost import CostStore, estimate_tokens, fetch_account_usage
+from cloud.cost import (CostStore, estimate_tokens, fetch_account_usage,
+                        price_usd_micros)
 from cloud.models import FALLBACKABLE, ModelRouter, is_fallbackable, record_fallback
 
 
@@ -192,8 +194,80 @@ class TestCostStore(unittest.TestCase):
             "fallback_count": 0,
             "actual_tokens": 0,
             "estimated_tokens": 0,
+            "cached_input_tokens": 0,
+            "unpriced_actual_tokens": 0,
         })
 
 
+class TestModelPrices(unittest.TestCase):
+    def test_list_prices(self):
+        # 2026-10-26 中继实测 5 调用回合：(58243-45960)*2
+        # + 45960*0.10 + 1138*10 = 40542 微美元。
+        self.assertEqual(
+            price_usd_micros("gpt-6.1-sol", 58243, 45960, 1138), 40542)
+        self.assertEqual(
+            price_usd_micros("gpt-6-sol", 1000, 200, 100),
+            int(round((800 * 2.00 + 200 * 0.20 + 100 * 10.00))))
+
+    def test_unknown_model_unpriced(self):
+        self.assertIsNone(price_usd_micros("gpt-5.6-sol", 100, 0, 100))
+        self.assertIsNone(price_usd_micros(None, 100, 0, 100))
+
+    def test_cached_clamped_to_input(self):
+        self.assertEqual(
+            price_usd_micros("gpt-6.1-sol", 100, 500, 100),
+            price_usd_micros("gpt-6.1-sol", 100, 100, 100))
+
+    def test_env_override(self):
+        with mock.patch.dict(os.environ, {
+                "CLOUD_MODEL_PRICES_JSON":
+                    '{"gpt-6.1-sol": [4.00, 0.20, 20.00]}'}):
+            self.assertEqual(
+                price_usd_micros("gpt-6.1-sol", 1000, 0, 0), 4000)
+        with mock.patch.dict(os.environ, {
+                "CLOUD_MODEL_PRICES_JSON": 'not json'}):
+            # 坏覆盖不炸，回落默认表。
+            self.assertEqual(
+                price_usd_micros("gpt-6.1-sol", 1000, 0, 0), 2000)
+
+
+class TestCachedColumn(unittest.TestCase):
+    def test_record_keeps_cached_and_unpriced(self):
+        store = CostStore(":memory:")
+        store.record("t1", "p1", "gpt-6.1-sol", input_tokens=1000,
+                     output_tokens=100, cached_input_tokens=400,
+                     usd_micros=2240, source="actual")
+        store.record("t2", "p1", "gpt-5.6-sol", input_tokens=500,
+                     output_tokens=50, cached_input_tokens=0,
+                     usd_micros=None, source="actual")
+        report = store.report_by_project("p1")
+        self.assertEqual(report["cached_input_tokens"], 400)
+        self.assertEqual(report["unpriced_actual_tokens"], 550)
+        self.assertEqual(report["usd_micros"], 2240)
+
+    def test_migration_adds_cached_column_to_old_table(self):
+        import sqlite3
+        import tempfile
+        path = os.path.join(tempfile.mkdtemp(), "c.db")
+        conn = sqlite3.connect(path)
+        conn.execute(
+            "CREATE TABLE cost_records (id INTEGER PRIMARY KEY"
+            " AUTOINCREMENT, task_id TEXT, project_id TEXT, model TEXT,"
+            " input_tokens INTEGER, output_tokens INTEGER,"
+            " total_tokens INTEGER, usd_micros INTEGER, source TEXT,"
+            " fallback_count INTEGER DEFAULT 0, day TEXT,"
+            " created_at REAL)")
+        conn.commit()
+        conn.close()
+        store = CostStore(path)
+        cols = {row[1] for row in store._conn.execute(
+            "PRAGMA table_info(cost_records)").fetchall()}
+        self.assertIn("cached_input_tokens", cols)
+        store.record("t1", "p1", "m", input_tokens=10, output_tokens=5,
+                     cached_input_tokens=3, source="actual")
+        self.assertEqual(store.report_by_task("t1")["cached_input_tokens"],
+                         3)
+
+
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_queue_sandbox.py

@@ -8,6 +8,7 @@
 top-level 目录无关。
 """
 
+import json
 import os
 import sys
 import unittest
@@ -385,5 +386,99 @@ class TestGoalExecutors(unittest.TestCase):
         self.assertIn("test -f x", seen[0])
 
 
+class TestUsageParsing(unittest.TestCase):
+    def test_result_from_runner_usage(self):
+        usage = {"input_tokens": 1000, "cached_input_tokens": 400,
+                 "cache_write_input_tokens": 0, "output_tokens": 100,
+                 "reasoning_output_tokens": 20, "total_tokens": 1100,
+                 "completions": 2}
+        fake = SimpleNamespace(
+            returncode=0,
+            stdout=json.dumps({"status": "completed", "text": "done",
+                               "usage": usage}),
+            stderr="")
+        result = LocalExecutor("/fake/host.py",
+                               runner=lambda cmd: fake
+                               )._result_from_runner("t1", fake)
+        self.assertEqual(result["usage"], usage)
+        self.assertEqual(result["status"], "completed")
+
+    def test_result_from_runner_usage_survives_long_text(self):
+        # usage 在 JSON 末尾：旧的截头解析在 text 很长时会丢 usage、
+        # 还会把 failed 误判成 completed；解析必须用完整 stdout。
+        usage = {"input_tokens": 100, "cached_input_tokens": 0,
+                 "cache_write_input_tokens": 0, "output_tokens": 10,
+                 "reasoning_output_tokens": 0, "total_tokens": 110,
+                 "completions": 1}
+        fake = SimpleNamespace(
+            returncode=0,
+            stdout=json.dumps({"status": "failed", "text": "x" * 5000,
+                               "usage": usage}),
+            stderr="")
+        result = LocalExecutor("/fake/host.py",
+                               runner=lambda cmd: fake
+                               )._result_from_runner("t1", fake)
+        self.assertEqual(result["status"], "failed")
+        self.assertEqual(result["usage"], usage)
+
+    def test_result_from_runner_dirty_usage_dropped(self):
+        for bad in ({"input_tokens": "abc"},
+                    {"input_tokens": 0, "output_tokens": 0,
+                     "total_tokens": 0},
+                    "not-a-dict"):
+            fake = SimpleNamespace(
+                returncode=0,
+                stdout=json.dumps({"status": "completed", "usage": bad}),
+                stderr="")
+            result = LocalExecutor("/fake/host.py",
+                                   runner=lambda cmd: fake
+                                   )._result_from_runner("t1", fake)
+            self.assertIsNone(result["usage"])
+
+    def test_goal_usage_line(self):
+        fake = SimpleNamespace(
+            returncode=0,
+            stdout="round 1: turn=completed verify_rc=0 verify_out=ok\n"
+                   "goal met after rounds\n"
+                   "usage input=100 cached=40 cache_write=0 output=50 "
+                   "reasoning=10 total=150 completions=2\n",
+            stderr="")
+        result = LocalExecutor("/fake/host.py",
+                               runner=lambda cmd: fake
+                               )._goal_result_from_runner("t1", fake)
+        self.assertEqual(result["usage"], {
+            "input_tokens": 100, "cached_input_tokens": 40,
+            "cache_write_input_tokens": 0, "output_tokens": 50,
+            "reasoning_output_tokens": 10, "total_tokens": 150,
+            "completions": 2})
+        fake2 = SimpleNamespace(
+            returncode=0, stdout="goal met after rounds\n", stderr="")
+        result2 = LocalExecutor("/fake/host.py",
+                                runner=lambda cmd: fake2
+                                )._goal_result_from_runner("t1", fake2)
+        self.assertIsNone(result2["usage"])
+
+    def test_docker_goal_usage_line(self):
+        def fake_runner(cmd, timeout=None):
+            return SimpleNamespace(
+                returncode=0,
+                stdout="round 1: turn=completed verify_rc=0 verify_out=ok\n"
+                       "goal met after rounds\n"
+                       "usage input=100 cached=40 cache_write=0 output=50 "
+                       "reasoning=10 total=150 completions=2\n",
+                stderr="")
+
+        with mock.patch("sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            executor = DockerExecutor(
+                "/fake/host.py", runner=fake_runner,
+                env={"XCM_API_KEY": "placeholder-not-a-real-key"})
+            result = executor.goal_task(
+                {"id": "t1", "prompt": "do it",
+                 "verify_command": "test -f x"})
+        self.assertEqual(result["usage"]["total_tokens"], 150)
+        self.assertEqual(result["usage"]["cached_input_tokens"], 40)
+
+
 if __name__ == "__main__":
     unittest.main()

cloud/tests/test_worker.py

@@ -374,6 +374,75 @@ class WorkerTestCase(unittest.TestCase):
         self.assertEqual(results[0]["status"], "failed")
         self.assertIn("不支持 goal", results[0]["error"])
 
+    # -- 成本：真实用量 --------------------------------------------------
+    def test_cost_actual_from_usage(self):
+        """执行器带回用量时记 actual：token 全是实测值，usd 按
+        价目表算（(1000-400)*2 + 400*0.10 + 100*10 = 2240 微美元）。"""
+        _insert_task(self.db_path, "A")
+        usage = {"input_tokens": 1000, "cached_input_tokens": 400,
+                 "cache_write_input_tokens": 0, "output_tokens": 100,
+                 "reasoning_output_tokens": 20, "total_tokens": 1100,
+                 "completions": 1}
+
+        def runner(cmd):
+            return SimpleNamespace(
+                returncode=0,
+                stdout=json.dumps({"status": "completed", "text": "done",
+                                   "usage": usage}),
+                stderr="")
+
+        worker = self._worker(runner=runner, model="gpt-6.1-sol")
+        results = worker.run_all()
+        self.assertEqual([r["status"] for r in results], ["completed"])
+        from cloud.cost import CostStore
+        report = CostStore(self.db_path).report_by_task("A")
+        self.assertEqual(report["actual_tokens"], 1100)
+        self.assertEqual(report["cached_input_tokens"], 400)
+        self.assertEqual(report["estimated_tokens"], 0)
+        self.assertEqual(report["usd_micros"], 2240)
+        self.assertEqual(report["unpriced_actual_tokens"], 0)
+
+    def test_cost_actual_unpriced_model(self):
+        """无价模型：token 照样记 actual，usd 记 None、报表里
+        进 unpriced_actual_tokens，绝不按 0 美元处理。"""
+        _insert_task(self.db_path, "A")
+        usage = {"input_tokens": 1000, "cached_input_tokens": 0,
+                 "cache_write_input_tokens": 0, "output_tokens": 100,
+                 "reasoning_output_tokens": 0, "total_tokens": 1100,
+                 "completions": 1}
+
+        def runner(cmd):
+            return SimpleNamespace(
+                returncode=0,
+                stdout=json.dumps({"status": "completed", "text": "done",
+                                   "usage": usage}),
+                stderr="")
+
+        worker = self._worker(runner=runner, model="gpt-5.6-sol")
+        results = worker.run_all()
+        self.assertEqual([r["status"] for r in results], ["completed"])
+        from cloud.cost import CostStore
+        report = CostStore(self.db_path).report_by_task("A")
+        self.assertEqual(report["actual_tokens"], 1100)
+        self.assertEqual(report["usd_micros"], 0)
+        self.assertEqual(report["unpriced_actual_tokens"], 1100)
+
+    def test_cost_estimated_when_no_usage(self):
+        """执行器没带回用量时回落字符估算记 estimated（旧行为）。"""
+        _insert_task(self.db_path, "A")
+
+        def runner(cmd):
+            return SimpleNamespace(returncode=0, stdout="plain text",
+                                   stderr="")
+
+        worker = self._worker(runner=runner, model="gpt-6.1-sol")
+        results = worker.run_all()
+        self.assertEqual([r["status"] for r in results], ["completed"])
+        from cloud.cost import CostStore
+        report = CostStore(self.db_path).report_by_task("A")
+        self.assertEqual(report["actual_tokens"], 0)
+        self.assertGreater(report["estimated_tokens"], 0)
+
     # -- 取消 ------------------------------------------------------------
     def test_cancel_queued_task_marks_interrupted(self):
         _insert_task(self.db_path, "A")

cloud/worker.py

@@ -26,9 +26,12 @@
 4. 写回：server 的 tasks 表没有 result/error 列，所以执行结果
    的持久化分三处，各复用已有结构、不另起字段名：
    - tasks 表：最终状态 + updated_at（取消时为 interrupted）；
-   - cost.py CostStore：每轮实际执行记一条 source="estimated"
+   - cost.py CostStore：每轮实际执行记一条成本记录；执行器带回
-     的成本记录（host 协议不暴露真实 token 数，估算绝不冒充
+     上游真实用量（host.py 从 tokenUsage/updated 事件汇总）时记
-     actual；total_tokens 由 CostStore.record 按 input+output 算）；
+     source="actual"（input/output/cached 全是实测值，usd 按价目
+     表算、无价模型记 None）；没带回用量时回落 cost.py 估算记
+     estimated，估算绝不冒充 actual；total_tokens 由
+     CostStore.record 按 input+output 算；
    - models.py：可 fallback 的失败（404/429/5xx/timeout，401/400
      不算）。默认（未配 --fallback-models）时 worker 只记录
      fallback 事件（record_fallback() 记 model_fallback 事件，
@@ -122,7 +125,7 @@ import time
 from types import SimpleNamespace
 
 from cloud.auth import AuthStore
-from cloud.cost import CostStore, estimate_tokens
+from cloud.cost import CostStore, estimate_tokens, price_usd_micros, sum_usage
 from cloud.db import connect as _db_connect
 from cloud.logging_config import get_logger, log_event, setup_logging
 from cloud.models import ModelRouter, is_fallbackable, record_fallback
@@ -1455,8 +1458,12 @@ class Worker:
         # fallback_models 时保持老行为（router 给出下一模型时只记事件
         # 不重试）；其它失败（含 worker 超时）按 max_retries 做同模型
         # 重试，退避等待后重跑一次 _run_attempt。
+        # 各次尝试的真实用量都要算进成本（失败尝试花的钱也是钱）：
+        # 主尝试与 fallback 尝试的结果都先累加，最后只记一条汇总。
+        usage_total = None
         for attempt in range(self.max_retries + 1):
             result, error = self._run_attempt(task)
+            usage_total = sum_usage(usage_total, (result or {}).get("usage"))
             if self._was_cancelled(task_id) or \
                     result.get("status") == "cancelled":
                 break
@@ -1493,6 +1500,8 @@ class Worker:
                 task = dict(task)
                 task["model"] = fb
                 result, error = self._run_attempt(task)
+                usage_total = sum_usage(usage_total,
+                                        (result or {}).get("usage"))
                 break
             if self.router is not None and host_status_try is not None \
                     and is_fallbackable(host_status_try) and \
@@ -1585,16 +1594,32 @@ class Worker:
             finished_at=finished_at)
         with self._db_lock:
             self.usage_turns += 1
-            # 成本：host 协议不暴露真实 token 数，用 cost.py 的估算结构
+            # 成本：执行器带回上游真实用量时记 actual（input/output/
-            # 记 estimated，绝不标 actual；fallback_count 复用同一记录；
+            # cached 全是实测值，usd 按价目表算、无价模型记 None，
-            # 模型记最后真正跑的那个（fallback 成功时就是 fallback 模型）。
+            # 报表里进 unpriced_actual_tokens 而不是 0 美元）；
-            # cost 记录共用一条连接，并发下必须持锁写，否则两个线程
+            # 没带回用量时回落字符估算记 estimated，绝不冒充 actual。
-            # 共用同一 sqlite 连接会直接报错。
+            # fallback_count 复用同一记录；模型记最后真正跑的那个
-            total = estimate_tokens(task.get("prompt") or "", stdout)
+            # （fallback 成功时就是 fallback 模型）。cost 记录共用
-            cost_id = self.costs.record(
+            # 一条连接，并发下必须持锁写，否则两个线程共用同一
-                task_id, task["project_id"], actual_model,
+            # sqlite 连接会直接报错。
-                input_tokens=total, output_tokens=0, usd_micros=None,
+            if usage_total and (usage_total.get("total_tokens") or 0) > 0:
-                source="estimated", fallback_count=fallback_count)
+                cost_id = self.costs.record(
+                    task_id, task["project_id"], actual_model,
+                    input_tokens=usage_total.get("input_tokens", 0),
+                    output_tokens=usage_total.get("output_tokens", 0),
+                    cached_input_tokens=usage_total.get(
+                        "cached_input_tokens", 0),
+                    usd_micros=price_usd_micros(
+                        actual_model, usage_total.get("input_tokens", 0),
+                        usage_total.get("cached_input_tokens", 0),
+                        usage_total.get("output_tokens", 0)),
+                    source="actual", fallback_count=fallback_count)
+            else:
+                total = estimate_tokens(task.get("prompt") or "", stdout)
+                cost_id = self.costs.record(
+                    task_id, task["project_id"], actual_model,
+                    input_tokens=total, output_tokens=0, usd_micros=None,
+                    source="estimated", fallback_count=fallback_count)
         if final_written:
             self._audit(task["project_id"],
                         "worker_completed" if final == "completed"

host.py

@@ -59,6 +59,56 @@ HERE = os.path.dirname(os.path.abspath(__file__))
 DEFAULT_STATE = os.path.join(HERE, "host-state.json")
 
 
+def _sum_turn_usage(events, thread_id, turn_id):
+    """从事件窗口里汇总一回合的上游真实用量。
+
+    app-server 在每个上游补全结束时发 thread/tokenUsage/updated，
+    params.tokenUsage.last 是该次补全的用量（2026-10-26 中继实测：
+    同一 turn 的多次 last 之和等于最后一次的 total，input/output/
+    cached/reasoning 四路都对得上）。按 threadId + turnId 过滤后
+    累加 last，得到本回合的真实用量；无匹配通知时返回 None（老
+    版本 server 或回合未产生补全时调用方回落估算，绝不把 0 当成
+    真实用量）。注意 totalTokens = inputTokens + outputTokens，
+    reasoning 包含在 output 里，累加时不另算。
+    """
+    fields = (("input_tokens", "inputTokens"),
+              ("cached_input_tokens", "cachedInputTokens"),
+              ("cache_write_input_tokens", "cacheWriteInputTokens"),
+              ("output_tokens", "outputTokens"),
+              ("reasoning_output_tokens", "reasoningOutputTokens"),
+              ("total_tokens", "totalTokens"))
+    total = None
+    for ev in events or []:
+        if ev.get("method") != "thread/tokenUsage/updated":
+            continue
+        p = ev.get("params") or {}
+        if p.get("threadId") != thread_id:
+            continue
+        if turn_id is not None and p.get("turnId") != turn_id:
+            continue
+        last = (p.get("tokenUsage") or {}).get("last") or {}
+        if total is None:
+            total = {k: 0 for k, _ in fields}
+            total["completions"] = 0
+        for key, wire in fields:
+            total[key] += int(last.get(wire) or 0)
+        total["completions"] += 1
+    return total
+
+
+def _add_usage(a, b):
+    """两份用量相加（任一为空时返回另一份；都为空返回 None）。"""
+    if not a:
+        return dict(b) if b else None
+    if not b:
+        return dict(a)
+    out = dict(a)
+    for k, v in b.items():
+        if isinstance(v, int):
+            out[k] = out.get(k, 0) + v
+    return out
+
+
 def default_codex_bin():
     if os.environ.get("CODEX_BIN"):
         return os.environ["CODEX_BIN"]
@@ -298,7 +348,10 @@ class Server:
             try:
                 done = q.get(timeout=timeout)
             except queue.Empty:
-                return {"status": "timeout", "text": "", "turn_id": turn_id}
+                with self.events_lock:
+                    evs = list(self.events[start_events:])
+                return {"status": "timeout", "text": "", "turn_id": turn_id,
+                        "usage": _sum_turn_usage(evs, thread_id, turn_id)}
             turn = (done.get("params") or {}).get("turn") or {}
             text = ""
             for item in turn.get("items") or []:
@@ -329,10 +382,16 @@ class Server:
                 # 没有原因，绝不因此把回合本身搞失败。
                 error = self._turn_error_from_history(
                     thread_id, turn.get("id") or turn_id)
+            # 真实用量：本回合窗口内的 tokenUsage/updated 通知累加。
+            # 失败/超时回合也一样记（已产生的补全就是已花的钱）。
+            with self.events_lock:
+                evs = list(self.events[start_events:])
+            usage = _sum_turn_usage(
+                evs, thread_id, turn.get("id") or turn_id)
             return {"status": status, "text": text,
                     "turn_id": turn.get("id") or turn_id,
                     "root_turn_id": turn.get("rootTurnId"),
-                    "error": error}
+                    "error": error, "usage": usage}
         finally:
             self.turn_queues.pop(thread_id, None)
 
@@ -1261,6 +1320,7 @@ def run_task_turn_with_retry(server, state, task, prompt, model, effort,
     每次重试前先查预算，超预算直接停，不烧钱。"""
     max_retries = int(task.get("max_retries") or 0)
     result = None
+    usage = None
     for attempt in range(max_retries + 1):
         if attempt > 0:
             task["retry_count"] = attempt
@@ -1277,8 +1337,13 @@ def run_task_turn_with_retry(server, state, task, prompt, model, effort,
             break
         result = run_task_turn(server, state, task, prompt_try,
                                model, effort, sandbox)
+        # 各次尝试的真实用量都要算（失败尝试花的钱也是钱），
+        # 最后一次的结果里带汇总，不只带最后一次。
+        usage = _add_usage(usage, result.get("usage"))
         if result["status"] == "completed":
             break
+    if usage is not None and result is not None:
+        result = dict(result, usage=usage)
     return result
 
 
@@ -1670,7 +1735,8 @@ def main():
             print(json.dumps({"task": args.name, "thread_id": task["thread_id"],
                               "cwd": task["cwd"], "status": task["status"],
                               "text": result.get("text", ""),
-                              "error": result.get("error")},
+                              "error": result.get("error"),
+                              "usage": result.get("usage")},
                              indent=2, ensure_ascii=False))
         elif args.cmd == "list":
             for name, t in state["tasks"].items():
@@ -1847,6 +1913,7 @@ def main():
             save_state(args.state, state)
             ok = False
             last_verify = ""
+            goal_usage = None
             for round_no in range(1, args.max_rounds + 1):
                 if round_no == 1:
                     prompt = args.task
@@ -1860,6 +1927,7 @@ def main():
                               f"上一轮验收输出：\n{last_verify}")
                 result = run_task_turn(srv, state, task, prompt,
                                        args.model, args.effort, args.sandbox)
+                goal_usage = _add_usage(goal_usage, result.get("usage"))
                 v = subprocess.run(args.verify, shell=True, cwd=task["cwd"],
                                    capture_output=True, text=True)
                 last_verify = (v.stdout + v.stderr).strip()[:1000] or \
@@ -1875,6 +1943,14 @@ def main():
                                {"status": "goal-unmet", "text": last_verify[:200]})
             save_state(args.state, state)
             print(f"goal {'met' if ok else 'unmet'} after rounds")
+            if goal_usage:
+                print(f"usage input={goal_usage['input_tokens']} "
+                      f"cached={goal_usage['cached_input_tokens']} "
+                      f"cache_write={goal_usage['cache_write_input_tokens']} "
+                      f"output={goal_usage['output_tokens']} "
+                      f"reasoning={goal_usage['reasoning_output_tokens']} "
+                      f"total={goal_usage['total_tokens']} "
+                      f"completions={goal_usage['completions']}")
         elif args.cmd == "parallel":
             jobs = json.load(open(args.jobs))
             # 依赖 + 优先级：先拓扑排序，缺依赖/成环直接报错不跑；

tests/test_host.py

@@ -572,6 +572,85 @@ class GithubPRTest(unittest.TestCase):
             self.assertIn("remote", r.stdout + r.stderr)
 
 
+class TurnUsageSumTest(unittest.TestCase):
+    def _note(self, thread_id, turn_id, last):
+        return {"method": "thread/tokenUsage/updated",
+                "params": {"threadId": thread_id, "turnId": turn_id,
+                           "tokenUsage": {"last": last}}}
+
+    def test_sums_last_across_completions(self):
+        # 2026-10-26 中继实测前两次通知：last 之和等于第二次的 total。
+        evs = [
+            self._note("t1", "u1", {
+                "inputTokens": 10906, "cachedInputTokens": 9192,
+                "cacheWriteInputTokens": 0, "outputTokens": 346,
+                "reasoningOutputTokens": 256, "totalTokens": 11252}),
+            self._note("t1", "u1", {
+                "inputTokens": 11041, "cachedInputTokens": 9192,
+                "cacheWriteInputTokens": 0, "outputTokens": 230,
+                "reasoningOutputTokens": 161, "totalTokens": 11271}),
+        ]
+        self.assertEqual(host._sum_turn_usage(evs, "t1", "u1"), {
+            "input_tokens": 21947, "cached_input_tokens": 18384,
+            "cache_write_input_tokens": 0, "output_tokens": 576,
+            "reasoning_output_tokens": 417, "total_tokens": 22523,
+            "completions": 2})
+
+    def test_filters_by_thread_and_turn(self):
+        evs = [
+            self._note("t1", "u1", {
+                "inputTokens": 10, "cachedInputTokens": 0,
+                "cacheWriteInputTokens": 0, "outputTokens": 5,
+                "reasoningOutputTokens": 0, "totalTokens": 15}),
+            self._note("t1", "u2", {
+                "inputTokens": 999, "cachedInputTokens": 0,
+                "cacheWriteInputTokens": 0, "outputTokens": 999,
+                "reasoningOutputTokens": 0, "totalTokens": 1998}),
+            self._note("t2", "u1", {
+                "inputTokens": 999, "cachedInputTokens": 0,
+                "cacheWriteInputTokens": 0, "outputTokens": 999,
+                "reasoningOutputTokens": 0, "totalTokens": 1998}),
+            {"method": "item/completed", "params": {}},
+        ]
+        usage = host._sum_turn_usage(evs, "t1", "u1")
+        self.assertEqual(usage["input_tokens"], 10)
+        self.assertEqual(usage["completions"], 1)
+        self.assertIsNone(host._sum_turn_usage(evs, "t1", "u9"))
+        self.assertIsNone(host._sum_turn_usage([], "t1", "u1"))
+
+    def test_retry_sums_usage_across_attempts(self):
+        class FakeServer:
+            def __init__(self, results):
+                self.results = list(results)
+
+            def run_turn(self, thread_id, prompt, model=None, effort=None,
+                         sandbox=None):
+                return self.results.pop(0)
+
+        task = {"name": "t", "thread_id": "x", "sandbox": None,
+                "max_retries": 1, "retry_count": 0, "turns": []}
+        state = {"tasks": {"t": task}, "automations": []}
+        srv = FakeServer([
+            {"status": "failed", "text": "", "usage": {
+                "input_tokens": 100, "cached_input_tokens": 0,
+                "cache_write_input_tokens": 0, "output_tokens": 10,
+                "reasoning_output_tokens": 0, "total_tokens": 110,
+                "completions": 1}},
+            {"status": "completed", "text": "ok", "usage": {
+                "input_tokens": 200, "cached_input_tokens": 50,
+                "cache_write_input_tokens": 0, "output_tokens": 20,
+                "reasoning_output_tokens": 5, "total_tokens": 220,
+                "completions": 1}},
+        ])
+        result = host.run_task_turn_with_retry(
+            srv, state, task, "do it", "m", "low", cfg={})
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["usage"]["input_tokens"], 300)
+        self.assertEqual(result["usage"]["cached_input_tokens"], 50)
+        self.assertEqual(result["usage"]["total_tokens"], 330)
+        self.assertEqual(result["usage"]["completions"], 2)
+
+
 if __name__ == "__main__":
     unittest.main()