diff --git a/cloud/README.md b/cloud/README.md index 22dc337..c439003 100644 --- a/cloud/README.md +++ b/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 协议 - 不暴露真实 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,单个失败只跳过, diff --git a/cloud/cost.py b/cloud/cost.py index 70cb869..c9a68a1 100644 --- a/cloud/cost.py +++ b/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 -``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: diff --git a/cloud/db.py b/cloud/db.py index 901f39c..1b75d74 100644 --- a/cloud/db.py +++ b/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, " diff --git a/cloud/sandbox.py b/cloud/sandbox.py index d6192ea..552b889 100644 --- a/cloud/sandbox.py +++ b/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, } diff --git a/cloud/server.py b/cloud/server.py index 13c9166..7d180af 100644 --- a/cloud/server.py +++ b/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() diff --git a/cloud/tests/test_models_cost.py b/cloud/tests/test_models_cost.py index 3b3f75c..0338156 100644 --- a/cloud/tests/test_models_cost.py +++ b/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() diff --git a/cloud/tests/test_queue_sandbox.py b/cloud/tests/test_queue_sandbox.py index 3535164..c7d6ac4 100644 --- a/cloud/tests/test_queue_sandbox.py +++ b/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() diff --git a/cloud/tests/test_worker.py b/cloud/tests/test_worker.py index 990ca70..86b62f8 100644 --- a/cloud/tests/test_worker.py +++ b/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") diff --git a/cloud/worker.py b/cloud/worker.py index 0c8e6ad..e18a0fd 100644 --- a/cloud/worker.py +++ b/cloud/worker.py @@ -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" diff --git a/host.py b/host.py index 2b6dfca..aa1150a 100644 --- a/host.py +++ b/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)) # 依赖 + 优先级:先拓扑排序,缺依赖/成环直接报错不跑; diff --git a/tests/test_host.py b/tests/test_host.py index c7eda55..be8b548 100644 --- a/tests/test_host.py +++ b/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()