diff --git a/packages/inference/package.json b/packages/inference/package.json index a7ae93d7d2..d60dde8195 100644 --- a/packages/inference/package.json +++ b/packages/inference/package.json @@ -159,6 +159,7 @@ "dependencies": { "@qvac/error": "^0.1.1", "@qvac/logging": "^0.1.1", + "@qvac/model-fit": "^0.7.0", "@qvac/rag": "^0.6.4", "@qvac/registry-client": "^0.6.1", "bare-abort-controller": "^1.1.2", @@ -172,6 +173,7 @@ "bare-os": "^3.9.3", "bare-path": "^3.1.1", "bare-rpc": "^1.3.8", + "bare-runtime": "^1.24.2", "bare-stream": "^2.13.3", "bare-url": "^2.4.5", "bare-zlib": "^1.4.0", @@ -182,6 +184,7 @@ "hyperswarm": "^4.17.0", "semver": "^7.8.5", "tar-stream": "^3.2.0", + "which-runtime": "^1.2.1", "zod": "^4.4.3" }, "peerDependencies": { diff --git a/packages/inference/src/model-fit/advisory-fit.ts b/packages/inference/src/model-fit/advisory-fit.ts new file mode 100644 index 0000000000..b33c69f1fa --- /dev/null +++ b/packages/inference/src/model-fit/advisory-fit.ts @@ -0,0 +1,259 @@ +import type { AbortSignal } from 'bare-abort-controller' +import type { FitLlamaResult } from '@qvac/model-fit/process' + +import { getEngineLogger } from '@/logging/index' +import type { Logger } from '@/logging/types' +import type { CanonicalModelType } from '@/schemas/index' +import { createLlamaFitRequest } from '@/model-fit/create-llama-fit-request' +import type { runIsolatedFit } from '@/model-fit/run-isolated-fit' + +/** + * Shorter than the supervisor's own 60s default: this check sits in front of a + * real load, so a wedged child must not hold the load for a full minute. Every + * expiry is `unknown` and the load continues. + */ +const ADVISORY_FIT_TIMEOUT_MS = 30_000 + +/** + * `@qvac/model-fit`'s own default margin. Made explicit here because the + * resident-model reserve below is *added* to it: setting `marginMiB` at all + * replaces the package default, so the base has to travel with the reserve. + */ +const ADVISORY_FIT_BASE_MARGIN_MIB = 1024 + +const BYTES_PER_MIB = 1024 * 1024 + +const ENABLED_VALUES = new Set(['1', 'true', 'on', 'yes']) + +/** + * `fit` and `does-not-fit` are projections of the load the SDK is about to run, + * not admission decisions. `@qvac/model-fit` duplicates the loader's policy for + * this experiment and the real loader neither consumes nor verifies the fitted + * plan, so neither verdict is denial-grade and no verdict changes the load. + */ +export type AdvisoryFitVerdict = 'fit' | 'does-not-fit' | 'unknown' + +export interface AdvisoryFitOutcome { + verdict: AdvisoryFitVerdict + /** Machine-readable explanation; never derived from log text. */ + reason: string + message?: string + plan?: { + nCtx: number + nGpuLayers: number + nGpuDevices: number + } +} + +export interface AdvisoryFitInput { + modelId: string + modelType: CanonicalModelType + modelPath: string + modelConfig: unknown + artifacts?: Record | undefined + isShardedModel: boolean +} + +/** + * Injection seams. Every field defaults to the real runtime dependency; tests + * substitute them rather than mocking modules. + */ +export interface AdvisoryFitOptions { + signal?: AbortSignal + enabled?: boolean + mobile?: boolean + timeoutMs?: number + runFit?: typeof runIsolatedFit + logger?: Logger + residentModelBytes?: () => Promise +} + +/** + * Opt-in while the result has no consumer: enabling it costs a child process + * and a full ggml backend registration on every supported load. + * + * The worker environment and the mobile runtime flag are imported lazily. Both + * modules reach Bare-only bindings, and resolving them eagerly would make this + * orchestration untestable outside a Bare runtime. + */ +async function resolveEnabled(explicit: boolean | undefined): Promise { + if (explicit !== undefined) return explicit + const { getValidatedEnv } = await import('@/runtime/env') + const value = getValidatedEnv().QVAC_ADVISORY_MODEL_FIT + return value !== undefined && ENABLED_VALUES.has(value.toLowerCase()) +} + +async function resolveMobile(explicit: boolean | undefined): Promise { + if (explicit !== undefined) return explicit + const { isMobile } = await import('@/runtime/state') + return isMobile() +} + +/** + * Sums the on-disk weight sizes of every model currently registered in this + * worker. The fit child is a fresh process, and Metal reports `free` as + * `recommendedMaxWorkingSetSize - currentAllocatedSize` *per process* + * (ggml-metal-device.m), so the child sees an idle device no matter what this + * worker holds resident. Measured consequence: a verdict that is correct on an + * idle machine admits a load that cannot decode once another model is loaded. + * + * Reserving the resident weight bytes through `marginMiB` folds that footprint + * back into the child's budget. Weight size is a lower bound — resident KV and + * compute buffers are not counted — so the verdict stays optimistic, but + * strictly less so than ignoring residency entirely. + * + * Advisory and fail-open like everything else here: any failure to stat a file + * contributes zero rather than an error. + */ +async function defaultResidentModelBytes(): Promise { + const [{ getAllModelIds, getModelInfo }, { promises: fsPromises }] = await Promise.all([ + import('@/runtime/model-registry'), + import('bare-fs') + ]) + let bytes = 0 + for (const id of getAllModelIds()) { + const info = getModelInfo(id) + if (info === null) continue + try { + const stats = (await fsPromises.stat(info.path)) as { size: number } + bytes += stats.size + } catch { + // Unreadable path: contribute nothing rather than fail the check. + } + } + return bytes +} + +/** + * Also lazy: the supervisor pulls the Bare process launcher and the packaged + * runner path, and a disabled check must not load either. + */ +async function resolveRunFit( + explicit: typeof runIsolatedFit | undefined +): Promise { + if (explicit !== undefined) return explicit + return (await import('@/model-fit/run-isolated-fit')).runIsolatedFit +} + +function unknown(reason: string, message?: string): AdvisoryFitOutcome { + return message === undefined + ? { verdict: 'unknown', reason } + : { verdict: 'unknown', reason, message } +} + +function classify(result: FitLlamaResult): AdvisoryFitOutcome { + if (result.status === 0) { + return { + verdict: 'fit', + reason: result.reason, + plan: { + nCtx: result.nCtx, + nGpuLayers: result.nGpuLayers, + nGpuDevices: result.nGpuDevices + } + } + } + if (result.status === 1) { + return { verdict: 'does-not-fit', reason: result.reason } + } + // `model-unreadable`, `no-backend-device`, and `unsupported-config` are all + // absence of evidence, not evidence of insufficiency. + return unknown(result.reason) +} + +function report(logger: Logger, input: AdvisoryFitInput, outcome: AdvisoryFitOutcome): void { + const prefix = `[advisory-fit:${input.modelType}:${input.modelId}]` + + if (outcome.verdict === 'fit') { + const plan = outcome.plan + logger.info( + `${prefix} projected to fit (advisory only)${ + plan === undefined + ? '' + : ` — nCtx ${plan.nCtx}, nGpuLayers ${plan.nGpuLayers} across ${plan.nGpuDevices} GPU device(s)` + }` + ) + return + } + + if (outcome.verdict === 'does-not-fit') { + logger.warn(`${prefix} projected not to fit (advisory only — the load continues unchanged)`) + return + } + + // `info`, not `debug`: this check only runs when it has been explicitly + // enabled to gather evidence, and "why was there no verdict" is the most + // useful thing it can report. `not-enabled` returns before reaching here. + logger.info( + `${prefix} no fit evidence: ${outcome.reason}${ + outcome.message === undefined ? '' : ` (${outcome.message})` + }` + ) +} + +/** + * Runs the advisory llama.cpp fit check for a load that is about to start. + * + * Fail-open by construction: an unsupported shape, a crashed or wedged child, a + * malformed response, and an unexpected internal error all resolve to `unknown` + * and the caller proceeds with the ordinary load path. This function never + * throws and never rejects. + */ +export async function runAdvisoryFitCheck( + input: AdvisoryFitInput, + options: AdvisoryFitOptions = {} +): Promise { + let logger: Logger | undefined = options.logger + try { + logger ??= getEngineLogger() + if (!(await resolveEnabled(options.enabled))) return unknown('not-enabled') + + const plan = createLlamaFitRequest({ + modelType: input.modelType, + modelPath: input.modelPath, + modelConfig: input.modelConfig, + artifacts: input.artifacts, + isShardedModel: input.isShardedModel, + isMobile: await resolveMobile(options.mobile) + }) + + if (!plan.supported) { + const outcome = unknown('unsupported-load', plan.detail) + report(logger, input, outcome) + return outcome + } + + const residentBytes = await (options.residentModelBytes ?? defaultResidentModelBytes)() + const residentReserveMiB = Math.ceil(residentBytes / BYTES_PER_MIB) + + const runFit = await resolveRunFit(options.runFit) + const result = await runFit( + plan.loadKind, + residentReserveMiB > 0 + ? { ...plan.config, marginMiB: ADVISORY_FIT_BASE_MARGIN_MIB + residentReserveMiB } + : plan.config, + { + timeoutMs: options.timeoutMs ?? ADVISORY_FIT_TIMEOUT_MS, + ...(options.signal !== undefined && { signal: options.signal }) + } + ) + + const outcome = + result.status === 'completed' + ? classify(result.result) + : unknown(result.reason, result.message) + report(logger, input, outcome) + return outcome + } catch (error) { + const outcome = unknown( + 'internal-error', + error instanceof Error ? `${error.name}: ${error.message}` : String(error) + ) + try { + if (logger !== undefined) report(logger, input, outcome) + } catch { + // A failing logger must not turn an advisory check into a load failure. + } + return outcome + } +} diff --git a/packages/inference/src/model-fit/create-llama-fit-request.ts b/packages/inference/src/model-fit/create-llama-fit-request.ts new file mode 100644 index 0000000000..8644a6aa31 --- /dev/null +++ b/packages/inference/src/model-fit/create-llama-fit-request.ts @@ -0,0 +1,204 @@ +import type { FitLlamaProcessConfig, LlamaLoadKind } from '@qvac/model-fit/process' + +import { + ModelType, + type CanonicalModelType, + type EmbedConfig, + type LlmConfig +} from '@/schemas/index' +import { transformLlmConfig } from '@/plugins/builtin/llamacpp-completion/transform' +import { transformEmbedConfig } from '@/plugins/builtin/llamacpp-embedding/transform' + +/** + * Keys `@qvac/model-fit` reads as load evidence, in the spelling the SDK's own + * completion/embedding transforms emit. The package canonicalizes `_` to `-`, + * so `ctx_size` and `ctx-size` reach the same native setting. + * + * This mirrors `SUPPORTED_LOAD_KEYS` in the package's `LlamaLoadConfig.cpp` + * intersected with what the two transforms can produce. It is a duplicated + * policy for the duration of the experiment: a key added to a load config + * without being classified here must not silently change the question the + * fitter answers, so `partitionParams` refuses the request instead. + */ +const FIT_LOAD_KEYS: Record = { + completion: [ + 'device', + 'ctx_size', + 'gpu_layers', + 'no_mmap', + 'parallel', + 'cache-type-k', + 'cache-type-v', + 'main-gpu', + 'split-mode', + 'tensor-split' + ], + embedding: [ + 'device', + 'gpu_layers', + 'batch_size', + 'flash_attn', + 'main-gpu', + 'split-mode', + 'tensor-split' + ] +} + +/** + * Keys that reach the addon but cannot move a load's memory footprint — + * sampling, generation, logging, and JS-side presentation settings. They are + * dropped rather than forwarded: `@qvac/model-fit` rejects everything outside + * its own allowlist, and a key it deliberately ignores must not turn a + * supported load into `unsupported-config`. + */ +const NON_FIT_KEYS: Record = { + completion: [ + 'temp', + 'top_p', + 'top_k', + 'seed', + 'predict', + 'presence_penalty', + 'frequency_penalty', + 'repeat_penalty', + 'reverse_prompt', + 'n_discarded', + 'tools', + 'verbosity', + 'reasoning_budget', + 'image_tile_mode', + 'image_no_upscale', + 'mmproj-use-gpu', + 'openclCacheDir' + ], + embedding: ['pooling', 'attention', 'embd_normalize', 'verbosity', 'openclCacheDir'] +} + +/** Load-config keys that describe a shape the fitter cannot answer for. */ +const UNSUPPORTED_KEYS: readonly string[] = ['lora'] + +/** + * llama.cpp's fitter constrains device memory but treats host memory as + * unlimited, so a CPU load is always projected to fit no matter how large the + * model is. Measured on a 24 GiB M4 Pro: an 18.3 GiB model at 32k context + * reports `fits` on `device: 'cpu'`. That answer carries no information, so it + * is refused here rather than spending a child process to produce it. + */ +function isCpuLoad(params: Record): boolean { + return params['device']?.toLowerCase() === 'cpu' +} + +export type LlamaFitRequestPlan = + | { supported: true; loadKind: LlamaLoadKind; config: FitLlamaProcessConfig } + | { supported: false; detail: string } + +export interface CreateLlamaFitRequestParams { + modelType: CanonicalModelType + modelPath: string + modelConfig: unknown + artifacts?: Record | undefined + isShardedModel: boolean + isMobile: boolean +} + +function loadKindFor(modelType: CanonicalModelType): LlamaLoadKind | undefined { + if (modelType === ModelType.llamacppCompletion) return 'completion' + if (modelType === ModelType.llamacppEmbedding) return 'embedding' + return undefined +} + +function unsupported(detail: string): LlamaFitRequestPlan { + return { supported: false, detail } +} + +function partitionParams( + loadKind: LlamaLoadKind, + transformed: Record +): { params: Record } | { detail: string } { + const forwarded = new Set(FIT_LOAD_KEYS[loadKind]) + const dropped = new Set(NON_FIT_KEYS[loadKind]) + const params: Record = {} + + for (const [key, value] of Object.entries(transformed)) { + if (UNSUPPORTED_KEYS.includes(key)) { + return { detail: `unsupported load setting: ${key}` } + } + if (forwarded.has(key)) { + params[key] = value + continue + } + if (dropped.has(key)) continue + // Neither fit evidence nor a known non-memory setting: this load carries a + // setting the SDK cannot classify, so it must not be answered for. + return { detail: `unclassified load setting: ${key}` } + } + + return { params } +} + +/** + * Pins the requested context so the fitter reports on the exact load the SDK + * is about to run. Left unset for an auto context (`0`, "use the model's + * trained context"), where the package's own floor applies and the fitter + * stays free to reduce. + */ +function contextFloor(params: Record): number | undefined { + const ctxSize = params['ctx_size'] + if (ctxSize === undefined) return undefined + const parsed = Number(ctxSize) + if (!Number.isSafeInteger(parsed) || parsed <= 0) return undefined + return parsed +} + +/** + * Builds a protocol-v2 fit request from the same resolved model config the real + * load is about to use, or explains why this load cannot be answered for. + * + * Structural shapes the SDK owns are refused here so no child process starts. + * Value-level policy (device names, symbolic GPU selection, context bounds) + * stays inside `@qvac/model-fit`, which reports `unsupported-config` for it. + */ +export function createLlamaFitRequest(params: CreateLlamaFitRequestParams): LlamaFitRequestPlan { + if (params.isMobile) { + return unsupported('mobile has no disposable process boundary') + } + + const loadKind = loadKindFor(params.modelType) + if (loadKind === undefined) { + return unsupported(`model type is not a llama.cpp load: ${params.modelType}`) + } + + if (params.isShardedModel) { + return unsupported('sharded models are not representable') + } + + if (loadKind === 'completion' && params.artifacts?.['projectionModelPath'] !== undefined) { + return unsupported('multimodal projection loads are not representable') + } + + const modelConfig = (params.modelConfig ?? {}) as Record + const transformed = + loadKind === 'completion' + ? transformLlmConfig(modelConfig as LlmConfig) + : (transformEmbedConfig(modelConfig as EmbedConfig) as unknown as Record) + + const partitioned = partitionParams(loadKind, transformed) + if ('detail' in partitioned) return unsupported(partitioned.detail) + + if (isCpuLoad(partitioned.params)) { + return unsupported('cpu loads carry no device-memory evidence') + } + + const nCtxMin = contextFloor(partitioned.params) + + return { + supported: true, + loadKind, + config: { + // Sharded loads are refused above, so the resolved path is the whole model. + modelPath: params.modelPath, + params: partitioned.params, + ...(nCtxMin !== undefined && { nCtxMin }) + } + } +} diff --git a/packages/inference/src/model-fit/run-isolated-fit.ts b/packages/inference/src/model-fit/run-isolated-fit.ts new file mode 100644 index 0000000000..cd7faf18a6 --- /dev/null +++ b/packages/inference/src/model-fit/run-isolated-fit.ts @@ -0,0 +1,560 @@ +import { + encodeFitLlamaProcessRequest, + FIT_PROCESS_MAX_RESPONSE_BYTES, + parseFitProcessResponse, + resolveFitProcessRunnerPath, + type FitLlamaProcessConfig, + type FitLlamaResult, + type LlamaLoadKind +} from '@qvac/model-fit/process' +import type { AbortSignal } from 'bare-abort-controller' +import env from 'bare-env' +import spawnBare from 'bare-runtime/spawn' +import { arch, isAndroid, isBrowser, isIOS, platform } from 'which-runtime' + +const DEFAULT_TIMEOUT_MS = 60_000 +const TERMINATION_GRACE_MS = 1_000 +const FINAL_KILL_GRACE_MS = 1_000 +const DRAIN_GRACE_MS = 1_000 +const STDERR_TAIL_BYTES = 16 * 1024 + +const FIT_ENVIRONMENT_KEYS = [ + 'HOME', + 'PATH', + 'TMPDIR', + 'TMP', + 'TEMP', + 'SystemRoot', + 'WINDIR', + 'LD_LIBRARY_PATH', + 'DYLD_LIBRARY_PATH', + 'VK_ICD_FILENAMES', + 'VK_DRIVER_FILES', + 'CUDA_VISIBLE_DEVICES', + 'HIP_VISIBLE_DEVICES', + 'ROCR_VISIBLE_DEVICES' +] as const + +export type IsolatedFitUnknownReason = + | 'unsupported-platform' + | 'spawn-failed' + | 'timeout' + | 'cancelled' + | 'crashed' + | 'invalid-response' + | 'invocation-error' + +export type IsolatedFitResult = + | { status: 'completed'; result: FitLlamaResult } + | { + status: 'unknown' + reason: IsolatedFitUnknownReason + message: string + stderrTail?: string + } + +export interface SpawnContext { + command: string + options: { + args: string[] + platform: string + arch: string + stdio: string[] + env: Record + } +} + +export interface RuntimeContext { + platform: string + arch: string + isAndroid: boolean + isBrowser: boolean + isIOS: boolean +} + +export interface RunIsolatedFitOptions { + timeoutMs?: number + signal?: AbortSignal + spawnProcess?: (context: SpawnContext) => ChildProcess + runtime?: RuntimeContext + environment?: Record + runnerPath?: string + runnerArgs?: string[] + terminationGraceMs?: number + finalKillGraceMs?: number + drainGraceMs?: number +} + +interface PendingTermination { + reason: IsolatedFitUnknownReason + message: string + crashCanOverride: boolean +} + +export interface ErrorEmitter { + on(event: 'error', listener: (error: Error) => void): unknown + off(event: 'error', listener: (error: Error) => void): unknown +} + +export interface ReadableChildStream { + setEncoding(encoding: 'utf8'): void + destroy(): void + on(event: 'error', listener: (error: Error) => void): unknown + on(event: 'data', listener: (chunk: string) => void): unknown + off(event: 'error', listener: (error: Error) => void): unknown + off(event: 'data', listener: (chunk: string) => void): unknown +} + +export interface WritableChildStream { + end(data?: string): void + destroy(): void + on(event: 'error', listener: (error: Error) => void): unknown + off(event: 'error', listener: (error: Error) => void): unknown +} + +export interface ChildProcess { + stdin: WritableChildStream | null + stdout: ReadableChildStream | null + stderr: ReadableChildStream | null + kill(signal?: string): boolean + on(event: 'error', listener: (error: Error) => void): unknown + on( + event: 'exit' | 'close', + listener: (code: number | null, signal: string | number | null) => void + ): unknown + off(event: 'error', listener: (error: Error) => void): unknown + off( + event: 'exit' | 'close', + listener: (code: number | null, signal: string | number | null) => void + ): unknown +} + +function ignoreLateError(): void {} + +function allowedEnvironment( + environment: Record +): Record { + const allowed: Record = {} + for (const key of FIT_ENVIRONMENT_KEYS) { + const value = environment[key] + if (value !== undefined) allowed[key] = value + } + return allowed +} + +// Local bare-buffer typings vary on the instance surface; the runtime value +// is always a real Buffer, so intersect with Uint8Array for length/subarray. +type TailBytes = Buffer & Uint8Array + +function appendTail(current: TailBytes, chunk: string): TailBytes { + const combined = Buffer.concat([current, Buffer.from(chunk)]) as TailBytes + return combined.length <= STDERR_TAIL_BYTES + ? combined + : (combined.subarray(combined.length - STDERR_TAIL_BYTES) as TailBytes) +} + +function unknown( + reason: IsolatedFitUnknownReason, + message: string, + stderrTail: TailBytes +): IsolatedFitResult { + return stderrTail.length === 0 + ? { status: 'unknown', reason, message } + : { status: 'unknown', reason, message, stderrTail: stderrTail.toString() } +} + +function formatError(error: unknown): string { + return error instanceof Error ? `${error.name}: ${error.message}` : String(error) +} + +function normalizeExitSignal(signal: string | number | null): string | null { + return signal === null || signal === 0 ? null : String(signal) +} + +function parseResponse(line: string, stderrTail: Buffer): IsolatedFitResult { + try { + const response = parseFitProcessResponse(JSON.parse(line)) + if (response.status === 'invocation-error') { + return unknown( + 'invocation-error', + `${response.error.name}: ${response.error.message}`, + stderrTail + ) + } + return { status: 'completed', result: response.result } + } catch (error) { + return unknown('invalid-response', formatError(error), stderrTail) + } +} + +export function runIsolatedFit( + loadKind: LlamaLoadKind, + config: FitLlamaProcessConfig, + options: RunIsolatedFitOptions = {} +): Promise { + const runtime = options.runtime ?? { + platform, + arch, + isAndroid, + isBrowser, + isIOS + } + if (runtime.isAndroid || runtime.isBrowser || runtime.isIOS) { + return Promise.resolve({ + status: 'unknown', + reason: 'unsupported-platform', + message: `Fit subprocess isolation is unavailable on ${runtime.platform}` + }) + } + + const spawnProcess = + options.spawnProcess ?? + ((context: SpawnContext) => + spawnBare(context.command, context.options) as unknown as ChildProcess) + const environment = options.environment ?? env + const timeoutMs = options.timeoutMs ?? DEFAULT_TIMEOUT_MS + const terminationGraceMs = options.terminationGraceMs ?? TERMINATION_GRACE_MS + const finalKillGraceMs = options.finalKillGraceMs ?? FINAL_KILL_GRACE_MS + const drainGraceMs = options.drainGraceMs ?? DRAIN_GRACE_MS + + return new Promise((resolve) => { + let child: ChildProcess | undefined + const streams: { + stdin: WritableChildStream | undefined + stdout: ReadableChildStream | undefined + stderr: ReadableChildStream | undefined + } = { + stdin: undefined, + stdout: undefined, + stderr: undefined + } + const timers: { + timeout: ReturnType | undefined + forceKill: ReturnType | undefined + finalKill: ReturnType | undefined + drain: ReturnType | undefined + } = { + timeout: undefined, + forceKill: undefined, + finalKill: undefined, + drain: undefined + } + let settled = false + let destroyingStreams = false + let stdout = '' + let stdoutFailure: string | undefined + let stderrTail: TailBytes = Buffer.alloc(0) as TailBytes + let pendingTermination: PendingTermination | undefined + let exitStatus: { code: number | null; signal: string | null } | undefined + + function safely(action: () => void): void { + try { + action() + } catch {} + } + + // A destroyed pipe or a reaped child can still emit EPIPE/ERR_STREAM_DESTROYED + // after the promise settles; without a listener that would take down the host, + // so the active handler is swapped for an inert one instead of removed. + function silenceErrors( + emitter: ErrorEmitter | undefined, + active: (error: Error) => void + ): void { + if (emitter === undefined) return + safely(() => emitter.on('error', ignoreLateError)) + safely(() => emitter.off('error', active)) + } + + function cleanup(): void { + if (timers.timeout !== undefined) clearTimeout(timers.timeout) + if (timers.forceKill !== undefined) clearTimeout(timers.forceKill) + if (timers.finalKill !== undefined) clearTimeout(timers.finalKill) + if (timers.drain !== undefined) clearTimeout(timers.drain) + options.signal?.removeEventListener('abort', onAbort) + safely(() => child?.off('exit', onExit)) + safely(() => child?.off('close', onClose)) + safely(() => streams.stdout?.off('data', onStdoutData)) + safely(() => streams.stderr?.off('data', onStderrData)) + silenceErrors(child, onChildError) + silenceErrors(streams.stdin, onStdinError) + silenceErrors(streams.stdout, onStdoutError) + silenceErrors(streams.stderr, onStderrError) + } + + function settle(result: IsolatedFitResult): void { + if (settled) return + settled = true + cleanup() + resolve(result) + } + + function kill(signal: string): void { + safely(() => { + child?.kill(signal) + }) + } + + function requestTermination( + reason: IsolatedFitUnknownReason, + message: string, + crashCanOverride = false + ): void { + if (settled || exitStatus !== undefined || pendingTermination !== undefined) return + pendingTermination = { reason, message, crashCanOverride } + timers.forceKill = setTimeout(() => { + if (settled || exitStatus !== undefined) return + kill('SIGKILL') + if (exitStatus !== undefined) return + timers.finalKill = setTimeout(() => { + if (settled || exitStatus !== undefined || pendingTermination === undefined) return + // Bound liveness when the child cannot be observed as reaped after SIGKILL. + destroyLocalStreams() + settle( + unknown( + pendingTermination.reason, + `${pendingTermination.message}; child did not report exit`, + stderrTail + ) + ) + }, finalKillGraceMs) + }, terminationGraceMs) + kill('SIGTERM') + } + + function destroyLocalStreams(): void { + destroyingStreams = true + safely(() => streams.stdin?.destroy()) + safely(() => streams.stdout?.destroy()) + safely(() => streams.stderr?.destroy()) + } + + function finalize(code: number | null, signal: string | null): void { + const newlineIndex = stdout.indexOf('\n') + const hasTrailingTerminator = newlineIndex === stdout.length - 1 + const invalidFraming = newlineIndex !== -1 && !hasTrailingTerminator + const line = hasTrailingTerminator ? stdout.slice(0, -1) : stdout + const response = + stdoutFailure === undefined && !invalidFraming && line !== '' + ? parseResponse(line, stderrTail) + : undefined + + if (pendingTermination !== undefined && pendingTermination.crashCanOverride === false) { + settle(unknown(pendingTermination.reason, pendingTermination.message, stderrTail)) + return + } + + // A signalled death outranks the runner's own diagnosis: the response was + // written before whatever killed the child, so it cannot describe the exit. + if ( + signal === null && + response?.status === 'unknown' && + response.reason === 'invocation-error' + ) { + settle(response) + return + } + + if (code !== 0 || signal !== null) { + settle( + unknown( + 'crashed', + `Fit subprocess exited with code ${String(code)} and signal ${String(signal)}`, + stderrTail + ) + ) + return + } + + if (pendingTermination !== undefined) { + settle(unknown(pendingTermination.reason, pendingTermination.message, stderrTail)) + return + } + + if (stdoutFailure !== undefined) { + settle(unknown('invalid-response', stdoutFailure, stderrTail)) + return + } + + if (invalidFraming || line === '') { + settle( + unknown( + 'invalid-response', + invalidFraming + ? 'Fit subprocess returned invalid line framing' + : `Fit subprocess exited with code ${String(code)} and signal ${String(signal)}`, + stderrTail + ) + ) + return + } + + settle(response ?? parseResponse(line, stderrTail)) + } + + function onAbort(): void { + requestTermination('cancelled', 'Fit subprocess was cancelled') + } + + function onChildError(error: Error): void { + requestTermination('spawn-failed', formatError(error)) + } + + function onExit(code: number | null, rawSignal: string | number | null): void { + if (exitStatus !== undefined) return + const signal = normalizeExitSignal(rawSignal) + exitStatus = { code, signal } + if (timers.timeout !== undefined) { + clearTimeout(timers.timeout) + timers.timeout = undefined + } + if (timers.forceKill !== undefined) { + clearTimeout(timers.forceKill) + timers.forceKill = undefined + } + if (timers.finalKill !== undefined) { + clearTimeout(timers.finalKill) + timers.finalKill = undefined + } + timers.drain = setTimeout(() => { + destroyLocalStreams() + finalize(code, signal) + }, drainGraceMs) + } + + function onClose(code: number | null, rawSignal: string | number | null): void { + const status = exitStatus ?? { code, signal: normalizeExitSignal(rawSignal) } + finalize(status.code, status.signal) + } + + function onStdoutData(chunk: string): void { + if (settled || pendingTermination !== undefined || destroyingStreams) { + return + } + + if (Buffer.byteLength(stdout) + Buffer.byteLength(chunk) > FIT_PROCESS_MAX_RESPONSE_BYTES) { + if (exitStatus !== undefined) { + stdoutFailure = 'Fit subprocess response exceeds 1 MiB' + return + } + requestTermination('invalid-response', 'Fit subprocess response exceeds 1 MiB') + return + } + stdout += chunk + const newlineIndex = stdout.indexOf('\n') + if (newlineIndex !== -1 && newlineIndex !== stdout.length - 1) { + if (exitStatus !== undefined) { + stdoutFailure = 'Fit subprocess returned invalid line framing' + } else { + requestTermination('invalid-response', 'Fit subprocess returned invalid line framing') + } + } + } + + function onStderrData(chunk: string): void { + if (settled || destroyingStreams) return + stderrTail = appendTail(stderrTail, chunk) + } + + function onStdinError(error: Error): void { + if (destroyingStreams) return + requestTermination('invalid-response', `Fit subprocess stdin failed: ${error.message}`, true) + } + + function onStdoutError(error: Error): void { + if (destroyingStreams) return + requestTermination('invalid-response', `Fit subprocess stdout failed: ${error.message}`) + } + + function onStderrError(error: Error): void { + if (destroyingStreams) return + requestTermination('invalid-response', `Fit subprocess stderr failed: ${error.message}`) + } + + try { + const stdio: string[] = + runtime.platform === 'win32' + ? ['overlapped', 'overlapped', 'overlapped'] + : ['pipe', 'pipe', 'pipe'] + child = spawnProcess({ + command: 'bare', + options: { + args: [ + options.runnerPath ?? resolveFitProcessRunnerPath(), + ...(options.runnerArgs ?? []) + ], + platform: runtime.platform, + arch: runtime.arch, + stdio, + env: allowedEnvironment(environment) + } + }) + } catch (error) { + settle(unknown('spawn-failed', formatError(error), stderrTail)) + return + } + + child.on('error', onChildError) + child.on('exit', onExit) + child.on('close', onClose) + + const stdin = child.stdin + const stdoutStream = child.stdout + const stderrStream = child.stderr + streams.stdin = stdin ?? undefined + streams.stdout = stdoutStream ?? undefined + streams.stderr = stderrStream ?? undefined + + options.signal?.addEventListener('abort', onAbort, { once: true }) + timers.timeout = setTimeout(() => { + requestTermination('timeout', `Fit subprocess exceeded ${timeoutMs}ms`) + }, timeoutMs) + + try { + streams.stdin?.on('error', onStdinError) + streams.stdout?.on('data', onStdoutData) + streams.stdout?.on('error', onStdoutError) + streams.stderr?.on('data', onStderrData) + streams.stderr?.on('error', onStderrError) + } catch (error) { + requestTermination( + 'invalid-response', + `Fit subprocess stream setup failed: ${formatError(error)}` + ) + destroyLocalStreams() + return + } + + if (stdin === null || stdoutStream === null || stderrStream === null) { + requestTermination('spawn-failed', 'Fit subprocess stdio pipes are unavailable') + destroyLocalStreams() + return + } + + try { + stdoutStream.setEncoding('utf8') + stderrStream.setEncoding('utf8') + } catch (error) { + requestTermination( + 'invalid-response', + `Fit subprocess stream setup failed: ${formatError(error)}` + ) + destroyLocalStreams() + return + } + + if (options.signal?.aborted === true) { + onAbort() + return + } + + try { + stdin.end(encodeFitLlamaProcessRequest(loadKind, config)) + } catch (error) { + requestTermination( + 'invalid-response', + `Fit subprocess request write failed: ${formatError(error)}` + ) + destroyLocalStreams() + } + }) +} diff --git a/packages/inference/src/plugins/builtin/llamacpp-embedding/plugin.ts b/packages/inference/src/plugins/builtin/llamacpp-embedding/plugin.ts index 5583e9013a..72e3720ef2 100644 --- a/packages/inference/src/plugins/builtin/llamacpp-embedding/plugin.ts +++ b/packages/inference/src/plugins/builtin/llamacpp-embedding/plugin.ts @@ -1,4 +1,4 @@ -import EmbedLlamacpp, { type GGMLConfig } from '@qvac/embed-llamacpp' +import EmbedLlamacpp from '@qvac/embed-llamacpp' import embedAddonLogging from '@qvac/embed-llamacpp/addonLogging' import { definePlugin, @@ -18,53 +18,7 @@ import { embed } from '@/plugins/ops/embed' import { forwardModelExecution } from '@/profiling/model-execution' import { isMobile } from '@/runtime/state' import { stripMultiGpuKeys } from '@/utils/multi-gpu-mobile' - -function transformEmbedConfig(embedConfig: EmbedConfig): GGMLConfig { - const config: GGMLConfig = { - device: embedConfig.device as 'gpu' | 'cpu', - gpu_layers: `${embedConfig.gpuLayers}` as `${number}`, - batch_size: `${embedConfig.batchSize}` as `${number}` - } - - if (embedConfig.flashAttention) { - config.flash_attn = embedConfig.flashAttention - } - - if (embedConfig.pooling) { - config.pooling = embedConfig.pooling - } - - if (embedConfig.attention) { - config.attention = embedConfig.attention - } - - if (typeof embedConfig.embdNormalize === 'number') { - config.embd_normalize = `${embedConfig.embdNormalize}` - } - - if (embedConfig.mainGpu !== undefined) { - config['main-gpu'] = - typeof embedConfig.mainGpu === 'number' ? `${embedConfig.mainGpu}` : embedConfig.mainGpu - } - - if (embedConfig.splitMode) { - config['split-mode'] = embedConfig.splitMode - } - - if (embedConfig.tensorSplit) { - config['tensor-split'] = embedConfig.tensorSplit - } - - if (typeof embedConfig.verbosity === 'number') { - config.verbosity = `${embedConfig.verbosity}` - } - - if (embedConfig.openclCacheDir) { - config.openclCacheDir = embedConfig.openclCacheDir - } - - return config -} +import { transformEmbedConfig } from '@/plugins/builtin/llamacpp-embedding/transform' function createEmbeddingsModel(modelId: string, modelPath: string, embedConfig: EmbedConfig) { const logger = createStreamLogger(modelId, ModelType.llamacppEmbedding) diff --git a/packages/inference/src/plugins/builtin/llamacpp-embedding/transform.ts b/packages/inference/src/plugins/builtin/llamacpp-embedding/transform.ts new file mode 100644 index 0000000000..965a0c4356 --- /dev/null +++ b/packages/inference/src/plugins/builtin/llamacpp-embedding/transform.ts @@ -0,0 +1,55 @@ +import type { GGMLConfig } from '@qvac/embed-llamacpp' +import { type EmbedConfig } from '@/schemas/index' + +/** + * Converts an EmbedConfig into the flat string-keyed map the C++ addon expects. + * + * Extracted from the plugin so the advisory fit check can build its request + * from the same transform the real embedding load uses. + */ +export function transformEmbedConfig(embedConfig: EmbedConfig): GGMLConfig { + const config: GGMLConfig = { + device: embedConfig.device as 'gpu' | 'cpu', + gpu_layers: `${embedConfig.gpuLayers}` as `${number}`, + batch_size: `${embedConfig.batchSize}` as `${number}` + } + + if (embedConfig.flashAttention) { + config.flash_attn = embedConfig.flashAttention + } + + if (embedConfig.pooling) { + config.pooling = embedConfig.pooling + } + + if (embedConfig.attention) { + config.attention = embedConfig.attention + } + + if (typeof embedConfig.embdNormalize === 'number') { + config.embd_normalize = `${embedConfig.embdNormalize}` + } + + if (embedConfig.mainGpu !== undefined) { + config['main-gpu'] = + typeof embedConfig.mainGpu === 'number' ? `${embedConfig.mainGpu}` : embedConfig.mainGpu + } + + if (embedConfig.splitMode) { + config['split-mode'] = embedConfig.splitMode + } + + if (embedConfig.tensorSplit) { + config['tensor-split'] = embedConfig.tensorSplit + } + + if (typeof embedConfig.verbosity === 'number') { + config.verbosity = `${embedConfig.verbosity}` + } + + if (embedConfig.openclCacheDir) { + config.openclCacheDir = embedConfig.openclCacheDir + } + + return config +} diff --git a/packages/inference/src/plugins/ops/load-model.ts b/packages/inference/src/plugins/ops/load-model.ts index c6c66ca95b..72ff2ea6e4 100644 --- a/packages/inference/src/plugins/ops/load-model.ts +++ b/packages/inference/src/plugins/ops/load-model.ts @@ -21,6 +21,7 @@ import { ModelFileLocateFailedError } from '@/errors/index' import { getPlugin } from '@/plugins/index' +import { runAdvisoryFitCheck } from '@/model-fit/advisory-fit' import { promises as fsPromises } from 'bare-fs' import path from 'bare-path' import { getEngineLogger } from '@/logging/index' @@ -93,6 +94,20 @@ export async function loadModel( } } + // Experimental, opt-in, and advisory: every outcome — including a projected + // insufficiency — continues to the ordinary load below. Runs after config + // resolution and path validation so it sees the same state the real load + // uses, and before `createModel()` so it never competes with the native + // load for device memory. + await runAdvisoryFitCheck({ + modelId, + modelType: modelType as CanonicalModelType, + modelPath, + modelConfig, + artifacts, + isShardedModel + }) + logger.info(`${modelType}: Loading model ${modelId}...`) startLogBuffering(modelId) diff --git a/packages/inference/src/runtime/env.ts b/packages/inference/src/runtime/env.ts index fa108f297c..ecb60648cf 100644 --- a/packages/inference/src/runtime/env.ts +++ b/packages/inference/src/runtime/env.ts @@ -2,7 +2,13 @@ import env from 'bare-env' import { z } from 'zod' const envSchema = z.object({ - HOME_DIR: z.string() + HOME_DIR: z.string(), + /** + * Opt-in for the experimental advisory llama.cpp fit check run before a + * completion/embedding load. Free-form rather than an enum so a typo cannot + * fail engine startup; only `1`/`true`/`on`/`yes` enable it. + */ + QVAC_ADVISORY_MODEL_FIT: z.string().optional() }) type Env = z.infer @@ -17,7 +23,8 @@ export function initEnv(): void { // Snap's HOME can be revision-scoped; SNAP_USER_COMMON is stable. env['SNAP_USER_COMMON'] ?? env['HOME'] ?? env['USERPROFILE'] ?? '/tmp' let envConfig: Record = { - HOME_DIR: defaultHomeDir + HOME_DIR: defaultHomeDir, + QVAC_ADVISORY_MODEL_FIT: env['QVAC_ADVISORY_MODEL_FIT'] } const isBareKit = typeof (globalThis as { BareKit?: unknown }).BareKit !== 'undefined' diff --git a/packages/inference/src/types/bare-runtime/index.d.ts b/packages/inference/src/types/bare-runtime/index.d.ts new file mode 100644 index 0000000000..6b9667af96 --- /dev/null +++ b/packages/inference/src/types/bare-runtime/index.d.ts @@ -0,0 +1,18 @@ +declare module 'bare-runtime/spawn' { + export interface SpawnOptions { + args?: string[] + platform?: string + arch?: string + stdio?: string[] + } + + export interface ChildProcess { + pid: number | null + killed: boolean + kill(signal?: string): boolean + on(event: 'exit', listener: (code: number | null, signal: string | null) => void): this + on(event: string, listener: (...args: unknown[]) => void): this + } + + export default function spawn(command: string, options?: SpawnOptions): ChildProcess +} diff --git a/packages/inference/src/types/which-runtime/index.d.ts b/packages/inference/src/types/which-runtime/index.d.ts new file mode 100644 index 0000000000..a6ed96f3b9 --- /dev/null +++ b/packages/inference/src/types/which-runtime/index.d.ts @@ -0,0 +1,11 @@ +declare module 'which-runtime' { + export const platform: string + export const arch: string + export const isAndroid: boolean + export const isBrowser: boolean + export const isIOS: boolean + export const isBare: boolean + export const isNode: boolean + export const isMobile: boolean + export const isBareKit: boolean +} diff --git a/packages/inference/test/fixtures/model-fit/fit-runner-fixture.ts b/packages/inference/test/fixtures/model-fit/fit-runner-fixture.ts new file mode 100644 index 0000000000..99be918562 --- /dev/null +++ b/packages/inference/test/fixtures/model-fit/fit-runner-fixture.ts @@ -0,0 +1,69 @@ +import process from 'bare-process' + +type FixtureMode = 'completed' | 'error' | 'hang' | 'abort' + +function write(stream: unknown, value: string): void { + const writable = stream as { write(value: string): void } + writable.write(value) +} + +function parseMode(value: string | undefined): FixtureMode { + switch (value) { + case 'completed': + case 'error': + case 'hang': + case 'abort': + return value + default: + throw new TypeError(`Unknown fixture mode: ${String(value)}`) + } +} + +const mode = parseMode(process.argv[2]) + +switch (mode) { + case 'completed': + write( + process.stdout, + `${JSON.stringify({ + version: 2, + status: 'completed', + result: { + status: 0, + fits: true, + reason: 'fits', + maxDevices: 1, + nDevices: 1, + nGpuDevices: 1, + nGpuLayers: 32, + nCtx: 4096, + nBatch: 512, + nUbatch: 512, + tensorSplit: [1], + buftOverrides: [], + splitMode: 1, + mainGpu: 0, + typeK: 1, + typeV: 1, + flashAttnType: 1 + } + })}\n` + ) + process.exitCode = 0 + break + case 'error': + write(process.stderr, 'fixture failed\n') + process.exitCode = 17 + break + case 'hang': + setInterval(() => {}, 1_000) + break + case 'abort': + process.kill(process.pid, 'SIGABRT') + break + default: { + const exhaustive: never = mode + write(process.stderr, `Unhandled fixture mode: ${String(exhaustive)}\n`) + process.exitCode = 2 + } +} diff --git a/packages/inference/test/model-fit-advisory-fit.test.ts b/packages/inference/test/model-fit-advisory-fit.test.ts new file mode 100644 index 0000000000..bbb1506917 --- /dev/null +++ b/packages/inference/test/model-fit-advisory-fit.test.ts @@ -0,0 +1,319 @@ +import test from 'brittle' +import { AbortController } from 'bare-abort-controller' +import type { FitLlamaResult } from '@qvac/model-fit/process' + +import type { Logger } from '@/logging/types' +import { ModelType } from '@/schemas/index' +import { runAdvisoryFitCheck } from '@/model-fit/advisory-fit' +import type { IsolatedFitResult } from '@/model-fit/run-isolated-fit' + +const COMPLETION_INPUT = { + modelId: 'llm-1', + modelType: ModelType.llamacppCompletion, + modelPath: '/models/model.gguf', + modelConfig: { ctx_size: 4096, gpu_layers: 99, device: 'gpu' }, + isShardedModel: false +} + +const FIT_PLAN: FitLlamaResult = { + status: 0, + fits: true, + reason: 'fits', + maxDevices: 1, + nDevices: 1, + nGpuDevices: 1, + nGpuLayers: 32, + nCtx: 4096, + nBatch: 512, + nUbatch: 512, + tensorSplit: [1], + buftOverrides: [], + splitMode: 1, + mainGpu: 0, + typeK: 1, + typeV: 1, + flashAttnType: 1 +} + +type LogLevelName = 'error' | 'warn' | 'info' | 'debug' | 'trace' + +interface Recorded { + level: LogLevelName + message: string +} + +function recordingLogger(): { logger: Logger; records: Recorded[] } { + const records: Recorded[] = [] + const record = + (level: LogLevelName) => + (...args: unknown[]) => { + records.push({ level, message: args.map(String).join(' ') }) + } + const logger = { + error: record('error'), + warn: record('warn'), + info: record('info'), + debug: record('debug'), + trace: record('trace'), + setLevel: () => {}, + getLevel: () => 2, + addTransport: () => {}, + setConsoleOutput: () => {} + } as unknown as Logger + return { logger, records } +} + +const zeroResident = () => Promise.resolve(0) + +function fitReturning(result: IsolatedFitResult) { + const calls: unknown[][] = [] + const runFit = (...args: unknown[]) => { + calls.push(args) + return Promise.resolve(result) + } + return { calls, runFit: runFit as never } +} + +test('advisory fit: is inert until explicitly enabled', async (t) => { + const { logger, records } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: false, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger + }) + + t.alike(outcome, { verdict: 'unknown', reason: 'not-enabled' }) + t.is(calls.length, 0) + t.is(records.length, 0) +}) + +test('advisory fit: reports a projected fit with its plan', async (t) => { + const { logger, records } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger + }) + + t.alike(outcome, { + verdict: 'fit', + reason: 'fits', + plan: { nCtx: 4096, nGpuLayers: 32, nGpuDevices: 1 } + }) + t.is(calls.length, 1) + t.is(calls[0]?.[0], 'completion') + t.is(records[0]?.level, 'info') + t.ok(records[0]?.message.includes('advisory only')) +}) + +test('advisory fit: reports a projected insufficiency without denying the load', async (t) => { + const { logger, records } = recordingLogger() + const { runFit } = fitReturning({ + status: 'completed', + result: { ...FIT_PLAN, status: 1, fits: false, reason: 'does-not-fit' } as never + }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger + }) + + t.alike(outcome, { verdict: 'does-not-fit', reason: 'does-not-fit' }) + t.is(records[0]?.level, 'warn') + t.ok(records[0]?.message.includes('the load continues unchanged')) +}) + +test('advisory fit: treats every non-verdict fit result as absent evidence', async (t) => { + for (const reason of ['model-unreadable', 'no-backend-device', 'unsupported-config']) { + const { logger } = recordingLogger() + const { runFit } = fitReturning({ + status: 'completed', + result: { ...FIT_PLAN, status: 2, fits: false, reason } as never + }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger + }) + + t.alike(outcome, { verdict: 'unknown', reason }) + } +}) + +test('advisory fit: treats every supervisor failure as absent evidence', async (t) => { + for (const reason of ['crashed', 'timeout', 'invalid-response', 'spawn-failed', 'cancelled']) { + const { logger, records } = recordingLogger() + const { runFit } = fitReturning({ + status: 'unknown', + reason: reason as never, + message: 'child failed' + }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger + }) + + t.alike(outcome, { verdict: 'unknown', reason, message: 'child failed' }) + t.is(records[0]?.level, 'info') + } +}) + +test('advisory fit: never launches a child for an unsupported load', async (t) => { + const { logger } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + const outcome = await runAdvisoryFitCheck( + { ...COMPLETION_INPUT, modelType: ModelType.ttsGgml }, + { enabled: true, mobile: false, runFit, logger } + ) + + t.is(outcome.verdict, 'unknown') + t.is(outcome.reason, 'unsupported-load') + t.is(calls.length, 0) +}) + +test('advisory fit: never launches a child on mobile', async (t) => { + const { logger } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: true, + residentModelBytes: zeroResident, + runFit, + logger + }) + + t.alike(outcome, { + verdict: 'unknown', + reason: 'unsupported-load', + message: 'mobile has no disposable process boundary' + }) + t.is(calls.length, 0) +}) + +test('advisory fit: absorbs a supervisor that rejects', async (t) => { + const { logger } = recordingLogger() + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit: (() => Promise.reject(new TypeError('supervisor exploded'))) as never, + logger + }) + + t.alike(outcome, { + verdict: 'unknown', + reason: 'internal-error', + message: 'TypeError: supervisor exploded' + }) +}) + +test('advisory fit: absorbs a supervisor that throws synchronously', async (t) => { + const { logger } = recordingLogger() + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit: (() => { + throw new RangeError('bad request') + }) as never, + logger + }) + + t.alike(outcome, { + verdict: 'unknown', + reason: 'internal-error', + message: 'RangeError: bad request' + }) +}) + +test('advisory fit: forwards the caller timeout and abort signal to the supervisor', async (t) => { + const { logger } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + const controller = new AbortController() + + await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger, + timeoutMs: 1_234, + signal: controller.signal + }) + + t.alike(calls[0]?.[2], { timeoutMs: 1_234, signal: controller.signal }) +}) + +test('advisory fit: reserves resident model bytes through the fit margin', async (t) => { + const { logger } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + // 10.5 GiB of resident weights -> ceil(10752 MiB) on top of the 1024 base. + residentModelBytes: () => Promise.resolve(10.5 * 1024 * 1024 * 1024), + runFit, + logger + }) + + const config = calls[0]?.[1] as { marginMiB?: number } + t.is(config.marginMiB, 1024 + 10752) +}) + +test('advisory fit: leaves the package default margin when nothing is resident', async (t) => { + const { logger } = recordingLogger() + const { calls, runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: zeroResident, + runFit, + logger + }) + + const config = calls[0]?.[1] as { marginMiB?: number } + t.absent('marginMiB' in config) +}) + +test('advisory fit: absorbs a resident-bytes probe that rejects', async (t) => { + const { logger } = recordingLogger() + const { runFit } = fitReturning({ status: 'completed', result: FIT_PLAN }) + + const outcome = await runAdvisoryFitCheck(COMPLETION_INPUT, { + enabled: true, + mobile: false, + residentModelBytes: () => Promise.reject(new TypeError('registry unavailable')), + runFit, + logger + }) + + t.alike(outcome, { + verdict: 'unknown', + reason: 'internal-error', + message: 'TypeError: registry unavailable' + }) +}) diff --git a/packages/inference/test/model-fit-create-llama-fit-request.test.ts b/packages/inference/test/model-fit-create-llama-fit-request.test.ts new file mode 100644 index 0000000000..51f549ec16 --- /dev/null +++ b/packages/inference/test/model-fit-create-llama-fit-request.test.ts @@ -0,0 +1,187 @@ +import test from 'brittle' + +import { ModelType } from '@/schemas/index' +import { createLlamaFitRequest } from '@/model-fit/create-llama-fit-request' + +const COMPLETION_CONFIG = { + ctx_size: 4096, + gpu_layers: 99, + device: 'gpu', + system_prompt: 'You are a helpful assistant.', + image_tile_mode: 'sequential', + temp: 0.8, + top_k: 40, + top_p: 0.9, + seed: -1, + predict: -1, + repeat_penalty: 1.1, + tools: false, + stop_sequences: [''], + n_discarded: 0, + parallel: 2, + no_mmap: true, + 'cache-type-k': 'q8_0', + 'cache-type-v': 'q8_0', + 'main-gpu': 0, + 'split-mode': 'layer', + 'tensor-split': '3,1', + openclCacheDir: '/tmp/opencl' +} + +const EMBEDDING_CONFIG = { + device: 'gpu', + gpuLayers: 99, + batchSize: 1024, + flashAttention: 'auto', + pooling: 'mean', + attention: 'non-causal', + embdNormalize: 2, + verbosity: 0, + openclCacheDir: '/tmp/opencl' +} + +function completionRequest(overrides: Record = {}) { + return createLlamaFitRequest({ + modelType: ModelType.llamacppCompletion, + modelPath: '/models/model.gguf', + modelConfig: { ...COMPLETION_CONFIG, ...overrides }, + isShardedModel: false, + isMobile: false + }) +} + +test('createLlamaFitRequest: forwards only fit-relevant completion load settings', (t) => { + const plan = completionRequest() + + t.ok(plan.supported) + if (!plan.supported) return + t.is(plan.loadKind, 'completion') + t.alike(plan.config.params, { + device: 'gpu', + ctx_size: '4096', + gpu_layers: '99', + no_mmap: 'true', + parallel: '2', + 'cache-type-k': 'q8_0', + 'cache-type-v': 'q8_0', + 'main-gpu': '0', + 'split-mode': 'layer', + 'tensor-split': '3,1' + }) +}) + +test('createLlamaFitRequest: pins the requested context as the reduction floor', (t) => { + const plan = completionRequest() + + t.ok(plan.supported) + if (!plan.supported) return + t.is(plan.config.nCtxMin, 4096) +}) + +test('createLlamaFitRequest: leaves the floor unset for an auto context', (t) => { + const plan = completionRequest({ ctx_size: 0 }) + + t.ok(plan.supported) + if (!plan.supported) return + t.is(plan.config.params['ctx_size'], '0') + t.absent('nCtxMin' in plan.config) +}) + +test('createLlamaFitRequest: refuses a load carrying an unclassified setting', (t) => { + t.alike(completionRequest({ some_new_load_knob: 7 }), { + supported: false, + detail: 'unclassified load setting: some_new_load_knob' + }) +}) + +test('createLlamaFitRequest: refuses a LoRA load', (t) => { + t.alike(completionRequest({ lora: '/adapters/style.gguf' }), { + supported: false, + detail: 'unsupported load setting: lora' + }) +}) + +test('createLlamaFitRequest: refuses a multimodal load', (t) => { + t.alike( + createLlamaFitRequest({ + modelType: ModelType.llamacppCompletion, + modelPath: '/models/model.gguf', + modelConfig: COMPLETION_CONFIG, + artifacts: { projectionModelPath: '/models/mmproj.gguf' }, + isShardedModel: false, + isMobile: false + }), + { supported: false, detail: 'multimodal projection loads are not representable' } + ) +}) + +test('createLlamaFitRequest: refuses a sharded load', (t) => { + t.alike( + createLlamaFitRequest({ + modelType: ModelType.llamacppCompletion, + modelPath: '/models/model-00001-of-00003.gguf', + modelConfig: COMPLETION_CONFIG, + isShardedModel: true, + isMobile: false + }), + { supported: false, detail: 'sharded models are not representable' } + ) +}) + +test('createLlamaFitRequest: refuses every load on mobile before inspecting it', (t) => { + t.alike( + createLlamaFitRequest({ + modelType: ModelType.llamacppCompletion, + modelPath: '/models/model.gguf', + modelConfig: { some_new_load_knob: 7 }, + isShardedModel: true, + isMobile: true + }), + { supported: false, detail: 'mobile has no disposable process boundary' } + ) +}) + +test('createLlamaFitRequest: refuses a model type that is not a llama.cpp load', (t) => { + t.alike( + createLlamaFitRequest({ + modelType: ModelType.whispercppTranscription, + modelPath: '/models/whisper.bin', + modelConfig: {}, + isShardedModel: false, + isMobile: false + }), + { + supported: false, + detail: `model type is not a llama.cpp load: ${ModelType.whispercppTranscription}` + } + ) +}) + +test('createLlamaFitRequest: forwards only fit-relevant embedding load settings', (t) => { + const plan = createLlamaFitRequest({ + modelType: ModelType.llamacppEmbedding, + modelPath: '/models/embed.gguf', + modelConfig: EMBEDDING_CONFIG, + isShardedModel: false, + isMobile: false + }) + + t.ok(plan.supported) + if (!plan.supported) return + t.is(plan.loadKind, 'embedding') + t.alike(plan.config.params, { + device: 'gpu', + gpu_layers: '99', + batch_size: '1024', + flash_attn: 'auto' + }) + // Embedding context is resolved by the package's own embedding policy. + t.absent('nCtxMin' in plan.config) +}) + +test('createLlamaFitRequest: refuses a CPU load, whose verdict carries no evidence', (t) => { + t.alike(completionRequest({ device: 'cpu' }), { + supported: false, + detail: 'cpu loads carry no device-memory evidence' + }) +}) diff --git a/packages/inference/test/model-fit-process.test.ts b/packages/inference/test/model-fit-process.test.ts new file mode 100644 index 0000000000..bca0bb20f4 --- /dev/null +++ b/packages/inference/test/model-fit-process.test.ts @@ -0,0 +1,69 @@ +import test from 'brittle' +import { fileURLToPath } from 'bare-url' + +import { runIsolatedFit } from '@/model-fit/run-isolated-fit' + +// The engine always runs under Bare, so unlike the pre-relocation SDK suite +// there is no Node-parent variant: this test IS the Bare parent, driving one +// real disposable child per case through the actual `bare` spawn path. +// +// Resolved against the compiled test's own location: the fixture compiles to +// ./fixtures/model-fit/fit-runner-fixture.js beside it under test/dist. +// `import.meta` is not modeled by this package's TS libs; the compiled test +// runs as ESM under Bare where it exists. +const testModuleUrl = (import.meta as unknown as { url: string }).url +const runnerFixturePath = fileURLToPath( + new URL('./fixtures/model-fit/fit-runner-fixture.js', testModuleUrl) +) + +function run(mode: 'completed' | 'error' | 'hang' | 'abort') { + return runIsolatedFit( + 'completion', + { modelPath: '/tmp/not-used.gguf', params: { device: 'gpu' } }, + { + runnerPath: runnerFixturePath, + runnerArgs: [mode], + timeoutMs: 2_000, + terminationGraceMs: 200, + finalKillGraceMs: 200 + } + ) +} + +test('process: a real child returning a valid response completes', async (t) => { + const result = await run('completed') + + t.is(result.status, 'completed') + if (result.status === 'completed') { + t.is(result.result.status, 0) + t.is(result.result.nCtx, 4096) + } +}) + +test('process: a real child exiting abnormally is unknown/crashed', async (t) => { + const result = await run('error') + + t.is(result.status, 'unknown') + if (result.status === 'unknown') { + t.is(result.reason, 'crashed') + t.ok((result.stderrTail ?? '').includes('fixture failed')) + } +}) + +test('process: a hung child is terminated and reported as timeout', async (t) => { + const result = await run('hang') + + t.is(result.status, 'unknown') + if (result.status === 'unknown') { + t.is(result.reason, 'timeout') + } +}) + +test('process: a child killed by a signal is unknown/crashed', async (t) => { + const result = await run('abort') + + t.is(result.status, 'unknown') + if (result.status === 'unknown') { + t.is(result.reason, 'crashed') + } +}) diff --git a/packages/inference/test/model-fit-run-isolated-fit.test.ts b/packages/inference/test/model-fit-run-isolated-fit.test.ts new file mode 100644 index 0000000000..980ae83e69 --- /dev/null +++ b/packages/inference/test/model-fit-run-isolated-fit.test.ts @@ -0,0 +1,1058 @@ +import test from 'brittle' +import EventEmitter from 'bare-events' + +import { AbortController, type AbortSignal } from 'bare-abort-controller' +import env from 'bare-env' +import { + FIT_PROCESS_MAX_RESPONSE_BYTES, + FIT_PROCESS_PROTOCOL_VERSION_V2 +} from '@qvac/model-fit/process' + +import { + runIsolatedFit, + type ChildProcess, + type ReadableChildStream, + type RunIsolatedFitOptions, + type SpawnContext, + type WritableChildStream +} from '@/model-fit/run-isolated-fit' + +const LOAD_KIND = 'completion' as const +const CONFIG = { + modelPath: '/models/test.gguf', + params: { device: 'gpu', 'ctx-size': '4096' }, + nCtxMin: 4096 +} +const RUNTIME = { + platform: 'darwin', + arch: 'arm64', + isAndroid: false, + isBrowser: false, + isIOS: false +} +const ALLOWED_DEFAULT_ENVIRONMENT = { + HOME: '/default/home', + PATH: '/default/bin', + TMPDIR: '/default/tmpdir', + TMP: '/default/tmp', + TEMP: '/default/temp', + SystemRoot: 'C:\\Windows', + WINDIR: 'C:\\Windows', + LD_LIBRARY_PATH: '/default/ld', + DYLD_LIBRARY_PATH: '/default/dyld', + VK_ICD_FILENAMES: '/default/icd.json', + VK_DRIVER_FILES: '/default/driver.json', + CUDA_VISIBLE_DEVICES: '0', + HIP_VISIBLE_DEVICES: '1', + ROCR_VISIBLE_DEVICES: '2' +} +const DEFAULT_ENVIRONMENT = { + ...ALLOWED_DEFAULT_ENVIRONMENT, + SECRET_TOKEN: 'must-not-leak' +} +const COMPLETED_RESULT = { + status: 0, + fits: true, + reason: 'fits', + maxDevices: 1, + nDevices: 1, + nGpuDevices: 1, + nGpuLayers: 32, + nCtx: 4096, + nBatch: 512, + nUbatch: 512, + tensorSplit: [1], + buftOverrides: [], + splitMode: 1, + mainGpu: 0, + typeK: 1, + typeV: 1, + flashAttnType: 1 +} as const + +class FakeReadable extends EventEmitter { + encoding: string | undefined + destroyed = false + throwOnSetEncoding = false + + setEncoding(encoding: 'utf8'): void { + if (this.throwOnSetEncoding) throw new TypeError('setEncoding failed') + this.encoding = encoding + } + + destroy(): void { + this.destroyed = true + } +} + +class FakeWritable extends EventEmitter { + writes: string[] = [] + destroyed = false + onEnd: (() => void) | undefined + throwOnEnd = false + + end(data?: string): void { + if (this.throwOnEnd) throw new TypeError('stdin end failed') + if (data !== undefined) this.writes.push(data) + this.onEnd?.() + } + + destroy(): void { + this.destroyed = true + } +} + +class FakeChild extends EventEmitter { + pid: number | null = 42 + killed = false + stdin: FakeWritable | null = new FakeWritable() + stdout: FakeReadable | null = new FakeReadable() + stderr: FakeReadable | null = new FakeReadable() + kills: string[] = [] + + kill(signal = 'SIGTERM'): boolean { + this.killed = true + this.kills.push(signal) + return true + } +} + +// The fakes are structurally what the supervisor consumes; bare-events' +// EventEmitter typings differ from the seam interfaces on `on`/`off` +// signatures, so hand them over through one explicit cast. +function asChild(child: FakeChild): ChildProcess { + return child as unknown as ChildProcess +} + +function optionsFor( + child: FakeChild, + overrides: RunIsolatedFitOptions = {} +): RunIsolatedFitOptions { + return { + runtime: RUNTIME, + environment: {}, + runnerPath: '/runner/process-runner.js', + spawnProcess: () => asChild(child), + ...overrides + } +} + +// bun:test's toMatchObject asserted a subset of keys; brittle's alike is an +// exact deep-compare. This keeps the original partial-match semantics. +function matchObject( + t: { alike: (a: unknown, b: unknown, msg?: string) => void }, + actual: unknown, + expected: Record +): void { + const source = (actual ?? {}) as Record + const picked: Record = {} + for (const key of Object.keys(expected)) { + picked[key] = source[key] + } + t.alike(picked, expected) +} + +function completedLine(result: unknown = COMPLETED_RESULT): string { + return `${JSON.stringify({ + version: FIT_PROCESS_PROTOCOL_VERSION_V2, + status: 'completed', + result + })}\n` +} + +function closeChild(child: FakeChild, code: number | null = 0, signal: string | null = null): void { + child.emit('exit', code, signal) + child.emit('close', code, signal) +} + +async function nextTurn(): Promise { + await new Promise((resolve) => setTimeout(resolve, 0)) +} + +test('returns unsupported-platform without spawning on mobile', async (t) => { + for (const unsupported of [ + { isAndroid: true, isBrowser: false, isIOS: false }, + { isAndroid: false, isBrowser: true, isIOS: false }, + { isAndroid: false, isBrowser: false, isIOS: true } + ]) { + let spawnCount = 0 + const result = await runIsolatedFit(LOAD_KIND, CONFIG, { + runtime: { ...RUNTIME, ...unsupported }, + spawnProcess: () => { + spawnCount++ + return asChild(new FakeChild()) + } + }) + + t.alike(result, { + status: 'unknown', + reason: 'unsupported-platform', + message: 'Fit subprocess isolation is unavailable on darwin' + }) + t.is(spawnCount, 0) + } +}) + +test('writes one versioned request and maps a valid response to completed', async (t) => { + const child = new FakeChild() + let handlersReadyAtWrite = false + child.stdin!.onEnd = () => { + handlersReadyAtWrite = + child.listenerCount('error') === 1 && + child.listenerCount('exit') === 1 && + child.listenerCount('close') === 1 && + child.stdin!.listenerCount('error') === 1 && + child.stdout!.listenerCount('data') === 1 && + child.stdout!.listenerCount('error') === 1 && + child.stderr!.listenerCount('data') === 1 && + child.stderr!.listenerCount('error') === 1 + child.stdout!.emit('data', completedLine()) + closeChild(child) + } + + const result = await runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + + t.is(handlersReadyAtWrite, true) + t.alike(child.stdin!.writes, [ + `${JSON.stringify({ + version: FIT_PROCESS_PROTOCOL_VERSION_V2, + loadKind: LOAD_KIND, + config: CONFIG + })}\n` + ]) + t.alike(result, { + status: 'completed', + result: COMPLETED_RESULT + }) +}) + +test('maps runner error responses to unknown invocation-error', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit( + 'data', + `${JSON.stringify({ + version: FIT_PROCESS_PROTOCOL_VERSION_V2, + status: 'invocation-error', + error: { name: 'RangeError', message: 'bad config' } + })}\n` + ) + closeChild(child, 1) + + t.alike(await promise, { + status: 'unknown', + reason: 'invocation-error', + message: 'RangeError: bad config' + }) +}) + +test('prefers crashed over a runner error response when the child died by signal', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit( + 'data', + `${JSON.stringify({ + version: FIT_PROCESS_PROTOCOL_VERSION_V2, + status: 'invocation-error', + error: { name: 'RangeError', message: 'bad config' } + })}\n` + ) + closeChild(child, null, 'SIGSEGV') + + t.alike(await promise, { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code null and signal SIGSEGV' + }) +}) + +test('maps spawn throws and error events to unknown spawn-failed', async (t) => { + const thrown = await runIsolatedFit(LOAD_KIND, CONFIG, { + ...optionsFor(new FakeChild()), + spawnProcess: () => { + throw new TypeError('binary unavailable') + } + }) + t.alike(thrown, { + status: 'unknown', + reason: 'spawn-failed', + message: 'TypeError: binary unavailable' + }) + + const child = new FakeChild() + const emittedPromise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.emit('error', new TypeError('launch failed')) + child.emit('close', null, null) + t.alike(await emittedPromise, { + status: 'unknown', + reason: 'spawn-failed', + message: 'TypeError: launch failed' + }) +}) + +test('maps non-zero exit without a response to unknown crashed', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + closeChild(child, 7) + + t.alike(await promise, { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code 7 and signal null' + }) +}) + +test('maps signal exit without a response to unknown crashed', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + closeChild(child, null, 'SIGSEGV') + + t.alike(await promise, { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code null and signal SIGSEGV' + }) +}) + +test('normalizes Bare numeric signal 0 as a successful exit', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit('data', completedLine()) + child.emit('exit', 0, 0) + child.emit('close', 0, 0) + + t.alike(await promise, { + status: 'completed', + result: COMPLETED_RESULT + }) +}) + +test('normalizes Bare numeric signal 6 as crashed', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.emit('exit', 0, 6) + child.emit('close', 0, 6) + + t.alike(await promise, { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code 0 and signal 6' + }) +}) + +test('rejects multiple response lines as unknown invalid-response', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit('data', `${completedLine()}${completedLine()}`) + closeChild(child) + + matchObject(t, await promise, { + status: 'unknown', + reason: 'invalid-response' + }) +}) + +test('rejects an additional blank line after a valid response', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit('data', `${completedLine()}\n`) + closeChild(child) + + t.alike(await promise, { + status: 'unknown', + reason: 'invalid-response', + message: 'Fit subprocess returned invalid line framing' + }) +}) + +test('rejects response output larger than 1 MiB as unknown invalid-response', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + let settled = false + void promise.then(() => { + settled = true + }) + + child.stdout!.emit('data', 'x'.repeat(FIT_PROCESS_MAX_RESPONSE_BYTES + 1)) + await nextTurn() + t.alike(child.kills, ['SIGTERM']) + t.is(settled, false) + + closeChild(child, null, 'SIGTERM') + t.alike(await promise, { + status: 'unknown', + reason: 'invalid-response', + message: 'Fit subprocess response exceeds 1 MiB' + }) +}) + +test('retains only the final 16 KiB of stderr', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stderr!.emit('data', `discard-${'a'.repeat(20_000)}`) + child.stderr!.emit('data', 'FINAL') + closeChild(child, 1) + + const result = await promise + matchObject(t, result, { + status: 'unknown', + reason: 'crashed' + }) + if (result.status !== 'unknown') throw new TypeError('expected unknown result') + t.is(Buffer.byteLength(result.stderrTail ?? '', 'utf8'), 16 * 1024) + t.is(result.stderrTail?.endsWith('FINAL'), true) + t.is(result.stderrTail?.startsWith('discard-'), false) +}) + +test('times out at the configured deadline, terminates, then force-kills', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { timeoutMs: 5, terminationGraceMs: 20 }) + ) + let settled = false + void promise.then(() => { + settled = true + }) + + await new Promise((resolve) => setTimeout(resolve, 10)) + t.alike(child.kills, ['SIGTERM']) + t.is(settled, false) + + await new Promise((resolve) => setTimeout(resolve, 20)) + t.alike(child.kills, ['SIGTERM', 'SIGKILL']) + t.is(settled, false) + + closeChild(child, null, 'SIGKILL') + t.alike(await promise, { + status: 'unknown', + reason: 'timeout', + message: 'Fit subprocess exceeded 5ms' + }) +}) + +test('cancellation terminates the child and returns unknown cancelled', async (t) => { + const child = new FakeChild() + const controller = new AbortController() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { signal: controller.signal }) + ) + let settled = false + void promise.then(() => { + settled = true + }) + + controller.abort(undefined) + await nextTurn() + t.alike(child.kills, ['SIGTERM']) + t.is(settled, false) + + closeChild(child, null, 'SIGTERM') + t.alike(await promise, { + status: 'unknown', + reason: 'cancelled', + message: 'Fit subprocess was cancelled' + }) +}) + +test('settles exactly once when response and exit race', async (t) => { + const child = new FakeChild() + let resolutions = 0 + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)).then((result) => { + resolutions++ + return result + }) + + child.stdout!.emit('data', completedLine()) + child.emit('error', new TypeError('late spawn error')) + closeChild(child, 9, null) + child.emit('close', 0, null) + + t.alike(await promise, { + status: 'unknown', + reason: 'spawn-failed', + message: 'TypeError: late spawn error' + }) + await nextTurn() + t.is(resolutions, 1) +}) + +test('uses one child per call and keeps concurrent responses isolated', async (t) => { + const children: FakeChild[] = [] + const spawnProcess = () => { + const child = new FakeChild() + children.push(child) + return asChild(child) + } + const first = runIsolatedFit( + LOAD_KIND, + { modelPath: '/models/first.gguf', params: { device: 'gpu' } }, + { ...optionsFor(new FakeChild()), spawnProcess } + ) + const second = runIsolatedFit( + LOAD_KIND, + { modelPath: '/models/second.gguf', params: { device: 'gpu' } }, + { ...optionsFor(new FakeChild()), spawnProcess } + ) + + t.is(children.length, 2) + children[1]!.stdout!.emit('data', completedLine({ ...COMPLETED_RESULT, nCtx: 2_048 })) + closeChild(children[1]!) + children[0]!.stdout!.emit('data', completedLine({ ...COMPLETED_RESULT, nCtx: 8_192 })) + closeChild(children[0]!) + + const firstResult = await first + matchObject(t, firstResult, { status: 'completed' }) + t.is(firstResult.status === 'completed' ? firstResult.result.nCtx : 0, 8_192) + const secondResult = await second + matchObject(t, secondResult, { status: 'completed' }) + t.is(secondResult.status === 'completed' ? secondResult.result.nCtx : 0, 2_048) +}) + +test('passes only approved environment variables to the child', async (t) => { + const child = new FakeChild() + let receivedOptions: SpawnContext['options'] | undefined + const promise = runIsolatedFit(LOAD_KIND, CONFIG, { + ...optionsFor(child), + runnerArgs: ['completed'], + environment: { + HOME: '/home/test', + CUDA_VISIBLE_DEVICES: '0', + SECRET_TOKEN: 'must-not-leak' + }, + spawnProcess: (context) => { + receivedOptions = context.options + return asChild(child) + } + }) + closeChild(child, 1) + await promise + + matchObject(t, receivedOptions, { + args: ['/runner/process-runner.js', 'completed'], + platform: 'darwin', + arch: 'arm64', + stdio: ['pipe', 'pipe', 'pipe'], + env: { + HOME: '/home/test', + CUDA_VISIBLE_DEVICES: '0' + } + }) +}) + +test('uses overlapped child pipes on Windows', async (t) => { + const child = new FakeChild() + let receivedOptions: SpawnContext['options'] | undefined + const promise = runIsolatedFit(LOAD_KIND, CONFIG, { + ...optionsFor(child), + runtime: { ...RUNTIME, platform: 'win32', arch: 'x64' }, + spawnProcess: (context) => { + receivedOptions = context.options + return asChild(child) + } + }) + closeChild(child, 1) + await promise + + t.alike(receivedOptions?.stdio, ['overlapped', 'overlapped', 'overlapped']) +}) + +test('uses the runtime-safe default environment source and preserves the 14-key allowlist', async (t) => { + const previousEnvironment: Record = {} + for (const [key, value] of Object.entries(DEFAULT_ENVIRONMENT)) { + previousEnvironment[key] = env[key] + env[key] = value + } + + const child = new FakeChild() + let receivedOptions: SpawnContext['options'] | undefined + try { + const promise = runIsolatedFit(LOAD_KIND, CONFIG, { + runtime: RUNTIME, + runnerPath: '/runner/process-runner.js', + spawnProcess: (context) => { + receivedOptions = context.options + return asChild(child) + } + }) + closeChild(child, 1) + await promise + + t.alike(receivedOptions?.env, ALLOWED_DEFAULT_ENVIRONMENT) + t.is(Object.keys(receivedOptions?.env ?? {}).length, 14) + } finally { + for (const [key, value] of Object.entries(previousEnvironment)) { + // bare-env's proxy rejects `delete`; blank the key instead. A '' entry + // still travels through the allowlist, but the assertions above ran + // before this cleanup and this is the final test in the file. + env[key] = value === undefined ? '' : value + } + } +}) + +test('maps malformed and unknown-version responses to invalid-response', async (t) => { + for (const line of [ + '{bad json}\n', + `${JSON.stringify({ + version: FIT_PROCESS_PROTOCOL_VERSION_V2 + 1, + status: 'completed', + result: COMPLETED_RESULT + })}\n`, + `${JSON.stringify({ + version: FIT_PROCESS_PROTOCOL_VERSION_V2, + status: 'unexpected' + })}\n` + ]) { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit('data', line) + closeChild(child) + matchObject(t, await promise, { + status: 'unknown', + reason: 'invalid-response' + }) + } +}) + +test('rejects completed responses without FitResult discriminants', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit('data', completedLine({})) + closeChild(child) + + matchObject(t, await promise, { + status: 'unknown', + reason: 'invalid-response' + }) +}) + +test('rejects successful FitResult responses with incomplete plans', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + const { tensorSplit: _tensorSplit, ...incompletePlan } = COMPLETED_RESULT + child.stdout!.emit('data', completedLine(incompletePlan)) + closeChild(child) + + matchObject(t, await promise, { + status: 'unknown', + reason: 'invalid-response' + }) +}) + +test('classifies nonzero exit with partial stdout as crashed', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + child.stdout!.emit('data', '{"version":1') + closeChild(child, 2) + + t.alike(await promise, { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code 2 and signal null' + }) +}) + +test('terminates on stdio errors and waits for close before settling', async (t) => { + for (const streamName of ['stdin', 'stdout', 'stderr'] as const) { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child)) + let settled = false + void promise.then(() => { + settled = true + }) + + child[streamName]!.emit('error', new TypeError(`${streamName} failed`)) + await nextTurn() + t.alike(child.kills, ['SIGTERM']) + t.is(settled, false) + + closeChild(child, null, 'SIGTERM') + t.alike( + await promise, + streamName === 'stdin' + ? { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code null and signal SIGTERM' + } + : { + status: 'unknown', + reason: 'invalid-response', + message: `Fit subprocess ${streamName} failed: ${streamName} failed` + } + ) + } +}) + +test('guards synchronous stream setup and request write failures', async (t) => { + for (const failure of ['setup', 'write'] as const) { + const child = new FakeChild() + if (failure === 'setup') child.stdout!.throwOnSetEncoding = true + if (failure === 'write') child.stdin!.throwOnEnd = true + + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child, { terminationGraceMs: 5 })) + let settled = false + void promise.then(() => { + settled = true + }) + + t.alike(child.kills, ['SIGTERM']) + t.is(settled, false) + closeChild(child, null, 'SIGTERM') + matchObject(t, await promise, { + status: 'unknown', + reason: 'invalid-response' + }) + } +}) + +test('keeps the first termination reason when a child error races timeout', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { timeoutMs: 5, terminationGraceMs: 5 }) + ) + let settled = false + void promise.then(() => { + settled = true + }) + + await new Promise((resolve) => setTimeout(resolve, 10)) + child.emit('error', new TypeError('late child error')) + t.is(settled, false) + await new Promise((resolve) => setTimeout(resolve, 10)) + t.alike(child.kills, ['SIGTERM', 'SIGKILL']) + + closeChild(child, null, 'SIGKILL') + t.alike(await promise, { + status: 'unknown', + reason: 'timeout', + message: 'Fit subprocess exceeded 5ms' + }) +}) + +test('rejects oversized stdout delivered after exit while pipes drain', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + timeoutMs: 5, + drainGraceMs: 50 + }) + ) + let settled = false + void promise.then(() => { + settled = true + }) + + child.stdout!.emit('data', completedLine()) + child.emit('exit', 0, null) + await new Promise((resolve) => setTimeout(resolve, 15)) + child.stdin!.emit('error', new TypeError('late EPIPE')) + child.emit('error', new TypeError('late child error')) + child.stdout!.emit('data', 'x'.repeat(FIT_PROCESS_MAX_RESPONSE_BYTES + 1)) + + t.alike(child.kills, []) + t.is(child.stdout!.destroyed, false) + t.is(settled, false) + child.emit('close', 0, null) + t.alike(await promise, { + status: 'unknown', + reason: 'invalid-response', + message: 'Fit subprocess response exceeds 1 MiB' + }) +}) + +test('accepts a valid response that drains after child exit', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + drainGraceMs: 50 + }) + ) + + child.emit('exit', 0, null) + child.stdout!.emit('data', completedLine()) + child.emit('close', 0, null) + + t.alike(await promise, { + status: 'completed', + result: COMPLETED_RESULT + }) +}) + +test('rejects a second response line delivered after child exit', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + drainGraceMs: 50 + }) + ) + + child.stdout!.emit('data', completedLine()) + child.emit('exit', 0, null) + child.stdout!.emit('data', '{"late":"inherited"}\n') + child.emit('close', 0, null) + + t.alike(await promise, { + status: 'unknown', + reason: 'invalid-response', + message: 'Fit subprocess returned invalid line framing' + }) +}) + +test('classifies early crash ahead of a racing stdin EPIPE', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + terminationGraceMs: 5, + finalKillGraceMs: 5 + }) + ) + + child.stdin!.emit('error', new TypeError('write EPIPE')) + closeChild(child, 1, null) + + t.alike(await promise, { + status: 'unknown', + reason: 'crashed', + message: 'Fit subprocess exited with code 1 and signal null' + }) +}) + +test('bounds a standalone stdin failure when the child never exits', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + terminationGraceMs: 5, + finalKillGraceMs: 5 + }) + ) + + child.stdin!.emit('error', new TypeError('write EPIPE')) + const outcome = await Promise.race([ + promise, + new Promise<'still-pending'>((resolve) => setTimeout(() => resolve('still-pending'), 30)) + ]) + if (outcome === 'still-pending') child.emit('close', null, 'SIGKILL') + t.alike(outcome, { + status: 'unknown', + reason: 'invalid-response', + message: 'Fit subprocess stdin failed: write EPIPE; child did not report exit' + }) +}) + +test('reaps a post-spawn child error before returning spawn-failed', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child, { terminationGraceMs: 5 })) + let settled = false + void promise.then(() => { + settled = true + }) + + child.emit('error', new TypeError('post-spawn failure')) + t.alike(child.kills, ['SIGTERM']) + t.is(settled, false) + + closeChild(child, null, 'SIGTERM') + t.alike(await promise, { + status: 'unknown', + reason: 'spawn-failed', + message: 'TypeError: post-spawn failure' + }) +}) + +test('settles after SIGKILL when the child reports neither exit nor close', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + timeoutMs: 5, + terminationGraceMs: 5, + finalKillGraceMs: 5 + }) + ) + + const result = await Promise.race([ + promise, + new Promise<'still-pending'>((resolve) => setTimeout(() => resolve('still-pending'), 30)) + ]) + if (result === 'still-pending') child.emit('close', null, 'SIGKILL') + t.alike(child.kills, ['SIGTERM', 'SIGKILL']) + t.is(child.stdin!.destroyed, true) + t.is(child.stdout!.destroyed, true) + t.is(child.stderr!.destroyed, true) + t.alike(result, { + status: 'unknown', + reason: 'timeout', + message: 'Fit subprocess exceeded 5ms; child did not report exit' + }) +}) + +test('bounds pipe drain after exit without close', async (t) => { + for (const scenario of ['parsed', 'crashed', 'pending'] as const) { + const child = new FakeChild() + const controller = new AbortController() + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + drainGraceMs: 5, + ...(scenario === 'pending' ? { signal: controller.signal } : {}) + }) + ) + + if (scenario === 'parsed') child.stdout!.emit('data', completedLine()) + if (scenario === 'crashed') child.stdout!.emit('data', '{"version":1') + if (scenario === 'pending') controller.abort(undefined) + child.emit('exit', scenario === 'parsed' ? 0 : null, scenario === 'parsed' ? null : 'SIGTERM') + + const outcome = await Promise.race([ + promise, + new Promise<'still-pending'>((resolve) => setTimeout(() => resolve('still-pending'), 30)) + ]) + if (outcome === 'still-pending') child.emit('close', 0, null) + + matchObject( + t, + outcome, + scenario === 'parsed' + ? { status: 'completed', result: COMPLETED_RESULT } + : scenario === 'crashed' + ? { status: 'unknown', reason: 'crashed' } + : { status: 'unknown', reason: 'cancelled' } + ) + t.is(child.stdin!.destroyed, true) + t.is(child.stdout!.destroyed, true) + t.is(child.stderr!.destroyed, true) + } +}) + +test('does not write a request when the signal is already aborted', async (t) => { + const child = new FakeChild() + const controller = new AbortController() + controller.abort(undefined) + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { + signal: controller.signal, + terminationGraceMs: 5 + }) + ) + + t.alike(child.stdin!.writes, []) + t.alike(child.kills, ['SIGTERM']) + closeChild(child, null, 'SIGTERM') + matchObject(t, await promise, { + status: 'unknown', + reason: 'cancelled' + }) +}) + +test('destroys available pipes when child stdio is incomplete', async (t) => { + const child = new FakeChild() + const stdin = child.stdin! + const stdout = child.stdout! + child.stderr = null + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child, { terminationGraceMs: 5 })) + + t.is(stdin.destroyed, true) + t.is(stdout.destroyed, true) + t.alike(child.kills, ['SIGTERM']) + closeChild(child, null, 'SIGTERM') + matchObject(t, await promise, { + status: 'unknown', + reason: 'spawn-failed' + }) +}) + +test('removes process, stream, abort listeners and timers after settlement', async (t) => { + const child = new FakeChild() + const controller = new AbortController() + let abortRemoves = 0 + const originalRemove = controller.signal.removeEventListener.bind(controller.signal) + controller.signal.removeEventListener = (( + ...args: Parameters + ) => { + abortRemoves++ + originalRemove(...args) + }) as AbortSignal['removeEventListener'] + + const promise = runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { signal: controller.signal, timeoutMs: 10 }) + ) + child.stdout!.emit('data', completedLine()) + closeChild(child) + await promise + + for (const emitter of [child, child.stdin!, child.stdout!, child.stderr!]) { + t.alike(emitter.eventNames(), ['error']) + t.is(emitter.listenerCount('error'), 1) + } + t.is(abortRemoves, 1) + await new Promise((resolve) => setTimeout(resolve, 30)) + t.alike(child.kills, []) +}) + +test('absorbs late child and stream errors emitted after settlement', async (t) => { + const child = new FakeChild() + const promise = runIsolatedFit(LOAD_KIND, CONFIG, optionsFor(child, { timeoutMs: 10 })) + child.stdout!.emit('data', completedLine()) + closeChild(child) + const result = await promise + + for (const emitter of [child, child.stdin!, child.stdout!, child.stderr!]) { + t.execution(() => emitter.emit('error', new TypeError('EPIPE'))) + t.execution(() => emitter.emit('error', new TypeError('ERR_STREAM_DESTROYED'))) + } + child.stdout!.emit('data', completedLine({ ...COMPLETED_RESULT, nCtx: 1 })) + await nextTurn() + + t.alike(result, { status: 'completed', result: COMPLETED_RESULT }) + t.alike(await promise, { status: 'completed', result: COMPLETED_RESULT }) + t.alike(child.kills, []) +}) + +test('absorbs stream errors raised by the post-SIGKILL destroy path', async (t) => { + const child = new FakeChild() + const stdin = child.stdin! + const stdout = child.stdout! + const stderr = child.stderr! + stdin.destroy = () => { + stdin.destroyed = true + stdin.emit('error', new TypeError('ERR_STREAM_DESTROYED')) + } + + const result = await runIsolatedFit( + LOAD_KIND, + CONFIG, + optionsFor(child, { timeoutMs: 5, terminationGraceMs: 5, finalKillGraceMs: 5 }) + ) + + t.alike(result, { + status: 'unknown', + reason: 'timeout', + message: 'Fit subprocess exceeded 5ms; child did not report exit' + }) + for (const emitter of [child, stdin, stdout, stderr]) { + t.execution(() => emitter.emit('error', new TypeError('late EPIPE'))) + t.alike(emitter.eventNames(), ['error']) + } +}) diff --git a/packages/sdk/examples/advisory-model-fit.ts b/packages/sdk/examples/advisory-model-fit.ts new file mode 100644 index 0000000000..9870a3cbd6 --- /dev/null +++ b/packages/sdk/examples/advisory-model-fit.ts @@ -0,0 +1,196 @@ +/** + * Advisory llama.cpp fit check (QVAC-22629) — experimental, opt-in. + * + * Before a completion or embedding load, the SDK can run `@qvac/model-fit` in + * one disposable Bare child and project whether the exact configuration it is + * about to load will fit in device memory. + * + * The result is ADVISORY. It never blocks a load. `does-not-fit` is logged and + * the ordinary load path runs unchanged, exactly as it would with the check + * switched off. Crashes, timeouts, malformed responses, unsupported + * configurations, and internal errors all resolve to "no evidence" and are + * equally non-blocking. Nothing consumes the verdict yet — this PR only + * produces it. + * + * Enable it with: + * + * QVAC_ADVISORY_MODEL_FIT=1 bun run examples/advisory-model-fit.ts + * + * With the flag unset the check never runs, costs one environment read, and + * loads none of the process machinery. + * + * The verdict is emitted on the SDK server log stream, not to stdout, so this + * example subscribes to `loggingStream({ id: SDK_LOG_ID })` and reprints the + * `[advisory-fit:…]` lines. + * + * --------------------------------------------------------------------------- + * What fits on this machine (Apple M4 Pro, 24 GiB unified memory) + * --------------------------------------------------------------------------- + * + * Measured with the fit addon on 2026-08-24, at the default 1024 MiB margin. + * Metal reports a working-set budget well below the full 24 GiB, so the usable + * ceiling for an all-GPU load is far lower than the raw RAM figure suggests. + * + * PROJECTED TO FIT — all layers on GPU + * Qwen3.5 0.8B Q4_K_M 0.5 GiB @ 4k ctx + * Qwen3.5 9B Q4_K_M 5.3 GiB @ 8k ctx + * Qwen3.5 9B Q6_K 7.0 GiB @ 32k ctx + * gpt-oss-20B Q4_K_M 10.8 GiB @ 32k ctx + * gte-large fp16 0.6 GiB (embedding, context pinned to 512) + * + * PROJECTED NOT TO FIT — try any of these to see a `does-not-fit` verdict + * gpt-oss-20B Q4_K_M 10.8 GiB @ 128k ctx with an f32 KV cache ← below + * Gemma 4 31B Q4_K_M 18.3 GiB @ 32k ctx + * Gemma 4 31B Q4_K_M 18.3 GiB @ 8k ctx + * Gemma 4 31B Q4_K_M 18.3 GiB @ 1k ctx ← the weights alone do not + * fit with gpu_layers: 99 + * ...and anything larger: a 70B at Q4 (~40 GiB), Qwen3.5 72B, Llama 3.3 + * 70B, or any 30B+ model at Q6/Q8 will not fit either. + * + * The first of those is the interesting one, and it is what this example uses: + * gpt-oss-20B at 128k context FITS with the default KV cache and DOES NOT FIT + * once `cache-type-k`/`cache-type-v` are set to `f32`. Same model, same + * context, same machine — the verdict tracks the configuration, not the file + * size. + * + * That verdict is also, measurably, WRONG: the configuration loads and runs at + * 54-69 tok/s. Which is the point of the example. Treat these verdicts as + * evidence to gather, not as answers to act on — they are not reliable in + * either direction yet. Gemma 4 31B is the opposite case: it loads and then + * fails at decode, so `loadModel` returning an id is not a usability signal + * either. + * + * Note the shape of that boundary: it is set by `gpu_layers`, not by the model + * alone. The SDK's schema default is `gpu_layers: 99`, i.e. "put everything on + * the GPU", so the question the fitter answers is always "does this fit + * entirely on the GPU?". Gemma 4 31B @ 32k reports `does-not-fit` under that + * default — but omit `gpu_layers` and the same model at the same context is + * projected to FIT with 48 of its layers offloaded and the rest on CPU. A + * `does-not-fit` therefore means "not at this placement", not "not on this + * machine". + */ + +import { + completion, + loadModel, + unloadModel, + loggingStream, + SDK_LOG_ID, + QWEN3_5_0_8B_MULTIMODAL_Q4_K_M, + GPT_OSS_20B_INST_Q4_K_M +} from '@qvac/sdk' + +// The oversized load below is expected to be reported as `does-not-fit` and +// then attempted anyway, because the check is advisory. It really does try to +// allocate, so it is opt-in on top of the feature flag itself. +const ATTEMPT_OVERSIZED = process.env['QVAC_FIT_DEMO_ATTEMPT_OVERSIZED'] === '1' + +if (process.env['QVAC_ADVISORY_MODEL_FIT'] === undefined) { + console.log('▸ QVAC_ADVISORY_MODEL_FIT is unset — the check will not run.') + console.log('▸ Re-run with: QVAC_ADVISORY_MODEL_FIT=1 bun run examples/advisory-model-fit.ts\n') +} + +// Reprint the worker's advisory verdicts. They arrive on the SDK server log +// stream; everything else on that stream is filtered out to keep this readable. +// +// Called once per phase rather than once for the process: a subscription +// currently stops delivering after any `unloadModel`, so a single one would go +// silent before the second verdict. Resubscribing after the unload works. +function watchVerdicts(): void { + void (async () => { + for await (const log of loggingStream({ id: SDK_LOG_ID })) { + if (log.message.includes('[advisory-fit:')) { + console.log(` ⟶ [${log.level.toUpperCase()}] ${log.message}`) + } + } + })().catch(() => { + // Stream terminated — normal on shutdown. + }) +} + +watchVerdicts() + +try { + // 1. A load the fitter projects to fit. The verdict carries the plan it + // projected: resolved context, offloaded layers, and GPU device count. + console.log('▸ Loading Qwen3.5 0.8B @ 4k — expected verdict: projected to fit') + const smallModelId = await loadModel({ + modelSrc: QWEN3_5_0_8B_MULTIMODAL_Q4_K_M, + modelConfig: { ctx_size: 4096 } + }) + console.log(`▸ Loaded ${smallModelId}\n`) + + const result = completion({ + modelId: smallModelId, + history: [{ role: 'user', content: 'Say hello in five words.' }], + stream: false, + generationParams: { predict: 48 } + }) + const final = await result.final + console.log(`▸ Completion still works normally: ${final.contentText.trim().slice(0, 120)}\n`) + + // Unloaded before the next phase, so the second load is measured on an idle + // machine. The fit check projects a single model in isolation and has no + // notion of what is already resident, so leaving this one loaded would change + // the outcome without changing the verdict. + await unloadModel({ modelId: smallModelId, clearStorage: false }) + watchVerdicts() + + // 2. A load the fitter projects NOT to fit. The point of this example is that + // the SDK reports the verdict and then loads anyway — the check is + // evidence, not admission control. + if (!ATTEMPT_OVERSIZED) { + console.log('▸ Skipping the oversized gpt-oss-20B load.') + console.log('▸ Set QVAC_FIT_DEMO_ATTEMPT_OVERSIZED=1 to let it run and watch the') + console.log(' load proceed past a `does-not-fit` verdict (allocates ~11 GiB).') + } else { + console.log('▸ Loading gpt-oss-20B @ 128k with an f32 KV cache') + console.log('▸ Expected verdict: projected NOT to fit') + console.log('▸ The load is attempted regardless. That is the fail-open contract:') + console.log(' the verdict is evidence, not admission control.\n') + const bigModelId = await loadModel({ + modelSrc: GPT_OSS_20B_INST_Q4_K_M, + modelConfig: { + ctx_size: 131072, + 'cache-type-k': 'f32', + 'cache-type-v': 'f32' + } + }) + console.log(`▸ Load returned ${bigModelId} — the advisory verdict did not block it`) + + // Loading is not the same as being usable: a model can load and then fail + // at decode time. Gemma 4 31B does exactly that on this machine. So run a + // real completion and report throughput rather than trusting the load. + try { + const check = completion({ + modelId: bigModelId, + history: [{ role: 'user', content: 'Name three colours. Answer briefly.' }], + stream: false, + generationParams: { predict: 40 } + }) + const checkFinal = await check.final + console.log( + `▸ ...and it actually runs: ${checkFinal.stats?.tokensPerSecond?.toFixed(1) ?? '?'} tok/s ` + + `— this verdict was a false negative` + ) + } catch (inferenceError) { + console.log( + `▸ ...but it cannot run: ${ + inferenceError instanceof Error ? inferenceError.message : String(inferenceError) + }` + ) + console.log('▸ The verdict was right, and `loadModel` succeeding did not mean usable.') + } + + await unloadModel({ modelId: bigModelId, clearStorage: false }) + } +} catch (error) { + // A failing load here is the native loader's own error, not the fit check. + // The check never throws and never converts a verdict into a load failure. + console.error('▸ Load failed:', error instanceof Error ? error.message : error) + process.exitCode = 1 +} + +// The log subscription is an open stream and would otherwise keep the process +// alive after the work is done. +process.exit(process.exitCode ?? 0)