@@ -1215,6 +1215,12 @@ async def _run_inner(
                     recorder.kernel_process_id = kernel_proc.id
                 except Exception as _silent_e:
                     logger.debug("suppressed: %s", _silent_e, exc_info=False)
+                try:
+                    _tb = getattr(kernel_proc, "token_budget", None)
+                    if _tb:
+                        recorder.token_limit = int(_tb)
+                except (TypeError, ValueError):
+                    pass
                 # PR1–PR4: configure Rust loop_guard (max rounds / ban worker orch / crew cap)
                 try:
                     if bool(getattr(settings, "agent_loop_guard_enabled", True)):
@@ -1541,8 +1547,10 @@ async def _release_kernel_slot(
             try:
                 if getattr(self, "last_iterations", None):
                     recorder._iterations = int(self.last_iterations)
-                if kernel_proc is not None:
-                    recorder.set_token_used(int(getattr(kernel_proc, "tokens_used", 0) or 0))
+                # token_used is accrued per LLM round from provider usage
+                # (note_llm_round_usage). Do not copy kernel_proc.tokens_used:
+                # that snapshot stays 0 when token_budget is None or the
+                # client holds a stale copy — which made completed runs look free.
             except Exception as _silent_e:
                 logger.debug("suppressed: %s", _silent_e, exc_info=False)
             _exit = str(getattr(self, "last_exit_reason", "") or "")
@@ -1577,13 +1585,6 @@ async def _release_kernel_slot(
         except asyncio.CancelledError:
             # P0：CancelledError 是 BaseException；清理段用 shield 防二次 cancel 打断漏槽
             async def _cleanup_cancel() -> None:
-                try:
-                    if kernel_proc is not None:
-                        recorder.set_token_used(
-                            int(getattr(kernel_proc, "tokens_used", 0) or 0)
-                        )
-                except Exception as _silent_e:
-                    logger.debug("suppressed: %s", _silent_e, exc_info=False)
                 try:
                     await recorder.cancel("cancelled")
                 except Exception as _silent_e:
@@ -1599,13 +1600,6 @@ async def _cleanup_cancel() -> None:
             # P0 审计 N1：except 退出时 Python 会 del e；shield 内闭包若引用 e 会 NameError → 槽位泄漏
             err_msg = str(e)
             async def _cleanup_fail() -> None:
-                try:
-                    if kernel_proc is not None:
-                        recorder.set_token_used(
-                            int(getattr(kernel_proc, "tokens_used", 0) or 0)
-                        )
-                except Exception as _silent_e:
-                    logger.debug("suppressed: %s", _silent_e, exc_info=False)
                 try:
                     await recorder.finish_fail(err_msg)
                 except Exception as _silent_e:
 
@@ -509,6 +509,15 @@ async def _close_thinking_stream() -> None:
     except Exception as _cost_e:
         logger.warning("cost_charge skip: %s", _cost_e)
 
+    # agent_runs.token_used: provider usage only. Kernel charge is a separate
+    # budget snapshot and is often 0 (token_budget is None, or a stale copy).
+    try:
+        rec = getattr(loop, "_run_recorder", None)
+        if rec is not None and hasattr(rec, "note_llm_round_usage"):
+            rec.note_llm_round_usage(stream_usage if stream_usage else None)
+    except Exception as _tu_e:
+        logger.debug("run token_used skip: %s", _tu_e)
+
     if kernel_proc is not None and kernel_proc.token_budget is not None:
         from backend.kernel import BudgetExceededError, get_kernel
 
 
@@ -57,6 +57,7 @@ def __init__(
         self._tool_calls = 0
         self._iterations = 0
         self._token_used = 0
+        self._token_used_source: str | None = None
 
     # ─────────── 内部工具 ───────────
 
@@ -253,24 +254,100 @@ async def _flush_counters(self, repo: Any = None) -> None:
         try:
             if repo is None:
                 repo = await self._repo()
-            await repo.update_run(
-                self.run_id,
-                {
-                    "total_tool_calls": int(self._tool_calls or 0),
-                    "total_iterations": int(self._iterations or 0),
-                    "token_used": int(self._token_used or 0),
-                },
-            )
+            payload: dict[str, Any] = {
+                "total_tool_calls": int(self._tool_calls or 0),
+                "total_iterations": int(self._iterations or 0),
+                "token_used": int(self._token_used or 0),
+            }
+            if self._token_used_source:
+                payload["meta"] = self._token_meta()
+            await repo.update_run(self.run_id, payload)
         except Exception as e:
             logger.debug("RunRecorder._flush_counters: %s", e)
 
     def set_token_used(self, used: int) -> None:
-        """同步 kernel 进程已用 token（finish 时落库）。"""
+        """Legacy absolute setter. Does not override provider-accrued totals.
+
+        Prefer ``note_llm_round_usage`` — kernel snapshots are often 0
+        (charge skipped when token_budget is None; Rust client returns copies).
+        """
+        if self._token_used_source in ("provider", "partial"):
+            return
         try:
             self._token_used = max(0, int(used or 0))
         except (TypeError, ValueError):
             pass
 
+    def note_llm_round_usage(self, usage: dict[str, Any] | None) -> None:
+        """Accrue provider-reported tokens for this run.
+
+        Empty/missing usage is recorded as ``omitted`` rather than a silent 0
+        that looks like a free run. Estimates are not written here.
+        """
+        try:
+            from backend.services.llm.usage_normalize import finalize_usage
+
+            u = finalize_usage(dict(usage) if isinstance(usage, dict) else {})
+        except Exception:
+            u = dict(usage) if isinstance(usage, dict) else {}
+        try:
+            prompt = int(u.get("prompt_tokens") or 0)
+            completion = int(u.get("completion_tokens") or 0)
+            total = int(u.get("total_tokens") or 0)
+        except (TypeError, ValueError):
+            prompt, completion, total = 0, 0, 0
+        n = total if total > 0 else max(0, prompt + completion)
+        if n > 0:
+            self._token_used = int(self._token_used or 0) + n
+            if self._token_used_source == "omitted":
+                self._token_used_source = "partial"
+            else:
+                self._token_used_source = "provider"
+            return
+        if self._token_used_source is None:
+            self._token_used_source = "omitted"
+        elif self._token_used_source == "provider":
+            self._token_used_source = "partial"
+
+    def apply_process_ledger(self, process_id: str | None = None) -> None:
+        """Copy provider-real usage from usage_ledger.by_process onto this run.
+
+        Live packed runs already wrote prompt/completion into
+        ``~/.tevarn/data/usage_ledger.json`` while ``agent_runs.token_used``
+        stayed 0 (kernel snapshot copy). Only prompt+completion are used —
+        estimated rounds store ``tokens`` with prompt=completion=0.
+        """
+        pid = (process_id or self.kernel_process_id or "").strip()
+        if not pid or pid == "system":
+            return
+        try:
+            from backend.services.usage_ledger import process_cost
+
+            bucket = process_cost(pid)
+        except Exception:
+            return
+        if not bucket:
+            return
+        try:
+            prompt = int(bucket.get("prompt") or 0)
+            completion = int(bucket.get("completion") or 0)
+        except (TypeError, ValueError):
+            return
+        if prompt <= 0 and completion <= 0:
+            return
+        n = prompt + completion
+        self._token_used = max(int(self._token_used or 0), n)
+        if self._token_used_source in (None, "omitted"):
+            self._token_used_source = "provider"
+        elif self._token_used_source == "partial":
+            self._token_used_source = "provider"
+
+    def _token_meta(self) -> dict[str, Any]:
+        meta = dict(self.meta or {})
+        if self._token_used_source:
+            meta["token_used_source"] = self._token_used_source
+        return meta
+
     # ─────────── 终态 ───────────
 
     async def _finish(
@@ -291,13 +368,16 @@ async def _finish(
             return
         try:
             repo = await self._repo()
+            self.apply_process_ledger()
             data: dict[str, Any] = {
                 "status": dst.value,
                 "ended_at": _utcnow(),
                 "total_tool_calls": self._tool_calls,
                 "total_iterations": int(self._iterations or 0),
                 "token_used": int(self._token_used or 0),
             }
+            if self._token_used_source:
+                data["meta"] = self._token_meta()
             if self.token_limit:
                 data["token_limit"] = int(self.token_limit)
             if final_summary:
 
@@ -23,13 +23,12 @@
 # file_read（T3）：executor 已按行边界分页并给出续读 offset，是截断的权威。
 # 这里再来一刀 head+tail 拼接会把「第 1-30 行 …省略… 末尾几行」交给模型，
 # 而模型以为自己读到了完整文件 —— 基于断裂视图改代码是静默错改的主因。
-# audit-fix(#7)：预算 21_000 → 12_000（压 context 占用）；executor 单次分页上限
-# FILE_READ_MAX_CHARS=20_000 不变，超出 12k 的读文件结果走 head+tail 兜底，
-# 模型应按 executor 给出的 offset 续读而非假设已读全文。
+# 预算必须高于 FILE_READ_MAX_CHARS（含 footer 余量），否则 12k 窗会把一个
+# 模块切成 ~250 行的 offset+= 片（slice-thrash）。超大文件仍由 executor 分页。
 # Aggressive defaults (H2/P1): prefer spill handle over fat LLM context.
 # file_read keeps higher budget so executor pagination remains authority.
 TOOL_RESULT_BUDGET: dict[str, int] = {
-    "file_read": 12_000,
+    "file_read": 100_000,
     "grep": 900,
     "glob": 600,
     # cargo/rustc diagnostics are long; 1200 forced head+tail thrash on E0xxx lists
 
@@ -12,7 +12,9 @@
 import re
 import sqlite3
 import sys
+import threading
 import urllib.parse
+from collections import OrderedDict
 from html import unescape
 from pathlib import Path
 from typing import Any
@@ -1515,9 +1517,50 @@ async def _maybe_cargo_stub_retry(
 
 
 # file_read 分页参数（T3）
-FILE_READ_DEFAULT_LIMIT = 1000        # 单次默认行数
-FILE_READ_MAX_CHARS = 20_000          # 单次输出字符上限（按行边界收口）
+# Default window must fit a typical module in one call. A ~11k char cap
+# (old TOOL_RESULT_BUDGET["file_read"]=12k minus footer) stopped after
+# ~250 lines on Tevarn Python files and forced offset+= slice thrash
+# (dispatcher.py ×6, permission_court.py ×6 on a live 2026-08-13 run).
+FILE_READ_DEFAULT_LIMIT = 2000        # 单次默认行数（够读一个中等模块）
+FILE_READ_MAX_CHARS = 96_000          # 单次输出字符上限（按行边界收口）
 FILE_READ_MAX_LINE_CHARS = 2_000      # 单行过长时的截断长度
+FILE_READ_CACHE_MAX = 64              # identical (path, offset, limit, mtime, size) hits
+
+_file_read_cache: OrderedDict[tuple, str] = OrderedDict()
+_file_read_cache_lock = threading.Lock()
+
+
+def _file_read_cache_clear() -> None:
+    """Test helper: drop identical-read cache."""
+    with _file_read_cache_lock:
+        _file_read_cache.clear()
+
+
+def _file_read_cache_key(
+    full_path: str, offset: int, limit: int
+) -> tuple[Any, ...] | None:
+    try:
+        st = os.stat(full_path)
+    except OSError:
+        return None
+    mtime_ns = int(getattr(st, "st_mtime_ns", int(st.st_mtime * 1_000_000_000)))
+    return (os.path.normcase(full_path), int(offset), int(limit), mtime_ns, int(st.st_size))
+
+
+def _file_read_cache_get(key: tuple[Any, ...]) -> str | None:
+    with _file_read_cache_lock:
+        val = _file_read_cache.get(key)
+        if val is not None:
+            _file_read_cache.move_to_end(key)
+        return val
+
+
+def _file_read_cache_put(key: tuple[Any, ...], value: str) -> None:
+    with _file_read_cache_lock:
+        _file_read_cache[key] = value
+        _file_read_cache.move_to_end(key)
+        while len(_file_read_cache) > FILE_READ_CACHE_MAX:
+            _file_read_cache.popitem(last=False)
 
 
 def _file_read_char_budget() -> int:
@@ -1671,15 +1714,29 @@ async def execute_file_read(config: dict[str, Any], arguments: dict[str, Any]) -
         return f"[Error] Not a file: {filepath}"
 
     try:
+        cache_key = _file_read_cache_key(full_path, offset, limit)
+        if cache_key is not None:
+            hit = _file_read_cache_get(cache_key)
+            if hit is not None:
+                return hit
         # 阻塞 I/O 丢进线程，让同轮并发的 tool call 真正并行（T1b）
-        return await asyncio.to_thread(
+        out = await asyncio.to_thread(
             _read_file_paginated,
             full_path,
             filepath,
             offset,
             limit,
             _file_read_char_budget(),
         )
+        if (
+            cache_key is not None
+            and isinstance(out, str)
+            and not out.startswith("[Error]")
+            and not out.startswith("[Security")
+            and not out.startswith("[Blocked]")
+        ):
+            _file_read_cache_put(cache_key, out)
+        return out
     except Exception as e:
         return f"[Error] {e}"
 
 
@@ -566,6 +566,31 @@ def merge_cache_panels(
     }
 
 
+def process_cost(process_id: str | None) -> dict[str, int]:
+    """Provider-real usage for one kernel process (usage_ledger.by_process).
+
+    Empty / missing / the shared ``system`` bucket → {}. Callers must not
+    treat a missing bucket as 0-used.
+    """
+    pid = (process_id or "").strip()
+    if not pid or pid == "system":
+        return {}
+    try:
+        snap = snapshot_cost()
+    except Exception:
+        return {}
+    bucket = (snap.get("by_process") or {}).get(pid)
+    if not isinstance(bucket, dict):
+        return {}
+    out: dict[str, int] = {}
+    for k in ("prompt", "completion", "tokens", "billable", "cache_read", "llm_rounds"):
+        try:
+            out[k] = max(0, int(bucket.get(k) or 0))
+        except (TypeError, ValueError):
+            out[k] = 0
+    return out
+
+
 def reset_for_tests() -> None:
     """Clear in-memory state (tests only)."""
     global _state, _path