chore(dev): token-lean workflow tools -- read guard, bulk reader, transcript report

Measured on the main session: 90% of development tokens were tool I/O. Three
tools from the Portal pattern:
- read_guard.py: PreToolUse hook that denies a whole-file Read over 350 lines and
  points to a ranged read or the bulk reader (pipe-tested: deny, range, small).
- bulk_read.py: answers a question about files with the cheap chat model, citing
  path:line, under a spend ceiling.
- token_report.py: where a session's tokens go, from its transcript.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo
This commit is contained in:
Marius Ciepluch
2026-09-29 17:23:30 +02:00
co-authored by Claude Fable 5.1
parent 38af0a86c2
commit 27812499b5
3 changed files with 166 additions and 0 deletions
+62
View File
@@ -0,0 +1,62 @@
"""Bulk reader: answer a question about one or more files with the cheap chat model, citing line numbers.
The Portal pattern (Spotify, 2026-09): reading five files to answer one question is I/O, not reasoning, so a
cheap model reads and the expensive one gets the answer. Files are sent in numbered chunks; the answer must cite
`path:line`, which is what a summary otherwise lacks. Reasoning stays with the caller: this reports what the
files say, it does not decide anything.
Usage: .venv/bin/python benchmarks/dev/bulk_read.py "<question>" FILE [FILE ...] [--max-usd 0.05]
"""
from __future__ import annotations
import argparse
import sys
from pathlib import Path
CHUNK_LINES = 400
PROMPT = (
"Answer the question using only the files below (numbered lines). Be specific and brief; cite every claim as "
"path:line. If the files do not answer it, say so.\n\nQuestion: {question}\n\n{files}"
)
def numbered(path: Path) -> list[str]:
lines = path.read_text(errors="ignore").splitlines()
chunks = []
for start in range(0, max(len(lines), 1), CHUNK_LINES):
body = "\n".join(f"{n + 1:5d}| {line}" for n, line in enumerate(lines[start : start + CHUNK_LINES], start=start))
chunks.append(f"--- {path} (lines {start + 1}-{min(start + CHUNK_LINES, len(lines))})\n{body}")
return chunks
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("question")
parser.add_argument("files", nargs="+", type=Path)
parser.add_argument("--max-usd", type=float, default=0.05)
args = parser.parse_args()
from openultrasast.config import load_config
from openultrasast.model.endpoint import DEFAULT_DETECTOR_MODEL
from openultrasast import tool_hunter
load_config()
client = tool_hunter.resolve_hunter_client()
if client is None:
print("no chat client: set DEEPSEEK_API_KEY", file=sys.stderr)
return 2
chunks = [chunk for path in args.files for chunk in numbered(path)]
answers = []
for chunk in chunks:
if float(client.cost_usd()) >= args.max_usd:
answers.append(f"[stopped at the ${args.max_usd} ceiling before: {chunk.splitlines()[0]}]")
break
response = client.complete(model=DEFAULT_DETECTOR_MODEL, messages=[{"role": "user", "content": PROMPT.format(question=args.question, files=chunk)}], tools=[], timeout_seconds=90)
answers.append((response.content or "").strip())
print("\n\n".join(a for a in answers if a))
print(f"\n[bulk_read: {len(chunks)} chunk(s), ${float(client.cost_usd()):.4f}]", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+45
View File
@@ -0,0 +1,45 @@
#!/usr/bin/env python3
"""PreToolUse hook: a whole-file Read of more than LIMIT lines is denied and redirected.
Measured 2026-09-29: 90% of a development session's tokens were tool I/O -- inline scripts and what they
printed -- not reasoning. Whole-file reads are the largest single item that a summary could replace. A Read with
`offset`/`limit` is a deliberate range and passes; a file at or under LIMIT lines passes.
Wired in .claude/settings.local.json (PreToolUse, matcher Read). Stdin: the hook JSON; stdout: a permission
decision only when denying. Never fails the tool: any error here exits 0 silently.
"""
import json
import sys
LIMIT = 350
def main() -> int:
try:
event = json.load(sys.stdin)
except (json.JSONDecodeError, OSError):
return 0
if event.get("tool_name") != "Read":
return 0
tool_input = event.get("tool_input") or {}
path = tool_input.get("file_path")
if not path or tool_input.get("limit") or tool_input.get("offset"):
return 0
try:
with open(path, "rb") as handle:
lines = sum(1 for _ in handle)
except OSError:
return 0
if lines <= LIMIT:
return 0
reason = (
f"{path} has {lines} lines (limit {LIMIT} for a whole-file read). Read a range with offset/limit, or ask "
f"the bulk reader for what you need: .venv/bin/python benchmarks/dev/bulk_read.py '<question>' {path}"
)
print(json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason}}))
return 0
if __name__ == "__main__":
raise SystemExit(main())
+59
View File
@@ -0,0 +1,59 @@
"""Where a Claude Code session's tokens go: tool input vs tool output vs text, from the transcript.
The 2026-09-29 measurement of this project's main session: tool_use input 47.5% (inline scripts), tool results
42%, assistant text 6%, thinking 2%, user 2%. Re-run after a workflow change to see whether it moved.
Usage: python benchmarks/dev/token_report.py [TRANSCRIPT.jsonl] (default: the newest under ~/.claude/projects)
"""
from __future__ import annotations
import collections
import json
import sys
from pathlib import Path
def newest() -> Path:
candidates = sorted(Path.home().glob(".claude/projects/*/*.jsonl"), key=lambda p: p.stat().st_mtime)
if not candidates:
raise SystemExit("no transcript found")
return candidates[-1]
def main() -> int:
path = Path(sys.argv[1]) if len(sys.argv) > 1 else newest()
chars: collections.Counter[str] = collections.Counter()
tools: collections.Counter[str] = collections.Counter()
for line in path.open():
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
content = (entry.get("message") or {}).get("content")
kind = entry.get("type", "?")
if isinstance(content, str):
chars[kind] += len(content)
elif isinstance(content, list):
for block in content:
block_type = block.get("type")
if block_type == "text":
chars[f"{kind}:text"] += len(block.get("text", ""))
elif block_type == "tool_use":
tools[str(block.get("name"))] += 1
chars["tool_use_input"] += len(json.dumps(block.get("input")))
elif block_type == "tool_result":
result = block.get("content")
chars["tool_result"] += len(result if isinstance(result, str) else json.dumps(result))
elif block_type == "thinking":
chars["thinking"] += len(block.get("thinking", ""))
total = sum(chars.values()) or 1
print(f"{path.name}: {total / 1e6:.1f}M chars unique (~{total / 4e6:.2f}M tokens once; re-sent every turn)")
for key, value in chars.most_common():
print(f" {key:16s} {value / 1e6:6.2f}M {value / total:5.1%}")
print("tool calls:", sum(tools.values()), dict(tools.most_common(5)))
return 0
if __name__ == "__main__":
raise SystemExit(main())