fix(models): apply multipass review findings

Adjudicated 13 findings from a 6-agent deep-review of #638 (api×2 +
data-integrity×2 + concurrency×2). 9 REAL + 1 PARTIAL + 3 NOISE.
This commit applies the 9 REAL fixes; PARTIAL (buffer backpressure
hardening) and NOISE (matches-existing-pattern) are deferred.

**F1 (Blocker) — Replication exclusion.** Added 'hdb_model_calls' to
`NON_REPLICATING_SYSTEM_TABLES` in `resources/databases.ts`. The new
table mimicked `getRawAnalyticsTable()`'s `audit: true` declaration
but missed that `hdb_raw_analytics` is in the exclusion list. Without
this, per-call rows would replicate to every cluster node — high-cost
audit-shape replication (the 150MB ProxiedRequestLog incident
pattern) and concurrent writes from two nodes could produce
identical process-local `getNextMonotonicTime()` IDs that LWW would
silently drop. One-line fix; matches the established convention.

**F2 (Important) — Pre-call errors record.** `resolveEmbedding` /
`resolveGenerative` / `requireCapability` were called BEFORE the
try/catch in `embed`/`generate`/`generateStream`, so pre-call
failures (capability mismatch, registry miss) bypassed the analytics
record. The class doc promises "both successful and failed calls
land in the table for billing visibility" — broken by these paths.
Moved resolution into the try block; failures now record with
`backend: 'unknown'` when the registry missed entirely, or with the
backend's name when it was found but the capability didn't match.
`error_code` is `'backend_not_found'`, `'capability_unsupported'`,
or `'pending_unsupported'` as appropriate.

**F3 (Important) — Context type drift.** Added `handlerPath?:
string` and `signal?: AbortSignal` to the `Context` interface in
`resources/ResourceInterface.ts`. These fields existed on the
`Request` object that becomes the ALS-bound Context for HTTP/WS
paths (via `transaction(request, ...)` at `REST.ts:92, 368`), but
weren't declared on the Context interface — the facade was reading
them via an `as any` cast. Drift bug latent until the next
refactor; documenting them on Context makes the contract explicit
and lets us drop the `as any` cast in `Models.ts`.

**F4 (Important) — Pending result duplicate row.** When a backend
returned `{ status: 'pending' }`, the facade wrote a success row
via `#record(...)`, THEN `unwrap()` threw inside the same try and
the catch wrote a second (failure) row. Replaced `unwrap()` with an
inline status check that throws BEFORE recording; pending now
produces exactly one row with `success: false` and
`error_code: 'pending_unsupported'`. Latent today (no Phase 1
backend emits pending) but ships in the contract.

**F5 (Important) — `isReadOnlyMode()` gate.** `flush()` and
`cleanup()` now short-circuit on `isReadOnlyMode()`, mirroring
`resources/analytics/write.ts:643-650`. Without this, read-only
nodes would log put/remove failures every 10s indefinitely.

**F6 (Important) — Async put rejection logging.** `Promise.allSettled`
was awaited but its results were discarded — async LMDB rejections
were silently swallowed (sync throws were already caught and
logged per-record). Iterate the settled results and `log.warn`
on each rejection so billing-visible record losses are observable.

**F13 (Important) — Stream completion flag.** `#wrapStream`'s
`finally` block recorded `success=true` whenever no exception
fired — including when the consumer `break`s the for-await loop
early. Added a `completed` flag set only after the inner for-await
exits normally; early termination now records `success=false` with
`error_code: 'aborted'`. The model did real work the caller didn't
consume; logging that as success poisoned billing aggregates.

**F11 (Nice-to-have) — Remove unregistered config key.** The writer
called `envGet('analytics.modelCallRetentionDays')` to override
retention, but the key isn't registered in `CONFIG_PARAM_MAP`, so
`getConfigValue` silently returned undefined. Removed the dead
envGet call; 90-day default is hardcoded with a TODO comment noting
that operator-tunable retention will land in Phase 2 alongside the
YAML→registry bootstrapper that owns models.* config.

**F12 (Nice-to-have) — id spread order.** Flipped `{ id: ...,
...record }` to `{ ...record, id: getNextMonotonicTime() }` so a
record that accidentally carries an `id` field can't override the
monotonic primary key. Defense-in-depth; no current caller hits
this.

Test updates:
- New: capability error writes a record (F2).
- New: pending result writes exactly ONE record (F4 — was silently a
  double-write before).
- New: ModelBackendNotFoundError records with backend='unknown' (F2).
- New: stream consumer `break` records success=false + 'aborted' (F13).
- New: pre-call generateStream failure records (F2).
- New: async put rejection (Promise.reject from primaryStore.put)
  doesn't silently swallow remaining records (F6).
- Existing tests updated where contract widened (capability test now
  asserts both the throw AND the record).

Coverage on `resources/models/`: 94.5% statements / 95.2% branches
(up from 94/92.6). Models.ts at 100% statements. All 19 Models.test
+ 12 analyticsTable.test + 8 backendRegistry.test + 11 TestBackend.test
pass.

NOT addressed in this commit (deferred):
- F7 PARTIAL (concurrent-flush backpressure hardening). Best-effort
  posture acknowledged in file header; matches analytics/write.ts.
- F8 NOISE (stale _table cache). Matches getRawAnalyticsTable
  pattern; system-DB recycling not a supported runtime op.
- F9 NOISE (eager singleton intervals). .unref()'d; matches
  existing analytics writer.
- F10 NOISE (async-gen abandonment without .return()). JS-spec
  violation by consumer; for-await consumers are safe.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: heskew

resources/ResourceInterface.ts

@@ -98,6 +98,19 @@ export interface Context {
 	_freezeRecords?: boolean; // until v5, we conditionally freeze records for back-compat
 	timestamp?: number;
 	includeExpensiveRecordCountEstimates?: boolean;
+	/**
+	 * Matched route path of the calling Resource, populated by the HTTP/WS entry points
+	 * before they hand the request into `transaction()`. Populated on the Request that
+	 * becomes the ALS-bound Context for that path; absent for ops-API, internal jobs,
+	 * and replication-driven contexts.
+	 */
+	handlerPath?: string;
+	/**
+	 * Abort signal carried through ALS so generator bodies can forward cancellation to
+	 * external work (e.g. `scope.models.generateStream({ signal })`). Populated on the
+	 * Request that becomes the ALS-bound Context for HTTP/WS paths via #513.
+	 */
+	signal?: AbortSignal;
 }
 
 export interface SourceContext<TRequestContext = Context, Record extends object = any> {

resources/databases.ts

@@ -78,6 +78,7 @@ export const NON_REPLICATING_SYSTEM_TABLES = [
 	'hdb_temp',
 	'hdb_certificate',
 	'hdb_raw_analytics',
+	'hdb_model_calls',
 	'hdb_session_will',
 	'hdb_job',
 	'hdb_info',

resources/models/Models.ts

@@ -22,12 +22,13 @@ type CallMethod = ModelCallRecord['method'];
  * Public `scope.models` facade. One instance per `Scope`.
  *
  * On every call:
- * - Resolves the configured backend via `backendRegistry` (config-driven).
+ * - Resolves the configured backend via `backendRegistry`.
  * - Reads the ALS-bound request `Context` to extract accounting context
  *   (tenantId, handlerPath) and an `AbortSignal`. Outside an ALS scope
  *   (app-init, internal jobs), accounting is empty and signal is undefined.
  * - Records the call to `hdb_model_calls` via the buffered writer — both
  *   successful and failed calls land in the table for billing visibility.
+ *   Pre-call resolution / capability errors land too, with `backend: 'unknown'`.
  *
  * The ALS pattern matches `resources/Table.ts:3517` and is rooted at
  * `resources/transaction.ts:6`.
@@ -40,68 +41,88 @@ export class Models implements ModelsContract {
 	}
 
 	async embed(input: string | string[], opts: EmbedOpts = {}): Promise<Float32Array[]> {
-		const backend = resolveEmbedding(opts.model);
-		requireCapability(backend, 'embed');
 		const { accounting, signal } = resolveCallContext(opts.signal);
-		const backendOpts: BackendOpts<EmbedOpts> = { ...opts, signal, accounting };
 		const startedAt = performance.now();
+		let backend: ModelBackend | undefined;
 		try {
+			backend = resolveEmbedding(opts.model);
+			requireCapability(backend, 'embed');
+			const backendOpts: BackendOpts<EmbedOpts> = { ...opts, signal, accounting };
 			const result = await backend.embed!(input, backendOpts);
+			// Throw on `pending` BEFORE recording success — otherwise we'd write a
+			// success row followed by a failure row from the catch (duplicate).
+			if (result.status !== 'completed') throw new ModelPendingNotSupportedError(backend.name);
 			this.#record(backend, 'embed', opts.model, accounting, undefined, result, startedAt);
-			return unwrap(backend, result);
+			return result.output;
 		} catch (err) {
 			this.#recordFailure(backend, 'embed', opts.model, accounting, undefined, startedAt, err);
 			throw err;
 		}
 	}
 
 	async generate(input: GenerateInput, opts: GenerateOpts = {}): Promise<GenerateResult> {
-		const backend = resolveGenerative(opts.model);
-		requireCapability(backend, 'generate');
 		const { accounting, signal } = resolveCallContext(opts.signal);
-		const backendOpts: BackendOpts<GenerateOpts> = { ...opts, signal, accounting };
 		const startedAt = performance.now();
+		let backend: ModelBackend | undefined;
 		try {
+			backend = resolveGenerative(opts.model);
+			requireCapability(backend, 'generate');
+			const backendOpts: BackendOpts<GenerateOpts> = { ...opts, signal, accounting };
 			const result = await backend.generate!(input, backendOpts);
+			if (result.status !== 'completed') throw new ModelPendingNotSupportedError(backend.name);
 			this.#record(backend, 'generate', opts.model, accounting, opts, result, startedAt);
-			return unwrap(backend, result);
+			return result.output;
 		} catch (err) {
 			this.#recordFailure(backend, 'generate', opts.model, accounting, opts, startedAt, err);
 			throw err;
 		}
 	}
 
 	generateStream(input: GenerateInput, opts: GenerateOpts = {}): AsyncIterable<GenerateChunk> {
-		const backend = resolveGenerative(opts.model);
-		requireCapability(backend, 'stream');
 		const { accounting, signal } = resolveCallContext(opts.signal);
+		const startedAt = performance.now();
+		let backend: ModelBackend;
+		try {
+			backend = resolveGenerative(opts.model);
+			requireCapability(backend, 'stream');
+		} catch (err) {
+			// Record pre-call failure synchronously so callers that hold but never
+			// iterate the returned iterable still produce a billing row, then rethrow.
+			this.#recordFailure(undefined, 'generateStream', opts.model, accounting, opts, startedAt, err);
+			throw err;
+		}
 		const backendOpts: BackendOpts<GenerateOpts> = { ...opts, signal, accounting };
-		return this.#wrapStream(backend, input, backendOpts, opts, accounting);
+		return this.#wrapStream(backend, input, backendOpts, opts, accounting, startedAt);
 	}
 
 	async *#wrapStream(
 		backend: ModelBackend,
 		input: GenerateInput,
 		backendOpts: BackendOpts<GenerateOpts>,
 		opts: GenerateOpts,
-		accounting: AccountingContext
+		accounting: AccountingContext,
+		startedAt: number
 	): AsyncIterable<GenerateChunk> {
-		const startedAt = performance.now();
-		let success = true;
 		let caught: unknown;
+		let completed = false;
 		try {
 			for await (const chunk of backend.generateStream!(input, backendOpts)) {
 				yield chunk;
 			}
+			completed = true;
 		} catch (err) {
-			success = false;
 			caught = err;
 			throw err;
 		} finally {
-			if (success) {
+			if (completed) {
 				this.#record(backend, 'generateStream', opts.model, accounting, opts, undefined, startedAt);
-			} else {
+			} else if (caught) {
 				this.#recordFailure(backend, 'generateStream', opts.model, accounting, opts, startedAt, caught);
+			} else {
+				// Stream terminated by the consumer (break / iter.return()) without an error
+				// from the backend. Treat as an aborted call rather than success — the model
+				// did real work that the caller didn't consume.
+				this.#recordFailure(backend, 'generateStream', opts.model, accounting, opts, startedAt, 'aborted');
 			}
 		}
 	}
@@ -120,23 +141,24 @@ export class Models implements ModelsContract {
 	}
 
 	#recordFailure(
-		backend: ModelBackend,
+		backend: ModelBackend | undefined,
 		method: CallMethod,
 		model: string | undefined,
 		accounting: AccountingContext,
 		opts: GenerateOpts | undefined,
 		startedAt: number,
-		err: unknown
+		errOrCode: unknown
 	): void {
+		const error_code = typeof errOrCode === 'string' ? errOrCode : classifyError(errOrCode);
 		this.#analyticsWriter.write({
 			...buildRecord(backend, method, model, accounting, opts, undefined, startedAt, false),
-			error_code: classifyError(err),
+			error_code,
 		});
 	}
 }
 
 function buildRecord(
-	backend: ModelBackend,
+	backend: ModelBackend | undefined,
 	method: CallMethod,
 	model: string | undefined,
 	accounting: AccountingContext,
@@ -146,7 +168,7 @@ function buildRecord(
 	success: boolean
 ): ModelCallRecord {
 	const record: ModelCallRecord = {
-		backend: backend.name,
+		backend: backend?.name ?? 'unknown',
 		method,
 		model,
 		tenant: accounting.tenantId,
@@ -166,7 +188,7 @@ function buildRecord(
 }
 
 function resolveCallContext(callerSignal?: AbortSignal): { accounting: AccountingContext; signal?: AbortSignal } {
-	const ctx = contextStorage.getStore() as any;
+	const ctx = contextStorage.getStore();
 	return {
 		accounting: {
 			tenantId: extractTenantId(ctx?.user),
@@ -184,16 +206,13 @@ function requireCapability(backend: ModelBackend, capability: 'embed' | 'generat
 	if (!backend.capabilities()[capability]) throw new ModelCapabilityError(backend.name, capability);
 }
 
-function unwrap<T>(backend: ModelBackend, result: ModelCallResult<T>): T {
-	if (result.status === 'completed') return result.output;
-	throw new ModelPendingNotSupportedError(backend.name);
-}
-
 function classifyError(err: unknown): string {
 	if (err && typeof err === 'object') {
 		const e = err as { name?: string; code?: string };
 		if (e.name === 'AbortError' || e.code === 'ABORT_ERR') return 'aborted';
 		if (e.name === 'ModelCapabilityError') return 'capability_unsupported';
+		if (e.name === 'ModelBackendNotFoundError') return 'backend_not_found';
+		if (e.name === 'ModelPendingNotSupportedError') return 'pending_unsupported';
 	}
 	return 'backend_error';
 }

resources/models/analyticsTable.ts

@@ -1,5 +1,4 @@
-import { table } from '../databases.ts';
-import { get as envGet } from '../../utility/environment/environmentManager.ts';
+import { table, isReadOnlyMode } from '../databases.ts';
 import { getNextMonotonicTime } from '../../utility/lmdb/commonUtility.ts';
 import harperLogger from '../../utility/logging/harper_logger.ts';
 
@@ -8,7 +7,11 @@ 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
-const DEFAULT_RETENTION_DAYS = 90;
+// 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).
+const DEFAULT_RETENTION_MS = 90 * 24 * 60 * 60 * 1000;
 
 /**
  * One row in `hdb_model_calls`. Field names are snake_case to match the table
@@ -109,7 +112,7 @@ export class ModelCallAnalyticsWriter {
 		const flushIntervalMs = opts.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS;
 		const cleanupIntervalMs = opts.cleanupIntervalMs ?? DEFAULT_CLEANUP_INTERVAL_MS;
 		this.#maxBufferSize = opts.maxBufferSize ?? DEFAULT_MAX_BUFFER_SIZE;
-		this.#retentionMs = opts.retentionMs ?? resolveRetentionMs();
+		this.#retentionMs = opts.retentionMs ?? DEFAULT_RETENTION_MS;
 		this.#getTable = opts.getTable ?? getModelCallsTable;
 		this.#flushTimer = setInterval(() => {
 			this.flush().catch((err) => log.warn?.(`Model-call analytics flush failed: ${err?.message ?? err}`));
@@ -123,7 +126,9 @@ export class ModelCallAnalyticsWriter {
 
 	write(record: ModelCallRecord): void {
 		if (this.#stopped) return;
-		this.#buffer.push({ id: getNextMonotonicTime(), ...record });
+		// id last so a record that accidentally carries an `id` field can't override
+		// the monotonic primary key.
+		this.#buffer.push({ ...record, id: getNextMonotonicTime() });
 		if (this.#buffer.length >= this.#maxBufferSize) {
 			// Out-of-cadence flush; swallow errors so a failing flush doesn't escape into the caller.
 			this.flush().catch((err) => log.warn?.(`Model-call analytics flush failed: ${err?.message ?? err}`));
@@ -132,26 +137,40 @@ export class ModelCallAnalyticsWriter {
 
 	async flush(): Promise<void> {
 		if (this.#buffer.length === 0) return;
+		// Read-only nodes (followers, recovery mode) shouldn't accumulate doomed writes —
+		// drop the buffer and skip. Matches `resources/analytics/write.ts:643-650`.
+		if (isReadOnlyMode()) {
+			this.#buffer = [];
+			return;
+		}
 		const batch = this.#buffer;
 		this.#buffer = [];
 		const tbl = this.#getTable();
-		const puts: Promise<unknown>[] = [];
+		const puts: Array<{ id: number; promise: Promise<unknown> }> = [];
 		for (const record of batch) {
 			try {
 				const result = tbl.primaryStore.put(record.id, record);
 				if (result && typeof (result as { then?: unknown }).then === 'function') {
-					puts.push(result as Promise<unknown>);
+					puts.push({ id: record.id, promise: result as Promise<unknown> });
 				}
 			} catch (err) {
 				log.warn?.(`Model-call analytics put failed for id=${record.id}: ${(err as Error)?.message ?? err}`);
 			}
 		}
 		if (puts.length > 0) {
-			await Promise.allSettled(puts);
+			const results = await Promise.allSettled(puts.map((p) => p.promise));
+			for (let i = 0; i < results.length; i++) {
+				const r = results[i];
+				if (r.status === 'rejected') {
+					const reason = r.reason as { message?: string } | undefined;
+					log.warn?.(`Model-call analytics async put failed for id=${puts[i].id}: ${reason?.message ?? r.reason}`);
+				}
+			}
 		}
 	}
 
 	async cleanup(): Promise<void> {
+		if (isReadOnlyMode()) return;
 		const end = Date.now() - this.#retentionMs;
 		const tbl = this.#getTable();
 		for (const key of tbl.primaryStore.getKeys({ start: false, end })) {
@@ -172,12 +191,6 @@ export class ModelCallAnalyticsWriter {
 	}
 }
 
-function resolveRetentionMs(): number {
-	const days = envGet('analytics.modelCallRetentionDays');
-	const n = typeof days === 'number' && days > 0 ? days : DEFAULT_RETENTION_DAYS;
-	return n * 24 * 60 * 60 * 1000;
-}
-
 let _writer: ModelCallAnalyticsWriter | undefined;
 /** Process-wide singleton writer. Constructed on first access. */
 export function getModelCallAnalyticsWriter(): ModelCallAnalyticsWriter {

unitTests/resources/models/Models.test.js

@@ -113,15 +113,19 @@ describe('Models facade', () => {
 			assert.strictEqual(seenSignal, ctxSignal);
 		});
 
-		it('throws ModelCapabilityError when the backend does not support embed', async () => {
+		it('throws ModelCapabilityError when the backend does not support embed AND records the failure', async () => {
 			setEmbedding('default', {
 				name: 'no-embed',
 				capabilities: () => ({ embed: false, generate: true, stream: false, tools: false, adapters: false }),
 			});
 			await assert.rejects(() => models.embed('x'), ModelCapabilityError);
+			assert.strictEqual(writer.records.length, 1);
+			assert.strictEqual(writer.records[0].backend, 'no-embed');
+			assert.strictEqual(writer.records[0].success, false);
+			assert.strictEqual(writer.records[0].error_code, 'capability_unsupported');
 		});
 
-		it('throws ModelPendingNotSupportedError when backend returns pending', async () => {
+		it('writes ONE record with success=false on pending result (not duplicate from unwrap+catch)', async () => {
 			setEmbedding('default', {
 				name: 'pending',
 				capabilities: () => ({ embed: true, generate: false, stream: false, tools: false, adapters: false }),
@@ -130,6 +134,19 @@ describe('Models facade', () => {
 				},
 			});
 			await assert.rejects(() => models.embed('x'), ModelPendingNotSupportedError);
+			assert.strictEqual(writer.records.length, 1, 'pending result must not produce duplicate rows');
+			assert.strictEqual(writer.records[0].success, false);
+			assert.strictEqual(writer.records[0].error_code, 'pending_unsupported');
+		});
+
+		it("records pre-call ModelBackendNotFoundError with backend='unknown' and error_code='backend_not_found'", async () => {
+			// No backend mapped for the requested logical name.
+			await assert.rejects(() => models.embed('x', { model: 'no-such-name' }));
+			assert.strictEqual(writer.records.length, 1);
+			assert.strictEqual(writer.records[0].backend, 'unknown');
+			assert.strictEqual(writer.records[0].model, 'no-such-name');
+			assert.strictEqual(writer.records[0].success, false);
+			assert.strictEqual(writer.records[0].error_code, 'backend_not_found');
 		});
 
 		it('writes an analytics record with success=false and sanitized error_code on backend failure', async () => {
@@ -225,5 +242,38 @@ describe('Models facade', () => {
 			assert.strictEqual(writer.records[0].success, false);
 			assert.strictEqual(writer.records[0].error_code, 'backend_error');
 		});
+
+		it("records success=false with error_code='aborted' when the consumer breaks early", async () => {
+			setGenerative('default', {
+				name: 'long-stream',
+				capabilities: () => ({ embed: false, generate: true, stream: true, tools: false, adapters: false }),
+				async *generateStream() {
+					yield { deltaContent: 'one ' };
+					yield { deltaContent: 'two ' };
+					yield { deltaContent: 'three ' };
+					yield { finishReason: 'stop' };
+				},
+			});
+			let count = 0;
+			for await (const _chunk of models.generateStream('x')) {
+				if (++count === 2) break;
+			}
+			assert.strictEqual(writer.records.length, 1);
+			assert.strictEqual(writer.records[0].success, false);
+			assert.strictEqual(writer.records[0].error_code, 'aborted');
+		});
+
+		it("records pre-call failure with backend='unknown' when the generative backend isn't registered", async () => {
+			await assert.rejects(async () => {
+				// eslint-disable-next-line no-unused-vars
+				for await (const _ of models.generateStream('x', { model: 'no-such-name' })) {
+					// will not iterate — resolve throws first
+				}
+			});
+			assert.strictEqual(writer.records.length, 1);
+			assert.strictEqual(writer.records[0].backend, 'unknown');
+			assert.strictEqual(writer.records[0].error_code, 'backend_not_found');
+			assert.strictEqual(writer.records[0].method, 'generateStream');
+		});
 	});
 });