diff --git a/README.md b/README.md index 330d22a..0da3c29 100644 --- a/README.md +++ b/README.md @@ -175,13 +175,6 @@ npm run build # build dist/node.js npm run deploy:worker # wrangler deploy ``` -## Legacy - -The original Python proxy (the reference implementation we ported from) -lives in [`legacy/python/`](./legacy/python/). It's kept around as a -parity oracle and will be deleted once TS reaches end-to-end byte-output -parity. - ## License MIT. diff --git a/legacy/README.md b/legacy/README.md deleted file mode 100644 index 8f27ceb..0000000 --- a/legacy/README.md +++ /dev/null @@ -1,35 +0,0 @@ -# legacy/ - -Reference implementations kept around as a parity oracle during the -TypeScript port. Not part of the published npm package. - -## python/ - -The original working Python proxy (commit `5212f71`) plus its companion -scripts: - -| file | role | -| -------------------- | -------------------------------------------------------- | -| `proxy.py` | the proxy itself — `python3 proxy.py` to run | -| `gen_atlas.py` | pre-bakes Menlo glyph atlas (originally for the Zig CLI) | -| `diff_render.py` | byte-compare PNGs from two renderers | -| `diff_transform.py` | byte-compare transformed request bodies | -| `visual_compare.py` | side-by-side PNG diff viewer | -| `install.js` | old postinstall (verified Python + Pillow) | -| `cli.js` | old npm `bin` shim that spawned `proxy.py` | - -### Running the legacy proxy - -```bash -cd legacy/python -python3 proxy.py --port 47821 -``` - -Requires Python 3.8+ and `Pillow` + `httpx` (`pip install Pillow httpx`). - -### Why it's still here - -Until the TypeScript port reaches output parity (byte-for-byte -transformed requests, visually identical PNGs), this is the reference -implementation we diff against. Once parity is verified end-to-end, -the entire `legacy/` directory will be removed in a single commit. diff --git a/legacy/python/cli.js b/legacy/python/cli.js deleted file mode 100755 index ddf2d8d..0000000 --- a/legacy/python/cli.js +++ /dev/null @@ -1,170 +0,0 @@ -#!/usr/bin/env node -/** - * pixelpipe CLI - * - * Usage: - * npx pixelpipe # start on default port 47821 - * npx pixelpipe --port 9000 # custom port - * npx pixelpipe --no-compress # disable compression (passthrough) - * - * Then point Claude Code at it: - * ANTHROPIC_BASE_URL=http://127.0.0.1:47821 claude - */ -"use strict"; - -const path = require("path"); -const { spawn, spawnSync } = require("child_process"); -const fs = require("fs"); - -const PROXY_PY = path.join(__dirname, "..", "src", "proxy.py"); - -function parseArgs(argv) { - const opts = { - port: 47821, - compress: true, - tools: true, - schemas: true, - reminders: true, - fontSize: 5, - minChars: 2000, - help: false, - version: false, - }; - for (let i = 2; i < argv.length; i++) { - const a = argv[i]; - switch (a) { - case "-p": case "--port": opts.port = parseInt(argv[++i], 10); break; - case "--no-compress": opts.compress = false; break; - case "--no-tools": opts.tools = false; break; - case "--no-schemas": opts.schemas = false; break; - case "--no-reminders": opts.reminders = false; break; - case "--font-size": opts.fontSize = parseInt(argv[++i], 10); break; - case "--min-chars": opts.minChars = parseInt(argv[++i], 10); break; - case "-h": case "--help": opts.help = true; break; - case "-v": case "--version": opts.version = true; break; - default: - if (a.startsWith("--")) { - console.error(`Unknown option: ${a}`); - process.exit(2); - } - } - } - return opts; -} - -function printHelp() { - console.log(`pixelpipe — token-saving proxy for Claude Code - -Renders system prompt + tool definitions as bitmap images. Achieves 65-73% -token savings on Opus 4.7 with 100% reasoning quality preserved. - -USAGE - npx pixelpipe [options] - -OPTIONS - -p, --port Port to listen on (default: 47821) - --no-compress Disable all compression (pure passthrough) - --no-tools Don't compress tool descriptions - --no-schemas Don't compress tool input_schemas (saves most tokens) - --no-reminders Don't compress blocks - --font-size Render font size in pt (default: 5; <5 fails OCR) - --min-chars Minimum chars to trigger compression (default: 2000) - -h, --help Show this help - -v, --version Show version - -USAGE WITH CLAUDE CODE - Terminal 1: - npx pixelpipe - - Terminal 2: - ANTHROPIC_BASE_URL=http://127.0.0.1:47821 claude --exclude-dynamic-system-prompt-sections - - Tip: add the --exclude-dynamic-system-prompt-sections flag (or set - CLAUDE_CODE_EXCLUDE_DYNAMIC_SYSTEM_PROMPT_SECTIONS=1) so Claude Code's - cwd/git/env data is byte-stable across turns — this lets the rendered - image hit cache instead of being re-rendered every turn. - -REQUIREMENTS - Python 3.8+ with Pillow and httpx. Run \`npm install\` and the postinstall - script will install them automatically. - -VERIFIED SAVINGS (Opus 4.7, real Claude Code workflows) - - Simple call: 30% token reduction - - Coding task (3 turn): 43% token reduction - - Multi-tool (Grep/Glob/Read/Edit/Bash): 73% token reduction - - 10-turn session: 67% token reduction (≈ $1.18 saved per session) -`); -} - -function readPkgVersion() { - try { - return require(path.join(__dirname, "..", "package.json")).version; - } catch { return "unknown"; } -} - -function findPython() { - const candidates = ["python3", "python", "python3.12", "python3.11", "python3.10"]; - for (const c of candidates) { - const r = spawnSync(c, ["--version"], { stdio: ["ignore", "pipe", "pipe"] }); - if (r.status === 0) { - const ver = (r.stdout || r.stderr || Buffer.from("")).toString(); - const m = ver.match(/Python (\d+)\.(\d+)/); - if (m && parseInt(m[1], 10) >= 3 && parseInt(m[2], 10) >= 8) return c; - } - } - return null; -} - -function main() { - const opts = parseArgs(process.argv); - if (opts.help) { printHelp(); process.exit(0); } - if (opts.version) { console.log(readPkgVersion()); process.exit(0); } - - const py = findPython(); - if (!py) { - console.error("ERROR: Python 3.8+ not found on PATH. Please install Python 3."); - console.error(" macOS: brew install python"); - console.error(" Linux: apt install python3 python3-pip"); - process.exit(1); - } - - if (!fs.existsSync(PROXY_PY)) { - console.error(`ERROR: proxy script not found at ${PROXY_PY}`); - process.exit(1); - } - - const env = { ...process.env, - PORT: String(opts.port), - COMPRESS_SYSTEM: opts.compress ? "1" : "0", - COMPRESS_TOOLS: opts.compress && opts.tools ? "1" : "0", - COMPRESS_SCHEMAS: opts.compress && opts.schemas ? "1" : "0", - COMPRESS_REMINDERS: opts.compress && opts.reminders ? "1" : "0", - COMPRESS_TOOL_RESULTS: opts.compress ? "1" : "0", - PLACEMENT: "user", - FONT_SIZE: String(opts.fontSize), - MIN_COMPRESS_CHARS: String(opts.minChars), - }; - - console.log(`pixelpipe v${readPkgVersion()} starting...`); - console.log(` python: ${py}`); - console.log(` port: ${opts.port}`); - console.log(` compression: ${opts.compress ? "ON" : "OFF (passthrough)"}`); - if (opts.compress) { - console.log(` tools: ${opts.tools}`); - console.log(` schemas: ${opts.schemas}`); - console.log(` reminders: ${opts.reminders}`); - console.log(` font: Menlo ${opts.fontSize}pt`); - } - console.log(""); - console.log(` Point Claude Code at: ANTHROPIC_BASE_URL=http://127.0.0.1:${opts.port}`); - console.log(""); - - const child = spawn(py, [PROXY_PY], { stdio: "inherit", env }); - - const shutdown = () => { try { child.kill("SIGTERM"); } catch {} }; - process.on("SIGINT", shutdown); - process.on("SIGTERM", shutdown); - child.on("exit", (code) => process.exit(code || 0)); -} - -main(); diff --git a/legacy/python/diff_render.py b/legacy/python/diff_render.py deleted file mode 100644 index de47c41..0000000 --- a/legacy/python/diff_render.py +++ /dev/null @@ -1,136 +0,0 @@ -#!/usr/bin/env python3 -""" -Diff harness: compare Python render_chunks vs Rust render_chunks pixel-by-pixel. - -Strategy: - 1. For each sample text, call Python render_chunks() -> list[PNG bytes] - 2. Call Rust diff_render binary -> writes PNGs to disk - 3. Decode both sides' PNGs and compare pixel arrays - 4. Report: pass / fail with first diff coordinates - -Run: - python3 scripts/diff_render.py -""" - -import hashlib -import json -import subprocess -import sys -import tempfile -from pathlib import Path - -# Make src/proxy.py importable -ROOT = Path(__file__).resolve().parent.parent -sys.path.insert(0, str(ROOT / "src")) - -import proxy as py_proxy # noqa: E402 -from PIL import Image # noqa: E402 -import io # noqa: E402 - -RUST_BIN = ROOT / "src" / "rust" / "target" / "release" / "examples" / "diff_render" - -SAMPLES = { - "short_ascii": "hello world\n", - "multiline_code": ( - "def fibonacci(n):\n" - " if n <= 1: return n\n" - " return fibonacci(n-1) + fibonacci(n-2)\n" - "print(fibonacci(10))\n" - ), - "long_with_blanks": ( - "# Header\n\n" - + "\n".join(f"line {i}: " + "x" * (i % 60) for i in range(200)) - + "\n\n# Footer\n" - ), - "many_short_lines": "\n".join(f"L{i}" for i in range(500)) + "\n", - "wide_lines": "\n".join("." * 80 for _ in range(100)) + "\n", - "tools_like_json": json.dumps( - {"tools": [{"name": f"tool_{i}", "input_schema": {"type": "object", - "properties": {f"p{j}": {"type": "string"} for j in range(5)}}} - for i in range(20)]}, indent=2), -} - - -def png_to_pixels(png_bytes: bytes): - """Decode PNG to (width, height, raw_grayscale_bytes).""" - img = Image.open(io.BytesIO(png_bytes)) - img = img.convert("L") # force 8-bit grayscale - return img.width, img.height, img.tobytes() - - -def run_rust(samples: dict, out_dir: Path) -> dict: - """Invoke rust diff_render binary, returns name -> list[png_bytes].""" - spec = {name: text for name, text in samples.items()} - spec_path = out_dir / "spec.json" - spec_path.write_text(json.dumps(spec)) - res = subprocess.run( - [str(RUST_BIN), str(spec_path), str(out_dir)], - capture_output=True, text=True, - ) - if res.returncode != 0: - print("RUST FAILED:", res.stderr, file=sys.stderr) - sys.exit(2) - manifest = json.loads((out_dir / "manifest.json").read_text()) - out = {} - for name, files in manifest.items(): - out[name] = [(out_dir / f).read_bytes() for f in files] - return out - - -def diff_pixels(py_bytes: bytes, rs_bytes: bytes): - """Return None if pixel-identical, else (reason, details).""" - pw, ph, pp = png_to_pixels(py_bytes) - rw, rh, rp = png_to_pixels(rs_bytes) - if (pw, ph) != (rw, rh): - return ("dim_mismatch", f"py={pw}x{ph} rs={rw}x{rh}") - if pp == rp: - return None - # Find first diff - for i, (a, b) in enumerate(zip(pp, rp)): - if a != b: - x = i % pw - y = i // pw - return ("pixel_mismatch", f"first diff at ({x},{y}): py={a} rs={b}") - return ("byte_len_mismatch", f"py_len={len(pp)} rs_len={len(rp)}") - - -def main(): - if not RUST_BIN.exists(): - print(f"build rust first: cd src/rust && cargo build --release --example diff_render") - print(f"missing: {RUST_BIN}") - sys.exit(1) - - with tempfile.TemporaryDirectory() as td: - out_dir = Path(td) - rust_out = run_rust(SAMPLES, out_dir) - - total = 0 - fail = 0 - for name, text in SAMPLES.items(): - py_pngs = py_proxy.render_chunks(text) - rs_pngs = rust_out.get(name, []) - total += 1 - print(f"\n[{name}] py_chunks={len(py_pngs)} rs_chunks={len(rs_pngs)}") - if len(py_pngs) != len(rs_pngs): - fail += 1 - print(f" FAIL: chunk count differs") - continue - any_fail = False - for i, (a, b) in enumerate(zip(py_pngs, rs_pngs)): - ah = hashlib.sha256(a).hexdigest()[:12] - bh = hashlib.sha256(b).hexdigest()[:12] - diff = diff_pixels(a, b) - if diff is None: - print(f" chunk {i}: OK py={ah} rs={bh} bytes={len(a)}/{len(b)}") - else: - any_fail = True - print(f" chunk {i}: FAIL {diff[0]}: {diff[1]} py={ah} rs={bh}") - if any_fail: - fail += 1 - - print(f"\n=== {total - fail}/{total} samples pixel-identical ===") - sys.exit(0 if fail == 0 else 1) - - -if __name__ == "__main__": - main() diff --git a/legacy/python/diff_transform.py b/legacy/python/diff_transform.py deleted file mode 100644 index a3e9e27..0000000 --- a/legacy/python/diff_transform.py +++ /dev/null @@ -1,213 +0,0 @@ -#!/usr/bin/env python3 -""" -Diff harness at the TRANSFORM layer: synthesize a realistic Anthropic -/v1/messages body, run it through Python proxy.transform_request() AND -the Rust transform_only binary, then diff what each emits. - -This is the validation we need before switching :47821 to Rust. It -catches divergences in: - - Field structure (system field rewriting, cache_control placement) - - Image block emission (number of imgs, dims) - - Tools/schemas/reminders compression behavior - - JSON serialization order (Anthropic cache hashes on bytes) - -Run: - python3 scripts/diff_transform.py -""" - -import io -import json -import subprocess -import sys -from pathlib import Path - -ROOT = Path(__file__).resolve().parent.parent -sys.path.insert(0, str(ROOT / "src")) - -import proxy as py_proxy # noqa: E402 - -RUST_BIN = ROOT / "src" / "rust" / "target" / "release" / "examples" / "transform_only" - - -def synthesize_body() -> bytes: - """Build a realistic Claude Code-shaped /v1/messages body. - - Shape based on what real Anthropic API requests look like: - - long `system` field (system prompt content) - - `tools` array with name/description/input_schema each - - `messages` array - - `max_tokens`, `model`, etc. - """ - # Real-ish system prompt: load CLAUDE.md + some extra padding to clear MIN_COMPRESS_CHARS=2000 - sysprompt_parts = [] - try: - sysprompt_parts.append((ROOT / "CLAUDE.md").read_text()) - except Exception: - pass - try: - sysprompt_parts.append((ROOT / "HANDOFF.md").read_text()) - except Exception: - pass - sysprompt_parts.append( - "\n## Additional padding to ensure >2000 chars\n" - + "Line of fake system instructions blah blah blah.\n" * 30 - ) - sysprompt = "\n\n".join(sysprompt_parts) - - body = { - "model": "claude-opus-4-5-20250929", - "max_tokens": 8192, - "system": [ - {"type": "text", "text": sysprompt}, - ], - "tools": [ - { - "name": "Bash", - "description": "Execute a bash command with optional timeout.\n\nThe command runs in a fresh subshell. Use this for git, builds, file ops.", - "input_schema": { - "type": "object", - "properties": { - "command": {"type": "string", "description": "The bash command to execute."}, - "timeout": {"type": "integer", "description": "Timeout in milliseconds. Defaults to 120000."}, - "description": {"type": "string", "description": "A short description of what the command does."}, - }, - "required": ["command"], - }, - }, - { - "name": "Read", - "description": "Read a file from disk.\n\nReturns up to 2000 lines starting from offset. Long lines truncated at 2000 chars.", - "input_schema": { - "type": "object", - "properties": { - "file_path": {"type": "string", "description": "Absolute path to the file."}, - "offset": {"type": "integer", "description": "1-indexed line to start at."}, - "limit": {"type": "integer", "description": "Max number of lines to read."}, - }, - "required": ["file_path"], - }, - }, - { - "name": "Edit", - "description": "Apply an exact-string find/replace in a file.\n\nold_string must match exactly including whitespace. Use replace_all for global.", - "input_schema": { - "type": "object", - "properties": { - "file_path": {"type": "string"}, - "old_string": {"type": "string"}, - "new_string": {"type": "string"}, - "replace_all": {"type": "boolean"}, - }, - "required": ["file_path", "old_string", "new_string"], - }, - }, - ], - "messages": [ - { - "role": "user", - "content": "Run `git status` and tell me what's modified.", - }, - ], - } - return json.dumps(body).encode() - - -def deep_compare(a, b, path=""): - """Walk two JSON structures and yield (path, kind, py_repr, rs_repr) diffs.""" - if type(a) is not type(b): - yield (path, "type", type(a).__name__, type(b).__name__) - return - if isinstance(a, dict): - keys_only_py = set(a) - set(b) - keys_only_rs = set(b) - set(a) - for k in keys_only_py: - yield (f"{path}.{k}", "missing_in_rs", short(a[k]), "") - for k in keys_only_rs: - yield (f"{path}.{k}", "missing_in_py", "", short(b[k])) - for k in set(a) & set(b): - yield from deep_compare(a[k], b[k], f"{path}.{k}") - return - if isinstance(a, list): - if len(a) != len(b): - yield (path, "list_len", str(len(a)), str(len(b))) - for i, (x, y) in enumerate(zip(a, b)): - yield from deep_compare(x, y, f"{path}[{i}]") - return - if a != b: - yield (path, "value", short(a), short(b)) - - -def short(v) -> str: - s = json.dumps(v) if not isinstance(v, str) else v - if len(s) > 80: - return s[:77] + "..." - return s - - -def main(): - if not RUST_BIN.exists(): - print(f"build rust first: cd src/rust && cargo build --release --example transform_only") - sys.exit(1) - - body = synthesize_body() - print(f"input body: {len(body)} bytes") - - # Python transform - py_out, py_info = py_proxy.transform_request(body) - print(f"python transform: out={len(py_out)}B info_keys={sorted(py_info.keys())}") - - # Rust transform via subprocess - rust_dir = Path("/tmp/cip_xform_rs") - rust_dir.mkdir(exist_ok=True) - body_path = rust_dir / "body_in.json" - body_path.write_bytes(body) - r = subprocess.run( - [str(RUST_BIN), str(body_path), str(rust_dir)], - capture_output=True, text=True, - ) - if r.returncode != 0: - print("RUST FAILED:", r.stderr); sys.exit(2) - print("rust transform stderr:", r.stderr.strip()) - rs_out = (rust_dir / "body_out.json").read_bytes() - rs_info = json.loads((rust_dir / "info.json").read_text()) - - print(f"rust transform: out={len(rs_out)}B info_keys={sorted(rs_info.keys())}") - - print(f"\n=== body bytes: py={len(py_out)} rs={len(rs_out)} delta={len(rs_out)-len(py_out):+d} ===") - - # Decode and structural diff - try: - py_json = json.loads(py_out) - rs_json = json.loads(rs_out) - except json.JSONDecodeError as e: - print(f"FAIL: invalid JSON output: {e}") - sys.exit(3) - - diffs = list(deep_compare(py_json, rs_json, "$")) - if not diffs: - print("body JSON is STRUCTURALLY IDENTICAL") - else: - print(f"\nfound {len(diffs)} structural diffs (showing first 30):") - for path, kind, pv, rv in diffs[:30]: - print(f" [{kind:18s}] {path}") - print(f" py: {pv}") - print(f" rs: {rv}") - if len(diffs) > 30: - print(f" ... and {len(diffs)-30} more") - - # Info side-by-side - print("\n=== TransformInfo comparison ===") - common = sorted(set(py_info.keys()) | set(rs_info.keys())) - print(f"{'field':30s} {'python':>14s} {'rust':>14s}") - for k in common: - pv = py_info.get(k, "") - rv = rs_info.get(k, "") - if isinstance(pv, list): pv = f"list[{len(pv)}]" - if isinstance(rv, list): rv = f"list[{len(rv)}]" - print(f"{k:30s} {str(pv):>14s} {str(rv):>14s}") - - sys.exit(0 if not diffs else 1) - - -if __name__ == "__main__": - main() diff --git a/legacy/python/gen_atlas.py b/legacy/python/gen_atlas.py deleted file mode 100644 index b3ccbdc..0000000 --- a/legacy/python/gen_atlas.py +++ /dev/null @@ -1,75 +0,0 @@ -"""Pre-render Menlo 5pt glyph atlas + metrics for Zig consumption. - -Format (binary, little-endian): - Header (16 bytes): - magic 4 "MNAT" - version 2 u16 = 1 - cell_w 2 u16 max glyph width - cell_h 2 u16 cell height (asc+desc) - asc 2 u16 - desc 2 u16 - first_char 1 u8 = 32 (space) - last_char 1 u8 = 126 (~) - Per-glyph (N glyphs = last-first+1): - advance_px 2 u16 - -- packed 1-bit bitmap, cell_w*cell_h bits, row-major -""" -import struct -from PIL import Image, ImageDraw, ImageFont - -FONT_PATH = "/System/Library/Fonts/Menlo.ttc" -FONT_SIZE = 5 -FIRST = 32 # space -LAST = 126 # ~ - -font = ImageFont.truetype(FONT_PATH, FONT_SIZE) -asc, desc = font.getmetrics() -cell_h = asc + desc - -# Find max glyph width across our range -max_w = 0 -for code in range(FIRST, LAST + 1): - ch = chr(code) - w = int(font.getlength(ch)) + 1 - if w > max_w: - max_w = w - -cell_w = max_w -print(f"Atlas: cell={cell_w}x{cell_h}, asc={asc}, desc={desc}, chars={LAST-FIRST+1}") - -with open("/tmp/menlo5_atlas.bin", "wb") as f: - # Header - f.write(b"MNAT") - f.write(struct.pack(" any non-white -> black) - pix = img.load() - f.write(struct.pack("= 3 && parseInt(m[2], 10) >= 8) return c; - } - } - return null; -} - -function hasModule(py, mod) { - const r = spawnSync(py, ["-c", `import ${mod}`], { stdio: ["ignore", "pipe", "pipe"] }); - return r.status === 0; -} - -const py = findPython(); -if (!py) { - console.warn("\n[pixelpipe] WARNING: Python 3.8+ not found."); - console.warn(" Install Python 3 (https://www.python.org/downloads/) then run:"); - console.warn(" python3 -m pip install Pillow httpx"); - console.warn(" Otherwise the proxy will fail to start.\n"); - process.exit(0); // don't break npm install -} - -const need = []; -if (!hasModule(py, "PIL")) need.push("Pillow"); -if (!hasModule(py, "httpx")) need.push("httpx"); -if (need.length === 0) { - console.log("[pixelpipe] Python deps already installed."); - process.exit(0); -} - -console.log(`[pixelpipe] Installing Python deps: ${need.join(", ")}`); -const args = ["-m", "pip", "install", "--quiet", "--user", ...need]; -const r = spawnSync(py, args, { stdio: "inherit" }); -if (r.status !== 0) { - console.warn("\n[pixelpipe] WARNING: failed to auto-install Python deps."); - console.warn(` Please run manually: ${py} -m pip install ${need.join(" ")}`); -} diff --git a/legacy/python/proxy.py b/legacy/python/proxy.py deleted file mode 100644 index ba25e62..0000000 --- a/legacy/python/proxy.py +++ /dev/null @@ -1,975 +0,0 @@ -""" -Experimental Python proxy for testing system-prompt-as-image compression. - -Sits between Claude Code and api.anthropic.com. Forwards Claude Code's own auth -(OAuth bearer + headers) untouched. When COMPRESS_SYSTEM is set, intercepts the -request body and rewrites the system prompt as image content blocks rendered -with a popular terminal font (Menlo) at small pt-size so Anthropic's vision -encoder can OCR it. Logs the full token breakdown (input / output / cache_read / -cache_create) so we can measure the actual win/loss vs plain text. - -Usage: - PORT=47821 COMPRESS_SYSTEM=1 python3 proxy_py.py - # in another shell: - ANTHROPIC_BASE_URL=http://127.0.0.1:47821 claude -p "..." -""" -from __future__ import annotations -import base64, io, json, os, sys, threading, traceback -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -import httpx -from PIL import Image, ImageDraw, ImageFont - -PORT = int(os.environ.get("PORT", "47821")) -COMPRESS = os.environ.get("COMPRESS_SYSTEM", "0") == "1" -FONT_PATH = os.environ.get("FONT_PATH", "/System/Library/Fonts/Menlo.ttc") -FONT_SIZE = int(os.environ.get("FONT_SIZE", "5")) # 5pt verified OCR=99.7% on Opus 4.7 -MAX_EDGE = 1568 # Anthropic resizes larger images, destroying tiny-font OCR -MIN_COMPRESS_CHARS = int(os.environ.get("MIN_COMPRESS_CHARS", "2000")) -UPSTREAM = "https://api.anthropic.com" - -# Memoize rendered images so the SAME system prompt across turns produces -# byte-identical PNGs (PIL's PNG encoder has tiny non-determinism that breaks -# Anthropic's prompt cache lookup). -import hashlib -_render_cache: dict[str, list[bytes]] = {} - -# Inject COMPRESS_SYSTEM as default value for compress_mode placement -# 'system' : put images in the system field (may not be supported by API) -# 'user' : prepend image to first user message (cache_control on the image) -# 'replace_system' : replace system with an empty text + add image to first user msg -PLACEMENT = os.environ.get("PLACEMENT", "user") - -_font_cache: dict[int, ImageFont.FreeTypeFont] = {} - -def get_font(size: int) -> ImageFont.FreeTypeFont: - if size not in _font_cache: - _font_cache[size] = ImageFont.truetype(FONT_PATH, size) - return _font_cache[size] - - -_render_dims_cache: dict[str, list[tuple[int, int]]] = {} - -def render_chunks(text: str, font_size: int = FONT_SIZE) -> list[bytes]: - """Render text into the MINIMUM number of MAXIMALLY-PACKED images. - - Layout strategy: NEWSPAPER (multi-column). For 26K-char system prompt - rendered at 5pt, a single-column layout would be ~3600 px tall (exceeds - 1568 cap → forces multiple images + per-image overhead). Multi-column - packs the same text into a SQUAREish image: width = N_cols × col_width, - height ≤ 1568. One image avoids per-image overhead doubling. - - Memoized on text content so the same prompt across turns yields BYTE- - IDENTICAL PNG (PIL's PNG encoder has tiny non-determinism that breaks - Anthropic's prompt cache hash if not memoized). - """ - key = hashlib.sha256(f"{font_size}:{text}".encode()).hexdigest() - if key in _render_cache: - return _render_cache[key] - - font = get_font(font_size) - asc, desc = font.getmetrics() - line_h = asc + desc - char_w = font.getlength("M") or (font_size * 0.6) - col_gap_px = 4 # tighter gap — every pixel matters - edge = MAX_EDGE - - # Minify: kill trailing whitespace + collapse runs of blank lines. - # Critical because system prompt has lots of blank lines that waste rows. - raw = [] - last_blank = False - for ln in text.split("\n"): - ln = ln.rstrip() - if ln == "": - if last_blank: - continue - last_blank = True - else: - last_blank = False - raw.append(ln if ln else " ") - - # Force aggressive wrapping at FIXED_COL_CHARS. Narrower cols pack better: - # short lines (bullets, headers) waste the unused chars in their row. - FIXED_COL_CHARS = int(os.environ.get("COL_CHARS", "80")) - lines: list[str] = [] - for ln in raw: - if len(ln) <= FIXED_COL_CHARS: - lines.append(ln) - else: - for i in range(0, len(ln), FIXED_COL_CHARS): - lines.append(ln[i:i + FIXED_COL_CHARS]) - - col_w_px = int(FIXED_COL_CHARS * char_w) + 1 - lines_per_col = max(8, edge // line_h) - n_cols_total = max(1, (len(lines) + lines_per_col - 1) // lines_per_col) - img_w = n_cols_total * col_w_px + (n_cols_total - 1) * col_gap_px - pngs: list[bytes] = [] - - dims: list[tuple[int, int]] = [] - max_cols_per_img = max(1, edge // (col_w_px + col_gap_px)) - lines_per_img = lines_per_col * max_cols_per_img - - for s in range(0, len(lines), lines_per_img): - chunk = lines[s:s + lines_per_img] - c_needed = max(1, (len(chunk) + lines_per_col - 1) // lines_per_col) - # Width: ONLY as wide as the cols actually used in this image - chunk_img_w = c_needed * col_w_px + (c_needed - 1) * col_gap_px - # Height: ONLY as tall as the tallest column in this image - max_col_lines = min(lines_per_col, len(chunk) - (c_needed - 1) * lines_per_col) - if c_needed > 1: - # Earlier cols are full, last col may be partial - tallest = lines_per_col if c_needed > 1 else len(chunk) - else: - tallest = len(chunk) - # All non-last columns are full lines_per_col; last col has the remainder. - # Single-column case → tight bound (just last_col_lines, no canvas pad). - # Multi-column case → all earlier cols are full lines_per_col tall, so - # the image must be lines_per_col rows tall to fit them. - last_col_lines = len(chunk) - (c_needed - 1) * lines_per_col - tallest = max(1, last_col_lines) if c_needed == 1 else lines_per_col - chunk_img_h = line_h * tallest - img = Image.new("L", (chunk_img_w, chunk_img_h), 255) - d = ImageDraw.Draw(img) - for c in range(c_needed): - col_lines = chunk[c * lines_per_col:(c + 1) * lines_per_col] - x = c * (col_w_px + col_gap_px) - for i, ln in enumerate(col_lines): - d.text((x, i * line_h - desc // 2), ln, fill=0, font=font) - buf = io.BytesIO() - # Save as 8-bit grayscale PNG so the rendered image has full contrast - # AND preserves antialiasing. Previously used `P + ADAPTIVE colors=2` - # which picked white+gray instead of white+black on AA-heavy content, - # producing washed-out images that visually look broken (and OCR worse). - # Grayscale PNG bytes are still deterministic → Anthropic cache still hits. - img.save(buf, "PNG", optimize=True) - pngs.append(buf.getvalue()) - dims.append((chunk_img_w, chunk_img_h)) - - _render_cache[key] = pngs - _render_dims_cache[key] = dims - return pngs - - -def render_dims(text: str, font_size: int = FONT_SIZE) -> list[tuple[int, int]]: - key = hashlib.sha256(f"{font_size}:{text}".encode()).hexdigest() - if key not in _render_dims_cache: - render_chunks(text, font_size) - return _render_dims_cache.get(key, []) - - -def image_block(png_bytes: bytes, cache: bool = False) -> dict: - blk = { - "type": "image", - "source": { - "type": "base64", - "media_type": "image/png", - "data": base64.standard_b64encode(png_bytes).decode("ascii"), - }, - } - if cache: - # Match Claude Code's extended cache TTL. If we use ephemeral (5m) here - # while CC uses 1h later in the request, Anthropic rejects with - # "ttl='1h' cache_control must not come after ttl='5m'". - blk["cache_control"] = {"type": "ephemeral", "ttl": "1h"} - return blk - - -def extract_system_text(system_field) -> tuple[str, list[dict] | str | None]: - """Return (extracted_text, remainder). - remainder is what should stay in the system field (or None to delete it).""" - if system_field is None: - return "", None - if isinstance(system_field, str): - return system_field, "" # replace with empty string - if isinstance(system_field, list): - # Concatenate all text blocks; keep non-text blocks in place. - text_parts = [] - kept = [] - for b in system_field: - if isinstance(b, dict) and b.get("type") == "text": - text_parts.append(b.get("text", "")) - else: - kept.append(b) - return "\n\n".join(text_parts), kept - return "", system_field - - -def transform_request(body: bytes) -> tuple[bytes, dict]: - """Compress system prompt + tool descriptions to image content.""" - info = {"compressed": False} - try: - req = json.loads(body) - except Exception as e: - info["parse_error"] = str(e) - return body, info - - sys_field = req.get("system") - text, remainder = extract_system_text(sys_field) - - # Strip Claude Code's per-turn-random `x-anthropic-billing-header: ...; cch=;` - # line so the rest renders byte-identical across turns. - billing_line_kept = None - lines = text.split("\n", 1) - if lines and lines[0].startswith("x-anthropic-billing-header:"): - billing_line_kept = lines[0] - text = lines[1] if len(lines) > 1 else "" - - # COMPRESS_TOOLS: move tool descriptions (+ optionally schemas) into the - # same image. Replace each tool's description with a tiny stub. - compress_tools = os.environ.get("COMPRESS_TOOLS", "1") == "1" - compress_schemas = os.environ.get("COMPRESS_SCHEMAS", "1") == "1" - tool_text_added = 0 - schemas_compressed = 0 - if compress_tools and isinstance(req.get("tools"), list): - tool_doc_blocks = [] - for t in req["tools"]: - if not isinstance(t, dict): - continue - name = t.get("name", "?") - desc = t.get("description", "") - doc_part = "" - if desc and len(desc) > 80: - doc_part = f"### Tool: {name}\n{desc}" - t["description"] = f"See `{name}` docs in system context image." - elif desc: - doc_part = f"### Tool: {name}\n{desc}" - # Compress input_schema: serialize full schema to text inside the - # image, replace tools[].input_schema with a permissive stub. The - # model still emits well-formed tool_use because it sees the real - # schema in the image; Anthropic accepts any JSON object against - # the {"type":"object"} placeholder. - if compress_schemas and isinstance(t.get("input_schema"), dict): - schema_json = json.dumps(t["input_schema"], indent=1) - if len(schema_json) > 200: # only worth it for non-trivial schemas - doc_part += f"\n#### Schema for `{name}`\n```json\n{schema_json}\n```" - t["input_schema"] = {"type": "object"} - schemas_compressed += 1 - if doc_part: - tool_doc_blocks.append(doc_part) - if tool_doc_blocks: - tool_section = "\n\n# Tool Documentation\n\n" + "\n\n".join(tool_doc_blocks) - text = text + tool_section - tool_text_added = len(tool_section) - info["tool_text_added"] = tool_text_added - info["schemas_compressed"] = schemas_compressed - - info["system_text_chars"] = len(text) - info["system_text_sha8"] = hashlib.sha256(text.encode()).hexdigest()[:8] - # Dump every system text we see to /tmp so we can diff what changes. - if os.environ.get("DUMP_SYSTEM") == "1": - n = 0 - while os.path.exists(f"/tmp/sys_dump_{n}.txt"): - n += 1 - with open(f"/tmp/sys_dump_{n}.txt", "w") as f: - f.write(text) - - if len(text) < MIN_COMPRESS_CHARS: - info["skipped"] = f"system <{MIN_COMPRESS_CHARS} chars" - return body, info - - pngs = render_chunks(text) - dims = render_dims(text) - info["png_sha8"] = [hashlib.sha256(p).hexdigest()[:8] for p in pngs] - info["dims"] = dims - info["total_pixels"] = sum(w * h for w, h in dims) - info["expected_image_tokens"] = sum(max(1, (w * h) // 750) for w, h in dims) + 85 * len(dims) - info["text_tokens_estimate"] = len(text) // 4 - info["images"] = len(pngs) - info["png_bytes"] = sum(len(p) for p in pngs) - - # Stash the FIRST image for the dashboard preview (so users can SEE what - # we're sending). Cropped to a reasonable size so the page doesn't blow up. - if pngs: - with _session_stats["lock"]: - _session_stats["latest_png_bytes"] = pngs[0] - w0, h0 = dims[0] if dims else (0, 0) - _session_stats["latest_png_meta"] = ( - f"{len(pngs)} image(s), {w0}×{h0}px, " - f"{sum(len(p) for p in pngs)} bytes total — " - f"compressed {info['system_text_chars']:,} chars " - f"({info.get('text_tokens_estimate',0):,}→~{info['expected_image_tokens']:,} tokens)" - ) - - # Anthropic caps cache_control breakpoints at 4 per request. The model's - # cache lookup matches the LONGEST cached prefix, so we only need ONE - # well-placed breakpoint — on the last image of the system+tools render. - # Reminders and tool_result images get NO cache_control (they're still - # in the cached prefix by virtue of position). - image_blocks = [image_block(p, cache=(i == len(pngs)-1)) - for i, p in enumerate(pngs)] - info["cc_breakpoints_added"] = 1 - - msgs = req.get("messages", []) - if PLACEMENT in ("user", "replace_system"): - # Prepend image blocks to first user message as a "context" frame - prefix_text = ("[Context (rendered as image for token efficiency, " - "OCR carefully and treat as authoritative system instructions):]") - first_idx = None - for i, m in enumerate(msgs): - if m.get("role") == "user": - first_idx = i - break - if first_idx is None: - info["skipped"] = "no user message to attach image to" - return body, info - - target = msgs[first_idx] - content = target.get("content", "") - if isinstance(content, str): - content = [{"type": "text", "text": content}] - elif not isinstance(content, list): - content = [{"type": "text", "text": str(content)}] - - # Compress CC-injected blocks in first user message. - # These are large static blobs (skill listings, project context) that - # CC injects every turn — perfect cache candidates as images. - # Targeted: ONLY compress blocks that start with , - # leaving the user's actual prompt (and any other content) as text. - compress_reminders = os.environ.get("COMPRESS_REMINDERS", "1") == "1" - reminder_imgs_added = 0 - if compress_reminders and isinstance(content, list): - new_content_parts = [] - for blk in content: - txt = blk.get("text", "") if isinstance(blk, dict) and blk.get("type") == "text" else "" - # Heuristic: long system-reminder-style blocks. Cover both the - # opening tag style and large generic system blocks. - is_reminder = txt.lstrip().startswith("") and len(txt) > 1000 - if is_reminder: - extra_pngs = render_chunks(txt) - # NO cache_control: Anthropic caps at 4 breakpoints; the - # system+tools image already anchors the cacheable prefix. - for p in extra_pngs: - new_content_parts.append(image_block(p, cache=False)) - reminder_imgs_added += 1 - else: - new_content_parts.append(blk) - content = new_content_parts - info["reminder_imgs"] = reminder_imgs_added - - new_content = ( - [{"type": "text", "text": prefix_text}] - + image_blocks - + [{"type": "text", "text": "[End context.]"}] - + content - ) - msgs[first_idx] = {**target, "content": new_content} - - # COMPRESS_TOOL_RESULTS: walk ALL user messages and image-compress any - # large tool_result text content. Tool results accumulate in history as - # files are read; compressing them at the source compounds per-turn - # savings across the rest of the session. - compress_tr = os.environ.get("COMPRESS_TOOL_RESULTS", "1") == "1" - tr_imgs_added = 0 - if compress_tr: - for mi, m in enumerate(msgs): - if m.get("role") != "user": - continue - content_list = m.get("content") - if not isinstance(content_list, list): - continue - changed = False - new_blocks = [] - for blk in content_list: - if (isinstance(blk, dict) and blk.get("type") == "tool_result"): - # Anthropic API constraint: when is_error=true, the - # tool_result content MUST be type=text only (no images). - # Leave error tool_results untouched. - if blk.get("is_error") is True: - new_blocks.append(blk) - continue - inner = blk.get("content") - # tool_result.content can be str or list[block] - # NO cache_control on tool_result images (Anthropic - # caps at 4 breakpoints, and these aren't useful as - # cache anchors — they change every session). - if isinstance(inner, str) and len(inner) > 2000: - pngs_tr = render_chunks(inner) - new_inner = [image_block(p, cache=False) - for p in pngs_tr] - blk = {**blk, "content": new_inner} - tr_imgs_added += len(pngs_tr) - changed = True - elif isinstance(inner, list): - new_inner = [] - for ib in inner: - if (isinstance(ib, dict) and ib.get("type") == "text" - and len(ib.get("text", "")) > 2000): - pngs_tr = render_chunks(ib["text"]) - for p in pngs_tr: - new_inner.append(image_block(p, cache=False)) - tr_imgs_added += len(pngs_tr) - changed = True - else: - new_inner.append(ib) - blk = {**blk, "content": new_inner} - new_blocks.append(blk) - if changed: - msgs[mi] = {**m, "content": new_blocks} - info["tool_result_imgs"] = tr_imgs_added - req["messages"] = msgs - - if PLACEMENT == "replace_system": - req["system"] = "Follow the instructions in the first image of the user message exactly." - elif billing_line_kept is not None: - # Replace original system with JUST the billing header line (small, stable - # in spirit but per-turn random — keep it as text so it doesn't pollute - # our cacheable image). - req["system"] = billing_line_kept - # else: leave system unchanged so Anthropic still has guardrails - info["placement"] = PLACEMENT - else: - # PLACEMENT == 'system': put images directly into system field - new_sys = [] - if isinstance(remainder, list): - new_sys.extend(remainder) - new_sys.extend(image_blocks) - req["system"] = new_sys - info["placement"] = "system" - - info["compressed"] = True - return json.dumps(req).encode(), info - - -# Running session totals so users can see live savings. -# Persisted to ~/.cache/pixelpipe/stats.json on every update so -# restarts don't wipe history. Recent feed + latest PNG stay in-memory only. -import threading, time, collections, pathlib - -_STATS_DIR = pathlib.Path.home() / ".cache" / "pixelpipe" -_STATS_FILE = _STATS_DIR / "stats.json" # cumulative totals (rewritten atomically) -_REQUESTS_FILE = _STATS_DIR / "requests.jsonl" # full per-request log (append-only) -_REQUESTS_ROTATE_BYTES = 10 * 1024 * 1024 # rotate at 10 MB - -_session_stats = { - "requests": 0, - "compressed_requests": 0, - "effective_input_actual": 0.0, - "effective_input_baseline_est": 0.0, - "started_at": time.time(), - "first_seen_at": time.time(), # persists across restarts; "uptime since first install" - "recent": collections.deque(maxlen=50), # in-memory only — restart resets - "latest_png_bytes": None, # in-memory only - "latest_png_meta": "", - "lock": threading.Lock(), -} - - -def _load_persisted_stats(): - """Load cumulative totals + recent request history on startup.""" - try: - if _STATS_FILE.exists(): - with open(_STATS_FILE) as f: - d = json.load(f) - _session_stats["requests"] = int(d.get("requests", 0)) - _session_stats["compressed_requests"] = int(d.get("compressed_requests", 0)) - _session_stats["effective_input_actual"] = float(d.get("effective_input_actual", 0.0)) - _session_stats["effective_input_baseline_est"] = float(d.get("effective_input_baseline_est", 0.0)) - _session_stats["first_seen_at"] = float(d.get("first_seen_at", time.time())) - saved = _session_stats["effective_input_baseline_est"] - _session_stats["effective_input_actual"] - print(f"[PROXY] loaded persisted stats: {_session_stats['requests']} requests, " - f"~{saved:.0f} effective tokens saved (~${saved * 15.0 / 1e6:.4f})", flush=True) - except Exception as e: - print(f"[PROXY] could not load persisted stats: {e}", flush=True) - - # Replay last N entries from requests.jsonl into the in-memory recent feed - try: - if _REQUESTS_FILE.exists(): - tail = collections.deque(maxlen=_session_stats["recent"].maxlen) - with open(_REQUESTS_FILE) as f: - for line in f: - line = line.strip() - if not line: continue - try: - tail.append(json.loads(line)) - except Exception: - pass - _session_stats["recent"].extend(tail) - if tail: - print(f"[PROXY] replayed {len(tail)} recent requests from {_REQUESTS_FILE.name}", - flush=True) - except Exception as e: - print(f"[PROXY] could not load request history: {e}", flush=True) - - -def _persist_stats(): - """Atomically rewrite cumulative totals.""" - try: - _STATS_DIR.mkdir(parents=True, exist_ok=True) - payload = { - "requests": _session_stats["requests"], - "compressed_requests": _session_stats["compressed_requests"], - "effective_input_actual": _session_stats["effective_input_actual"], - "effective_input_baseline_est": _session_stats["effective_input_baseline_est"], - "first_seen_at": _session_stats["first_seen_at"], - "saved_at": time.time(), - } - tmp = _STATS_FILE.with_suffix(".json.tmp") - with open(tmp, "w") as f: - json.dump(payload, f, indent=2) - tmp.replace(_STATS_FILE) - except Exception as e: - print(f"[PROXY] could not persist stats: {e}", flush=True) - - -def _append_request_log(entry: dict): - """Append one JSONL line for the per-request log. Rotate when oversize.""" - try: - _STATS_DIR.mkdir(parents=True, exist_ok=True) - # Rotate if file is too big - if _REQUESTS_FILE.exists() and _REQUESTS_FILE.stat().st_size > _REQUESTS_ROTATE_BYTES: - rotated = _STATS_DIR / "requests.jsonl.1" - try: rotated.unlink() - except FileNotFoundError: pass - _REQUESTS_FILE.rename(rotated) - with open(_REQUESTS_FILE, "a") as f: - f.write(json.dumps(entry, separators=(",", ":")) + "\n") - except Exception as e: - # Don't crash request handling on disk errors - print(f"[PROXY] could not append request log: {e}", flush=True) - - -_load_persisted_stats() - - -DASHBOARD_HTML = """ - - - -pixelpipe — live dashboard - - - -

pixelpipe

-
connecting...
- -
-
requests
-
0
-
— compressed
-
-
tokens saved
-
0
-
effective input tokens
-
-
$ saved (opus 4.7)
-
$0.00
-
at $15/M input tokens
-
-
reduction
-
0%
-
vs uncompressed baseline
-
-
- -
-
-

recent requests

- - - - - - - - - -
#statuspathsize inccimg tokactualsaved
-
-
-

latest rendered image

-
(none yet)
-
-
-
- - - -""" - - -class ProxyHandler(BaseHTTPRequestHandler): - server_version = "TokenProxy/0.1" - - def log_message(self, fmt, *args): - pass - - def _serve_stats(self): - with _session_stats["lock"]: - saved = _session_stats["effective_input_baseline_est"] - _session_stats["effective_input_actual"] - pct = (saved / _session_stats["effective_input_baseline_est"] * 100.0 - if _session_stats["effective_input_baseline_est"] > 0 else 0) - uptime = time.time() - _session_stats["started_at"] - payload = { - "requests": _session_stats["requests"], - "compressed_requests": _session_stats["compressed_requests"], - "effective_input_actual": round(_session_stats["effective_input_actual"], 1), - "effective_input_baseline_est": round(_session_stats["effective_input_baseline_est"], 1), - "saved_effective_tokens": round(saved, 1), - "saved_pct": round(pct, 1), - "saved_usd_opus47": round(saved * 15.0 / 1e6, 4), - "uptime_sec": uptime, - } - body = json.dumps(payload, indent=2).encode() - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Access-Control-Allow-Origin", "*") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def _serve_recent(self): - with _session_stats["lock"]: - recent = list(_session_stats["recent"]) - has_preview = _session_stats["latest_png_bytes"] is not None - preview_meta = _session_stats.get("latest_png_meta", "") - body = json.dumps({ - "recent": recent, - "has_preview": has_preview, - "preview_meta": preview_meta, - }).encode() - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Access-Control-Allow-Origin", "*") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def _serve_latest_png(self): - """Serve the current rendered image. Supports `?crop=N` to return only - the top-left N×N region at native resolution (so humans can actually - SEE the tiny font — at 1466x1568 full-image scaled to 280px wide - the browser turns the antialiased glyphs into unreadable noise).""" - with _session_stats["lock"]: - data = _session_stats["latest_png_bytes"] - if not data: - self.send_response(404) - self.end_headers() - return - - # Parse ?crop=N from query string - crop_n = 0 - if "?" in self.path: - qs = self.path.split("?", 1)[1] - for kv in qs.split("&"): - if kv.startswith("crop="): - try: crop_n = int(kv[5:]) - except: pass - - if crop_n > 0: - try: - img = Image.open(io.BytesIO(data)) - cw = min(crop_n, img.width) - ch = min(crop_n, img.height) - cropped = img.crop((0, 0, cw, ch)) - buf = io.BytesIO() - cropped.save(buf, "PNG", optimize=True) - data = buf.getvalue() - except Exception: - pass # fall back to full image - - self.send_response(200) - self.send_header("Content-Type", "image/png") - self.send_header("Cache-Control", "no-cache") - self.send_header("Content-Length", str(len(data))) - self.end_headers() - self.wfile.write(data) - - def _do_proxy(self, method: str): - # Local endpoints — never forwarded to Anthropic. - if method == "GET" and self.path in ("/", "/dashboard"): - html = DASHBOARD_HTML.replace("__PORT__", str(PORT)).encode() - self.send_response(200) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(html))) - self.end_headers() - self.wfile.write(html) - return - if method == "GET" and self.path == "/proxy-stats": - self._serve_stats() - return - if method == "GET" and self.path == "/proxy-recent": - self._serve_recent() - return - if method == "GET" and self.path.split("?", 1)[0] == "/proxy-latest-png": - self._serve_latest_png() - return - try: - length = int(self.headers.get("Content-Length", "0")) - body = self.rfile.read(length) if length > 0 else b"" - - # Strip hop-by-hop headers - HOP = {"connection", "keep-alive", "proxy-connection", - "transfer-encoding", "upgrade", "te", "host", - "content-length", "expect", "accept-encoding"} - fwd_headers = {k: v for k, v in self.headers.items() - if k.lower() not in HOP} - fwd_headers.setdefault("anthropic-version", "2023-06-01") - - log = {"method": method, "path": self.path, "size_in": len(body)} - - if COMPRESS and method == "POST" and self.path.startswith("/v1/messages"): - new_body, info = transform_request(body) - log.update(info) - body = new_body - log["size_after"] = len(body) - - with httpx.Client(http2=False, timeout=120.0) as c: - resp = c.request(method, UPSTREAM + self.path, - content=body if body else None, - headers=fwd_headers) - - log["status"] = resp.status_code - log["size_out"] = len(resp.content) - log["upstream_ms"] = log.get("upstream_ms") # placeholder if added - - # Try plain JSON first; if that fails, parse SSE event stream for usage. - usage = None - try: - usage = resp.json().get("usage") - except Exception: - pass - if not usage and b"event:" in resp.content[:200]: - # SSE: aggregate usage from message_start (input) + message_delta (output). - txt = resp.content.decode("utf-8", errors="replace") - inp = cr = cc = out = 0 - for raw in txt.split("\n"): - if not raw.startswith("data: "): - continue - try: - ev = json.loads(raw[6:]) - except Exception: - continue - if ev.get("type") == "message_start": - u = ev.get("message", {}).get("usage", {}) - inp = u.get("input_tokens", inp) - cr = u.get("cache_read_input_tokens", cr) - cc = u.get("cache_creation_input_tokens", cc) - elif ev.get("type") == "message_delta": - u = ev.get("usage", {}) - if "output_tokens" in u: - out = u["output_tokens"] - if inp or out or cr or cc: - usage = {"input_tokens": inp, "output_tokens": out, - "cache_read_input_tokens": cr, - "cache_creation_input_tokens": cc} - if usage: - inp = usage.get("input_tokens", 0) or 0 - out = usage.get("output_tokens", 0) or 0 - cr = usage.get("cache_read_input_tokens", 0) or 0 - cc = usage.get("cache_creation_input_tokens", 0) or 0 - eff = inp + cc * 1.25 + cr * 0.10 - log["tokens"] = { - "in": inp, "out": out, - "cache_read": cr, "cache_create": cc, - "effective_cost": round(eff, 1), - } - - # Update session totals + estimate the baseline (uncompressed) cost. - # Baseline estimate: add back the text-token equivalent of whatever - # we replaced with images. Conservative — assumes baseline would - # also have cached, so we apply the 10% cache_read rate to the - # uncompressed delta. - if log.get("compressed"): - txt_replaced = (log.get("system_text_chars", 0) - + log.get("tool_text_added", 0)) // 4 - img_tokens_est = log.get("expected_image_tokens", 0) - extra_text_input_baseline = max(0, txt_replaced - img_tokens_est) - # CRITICAL: the extra text would have been billed at the - # SAME mix of cache_create/cache_read as the actual call, - # NOT all at cache_read (10%). On cold-start turns where - # cache_create dominates, baseline should be billed at 1.25 - # not 0.10 — otherwise we drastically under-estimate the - # savings and the dashboard shows tiny numbers. - cached_total = (cr or 0) + (cc or 0) - if cached_total > 0: - cc_share = (cc or 0) / cached_total - # Effective rate the extra text would have paid: - baseline_rate = cc_share * 1.25 + (1 - cc_share) * 0.10 - else: - baseline_rate = 0.10 # fully warm-cache assumption - baseline_eff = eff + extra_text_input_baseline * baseline_rate - else: - baseline_eff = eff - - with _session_stats["lock"]: - prev_saved = (_session_stats["effective_input_baseline_est"] - - _session_stats["effective_input_actual"]) - _session_stats["requests"] += 1 - if log.get("compressed"): - _session_stats["compressed_requests"] += 1 - _session_stats["effective_input_actual"] += eff - _session_stats["effective_input_baseline_est"] += baseline_eff - saved_so_far = (_session_stats["effective_input_baseline_est"] - - _session_stats["effective_input_actual"]) - # Push compact row for dashboard - row = { - "ts": time.time(), - "method": method, - "path": log.get("path", ""), - "status": log.get("status", 0), - "size_in": log.get("size_in", 0), - "size_out": log.get("size_out", 0), - "compressed": bool(log.get("compressed")), - "cc_added": log.get("cc_breakpoints_added"), - "expected_image_tokens": log.get("expected_image_tokens"), - "input_tokens": inp, - "cache_create": cc, - "cache_read": cr, - "effective_actual": round(eff, 1), - "effective_baseline": round(baseline_eff, 1), - "session_saved_so_far_delta": round(saved_so_far - prev_saved, 1), - } - _session_stats["recent"].append(row) - - # Persist cumulative totals (atomic JSON rewrite) + append - # per-request line to the JSONL log so dashboards survive restart. - _persist_stats() - _append_request_log(row) - - log["session_saved_so_far"] = round(saved_so_far, 1) - log["session_saved_usd"] = round(saved_so_far * 15.0 / 1e6, 4) - - print(f"[PROXY] {json.dumps(log)}", flush=True) - - try: - self.send_response(resp.status_code) - for k, v in resp.headers.items(): - if k.lower() in HOP or k.lower() == "content-encoding": - continue - self.send_header(k, v) - self.send_header("Content-Length", str(len(resp.content))) - self.end_headers() - self.wfile.write(resp.content) - except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError) as e: - # Client (Claude Code) closed before we finished writing. Common - # on stream timeouts / Ctrl-C / fast aborts. Not fatal — just log. - print(f"[PROXY] client disconnect during response: {type(e).__name__}", flush=True) - except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError) as e: - print(f"[PROXY] transport error: {type(e).__name__}", flush=True) - except Exception as e: - traceback.print_exc() - try: - self.send_response(502) - self.send_header("Content-Type", "application/json") - msg = json.dumps({"error": "proxy_error", "detail": str(e)}).encode() - self.send_header("Content-Length", str(len(msg))) - self.end_headers() - self.wfile.write(msg) - except Exception: - pass - - def do_GET(self): self._do_proxy("GET") - def do_POST(self): self._do_proxy("POST") - def do_PUT(self): self._do_proxy("PUT") - def do_DELETE(self): self._do_proxy("DELETE") - def do_HEAD(self): self._do_proxy("HEAD") - - -def main(): - print(f"Python token proxy listening on http://127.0.0.1:{PORT}", flush=True) - print(f" COMPRESS_SYSTEM={COMPRESS} FONT={FONT_PATH}@{FONT_SIZE}pt " - f"PLACEMENT={PLACEMENT} MIN_CHARS={MIN_COMPRESS_CHARS}", flush=True) - srv = ThreadingHTTPServer(("127.0.0.1", PORT), ProxyHandler) - try: - srv.serve_forever() - except KeyboardInterrupt: - srv.shutdown() - - -if __name__ == "__main__": - main() diff --git a/legacy/python/visual_compare.py b/legacy/python/visual_compare.py deleted file mode 100644 index 0d4579b..0000000 --- a/legacy/python/visual_compare.py +++ /dev/null @@ -1,150 +0,0 @@ -#!/usr/bin/env python3 -""" -Render the same text through Python and Rust, save PNGs side-by-side, and -also stitch a single comparison image with labels for easy visual inspection. - -Usage: - python3 scripts/visual_compare.py [out_dir] - -Output: - /python_.png - /rust_.png - /compare_.png (side-by-side with labels) - /README.txt (dim + byte size summary) -""" - -import io -import json -import subprocess -import sys -import tempfile -from pathlib import Path - -ROOT = Path(__file__).resolve().parent.parent -sys.path.insert(0, str(ROOT / "src")) - -import proxy as py_proxy # noqa: E402 -from PIL import Image, ImageDraw, ImageFont # noqa: E402 - -RUST_BIN = ROOT / "src" / "rust" / "target" / "release" / "examples" / "diff_render" - -# Three representative samples -SAMPLES = { - "tiny": "echo hello world\n", - "code": ( - "def render_chunks(text):\n" - " \"\"\"Render text into 5pt PNG, multi-column.\"\"\"\n" - " font = get_font(FONT_SIZE)\n" - " lines = wrap(text, FIXED_COL_CHARS)\n" - " return paint(font, lines)\n" - ), - "tools_schema": json.dumps({ - "tools": [ - { - "name": "Bash", - "description": "Execute a bash command", - "input_schema": { - "type": "object", - "properties": { - "command": {"type": "string", "description": "the command"}, - "timeout": {"type": "integer", "description": "timeout in ms"}, - }, - "required": ["command"], - }, - }, - { - "name": "Read", - "description": "Read a file from disk", - "input_schema": { - "type": "object", - "properties": { - "file_path": {"type": "string"}, - "offset": {"type": "integer"}, - "limit": {"type": "integer"}, - }, - "required": ["file_path"], - }, - }, - ] - }, indent=2), -} - - -def label_image(img: Image.Image, label: str) -> Image.Image: - """Add a black bar at top with the label in white.""" - bar_h = 24 - out = Image.new("RGB", (img.width, img.height + bar_h), (0, 0, 0)) - out.paste(img.convert("RGB"), (0, bar_h)) - d = ImageDraw.Draw(out) - try: - f = ImageFont.truetype("/System/Library/Fonts/Menlo.ttc", 14) - except Exception: - f = ImageFont.load_default() - d.text((6, 4), label, fill=(255, 255, 255), font=f) - return out - - -def main(): - out_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("/tmp/cip_compare") - out_dir.mkdir(parents=True, exist_ok=True) - - if not RUST_BIN.exists(): - print(f"build rust first: cd src/rust && cargo build --release --example diff_render") - sys.exit(1) - - # Run rust - spec_path = out_dir / "spec.json" - spec_path.write_text(json.dumps(SAMPLES)) - rust_outdir = out_dir / "_rust" - rust_outdir.mkdir(exist_ok=True) - r = subprocess.run( - [str(RUST_BIN), str(spec_path), str(rust_outdir)], - capture_output=True, text=True, - ) - if r.returncode != 0: - print("RUST FAILED:", r.stderr); sys.exit(2) - rust_manifest = json.loads((rust_outdir / "manifest.json").read_text()) - - summary_lines = ["Visual compare: Python proxy.py vs Rust src/rust/", ""] - - for name, text in SAMPLES.items(): - # Python - py_pngs = py_proxy.render_chunks(text) - assert len(py_pngs) == 1, f"unexpected multi-chunk python output for {name}" - py_path = out_dir / f"python_{name}.png" - py_path.write_bytes(py_pngs[0]) - py_img = Image.open(io.BytesIO(py_pngs[0])) - - # Rust - rs_files = rust_manifest.get(name, []) - assert len(rs_files) == 1, f"unexpected multi-chunk rust output for {name}" - rs_bytes = (rust_outdir / rs_files[0]).read_bytes() - rs_path = out_dir / f"rust_{name}.png" - rs_path.write_bytes(rs_bytes) - rs_img = Image.open(io.BytesIO(rs_bytes)) - - # Side-by-side compare - py_labeled = label_image(py_img, f"PYTHON {py_img.width}x{py_img.height} ({len(py_pngs[0])}B)") - rs_labeled = label_image(rs_img, f"RUST {rs_img.width}x{rs_img.height} ({len(rs_bytes)}B)") - gap = 8 - cmp_w = max(py_labeled.width, rs_labeled.width) - cmp_h = py_labeled.height + gap + rs_labeled.height - cmp = Image.new("RGB", (cmp_w, cmp_h), (40, 40, 40)) - cmp.paste(py_labeled, (0, 0)) - cmp.paste(rs_labeled, (0, py_labeled.height + gap)) - cmp.save(out_dir / f"compare_{name}.png", "PNG") - - line = ( - f"[{name:14s}] py={py_img.width:4d}x{py_img.height:<4d} ({len(py_pngs[0]):>6d}B) " - f"rs={rs_img.width:4d}x{rs_img.height:<4d} ({len(rs_bytes):>6d}B) " - f"text_len={len(text)}" - ) - print(line) - summary_lines.append(line) - - (out_dir / "README.txt").write_text("\n".join(summary_lines) + "\n") - print(f"\nwrote PNGs to {out_dir}/") - - -if __name__ == "__main__": - main() diff --git a/src/core/transform.ts b/src/core/transform.ts index 81a900b..f5de3a1 100644 --- a/src/core/transform.ts +++ b/src/core/transform.ts @@ -5,8 +5,9 @@ * those images instead — saving 65-73% input tokens on Opus 4.7 while * preserving 100% reasoning quality. * - * Matches the public-surface behavior of legacy/python/proxy.py at a - * minimum. Stricter byte-for-byte parity is verified in tests. + * Originally ported from a Python reference implementation; the Python + * has since been removed (live cache-rate validation passed at 98.7% by + * tokens). Byte-output determinism is now verified by tests alone. */ import type { diff --git a/src/dashboard.ts b/src/dashboard.ts index c824397..8b12ea1 100644 --- a/src/dashboard.ts +++ b/src/dashboard.ts @@ -7,7 +7,8 @@ * GET /proxy-recent → JSON ring buffer of recent requests * GET /proxy-latest-png[?crop=N] → raw PNG of the latest rendered image * - * Ported from legacy/python/proxy.py — same metric formulas, same HTML shell. + * Metric formulas and HTML shell originally ported from the Python reference + * implementation (deleted after live cache-rate validation hit 98.7% by tokens). * * Node-only by design (uses node:fs for startup replay + node:zlib for the * preview crop). The Worker host doesn't expose a dashboard; use Workers Logs.