diff --git a/benchmarks/dev/bulk_read.py b/benchmarks/dev/bulk_read.py new file mode 100644 index 0000000..04c3157 --- /dev/null +++ b/benchmarks/dev/bulk_read.py @@ -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 "" 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()) diff --git a/benchmarks/dev/read_guard.py b/benchmarks/dev/read_guard.py new file mode 100644 index 0000000..c6fa3cc --- /dev/null +++ b/benchmarks/dev/read_guard.py @@ -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 '' {path}" + ) + print(json.dumps({"hookSpecificOutput": {"hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason}})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/dev/token_report.py b/benchmarks/dev/token_report.py new file mode 100644 index 0000000..51bf6a2 --- /dev/null +++ b/benchmarks/dev/token_report.py @@ -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())