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

此前每任务只记一条 estimated（字符数/4，金额为空），注释称 host 协议不暴露真实 token 数。本轮实测推翻了该前提： - account/usage/read 在中继上是死的（要求 ChatGPT 账号登录），不是来源，已在 cost.py docstring 写明死胡同 - app-server 每上游补全发 thread/tokenUsage/updated，last 是该次补全的用量；中继 5 调用回合实测 last 之和等于最后 total （input/output/cached/reasoning 四路都对得上）实现：host.py run_turn 从事件窗口按 threadId+turnId 汇总用量，重试/goal 各轮累加，dispatch/run JSON 与 goal 用量行透传； sandbox 三路解析（dispatch JSON 解析改用完整 stdout，顺带修了长 text 下 failed 被误判 completed 的潜 bug）；worker 每任务记一条 actual（重试/fallback 用量累加进同一条），无用量时回落 estimated；cost.py 价目表（默认 OpenAI 挂牌价，env 可覆盖，无价模型 usd 记 None、报表进 unpriced_actual_tokens 而非 0 美元）， cost_records 加 cached_input_tokens 列（SQLite/PG 两支迁移）；报表与按天行新增 cached_input_tokens/unpriced_actual_tokens。测试：host 32、cloud 483 全过。
2026-10-26 18:41:27 +08:00
1 parent 353d6c1aa8
commit 63823fc2e3
@@ -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 协议
-  不暴露真实 token 数），fallback 不改变成本来源，只改变实际跑的
-  模型与 `actual_model`/`fallback_count` 的记录。
+  fallback 模型失败原因。成本来源与 fallback 无关（见下文“真实成本”）。
+- 真实成本（逐任务 actual）：host.py 在每回合结束时从 app-server
+  的 `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，单个失败只跳过，
@@ -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
-``estimate_tokens``) and are stored separately as ``source="estimated"``
-so estimates are never mixed into actual totals.
+available: host.py collects the app-server's thread/tokenUsage/updated
+events during each turn (one per upstream completion, summed over the
+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):
-        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 (
+    _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,
+                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(
-            """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.execute(self._COST_DDL)
        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",
-               fallback_count=0):
+               output_tokens=0, cached_input_tokens=0, usd_micros=None,
+               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,
-                    created_at)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                    cached_input_tokens, total_tokens, usd_micros, source,
+                    fallback_count, day, created_at)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                (task_id, project_id, model, input_tokens, output_tokens,
-                 total_tokens, usd_micros, source, int(fallback_count or 0),
-                 day, created_at),
+                 cached_input_tokens, total_tokens, usd_micros, source,
+                 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:
@@ -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, "
@@ -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,
    }


@@ -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()
@@ -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()
@@ -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()
@@ -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")
@@ -26,9 +26,12 @@
 4. 写回：server 的 tasks 表没有 result/error 列，所以执行结果
   的持久化分三处，各复用已有结构、不另起字段名：
   - tasks 表：最终状态 + updated_at（取消时为 interrupted）；
-   - cost.py CostStore：每轮实际执行记一条 source="estimated"
-     的成本记录（host 协议不暴露真实 token 数，估算绝不冒充
-     actual；total_tokens 由 CostStore.record 按 input+output 算）；
+   - cost.py CostStore：每轮实际执行记一条成本记录；执行器带回
+     上游真实用量（host.py 从 tokenUsage/updated 事件汇总）时记
+     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 的估算结构
-            # 记 estimated，绝不标 actual；fallback_count 复用同一记录；
-            # 模型记最后真正跑的那个（fallback 成功时就是 fallback 模型）。
-            # cost 记录共用一条连接，并发下必须持锁写，否则两个线程
-            # 共用同一 sqlite 连接会直接报错。
-            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)
+            # 成本：执行器带回上游真实用量时记 actual（input/output/
+            # cached 全是实测值，usd 按价目表算、无价模型记 None，
+            # 报表里进 unpriced_actual_tokens 而不是 0 美元）；
+            # 没带回用量时回落字符估算记 estimated，绝不冒充 actual。
+            # fallback_count 复用同一记录；模型记最后真正跑的那个
+            # （fallback 成功时就是 fallback 模型）。cost 记录共用
+            # 一条连接，并发下必须持锁写，否则两个线程共用同一
+            # sqlite 连接会直接报错。
+            if usage_total and (usage_total.get("total_tokens") or 0) > 0:
+                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"