review fixes: structural test, URL-parsed native-OpenAI check, bounded-reader hardening, total tool-arg cap

Gemini review findings adjudicated and applied: the analyticsTable
schema test asserts on the exported attribute list instead of
regexing the source; #isNativeOpenAI parses the URL hostname (port
and spoofed-suffix safe); readBoundedJson wraps stream errors in the
backend error class, cancels the body before throwing on the cap,
and reuses a module-level TextDecoder; an 8 MiB total tool-argument
cap bounds the accumulator maps across entries in all three
streaming parsers.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: heskew

components/anthropic/index.ts

@@ -65,6 +65,8 @@ const MAX_UPSTREAM_ERROR_MESSAGE_CHARS = 500;
 // `index` from content_block_start events; a hostile upstream can open unbounded
 // entries if content_block_stop never arrives for earlier indices.
 const MAX_TOOL_CALL_ACCUMULATOR_ENTRIES = 128;
+// Total tool-call argument chars across all content blocks in one stream.
+const MAX_TOTAL_TOOL_CALL_ARGS_CHARS = 8 * 1024 * 1024; // 8 MiB
 
 const log = harperLogger.forComponent('anthropic').conditional;
 
@@ -144,6 +146,7 @@ export class AnthropicBackend implements ModelBackend {
 		// partial strings.
 		const toolBuf = new Map<number, AnthropicToolCallAccumulator>();
 		let finalFinishReason: GenerateResult['finishReason'] | undefined;
+		let totalArgChars = 0;
 
 		for await (const event of readSse(res.body)) {
 			const chunk: GenerateChunk = {};
@@ -196,6 +199,12 @@ export class AnthropicBackend implements ModelBackend {
 								`Anthropic tool-call arguments exceed ${MAX_TOOL_CALL_ARGS_CHARS} chars (index ${event.index})`
 							);
 						}
+						totalArgChars += event.delta.partial_json.length;
+						if (totalArgChars > MAX_TOTAL_TOOL_CALL_ARGS_CHARS) {
+							throw new AnthropicBackendError(
+								`Anthropic tool-call arguments exceed total stream cap of ${MAX_TOTAL_TOOL_CALL_ARGS_CHARS} chars`
+							);
+						}
 						acc.argumentsBuf += event.delta.partial_json;
 					}
 				}

components/bedrock/index.ts

@@ -50,6 +50,8 @@ const MAX_TOOL_CALL_ARGS_CHARS = 1 << 20;
 // `index` values could otherwise allocate unbounded map entries if
 // content_block_stop events never arrive. Matches the cap in the direct backends.
 const MAX_TOOL_CALL_ACCUMULATOR_ENTRIES = 128;
+// Total tool-call argument chars across all content blocks in one stream.
+const MAX_TOTAL_TOOL_CALL_ARGS_CHARS = 8 * 1024 * 1024; // 8 MiB
 
 const log = harperLogger.forComponent('bedrock').conditional;
 
@@ -371,7 +373,12 @@ function rejectNovaModel(modelId: string): void {
 	}
 }
 
-function buildGenerateBody(family: Family, modelId: string, input: GenerateInput, opts: BackendOpts<GenerateOpts>): object {
+function buildGenerateBody(
+	family: Family,
+	modelId: string,
+	input: GenerateInput,
+	opts: BackendOpts<GenerateOpts>
+): object {
 	if (family === 'anthropic') return buildAnthropicBody(input, opts);
 	if (family === 'meta') return buildLlamaBody(input, opts);
 	if (family === 'amazon') {
@@ -551,6 +558,7 @@ async function* parseAnthropicStream(
 	const decoder = new TextDecoder('utf-8');
 	const toolBuf = new Map<number, { id: string; name: string; argumentsBuf: string }>();
 	let finalFinishReason: GenerateResult['finishReason'] | undefined;
+	let totalArgChars = 0;
 
 	for await (const event of body) {
 		if (!event.chunk?.bytes) continue;
@@ -607,6 +615,12 @@ async function* parseAnthropicStream(
 							`Bedrock tool-call arguments exceed ${MAX_TOOL_CALL_ARGS_CHARS} chars (index ${index})`
 						);
 					}
+					totalArgChars += delta.partial_json.length;
+					if (totalArgChars > MAX_TOTAL_TOOL_CALL_ARGS_CHARS) {
+						throw new BedrockBackendError(
+							`Bedrock tool-call arguments exceed total stream cap of ${MAX_TOTAL_TOOL_CALL_ARGS_CHARS} chars`
+						);
+					}
 					acc.argumentsBuf += delta.partial_json;
 				}
 			}

components/openai/index.ts

@@ -67,6 +67,10 @@ const MAX_UPSTREAM_ERROR_MESSAGE_CHARS = 500;
 // upstream can allocate unbounded map entries (one per index value). Real
 // responses use single-digit counts.
 const MAX_TOOL_CALL_ACCUMULATOR_ENTRIES = 128;
+// Total tool-call argument chars across all entries in one stream. The per-entry
+// cap (1 MiB) plus the 128-entry cap still allows ~128 MiB accumulated; this cap
+// keeps any single stream well-bounded. Real responses use tens of KB.
+const MAX_TOTAL_TOOL_CALL_ARGS_CHARS = 8 * 1024 * 1024; // 8 MiB
 
 const log = harperLogger.forComponent('openai').conditional;
 
@@ -115,7 +119,11 @@ export class OpenAIBackend implements ModelBackend {
 	constructor(config: OpenAIBackendConfig = {}, fetchImpl: typeof fetch = fetch) {
 		this.#apiKey = requireCredential(config.apiKey, 'OpenAI', 'apiKey', OpenAIBackendError);
 		this.#baseUrl = normalizeOrigin(config.baseUrl, { host: DEFAULT_BASE_URL, secure: true });
-		this.#isNativeOpenAI = this.#baseUrl.startsWith('https://api.openai.com');
+		try {
+			this.#isNativeOpenAI = new URL(this.#baseUrl).hostname === 'api.openai.com';
+		} catch {
+			this.#isNativeOpenAI = false;
+		}
 		this.#defaultModel = config.model;
 		this.#organization = config.organization;
 		this.#requestTimeoutMs = config.requestTimeoutMs;
@@ -196,6 +204,7 @@ export class OpenAIBackend implements ModelBackend {
 		// never a partial string.
 		const toolBuf = new Map<number, ToolCallAccumulator>();
 		let finalFinishReason: GenerateResult['finishReason'] | undefined;
+		let totalArgChars = 0;
 
 		for await (const event of readSse(res.body)) {
 			const choice = event.choices?.[0];
@@ -207,7 +216,7 @@ export class OpenAIBackend implements ModelBackend {
 			}
 			if (Array.isArray(delta?.tool_calls)) {
 				for (const tcDelta of delta.tool_calls) {
-					accumulateToolCallDelta(toolBuf, tcDelta);
+					totalArgChars = accumulateToolCallDelta(toolBuf, tcDelta, totalArgChars);
 				}
 			}
 			if (choice.finish_reason) {
@@ -448,7 +457,11 @@ interface ToolCallAccumulator {
 	argumentsBuf: string;
 }
 
-function accumulateToolCallDelta(buf: Map<number, ToolCallAccumulator>, delta: OpenAIToolCallDelta): void {
+function accumulateToolCallDelta(
+	buf: Map<number, ToolCallAccumulator>,
+	delta: OpenAIToolCallDelta,
+	totalArgChars: number
+): number {
 	const index = typeof delta.index === 'number' ? delta.index : 0;
 	let acc = buf.get(index);
 	if (!acc) {
@@ -465,16 +478,23 @@ function accumulateToolCallDelta(buf: Map<number, ToolCallAccumulator>, delta: O
 	if (delta.id) acc.id = delta.id;
 	if (delta.function?.name) acc.name = delta.function.name;
 	if (typeof delta.function?.arguments === 'string') {
-		// Defend against an unbounded accumulator: the per-event SSE buffer cap
-		// stops a single oversize event, but tool-call arguments are *built up*
-		// across many sub-cap events. Throw before V8 hits string-length limits.
+		// Per-entry cap: the per-event SSE buffer cap stops a single oversize event,
+		// but tool-call arguments are *built up* across many sub-cap events.
 		if (acc.argumentsBuf.length + delta.function.arguments.length > MAX_TOOL_CALL_ARGS_CHARS) {
 			throw new OpenAIBackendError(
 				`OpenAI tool-call arguments exceed ${MAX_TOOL_CALL_ARGS_CHARS} chars (index ${index})`
 			);
 		}
+		// Total-stream cap: 128 entries each at 1 MiB still allows ~128 MiB accumulated.
+		totalArgChars += delta.function.arguments.length;
+		if (totalArgChars > MAX_TOTAL_TOOL_CALL_ARGS_CHARS) {
+			throw new OpenAIBackendError(
+				`OpenAI tool-call arguments exceed total stream cap of ${MAX_TOTAL_TOOL_CALL_ARGS_CHARS} chars`
+			);
+		}
 		acc.argumentsBuf += delta.function.arguments;
 	}
+	return totalArgChars;
 }
 
 function flushToolCallBuffer(buf: Map<number, ToolCallAccumulator>): Partial<ToolCall>[] {

resources/models/analyticsTable.ts

@@ -7,10 +7,7 @@ const log = harperLogger.forComponent('models').conditional;
 const DEFAULT_FLUSH_INTERVAL_MS = 10_000; // 10s
 const DEFAULT_MAX_BUFFER_SIZE = 1000;
 const DEFAULT_CLEANUP_INTERVAL_MS = 60 * 60 * 1000; // 1h
-// 90-day default tuned for billing windows. Operator-tunable config key will land
-// in Phase 2 alongside the YAML→registry bootstrapper (Harper's `getConfigValue`
-// only reads keys registered in `CONFIG_PARAM_MAP`, so we defer config plumbing
-// until the first real backend ships and the key has a documented owner).
+// 90-day default tuned for billing windows.
 const DEFAULT_RETENTION_MS = 90 * 24 * 60 * 60 * 1000;
 
 /**
@@ -41,6 +38,35 @@ interface BufferedRecord extends ModelCallRecord {
 }
 
 let _table: any;
+
+/**
+ * Schema attributes for `hdb_model_calls`. Exported so tests can assert on
+ * the structural shape (e.g. no spurious `indexed: true`) without parsing
+ * TypeScript source.
+ *
+ * flush() writes via tbl.primaryStore.put and cleanup() removes via
+ * primaryStore.remove — both bypass updateIndices, so secondary indexes
+ * would stay permanently empty. Match hdb_raw_analytics (write.ts) which
+ * intentionally omits `indexed` from all non-PK attributes for the same reason.
+ */
+export const MODEL_CALL_ATTRIBUTES = [
+	{ name: 'id', isPrimaryKey: true },
+	{ name: 'tenant', type: 'string' },
+	{ name: 'app', type: 'string' },
+	{ name: 'model', type: 'string' },
+	{ name: 'backend', type: 'string' },
+	{ name: 'method', type: 'string' },
+	{ name: 'adapter', type: 'string' },
+	{ name: 'conversation_id', type: 'string' },
+	{ name: 'prompt_tokens', type: 'number' },
+	{ name: 'completion_tokens', type: 'number' },
+	{ name: 'embedding_tokens', type: 'number' },
+	{ name: 'gpu_ms', type: 'number' },
+	{ name: 'latency_ms', type: 'number' },
+	{ name: 'success', type: 'boolean' },
+	{ name: 'error_code', type: 'string' },
+];
+
 /**
  * Lazy-getter for `hdb_model_calls`. Matches the convention used by
  * `getRawAnalyticsTable()` / `getAnalyticsTable()` in
@@ -54,27 +80,7 @@ export function getModelCallsTable(): any {
 		database: 'system',
 		audit: true,
 		trackDeletes: false,
-		attributes: [
-			// flush() writes via tbl.primaryStore.put and cleanup() removes via
-			// primaryStore.remove — both bypass updateIndices, so secondary indexes
-			// would stay permanently empty. Match hdb_raw_analytics (write.ts) which
-			// intentionally omits `indexed` from all non-PK attributes for the same reason.
-			{ name: 'id', isPrimaryKey: true },
-			{ name: 'tenant', type: 'string' },
-			{ name: 'app', type: 'string' },
-			{ name: 'model', type: 'string' },
-			{ name: 'backend', type: 'string' },
-			{ name: 'method', type: 'string' },
-			{ name: 'adapter', type: 'string' },
-			{ name: 'conversation_id', type: 'string' },
-			{ name: 'prompt_tokens', type: 'number' },
-			{ name: 'completion_tokens', type: 'number' },
-			{ name: 'embedding_tokens', type: 'number' },
-			{ name: 'gpu_ms', type: 'number' },
-			{ name: 'latency_ms', type: 'number' },
-			{ name: 'success', type: 'boolean' },
-			{ name: 'error_code', type: 'string' },
-		],
+		attributes: MODEL_CALL_ATTRIBUTES,
 	});
 	return _table;
 }

resources/models/backendHelpers.ts

@@ -54,6 +54,9 @@ export function assignFiniteTokenCount(
 export const MAX_RESPONSE_BODY_BYTES = 64 << 20; // 64 MiB
 export const MAX_ERROR_BODY_BYTES = 256 << 10; // 256 KiB
 
+// Module-level TextDecoder avoids per-call allocation in the streaming read path.
+const BODY_DECODER = new TextDecoder('utf-8');
+
 /**
  * Read at most `maxBytes` from `res.body`, then JSON.parse. Throws the
  * caller's error class — never a bare `SyntaxError` or `RangeError` — so
@@ -74,29 +77,42 @@ export async function readBoundedJson<T>(
 	}
 	const chunks: Uint8Array[] = [];
 	let totalBytes = 0;
-	for await (const chunk of res.body as unknown as AsyncIterable<Uint8Array>) {
-		totalBytes += chunk.byteLength;
-		if (totalBytes > maxBytes) {
-			throw new Err(
-				`${endpoint} response body exceeds ${maxBytes}-byte limit (received >${totalBytes} bytes); ` +
-					'rejecting to prevent unbounded memory use'
-			);
+	try {
+		for await (const chunk of res.body as unknown as AsyncIterable<Uint8Array>) {
+			totalBytes += chunk.byteLength;
+			if (totalBytes > maxBytes) {
+				// Release the connection promptly before throwing.
+				res.body?.cancel?.().catch(() => {});
+				throw new Err(
+					`${endpoint} response body exceeds ${maxBytes}-byte limit (received >${totalBytes} bytes); ` +
+						'rejecting to prevent unbounded memory use'
+				);
+			}
+			chunks.push(chunk);
 		}
-		chunks.push(chunk);
+	} catch (err) {
+		// Re-wrap unexpected stream errors (network abort, socket reset, etc.)
+		// so callers always see the backend's Err class, not a raw DOMException
+		// or Node stream error. Don't rewrap errors we already constructed above.
+		if (err instanceof Err) throw err;
+		throw new Err(`${endpoint} stream error while reading response body: ${(err as Error)?.message ?? err}`);
 	}
-	const merged = totalBytes === 0 ? '' : new TextDecoder('utf-8').decode(
-		chunks.length === 1
-			? chunks[0]
-			: (() => {
-				const buf = new Uint8Array(totalBytes);
-				let offset = 0;
-				for (const c of chunks) {
-					buf.set(c, offset);
-					offset += c.byteLength;
-				}
-				return buf;
-			})()
-	);
+	const merged =
+		totalBytes === 0
+			? ''
+			: BODY_DECODER.decode(
+					chunks.length === 1
+						? chunks[0]
+						: (() => {
+								const buf = new Uint8Array(totalBytes);
+								let offset = 0;
+								for (const c of chunks) {
+									buf.set(c, offset);
+									offset += c.byteLength;
+								}
+								return buf;
+							})()
+				);
 	try {
 		return JSON.parse(merged) as T;
 	} catch {