feat: enhance token usage tracking with cache hit rate details

src/bub/builtin/codex_provider.py

@@ -365,22 +365,28 @@ def _terminal_chunk(self, *, response: Any) -> ChatCompletionChunk:
         })
 
 
-def _completion_usage_from_response_usage(usage: Any) -> dict[str, int]:
+def _completion_usage_from_response_usage(usage: Any) -> dict[str, Any]:
     payload = usage.model_dump(exclude_none=True) if hasattr(usage, "model_dump") else usage
     if not isinstance(payload, dict):
         payload = {
             "input_tokens": getattr(usage, "input_tokens", 0),
             "output_tokens": getattr(usage, "output_tokens", 0),
             "total_tokens": getattr(usage, "total_tokens", 0),
+            "input_tokens_details": getattr(usage, "input_tokens_details", None),
         }
     prompt_tokens = int(payload.get("input_tokens") or payload.get("prompt_tokens") or 0)
     completion_tokens = int(payload.get("output_tokens") or payload.get("completion_tokens") or 0)
     total_tokens = int(payload.get("total_tokens") or prompt_tokens + completion_tokens)
-    return {
+    completion_usage: dict[str, Any] = {
         "prompt_tokens": prompt_tokens,
         "completion_tokens": completion_tokens,
         "total_tokens": total_tokens,
     }
+    input_details = _mapping_from_value(payload.get("input_tokens_details"))
+    cached_tokens = input_details.get("cached_tokens")
+    if isinstance(cached_tokens, int) and not isinstance(cached_tokens, bool):
+        completion_usage["prompt_tokens_details"] = {"cached_tokens": cached_tokens}
+    return completion_usage
 
 
 def _string_attr(obj: Any, name: str) -> str:

src/bub/builtin/tape.py

@@ -4,7 +4,7 @@
 import hashlib
 import inspect
 import json
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Mapping
 from dataclasses import asdict, dataclass, field, replace
 from datetime import UTC, datetime
 from pathlib import Path
@@ -33,6 +33,7 @@ class TapeInfo:
     last_anchor: str | None
     entries_since_last_anchor: int
     last_token_usage: int | None
+    last_token_cache_hit_rate: float | None
 
 
 @dataclass(frozen=True)
@@ -77,20 +78,37 @@ async def info(self) -> TapeInfo:
             last_anchor = None
             entries_since_last_anchor = len(entries)
         last_token_usage: int | None = None
+        last_token_cache_hit_rate: float | None = None
         for entry in reversed(entries):
             if entry.kind == "event" and entry.payload.get("name") == "run":
-                with contextlib.suppress(AttributeError):
-                    token_usage = entry.payload.get("data", {}).get("usage", {}).get("total_tokens")
-                    if token_usage and isinstance(token_usage, int):
-                        last_token_usage = token_usage
-                        break
+                data = entry.payload.get("data")
+                usage = data.get("usage") if isinstance(data, Mapping) else None
+                if not isinstance(usage, Mapping):
+                    continue
+                token_usage = usage.get("total_tokens")
+                if not isinstance(token_usage, int) or isinstance(token_usage, bool):
+                    continue
+                last_token_usage = token_usage
+                prompt_tokens = usage.get("prompt_tokens")
+                prompt_details = usage.get("prompt_tokens_details")
+                cached_tokens = prompt_details.get("cached_tokens") if isinstance(prompt_details, Mapping) else None
+                if (
+                    isinstance(prompt_tokens, int)
+                    and not isinstance(prompt_tokens, bool)
+                    and prompt_tokens > 0
+                    and isinstance(cached_tokens, int)
+                    and not isinstance(cached_tokens, bool)
+                ):
+                    last_token_cache_hit_rate = cached_tokens / prompt_tokens
+                break
         return TapeInfo(
             name=self.name,
             entries=len(entries),
             anchors=len(anchors),
             last_anchor=str(last_anchor) if last_anchor else None,
             entries_since_last_anchor=entries_since_last_anchor,
             last_token_usage=last_token_usage,
+            last_token_cache_hit_rate=last_token_cache_hit_rate,
         )
 
     async def ensure_bootstrap_anchor(self) -> None:

src/bub/builtin/tools.py

@@ -273,13 +273,15 @@ def skill_describe(name: str, *, context: ToolContext) -> str:
 async def tape_info(context: ToolContext) -> str:
     """Get information about the current tape, such as number of entries and anchors."""
     info = await context.tape.info()
+    cache_hit_rate = f"{info.last_token_cache_hit_rate:.2%}" if info.last_token_cache_hit_rate is not None else "None"
     return (
         f"name: {info.name}\n"
         f"entries: {info.entries}\n"
         f"anchors: {info.anchors}\n"
         f"last_anchor: {info.last_anchor}\n"
         f"entries_since_last_anchor: {info.entries_since_last_anchor}\n"
-        f"last_token_usage: {info.last_token_usage}"
+        f"last_token_usage: {info.last_token_usage}\n"
+        f"last_token_cache_hit_rate: {cache_hit_rate}"
     )
 
 

tests/test_builtin_codex.py

@@ -256,7 +256,12 @@ async def _codex_response_events():
             id="resp_123",
             created_at=1,
             model="gpt-5-codex",
-            usage=SimpleNamespace(input_tokens=3, output_tokens=2, total_tokens=5),
+            usage=SimpleNamespace(
+                input_tokens=3,
+                output_tokens=2,
+                total_tokens=5,
+                input_tokens_details={"cached_tokens": 2},
+            ),
         ),
     )
 
@@ -279,6 +284,8 @@ async def test_codex_completion_stream_maps_response_events_to_completion_chunks
     assert chunks[-1].usage is not None
     assert chunks[-1].usage.prompt_tokens == 3
     assert chunks[-1].usage.completion_tokens == 2
+    assert chunks[-1].usage.prompt_tokens_details is not None
+    assert chunks[-1].usage.prompt_tokens_details.cached_tokens == 2
 
 
 async def _codex_tool_response_events():

tests/test_builtin_tape.py

@@ -32,3 +32,41 @@ async def test_tape_fork_binds_temporary_fork_store_to_scoped_tape(tmp_path: Pat
     assert isinstance(second_store, ForkTapeStore)
     assert second_store is not first_store
     assert [entry.payload["data"]["value"] for entry in parent.read("test-tape") or []] == [1]
+
+
+@pytest.mark.asyncio
+async def test_tape_info_reports_last_token_cache_hit_rate(tmp_path: Path) -> None:
+    tape = Tape(tmp_path, AsyncTapeStoreAdapter(InMemoryTapeStore()), TapeContext()).scoped("test-tape")
+    await tape.record_chat(
+        run_id="run-1",
+        system_prompt=None,
+        new_messages=[],
+        response_text=None,
+        usage={
+            "prompt_tokens": 80,
+            "completion_tokens": 20,
+            "total_tokens": 100,
+            "prompt_tokens_details": {"cached_tokens": 60},
+        },
+    )
+
+    info = await tape.info()
+
+    assert info.last_token_usage == 100
+    assert info.last_token_cache_hit_rate == 0.75
+
+
+@pytest.mark.asyncio
+async def test_tape_info_omits_cache_hit_rate_when_usage_has_no_cache_details(tmp_path: Path) -> None:
+    tape = Tape(tmp_path, AsyncTapeStoreAdapter(InMemoryTapeStore()), TapeContext()).scoped("test-tape")
+    await tape.record_chat(
+        run_id="run-1",
+        system_prompt=None,
+        new_messages=[],
+        response_text=None,
+        usage={"prompt_tokens": 80, "completion_tokens": 20, "total_tokens": 100},
+    )
+
+    info = await tape.info()
+
+    assert info.last_token_cache_hit_rate is None

tests/test_builtin_tools.py

@@ -21,6 +21,7 @@
     render_tools_prompt,
     resolve_tool_names,
     set_model,
+    tape_info,
 )
 from bub.errors import ErrorKind
 from bub.tape import AsyncTapeStoreAdapter, InMemoryTapeStore, TapeContext
@@ -36,6 +37,27 @@ def _python_shell(code: str) -> str:
     return f"{shlex.quote(sys.executable)} -c {shlex.quote(code)}"
 
 
+@pytest.mark.asyncio
+async def test_tape_info_formats_token_cache_hit_rate(tmp_path) -> None:
+    context = _tool_context(tmp_path)
+    await context.tape.record_chat(
+        run_id="run-1",
+        system_prompt=None,
+        new_messages=[],
+        response_text=None,
+        usage={
+            "prompt_tokens": 8,
+            "completion_tokens": 2,
+            "total_tokens": 10,
+            "prompt_tokens_details": {"cached_tokens": 3},
+        },
+    )
+
+    result = await tape_info.run(context=context)
+
+    assert "last_token_cache_hit_rate: 37.50%" in result
+
+
 def test_completion_tools_builds_any_llm_payload() -> None:
     parameters = {
         "type": "object",

website/src/content/docs/docs/tutorials/observability.mdx

@@ -54,6 +54,7 @@ anchors: 2
 last_anchor: session/start
 entries_since_last_anchor: 44
 last_token_usage: 7458
+last_token_cache_hit_rate: 72.50%
 ```
 
 ![A terminal screenshot showing `tape.info` after a Bub task run](/docs/observability/tape-info.png)
@@ -64,6 +65,7 @@ The fields are useful when the model starts behaving oddly:
 - `anchors` and `last_anchor` tell you whether the tape has a checkpoint for context reconstruction.
 - `entries_since_last_anchor` tells you whether a handoff could shorten the next prompt.
 - `last_token_usage` appears when token usage has been recorded by the model path.
+- `last_token_cache_hit_rate` shows the cached share of prompt tokens for that same model call when available.
 
 Because Bub uses the [tape model](/docs/concepts/tape-and-context/) from [tape.systems](https://tape.systems), the runtime can inspect its own operational record. Bub can answer questions about what happened because the tape is the same state it uses to rebuild context.
 

website/src/content/docs/docs/tutorials/tapestore-sqlalchemy.mdx

@@ -69,6 +69,7 @@ anchors: 1
 last_anchor: session/start
 entries_since_last_anchor: 8
 last_token_usage: 4106
+last_token_cache_hit_rate: 72.50%
 ```
 
 The exact tape name and counts depend on your workspace and model call, but the command should report a tape name, at least one anchor, and entries written after `session/start`.

website/src/content/docs/zh-cn/docs/tutorials/observability.mdx

@@ -54,6 +54,7 @@ anchors: 2
 last_anchor: session/start
 entries_since_last_anchor: 44
 last_token_usage: 7458
+last_token_cache_hit_rate: 72.50%
 ```
 
 ![展示 Bub task run 之后 `tape.info` 输出的终端截图](/docs/observability/tape-info.png)
@@ -64,6 +65,7 @@ last_token_usage: 7458
 - `anchors` 与 `last_anchor` 表示 tape 是否已有用于重建 context 的 checkpoint。
 - `entries_since_last_anchor` 表示是否可以通过 handoff 缩短下一次 prompt。
 - `last_token_usage` 会在模型路径记录 token usage 后出现。
+- `last_token_cache_hit_rate` 表示同一次模型调用中命中缓存的 prompt token 比例（provider 提供明细时可用）。
 
 由于 Bub 使用来自 [tape.systems](https://tape.systems) 的 [tape 模型](/zh-cn/docs/concepts/tape-and-context/)，运行时可以检查自己的操作记录。Bub 能回答发生了什么，是因为 tape 正是它重建 context 时使用的状态。
 

website/src/content/docs/zh-cn/docs/tutorials/tapestore-sqlalchemy.mdx

@@ -69,6 +69,7 @@ anchors: 1
 last_anchor: session/start
 entries_since_last_anchor: 8
 last_token_usage: 4106
+last_token_cache_hit_rate: 72.50%
 ```
 
 具体 tape name 和数量取决于你的 workspace 与模型调用，但命令应当输出 tape name、至少一个 anchor，以及 `session/start` 之后写入的 entries。