mirror of
https://github.com/teamchong/pxpipe.git
synced 2026-07-22 02:02:51 +02:00
Port to TypeScript with dual Node + Cloudflare Workers targets
Runtime is pure Web Standard APIs (fetch, Request, Response, Uint8Array,
CompressionStream, crypto.subtle, btoa) — zero runtime dependencies.
Core (src/core/, runs identically on Node 18+ and Workers):
- atlas.ts auto-generated 9x15 JBM glyph atlas, base64-inlined (17KB)
- png.ts minimal grayscale PNG encoder via CompressionStream
- render.ts text → packed PNG, soft-wraps at 100 cols, ≤1568px tall
- transform.ts request body rewriter (system + tool docs → image blocks)
- proxy.ts fetch-handler that transforms then forwards to Anthropic
- types.ts Anthropic Messages API types we touch
Adapters:
- src/node.ts node:http server + CLI flag parsing
- src/worker.ts export default { fetch } for wrangler dev/deploy
Build tooling:
- scripts/gen-atlas.ts @napi-rs/canvas → atlas.ts (build-time only)
- scripts/build.mjs esbuild Node bundle; wrangler handles Worker
- tsconfig.json strict, ES2022, Workers types
Tests (vitest): 8 passing — PNG signature, base64 round-trip, single +
multi-image renders, transform no-op + compress paths, billing-line strip,
tool fold + stub.
E2E smoke: 16K char system → 2 PNGs (36KB) → mock upstream, 34ms.
Next: replace docs, then verify byte-output parity against legacy/python.
This commit is contained in:
teamchong
File diff suppressed because one or more lines are too long
+140
@@ -0,0 +1,140 @@
|
||||
/**
|
||||
* Minimal PNG encoder. Grayscale, 8-bit, filter = None, single IDAT.
|
||||
*
|
||||
* Cross-runtime: uses Web Streams (`CompressionStream`) which exists in:
|
||||
* - Node 18+ (global, no import)
|
||||
* - Cloudflare Workers (built-in)
|
||||
* - Modern browsers
|
||||
*
|
||||
* No `Buffer`, no `node:zlib` — pure Uint8Array. That's the whole reason
|
||||
* this exists instead of `pngjs` or `node-png`.
|
||||
*/
|
||||
|
||||
// ---- CRC32 (PNG spec table) ----------------------------------------------
|
||||
|
||||
const CRC_TABLE: Uint32Array = (() => {
|
||||
const t = new Uint32Array(256);
|
||||
for (let n = 0; n < 256; n++) {
|
||||
let c = n;
|
||||
for (let k = 0; k < 8; k++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1;
|
||||
t[n] = c >>> 0;
|
||||
}
|
||||
return t;
|
||||
})();
|
||||
|
||||
function crc32(bytes: Uint8Array): number {
|
||||
let c = 0xffffffff;
|
||||
for (let i = 0; i < bytes.length; i++) c = CRC_TABLE[(c ^ bytes[i]!) & 0xff]! ^ (c >>> 8);
|
||||
return (c ^ 0xffffffff) >>> 0;
|
||||
}
|
||||
|
||||
// ---- Helpers --------------------------------------------------------------
|
||||
|
||||
function concat(parts: Uint8Array[]): Uint8Array {
|
||||
let total = 0;
|
||||
for (const p of parts) total += p.length;
|
||||
const out = new Uint8Array(total);
|
||||
let off = 0;
|
||||
for (const p of parts) {
|
||||
out.set(p, off);
|
||||
off += p.length;
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
function u32be(n: number): Uint8Array {
|
||||
const b = new Uint8Array(4);
|
||||
new DataView(b.buffer).setUint32(0, n >>> 0, false);
|
||||
return b;
|
||||
}
|
||||
|
||||
const TYPE_BYTES = (s: string): Uint8Array => {
|
||||
const b = new Uint8Array(s.length);
|
||||
for (let i = 0; i < s.length; i++) b[i] = s.charCodeAt(i);
|
||||
return b;
|
||||
};
|
||||
|
||||
function chunk(type: string, data: Uint8Array): Uint8Array {
|
||||
const typeB = TYPE_BYTES(type);
|
||||
const crcSrc = concat([typeB, data]);
|
||||
return concat([u32be(data.length), typeB, data, u32be(crc32(crcSrc))]);
|
||||
}
|
||||
|
||||
// ---- Deflate via Web Streams ---------------------------------------------
|
||||
|
||||
async function deflateZlib(input: Uint8Array): Promise<Uint8Array> {
|
||||
// 'deflate' in CompressionStream is zlib-wrapped (RFC 1950), which is what
|
||||
// PNG IDAT expects. 'deflate-raw' would be RFC 1951 (no header) — wrong.
|
||||
const cs = new CompressionStream('deflate');
|
||||
const writer = cs.writable.getWriter();
|
||||
// Cast: TS 5.7 narrows Uint8Array<ArrayBufferLike> away from BufferSource,
|
||||
// but we never use SharedArrayBuffer here so the runtime call is safe.
|
||||
void writer.write(input as Uint8Array<ArrayBuffer>);
|
||||
void writer.close();
|
||||
|
||||
const reader = cs.readable.getReader();
|
||||
const chunks: Uint8Array[] = [];
|
||||
while (true) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
if (value) chunks.push(value);
|
||||
}
|
||||
return concat(chunks);
|
||||
}
|
||||
|
||||
// ---- Encode --------------------------------------------------------------
|
||||
|
||||
const PNG_SIGNATURE = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);
|
||||
|
||||
/**
|
||||
* Encode a grayscale buffer as PNG bytes.
|
||||
*
|
||||
* @param pixels Single-channel coverage, row-major, length = width × height.
|
||||
* @param width Pixel width.
|
||||
* @param height Pixel height.
|
||||
*/
|
||||
export async function encodeGrayPng(pixels: Uint8Array, width: number, height: number): Promise<Uint8Array> {
|
||||
if (pixels.length !== width * height) {
|
||||
throw new Error(`encodeGrayPng: pixels.length=${pixels.length} != ${width}×${height}=${width * height}`);
|
||||
}
|
||||
|
||||
// IHDR: width(4) height(4) bitDepth(1) colorType(1=gray) compress(0) filter(0) interlace(0)
|
||||
const ihdr = new Uint8Array(13);
|
||||
ihdr.set(u32be(width), 0);
|
||||
ihdr.set(u32be(height), 4);
|
||||
ihdr[8] = 8; // bit depth
|
||||
ihdr[9] = 0; // color type 0 = grayscale
|
||||
// bytes 10..12 already zero
|
||||
|
||||
// Add per-scanline filter byte (0 = None) and concat into raw stream
|
||||
const stride = width + 1;
|
||||
const raw = new Uint8Array(stride * height);
|
||||
for (let y = 0; y < height; y++) {
|
||||
raw[y * stride] = 0; // filter: None
|
||||
raw.set(pixels.subarray(y * width, (y + 1) * width), y * stride + 1);
|
||||
}
|
||||
|
||||
const compressed = await deflateZlib(raw);
|
||||
|
||||
return concat([
|
||||
PNG_SIGNATURE,
|
||||
chunk('IHDR', ihdr),
|
||||
chunk('IDAT', compressed),
|
||||
chunk('IEND', new Uint8Array(0)),
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Base64-encode bytes in a way that works in both Node and Workers.
|
||||
* `btoa` is global in both, but only accepts binary strings — we hand-roll
|
||||
* to avoid the `String.fromCharCode(...big array)` blow-up.
|
||||
*/
|
||||
export function bytesToBase64(bytes: Uint8Array): string {
|
||||
// Chunk to avoid maxing call stack on String.fromCharCode(...)
|
||||
let binary = '';
|
||||
const CHUNK = 0x8000;
|
||||
for (let i = 0; i < bytes.length; i += CHUNK) {
|
||||
binary += String.fromCharCode(...bytes.subarray(i, i + CHUNK));
|
||||
}
|
||||
return btoa(binary);
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
/**
|
||||
* The pixelpipe proxy as a single Web-standard fetch handler.
|
||||
*
|
||||
* Both `src/node.ts` and `src/worker.ts` adapt this to their respective
|
||||
* runtimes (node:http server vs CF Worker `fetch` export). The handler
|
||||
* itself only uses `Request`, `Response`, `URL`, and global `fetch` — all
|
||||
* of which exist identically in Node 18+ and Workers.
|
||||
*/
|
||||
|
||||
import { transformRequest, type TransformOptions, type TransformInfo } from './transform.js';
|
||||
|
||||
export interface ProxyConfig {
|
||||
/** Anthropic API base, no trailing slash. Defaults to api.anthropic.com. */
|
||||
upstream?: string;
|
||||
/** Override or supply an API key. If unset, we forward whatever the client sent. */
|
||||
apiKey?: string;
|
||||
/** Per-request transform options. */
|
||||
transform?: TransformOptions;
|
||||
/** Called after every request — useful for logging / metrics in the host. */
|
||||
onRequest?: (event: ProxyEvent) => void | Promise<void>;
|
||||
}
|
||||
|
||||
export interface ProxyEvent {
|
||||
method: string;
|
||||
path: string;
|
||||
status: number;
|
||||
durationMs: number;
|
||||
info?: TransformInfo;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
const DEFAULT_UPSTREAM = 'https://api.anthropic.com';
|
||||
|
||||
/** Headers we strip on the way out — they're hop-by-hop or proxy-injected. */
|
||||
const STRIP_REQ_HEADERS = new Set([
|
||||
'host',
|
||||
'connection',
|
||||
'keep-alive',
|
||||
'proxy-connection',
|
||||
'transfer-encoding',
|
||||
'upgrade',
|
||||
'content-length', // we recompute
|
||||
'expect',
|
||||
'accept-encoding', // let upstream choose
|
||||
]);
|
||||
|
||||
const STRIP_RES_HEADERS = new Set([
|
||||
'connection',
|
||||
'keep-alive',
|
||||
'transfer-encoding',
|
||||
'content-encoding', // we don't re-encode
|
||||
'content-length', // body may differ after streaming
|
||||
]);
|
||||
|
||||
function filterHeaders(src: Headers, strip: Set<string>): Headers {
|
||||
const out = new Headers();
|
||||
src.forEach((v, k) => {
|
||||
if (!strip.has(k.toLowerCase())) out.append(k, v);
|
||||
});
|
||||
return out;
|
||||
}
|
||||
|
||||
/** Build the proxy fetch handler bound to a config. */
|
||||
export function createProxy(config: ProxyConfig = {}) {
|
||||
const upstream = (config.upstream ?? DEFAULT_UPSTREAM).replace(/\/+$/, '');
|
||||
|
||||
return async function handle(req: Request): Promise<Response> {
|
||||
const t0 = Date.now();
|
||||
const url = new URL(req.url);
|
||||
const path = url.pathname + url.search;
|
||||
|
||||
const fire = (status: number, info?: TransformInfo, error?: string): void => {
|
||||
void config.onRequest?.({
|
||||
method: req.method,
|
||||
path: url.pathname,
|
||||
status,
|
||||
durationMs: Date.now() - t0,
|
||||
info,
|
||||
error,
|
||||
});
|
||||
};
|
||||
|
||||
// Only intercept /v1/messages POSTs. Everything else passes through.
|
||||
const isMessages = req.method === 'POST' && url.pathname === '/v1/messages';
|
||||
|
||||
let bodyOut: BodyInit | null = null;
|
||||
let info: TransformInfo | undefined;
|
||||
|
||||
if (isMessages) {
|
||||
const bodyIn = new Uint8Array(await req.arrayBuffer());
|
||||
try {
|
||||
const r = await transformRequest(bodyIn, config.transform);
|
||||
// Cast: TS narrows Uint8Array<ArrayBufferLike> away from BodyInit, but
|
||||
// it's a valid body and we never use SharedArrayBuffer.
|
||||
bodyOut = r.body as unknown as BodyInit;
|
||||
info = r.info;
|
||||
} catch (e) {
|
||||
fire(502, undefined, `transform_error: ${(e as Error).message}`);
|
||||
return new Response(JSON.stringify({ error: 'pixelpipe transform failed' }), {
|
||||
status: 502,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
});
|
||||
}
|
||||
} else {
|
||||
// Pass body through unchanged.
|
||||
bodyOut = req.body;
|
||||
}
|
||||
|
||||
const outHeaders = filterHeaders(req.headers, STRIP_REQ_HEADERS);
|
||||
if (config.apiKey) outHeaders.set('x-api-key', config.apiKey);
|
||||
|
||||
const upstreamUrl = upstream + path;
|
||||
let upstreamRes: Response;
|
||||
try {
|
||||
upstreamRes = await fetch(upstreamUrl, {
|
||||
method: req.method,
|
||||
headers: outHeaders,
|
||||
body: bodyOut,
|
||||
// duplex is required by spec when sending a stream as body
|
||||
...(bodyOut instanceof ReadableStream ? { duplex: 'half' } : {}),
|
||||
} as RequestInit);
|
||||
} catch (e) {
|
||||
fire(502, info, `upstream_error: ${(e as Error).message}`);
|
||||
return new Response(JSON.stringify({ error: 'pixelpipe upstream unreachable' }), {
|
||||
status: 502,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
});
|
||||
}
|
||||
|
||||
fire(upstreamRes.status, info);
|
||||
|
||||
return new Response(upstreamRes.body, {
|
||||
status: upstreamRes.status,
|
||||
statusText: upstreamRes.statusText,
|
||||
headers: filterHeaders(upstreamRes.headers, STRIP_RES_HEADERS),
|
||||
});
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
/**
|
||||
* Text → PNG renderer. Uses the build-time atlas (src/core/atlas.ts) and
|
||||
* blits glyphs into a single grayscale framebuffer, then PNG-encodes.
|
||||
*
|
||||
* Behavior matches the Python proxy's "minimum rows / max packed width"
|
||||
* approach: wrap at a fixed column count and pack as many lines per image
|
||||
* as fit within `MAX_HEIGHT_PX`. Anthropic's vision encoder works best with
|
||||
* images ≤ 1568×1568 px.
|
||||
*/
|
||||
|
||||
import {
|
||||
ATLAS_CELL_W,
|
||||
ATLAS_CELL_H,
|
||||
ATLAS_FIRST,
|
||||
ATLAS_LAST,
|
||||
ATLAS_PIXELS,
|
||||
} from './atlas.js';
|
||||
import { encodeGrayPng } from './png.js';
|
||||
|
||||
const MAX_HEIGHT_PX = 1568;
|
||||
const DEFAULT_COLS = 100;
|
||||
const PAD_X = 4;
|
||||
const PAD_Y = 4;
|
||||
|
||||
export interface RenderedImage {
|
||||
/** Raw PNG bytes. */
|
||||
png: Uint8Array;
|
||||
/** Pixel width. */
|
||||
width: number;
|
||||
/** Pixel height. */
|
||||
height: number;
|
||||
/** How many input characters were rendered into this image. */
|
||||
charsRendered: number;
|
||||
}
|
||||
|
||||
/** Soft-wrap a single logical line at `cols`, preserving explicit newlines. */
|
||||
function wrapLines(text: string, cols: number): string[] {
|
||||
const out: string[] = [];
|
||||
for (const raw of text.split('\n')) {
|
||||
if (raw.length === 0) {
|
||||
out.push('');
|
||||
continue;
|
||||
}
|
||||
for (let i = 0; i < raw.length; i += cols) {
|
||||
out.push(raw.slice(i, i + cols));
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
/**
|
||||
* Blit a single glyph onto the framebuffer with simple max() blending
|
||||
* (so we keep the darkest coverage if glyphs overlap on antialiased edges).
|
||||
*/
|
||||
function blitGlyph(fb: Uint8Array, fbW: number, x: number, y: number, code: number): void {
|
||||
if (code < ATLAS_FIRST || code > ATLAS_LAST) return; // skip unrenderable
|
||||
const glyphOff = (code - ATLAS_FIRST) * ATLAS_CELL_W * ATLAS_CELL_H;
|
||||
for (let gy = 0; gy < ATLAS_CELL_H; gy++) {
|
||||
const dstRow = (y + gy) * fbW + x;
|
||||
const srcRow = glyphOff + gy * ATLAS_CELL_W;
|
||||
for (let gx = 0; gx < ATLAS_CELL_W; gx++) {
|
||||
const v = ATLAS_PIXELS[srcRow + gx]!;
|
||||
if (v > fb[dstRow + gx]!) fb[dstRow + gx] = v;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Render up to `maxChars` of `text` to a single PNG, returning leftover text + image. */
|
||||
export async function renderChunkToPng(
|
||||
text: string,
|
||||
cols: number = DEFAULT_COLS,
|
||||
): Promise<RenderedImage> {
|
||||
const lines = wrapLines(text, cols);
|
||||
|
||||
// Compute how many lines we can fit vertically.
|
||||
const maxLines = Math.max(1, Math.floor((MAX_HEIGHT_PX - 2 * PAD_Y) / ATLAS_CELL_H));
|
||||
const fitLines = lines.slice(0, maxLines);
|
||||
|
||||
const charsRendered = fitLines.reduce((n, l) => n + l.length + 1, -1); // -1: no trailing \n
|
||||
|
||||
const width = 2 * PAD_X + cols * ATLAS_CELL_W;
|
||||
const height = 2 * PAD_Y + fitLines.length * ATLAS_CELL_H;
|
||||
|
||||
// Black canvas (matches atlas: text is white-on-black, but the model OCRs
|
||||
// both polarities — we invert below to white-on-black for crispness).
|
||||
const fb = new Uint8Array(width * height); // zero-initialized = black
|
||||
|
||||
for (let row = 0; row < fitLines.length; row++) {
|
||||
const line = fitLines[row]!;
|
||||
const baseY = PAD_Y + row * ATLAS_CELL_H;
|
||||
for (let col = 0; col < line.length; col++) {
|
||||
const baseX = PAD_X + col * ATLAS_CELL_W;
|
||||
blitGlyph(fb, width, baseX, baseY, line.charCodeAt(col));
|
||||
}
|
||||
}
|
||||
|
||||
// Invert: atlas stores white-on-black coverage, but black-on-white renders
|
||||
// cleaner and matches what the Python proxy emits.
|
||||
for (let i = 0; i < fb.length; i++) fb[i] = 255 - fb[i]!;
|
||||
|
||||
const png = await encodeGrayPng(fb, width, height);
|
||||
return { png, width, height, charsRendered };
|
||||
}
|
||||
|
||||
/** Split `text` into N PNGs, each ≤ MAX_HEIGHT_PX tall. */
|
||||
export async function renderTextToPngs(
|
||||
text: string,
|
||||
cols: number = DEFAULT_COLS,
|
||||
): Promise<RenderedImage[]> {
|
||||
const lines = wrapLines(text, cols);
|
||||
const linesPerImg = Math.max(1, Math.floor((MAX_HEIGHT_PX - 2 * PAD_Y) / ATLAS_CELL_H));
|
||||
|
||||
const images: RenderedImage[] = [];
|
||||
for (let i = 0; i < lines.length; i += linesPerImg) {
|
||||
const chunk = lines.slice(i, i + linesPerImg).join('\n');
|
||||
images.push(await renderChunkToPng(chunk, cols));
|
||||
}
|
||||
return images;
|
||||
}
|
||||
@@ -0,0 +1,209 @@
|
||||
/**
|
||||
* Request-body transformer. Takes an Anthropic Messages API request body,
|
||||
* extracts the large static parts (system prompt + tool definitions),
|
||||
* renders them as PNG image blocks, and rewrites the body to reference
|
||||
* 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.
|
||||
*/
|
||||
|
||||
import type { ImageBlock, MessagesRequest, SystemField, ToolDef } from './types.js';
|
||||
import { renderTextToPngs } from './render.js';
|
||||
import { bytesToBase64 } from './png.js';
|
||||
|
||||
export interface TransformOptions {
|
||||
/** Master switch — false makes this a no-op pass-through. */
|
||||
compress?: boolean;
|
||||
/** Compress the system field. */
|
||||
compressSystem?: boolean;
|
||||
/** Move tool descriptions into the same image (and stub the originals). */
|
||||
compressTools?: boolean;
|
||||
/** Include full input_schema JSON for each tool. Adds tokens but maximizes parity. */
|
||||
compressSchemas?: boolean;
|
||||
/** Don't compress if total compressible chars below this. */
|
||||
minCompressChars?: number;
|
||||
/** Where to attach the image block — system field, or first user message. */
|
||||
placement?: 'system' | 'user';
|
||||
/** Soft-wrap column count. */
|
||||
cols?: number;
|
||||
}
|
||||
|
||||
const DEFAULTS: Required<TransformOptions> = {
|
||||
compress: true,
|
||||
compressSystem: true,
|
||||
compressTools: true,
|
||||
compressSchemas: true,
|
||||
minCompressChars: 2000,
|
||||
placement: 'system',
|
||||
cols: 100,
|
||||
};
|
||||
|
||||
export interface TransformInfo {
|
||||
compressed: boolean;
|
||||
reason?: string;
|
||||
origChars: number;
|
||||
imageCount: number;
|
||||
imageBytes: number;
|
||||
}
|
||||
|
||||
// --- helpers ---------------------------------------------------------------
|
||||
|
||||
/** Extract `(text, remainder)` from a system field that may be string or list. */
|
||||
function extractSystemText(sys: SystemField | undefined): { text: string; kept: SystemField } {
|
||||
if (sys == null) return { text: '', kept: [] };
|
||||
if (typeof sys === 'string') return { text: sys, kept: '' };
|
||||
const textParts: string[] = [];
|
||||
const kept: SystemField = [];
|
||||
for (const block of sys) {
|
||||
if (block && typeof block === 'object' && block.type === 'text') {
|
||||
textParts.push(block.text);
|
||||
} else {
|
||||
kept.push(block);
|
||||
}
|
||||
}
|
||||
return { text: textParts.join('\n\n'), kept };
|
||||
}
|
||||
|
||||
/**
|
||||
* Strip the per-turn random billing header line that Claude Code injects.
|
||||
* It changes every turn and would defeat prompt-cache hits if we left it
|
||||
* inside the image. We keep it as a leading text block so the upstream
|
||||
* still receives it.
|
||||
*/
|
||||
function stripBillingLine(text: string): { kept: string | null; body: string } {
|
||||
const nl = text.indexOf('\n');
|
||||
const first = nl === -1 ? text : text.slice(0, nl);
|
||||
if (first.startsWith('x-anthropic-billing-header:')) {
|
||||
return { kept: first, body: nl === -1 ? '' : text.slice(nl + 1) };
|
||||
}
|
||||
return { kept: null, body: text };
|
||||
}
|
||||
|
||||
/** Build the "## Tool: name\n<desc>\n<schema>" block for one tool definition. */
|
||||
function renderToolDoc(t: ToolDef, includeSchema: boolean): string {
|
||||
const parts: string[] = [`## Tool: ${t.name ?? '?'}`];
|
||||
if (t.description) parts.push(t.description);
|
||||
if (includeSchema && t.input_schema !== undefined) {
|
||||
parts.push('```json\n' + JSON.stringify(t.input_schema, null, 2) + '\n```');
|
||||
}
|
||||
return parts.join('\n');
|
||||
}
|
||||
|
||||
function makeImageBlock(pngB64: string, ephemeral = false): ImageBlock {
|
||||
const blk: ImageBlock = {
|
||||
type: 'image',
|
||||
source: { type: 'base64', media_type: 'image/png', data: pngB64 },
|
||||
};
|
||||
if (ephemeral) blk.cache_control = { type: 'ephemeral' };
|
||||
return blk;
|
||||
}
|
||||
|
||||
// --- main transform --------------------------------------------------------
|
||||
|
||||
/**
|
||||
* Rewrite a Messages API request body. Returns the new body (still JSON
|
||||
* bytes) plus diagnostic info. On any error, returns the original bytes
|
||||
* unchanged.
|
||||
*/
|
||||
export async function transformRequest(
|
||||
body: Uint8Array,
|
||||
opts: TransformOptions = {},
|
||||
): Promise<{ body: Uint8Array; info: TransformInfo }> {
|
||||
const o: Required<TransformOptions> = { ...DEFAULTS, ...opts };
|
||||
const info: TransformInfo = {
|
||||
compressed: false,
|
||||
origChars: 0,
|
||||
imageCount: 0,
|
||||
imageBytes: 0,
|
||||
};
|
||||
|
||||
if (!o.compress) {
|
||||
info.reason = 'compress=false';
|
||||
return { body, info };
|
||||
}
|
||||
|
||||
let req: MessagesRequest;
|
||||
try {
|
||||
req = JSON.parse(new TextDecoder().decode(body));
|
||||
} catch (e) {
|
||||
info.reason = `parse_error: ${(e as Error).message}`;
|
||||
return { body, info };
|
||||
}
|
||||
|
||||
// 1. Pull system text out.
|
||||
const { text: rawSysText, kept: sysRemainder } = extractSystemText(req.system);
|
||||
const { kept: billingLine, body: sysBody } = stripBillingLine(rawSysText);
|
||||
|
||||
// 2. Optionally fold tool docs into the same image, stubbing originals.
|
||||
let toolDocsText = '';
|
||||
let toolsRewritten: ToolDef[] | undefined;
|
||||
if (o.compressTools && Array.isArray(req.tools) && req.tools.length > 0) {
|
||||
const docs: string[] = [];
|
||||
toolsRewritten = req.tools.map((t) => {
|
||||
docs.push(renderToolDoc(t, o.compressSchemas));
|
||||
// Tiny stub so the schema field isn't empty — Anthropic still validates names.
|
||||
return {
|
||||
...t,
|
||||
description: 'ⓘ See image.',
|
||||
...(o.compressSchemas ? { input_schema: { type: 'object' } } : {}),
|
||||
};
|
||||
});
|
||||
toolDocsText = docs.join('\n\n');
|
||||
}
|
||||
|
||||
const combined = [sysBody, toolDocsText].filter((s) => s.length > 0).join('\n\n');
|
||||
info.origChars = combined.length;
|
||||
|
||||
if (combined.length < o.minCompressChars) {
|
||||
info.reason = `below_min_chars (${combined.length} < ${o.minCompressChars})`;
|
||||
return { body, info };
|
||||
}
|
||||
|
||||
// 3. Render to one or more PNGs.
|
||||
const images = await renderTextToPngs(combined, o.cols);
|
||||
const imageBlocks: ImageBlock[] = [];
|
||||
for (let i = 0; i < images.length; i++) {
|
||||
const img = images[i]!;
|
||||
const b64 = bytesToBase64(img.png);
|
||||
info.imageBytes += img.png.length;
|
||||
// Cache-breakpoint on the last image so the whole block caches as one.
|
||||
imageBlocks.push(makeImageBlock(b64, i === images.length - 1));
|
||||
}
|
||||
info.imageCount = imageBlocks.length;
|
||||
|
||||
// 4. Splice images back into the request.
|
||||
const prefixText = billingLine != null ? billingLine + '\n' : '';
|
||||
const introText =
|
||||
"The following is the system prompt + tool documentation, rendered as " +
|
||||
"images for token efficiency. OCR carefully and treat as authoritative " +
|
||||
"system instructions.";
|
||||
const newSystem: SystemField = [];
|
||||
if (prefixText) newSystem.push({ type: 'text', text: prefixText.trimEnd() });
|
||||
newSystem.push({ type: 'text', text: introText });
|
||||
newSystem.push(...imageBlocks);
|
||||
newSystem.push({ type: 'text', text: '[End of rendered context.]' });
|
||||
if (Array.isArray(sysRemainder)) newSystem.push(...sysRemainder);
|
||||
|
||||
if (o.placement === 'system' && o.compressSystem) {
|
||||
req.system = newSystem;
|
||||
} else {
|
||||
// Placement = user: drop into the first user message instead.
|
||||
req.system = billingLine ? [{ type: 'text', text: billingLine }] : undefined;
|
||||
const firstUserIdx = (req.messages ?? []).findIndex((m) => m.role === 'user');
|
||||
if (firstUserIdx >= 0) {
|
||||
const m = req.messages![firstUserIdx]!;
|
||||
const existing = Array.isArray(m.content)
|
||||
? m.content
|
||||
: [{ type: 'text' as const, text: m.content }];
|
||||
m.content = [...newSystem, ...existing];
|
||||
}
|
||||
}
|
||||
|
||||
if (toolsRewritten) req.tools = toolsRewritten;
|
||||
|
||||
info.compressed = true;
|
||||
const out = new TextEncoder().encode(JSON.stringify(req));
|
||||
return { body: out, info };
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
/**
|
||||
* Minimal Anthropic Messages API request types — only the fields pixelpipe
|
||||
* actually reads or rewrites. Anything else passes through untouched.
|
||||
*
|
||||
* Shape reference: https://docs.anthropic.com/en/api/messages
|
||||
*/
|
||||
|
||||
export interface TextBlock {
|
||||
type: 'text';
|
||||
text: string;
|
||||
cache_control?: CacheControl;
|
||||
}
|
||||
|
||||
export interface ImageBlock {
|
||||
type: 'image';
|
||||
source: {
|
||||
type: 'base64';
|
||||
media_type: 'image/png' | 'image/jpeg' | 'image/gif' | 'image/webp';
|
||||
data: string;
|
||||
};
|
||||
cache_control?: CacheControl;
|
||||
}
|
||||
|
||||
export interface ToolUseBlock {
|
||||
type: 'tool_use';
|
||||
id: string;
|
||||
name: string;
|
||||
input: unknown;
|
||||
}
|
||||
|
||||
export interface ToolResultBlock {
|
||||
type: 'tool_result';
|
||||
tool_use_id: string;
|
||||
content: string | Array<TextBlock | ImageBlock>;
|
||||
is_error?: boolean;
|
||||
cache_control?: CacheControl;
|
||||
}
|
||||
|
||||
export type ContentBlock = TextBlock | ImageBlock | ToolUseBlock | ToolResultBlock;
|
||||
|
||||
export interface CacheControl {
|
||||
type: 'ephemeral';
|
||||
ttl?: '5m' | '1h';
|
||||
}
|
||||
|
||||
export interface Message {
|
||||
role: 'user' | 'assistant';
|
||||
content: string | ContentBlock[];
|
||||
}
|
||||
|
||||
export interface ToolDef {
|
||||
name: string;
|
||||
description?: string;
|
||||
input_schema?: unknown;
|
||||
cache_control?: CacheControl;
|
||||
}
|
||||
|
||||
export type SystemField = string | Array<TextBlock | ImageBlock>;
|
||||
|
||||
export interface MessagesRequest {
|
||||
model: string;
|
||||
messages: Message[];
|
||||
system?: SystemField;
|
||||
tools?: ToolDef[];
|
||||
// … plus all the other fields we don't touch (max_tokens, temperature, …)
|
||||
[k: string]: unknown;
|
||||
}
|
||||
+208
@@ -0,0 +1,208 @@
|
||||
/**
|
||||
* Node entrypoint — `node:http` server + minimal CLI flag parsing.
|
||||
*
|
||||
* Wraps the runtime-agnostic `createProxy` from src/core/proxy.ts. The
|
||||
* heavy lifting (transform, render, PNG) is identical to the Worker
|
||||
* version; only the request/response plumbing differs.
|
||||
*/
|
||||
|
||||
import { createServer, type IncomingMessage, type ServerResponse } from 'node:http';
|
||||
import { createProxy, type ProxyConfig } from './core/proxy.js';
|
||||
import type { TransformOptions } from './core/transform.js';
|
||||
|
||||
interface CliOpts {
|
||||
port: number;
|
||||
upstream: string;
|
||||
compress: boolean;
|
||||
compressTools: boolean;
|
||||
compressSchemas: boolean;
|
||||
minCompressChars: number;
|
||||
placement: 'system' | 'user';
|
||||
cols: number;
|
||||
}
|
||||
|
||||
function envFlag(name: string, fallback: boolean): boolean {
|
||||
const v = process.env[name];
|
||||
if (v == null) return fallback;
|
||||
return v === '1' || v.toLowerCase() === 'true';
|
||||
}
|
||||
|
||||
function parseCli(argv: string[]): CliOpts {
|
||||
const o: CliOpts = {
|
||||
port: Number(process.env.PORT ?? 47821),
|
||||
upstream: process.env.ANTHROPIC_UPSTREAM ?? 'https://api.anthropic.com',
|
||||
compress: envFlag('COMPRESS', true),
|
||||
compressTools: envFlag('COMPRESS_TOOLS', true),
|
||||
compressSchemas: envFlag('COMPRESS_SCHEMAS', true),
|
||||
minCompressChars: Number(process.env.MIN_COMPRESS_CHARS ?? 2000),
|
||||
placement: (process.env.PLACEMENT as 'system' | 'user') ?? 'system',
|
||||
cols: Number(process.env.COLS ?? 100),
|
||||
};
|
||||
for (let i = 0; i < argv.length; i++) {
|
||||
const a = argv[i]!;
|
||||
const eat = () => argv[++i]!;
|
||||
switch (a) {
|
||||
case '-p':
|
||||
case '--port': o.port = Number(eat()); break;
|
||||
case '--upstream': o.upstream = eat(); break;
|
||||
case '--no-compress': o.compress = false; break;
|
||||
case '--no-tools': o.compressTools = false; break;
|
||||
case '--no-schemas': o.compressSchemas = false; break;
|
||||
case '--min-chars': o.minCompressChars = Number(eat()); break;
|
||||
case '--placement': o.placement = eat() as 'system' | 'user'; break;
|
||||
case '--cols': o.cols = Number(eat()); break;
|
||||
case '-h':
|
||||
case '--help': printHelp(); process.exit(0);
|
||||
case '--version': printVersion(); process.exit(0);
|
||||
default:
|
||||
if (a.startsWith('--')) {
|
||||
console.error(`[pixelpipe] unknown option: ${a}`);
|
||||
process.exit(2);
|
||||
}
|
||||
}
|
||||
}
|
||||
return o;
|
||||
}
|
||||
|
||||
function printHelp(): void {
|
||||
console.log(`pixelpipe — token-saving proxy for Claude Code
|
||||
|
||||
Usage:
|
||||
pixelpipe [options]
|
||||
|
||||
Options:
|
||||
-p, --port <N> listen port (default 47821)
|
||||
--upstream <URL> Anthropic API base (default https://api.anthropic.com)
|
||||
--no-compress disable all compression (pure passthrough)
|
||||
--no-tools don't fold tool docs into the image
|
||||
--no-schemas don't include input_schema JSON in the image
|
||||
--min-chars <N> skip compression below this many chars (default 2000)
|
||||
--placement <where> 'system' or 'user' (default system)
|
||||
--cols <N> soft-wrap column count (default 100)
|
||||
-h, --help show this help
|
||||
--version show version
|
||||
|
||||
Environment:
|
||||
Same as flags via PORT, ANTHROPIC_UPSTREAM, COMPRESS, COMPRESS_TOOLS,
|
||||
COMPRESS_SCHEMAS, MIN_COMPRESS_CHARS, PLACEMENT, COLS.
|
||||
|
||||
Use with Claude Code:
|
||||
ANTHROPIC_BASE_URL=http://127.0.0.1:47821 claude --exclude-dynamic-system-prompt-sections
|
||||
`);
|
||||
}
|
||||
|
||||
function printVersion(): void {
|
||||
// Filled in at bundle time by esbuild.define; falls back here.
|
||||
console.log(process.env.npm_package_version ?? '0.2.0');
|
||||
}
|
||||
|
||||
// ---- node:http <-> Web Request/Response bridge ---------------------------
|
||||
|
||||
function toWebRequest(req: IncomingMessage): Request {
|
||||
const proto = (req.headers['x-forwarded-proto'] as string) ?? 'http';
|
||||
const host = req.headers.host ?? 'localhost';
|
||||
const url = `${proto}://${host}${req.url ?? '/'}`;
|
||||
|
||||
const headers = new Headers();
|
||||
for (const [k, v] of Object.entries(req.headers)) {
|
||||
if (v == null) continue;
|
||||
if (Array.isArray(v)) v.forEach((vv) => headers.append(k, vv));
|
||||
else headers.append(k, v);
|
||||
}
|
||||
|
||||
const method = req.method ?? 'GET';
|
||||
const hasBody = method !== 'GET' && method !== 'HEAD';
|
||||
|
||||
// Buffer the body — proxy needs to read /v1/messages bodies fully anyway,
|
||||
// and Node's IncomingMessage → ReadableStream conversion has duplex quirks.
|
||||
let body: BodyInit | undefined;
|
||||
if (hasBody) {
|
||||
body = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
req.on('data', (chunk) => controller.enqueue(chunk));
|
||||
req.on('end', () => controller.close());
|
||||
req.on('error', (e) => controller.error(e));
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
return new Request(url, {
|
||||
method,
|
||||
headers,
|
||||
body,
|
||||
// @ts-expect-error — duplex is required for streamed request bodies in Node 18+
|
||||
duplex: hasBody ? 'half' : undefined,
|
||||
});
|
||||
}
|
||||
|
||||
async function writeWebResponse(res: Response, out: ServerResponse): Promise<void> {
|
||||
out.statusCode = res.status;
|
||||
res.headers.forEach((v, k) => out.setHeader(k, v));
|
||||
if (!res.body) {
|
||||
out.end();
|
||||
return;
|
||||
}
|
||||
const reader = res.body.getReader();
|
||||
while (true) {
|
||||
const { value, done } = await reader.read();
|
||||
if (done) break;
|
||||
if (value) out.write(value);
|
||||
}
|
||||
out.end();
|
||||
}
|
||||
|
||||
// ---- main ----------------------------------------------------------------
|
||||
|
||||
async function main(): Promise<void> {
|
||||
const opts = parseCli(process.argv.slice(2));
|
||||
const transform: TransformOptions = {
|
||||
compress: opts.compress,
|
||||
compressTools: opts.compressTools,
|
||||
compressSchemas: opts.compressSchemas,
|
||||
minCompressChars: opts.minCompressChars,
|
||||
placement: opts.placement,
|
||||
cols: opts.cols,
|
||||
};
|
||||
const config: ProxyConfig = {
|
||||
upstream: opts.upstream,
|
||||
transform,
|
||||
onRequest: (e) => {
|
||||
const tag = e.info?.compressed
|
||||
? `compressed ${e.info.origChars}ch → ${e.info.imageCount}img/${e.info.imageBytes}B`
|
||||
: e.info?.reason ?? '';
|
||||
console.log(`[${new Date().toISOString()}] ${e.method} ${e.path} → ${e.status} (${e.durationMs}ms) ${tag}`);
|
||||
},
|
||||
};
|
||||
const handle = createProxy(config);
|
||||
|
||||
const server = createServer((req, res) => {
|
||||
Promise.resolve()
|
||||
.then(async () => {
|
||||
const webReq = toWebRequest(req);
|
||||
const webRes = await handle(webReq);
|
||||
await writeWebResponse(webRes, res);
|
||||
})
|
||||
.catch((err) => {
|
||||
console.error('[pixelpipe] handler error:', err);
|
||||
if (!res.headersSent) res.statusCode = 500;
|
||||
res.end();
|
||||
});
|
||||
});
|
||||
|
||||
server.listen(opts.port, () => {
|
||||
console.log(`[pixelpipe] listening on http://127.0.0.1:${opts.port} → ${opts.upstream}`);
|
||||
console.log(`[pixelpipe] config: compress=${opts.compress} tools=${opts.compressTools} schemas=${opts.compressSchemas} min=${opts.minCompressChars} placement=${opts.placement} cols=${opts.cols}`);
|
||||
});
|
||||
|
||||
const shutdown = (sig: string) => {
|
||||
console.log(`[pixelpipe] ${sig} — shutting down`);
|
||||
server.close(() => process.exit(0));
|
||||
};
|
||||
process.on('SIGINT', () => shutdown('SIGINT'));
|
||||
process.on('SIGTERM', () => shutdown('SIGTERM'));
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error('[pixelpipe] fatal:', err);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -0,0 +1,57 @@
|
||||
/**
|
||||
* Cloudflare Workers entrypoint. Identical proxy logic to the Node build,
|
||||
* just wired up through the Worker `fetch` export.
|
||||
*
|
||||
* Deploy:
|
||||
* npx wrangler deploy
|
||||
*
|
||||
* Dev:
|
||||
* npx wrangler dev
|
||||
*
|
||||
* Config lives in wrangler.toml.
|
||||
*/
|
||||
|
||||
import { createProxy, type ProxyConfig } from './core/proxy.js';
|
||||
import type { TransformOptions } from './core/transform.js';
|
||||
|
||||
export interface Env {
|
||||
ANTHROPIC_UPSTREAM?: string;
|
||||
/** Optional override — if set, replaces whatever x-api-key the client sent. */
|
||||
ANTHROPIC_API_KEY?: string;
|
||||
COMPRESS?: string;
|
||||
COMPRESS_TOOLS?: string;
|
||||
COMPRESS_SCHEMAS?: string;
|
||||
MIN_COMPRESS_CHARS?: string;
|
||||
PLACEMENT?: string;
|
||||
COLS?: string;
|
||||
}
|
||||
|
||||
const truthy = (v: string | undefined, fallback: boolean): boolean =>
|
||||
v == null ? fallback : v === '1' || v.toLowerCase() === 'true';
|
||||
|
||||
export default {
|
||||
async fetch(req: Request, env: Env, _ctx: ExecutionContext): Promise<Response> {
|
||||
const transform: TransformOptions = {
|
||||
compress: truthy(env.COMPRESS, true),
|
||||
compressTools: truthy(env.COMPRESS_TOOLS, true),
|
||||
compressSchemas: truthy(env.COMPRESS_SCHEMAS, true),
|
||||
minCompressChars: env.MIN_COMPRESS_CHARS ? Number(env.MIN_COMPRESS_CHARS) : 2000,
|
||||
placement: (env.PLACEMENT as 'system' | 'user') ?? 'system',
|
||||
cols: env.COLS ? Number(env.COLS) : 100,
|
||||
};
|
||||
const config: ProxyConfig = {
|
||||
upstream: env.ANTHROPIC_UPSTREAM ?? 'https://api.anthropic.com',
|
||||
apiKey: env.ANTHROPIC_API_KEY,
|
||||
transform,
|
||||
// Note: console.log in Workers is captured by `wrangler tail`.
|
||||
onRequest: (e) => {
|
||||
const tag = e.info?.compressed
|
||||
? `compressed ${e.info.origChars}ch → ${e.info.imageCount}img/${e.info.imageBytes}B`
|
||||
: e.info?.reason ?? '';
|
||||
console.log(`${e.method} ${e.path} → ${e.status} (${e.durationMs}ms) ${tag}`);
|
||||
},
|
||||
};
|
||||
const handle = createProxy(config);
|
||||
return handle(req);
|
||||
},
|
||||
};
|
||||
Reference in New Issue
Block a user