#!/usr/bin/env node /** * server.mjs — OLP HTTP listener and request dispatcher * * Authority (entry surface): OpenAI Chat Completions API * https://platform.openai.com/docs/api-reference/chat/create * Authority (IR): ADR 0003 * Authority (provider dispatch): ADR 0002 * Authority (cache layer): ADR 0005 * * Design principles (OCP precedent, ESM/.mjs, http built-ins, no external deps): * - Node ESM, no build step, no bundler * - http built-in only (no Express/Fastify) * - Zero runtime npm dependencies in the proxy core * * Env vars: * OLP_PORT — listen port (default: 3456) */ import { createServer } from 'node:http'; import { readFileSync } from 'node:fs'; import { fileURLToPath } from 'node:url'; import { dirname, join } from 'node:path'; import { openAIToIR, BadRequestError } from './lib/ir/openai-to-ir.mjs'; import { irChunkToOpenAISSE, irResponseToOpenAINonStream, generateRequestId, SSE_DONE, } from './lib/ir/ir-to-openai.mjs'; import { loadProviders, getProviderForModel, listAllProviderNames } from './lib/providers/index.mjs'; import { ProviderError } from './lib/providers/base.mjs'; import { computeCacheKey, hasCacheControl, extractCacheControlMarkers } from './lib/cache/keys.mjs'; import { CacheStore } from './lib/cache/store.mjs'; import { evaluateHardTriggers, executeWithFallback, buildDefaultChain, loadFallbackConfigSync, } from './lib/fallback/engine.mjs'; // ── Config ──────────────────────────────────────────────────────────────── const __dirname = dirname(fileURLToPath(import.meta.url)); const pkg = JSON.parse(readFileSync(join(__dirname, 'package.json'), 'utf8')); const VERSION = pkg.version; const PORT = parseInt(process.env.OLP_PORT ?? '3456', 10); const BODY_LIMIT = 5 * 1024 * 1024; // 5 MB // ── Startup config ──────────────────────────────────────────────────────── // Read ~/.olp/config.json once at startup. Provides: // - providers.enabled → which providers are loaded (ADR 0002 § Disable model) // - routing.chains → fallback chain config (ADR 0004 § D9) // - routing.soft_triggers → soft trigger thresholds (ADR 0004) // If the file is absent or malformed, defaults are safe: // empty providersEnabled → all 503 (ALIGNMENT.md § v0.1 zero-Enabled-Providers posture) // empty chains/soft_triggers → single-hop mode const _startupConfig = loadFallbackConfigSync(); // ── Provider registry ───────────────────────────────────────────────────── // ALIGNMENT.md § Provider Inventory: 0 Enabled Providers at v0.1 unless // ~/.olp/config.json has providers.enabled.X = true. // ADR 0002 § Disable model: enabled toggle in config.json transitions Candidate → Enabled. const loadedProviders = loadProviders({ enabled: _startupConfig.providersEnabled ?? {} }); // ── Fallback config ─────────────────────────────────────────────────────── // Read ~/.olp/config.json routing.chains at startup. Empty at v0.1. // Per ADR 0004 § D9: fallback engine is wired; activates when user populates chains. // Tests may inject a synthetic fallbackConfig via __setFallbackConfig(). let _fallbackConfig = _startupConfig; /** @internal — test seam: inject a synthetic fallback config (no file I/O) */ export function __setFallbackConfig(config) { _fallbackConfig = config ?? { chains: {}, soft_triggers: {} }; } /** @internal — reset to file-based config */ export function __resetFallbackConfig() { _fallbackConfig = loadFallbackConfigSync(); } /** * @internal — test seam: reload loadedProviders to match a given enabledMap. * Mirrors __setFallbackConfig. Allows tests to exercise the production * loadProviders() code path without touching the config file. * * Usage: __setProvidersEnabled({ anthropic: true }) before creating a server. * Reset: __resetProvidersEnabled() or __setProvidersEnabled({}) to clear all. * * @param {Record} enabledMap */ export function __setProvidersEnabled(enabledMap) { const next = loadProviders({ enabled: enabledMap ?? {} }); // Mutate the shared map in-place so existing references see the update. loadedProviders.clear(); for (const [name, p] of next) { loadedProviders.set(name, p); } } /** @internal — reset loadedProviders to the startup-config state */ export function __resetProvidersEnabled() { const startup = loadFallbackConfigSync(); const next = loadProviders({ enabled: startup.providersEnabled ?? {} }); loadedProviders.clear(); for (const [name, p] of next) { loadedProviders.set(name, p); } } /** @internal — clear the cache store (for tests that need a fresh cache state) */ export function __clearCache() { cacheStore.clear(); } // ── Cache layer ─────────────────────────────────────────────────────────── // D1 per-key isolation + D4 singleflight per ADR 0005. // keyId: '__anonymous__' at D5 — Phase 2 multi-key infrastructure wires in // the real OLP API key ID here. export const cacheStore = new CacheStore(); // ── Logging ─────────────────────────────────────────────────────────────── function logEvent(level, event, data = {}) { const entry = { ts: new Date().toISOString(), level, event, ...data }; if (level === 'error' || level === 'warn') { process.stderr.write(JSON.stringify(entry) + '\n'); } else { process.stdout.write(JSON.stringify(entry) + '\n'); } } // ── Body reader ─────────────────────────────────────────────────────────── /** * Reads and JSON-parses the request body. * Enforces the 5MB body limit. * Throws on parse failure or oversized body. * * @param {import('node:http').IncomingMessage} req * @returns {Promise} */ function readJSON(req) { return new Promise((resolve, reject) => { let body = ''; let size = 0; req.on('data', chunk => { size += chunk.length; if (size > BODY_LIMIT) { reject(Object.assign(new Error('Request body too large (limit 5MB)'), { statusCode: 413 })); req.destroy(); return; } body += chunk; }); req.on('end', () => { try { resolve(JSON.parse(body)); } catch { reject(Object.assign(new Error('Invalid JSON in request body'), { statusCode: 400 })); } }); req.on('error', reject); }); } // ── Response helpers ────────────────────────────────────────────────────── /** * @param {import('node:http').ServerResponse} res * @param {number} status * @param {object} body * @param {Record} [extraHeaders] */ function sendJSON(res, status, body, extraHeaders = {}) { const payload = JSON.stringify(body); res.writeHead(status, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload), ...extraHeaders, }); res.end(payload); } /** * OpenAI-format error response helper. * @param {import('node:http').ServerResponse} res * @param {number} status * @param {string} message * @param {string} type */ function sendError(res, status, message, type) { sendJSON(res, status, { error: { message, type } }); } // ── OLP response headers ────────────────────────────────────────────────── /** * Returns the standard OLP diagnostic headers. * Per spec §4.7 and ADR 0004 § Observability headers: * X-OLP-Provider-Used, X-OLP-Model-Used, X-OLP-Fallback-Hops, * X-OLP-Cache, X-OLP-Latency-Ms. * Fallback-Hops reflects which chain index served the request (0=primary). * Cache reflects actual hit/miss/bypass status from the cache layer (ADR 0005). * * @param {object} opts * @param {string} opts.providerUsed * @param {string} opts.modelUsed * @param {number} opts.startMs * @param {'hit'|'miss'|'bypass'} [opts.cacheStatus='miss'] * @param {number} [opts.fallbackHops=0] — from executeWithFallback result * @returns {Record} */ function olpHeaders({ providerUsed, modelUsed, startMs, cacheStatus = 'miss', fallbackHops = 0 }) { return { 'X-OLP-Provider-Used': providerUsed, 'X-OLP-Model-Used': modelUsed, 'X-OLP-Fallback-Hops': String(fallbackHops), 'X-OLP-Cache': cacheStatus, 'X-OLP-Latency-Ms': String(Date.now() - startMs), }; } // ── Route handlers ──────────────────────────────────────────────────────── /** * GET /health * Returns server health including count of loaded providers. */ function handleHealth(req, res) { const enabled = loadedProviders.size; const available = listAllProviderNames().length; sendJSON(res, 200, { ok: true, version: VERSION, providers: { enabled, available }, }); } /** * GET /v1/models * Returns an empty data array at D3. * Will be populated from models-registry.json + loaded providers in Phase 1 Day 2. */ function handleModels(req, res) { sendJSON(res, 200, { object: 'list', data: [] }); } /** * POST /v1/chat/completions * Core dispatch path: OpenAI request → IR → fallback engine → provider.spawn → OpenAI response. * * D9: Fallback engine (ADR 0004) is wired between IR construction and provider.spawn. * Chain advancement, soft/hard trigger evaluation, and first-chunk safety are all * handled by executeWithFallback(). At v0.1 with empty routing.chains config, this * is a transparent single-hop pass-through. Multi-hop fallback activates when the * user populates ~/.olp/config.json. * * @param {import('node:http').IncomingMessage} req * @param {import('node:http').ServerResponse} res */ async function handleChatCompletions(req, res) { const startMs = Date.now(); // Require JSON content-type const ct = req.headers['content-type'] ?? ''; if (!ct.includes('application/json')) { return sendError(res, 415, 'Content-Type must be application/json', 'invalid_request_error'); } let body; try { body = await readJSON(req); } catch (e) { return sendError(res, e.statusCode ?? 400, e.message, 'invalid_request_error'); } // Translate OpenAI → IR (ADR 0003) let ir; try { ir = openAIToIR(body); } catch (e) { if (e instanceof BadRequestError) { return sendError(res, 400, e.message, 'invalid_request_error'); } throw e; } // Auth context is null at D5/D9 — providers fall back to their own credential // discovery (env var, keychain, credentials file). Phase 2 multi-key // infrastructure will pass a real authContext carrying the per-key OLP token. const authContext = null; // ── Fallback engine: build chain (ADR 0004) ───────────────────────────── // buildDefaultChain returns null if no enabled provider serves this model. // Per ADR 0004 § D9: at v0.1, chain is single-hop (no fallback) unless the // user has populated ~/.olp/config.json routing.chains. const chain = buildDefaultChain( ir.model, loadedProviders, _fallbackConfig.chains, _fallbackConfig.soft_triggers, ); if (!chain) { // ALIGNMENT.md: 0 Enabled Providers at v0.1 → 503 per spec return sendError( res, 503, `No enabled providers for model ${ir.model}. See README § Supported Providers.`, 'no_enabled_provider', ); } const requestId = generateRequestId(); // ── Cache layer (ADR 0005) ────────────────────────────────────────────── // keyId: '__anonymous__' at D5/D9. Phase 2 multi-key infrastructure wires the // real OLP API key ID here for D1 per-key isolation. const keyId = '__anonymous__'; // D2 bypass (per-hop, per ADR 0005 § D2): // cache_control markers bypass OLP's response cache ONLY when the active hop // provider is Anthropic. For non-Anthropic hops the markers are noop'd // (openai-to-ir already strips them from the IR before provider translation, // so they never reach a non-Anthropic plugin — no separate strip needed here). // // Pre-compute whether the raw request carries any cache_control markers at all. // The per-hop decision is: markers present AND hop is Anthropic → bypass. const hasCacheControlMarkers = hasCacheControl(ir) || extractCacheControlMarkers(body?.messages ?? []).length > 0; /** * Returns true if OLP's response cache should be bypassed for the given hop. * Per ADR 0005 § D2: bypass only when provider is Anthropic AND markers present. * * @param {string} hopProviderName — e.g. 'anthropic', 'openai', 'mistral' * @returns {boolean} */ function shouldBypassCacheForHop(hopProviderName) { return hasCacheControlMarkers && hopProviderName === 'anthropic'; } // ── executeHopFn: per-hop spawn + cache wrapper ───────────────────────── // This is the function executeWithFallback calls for each chain hop. // Each hop gets its own (provider, model) cache key per ADR 0005 § Per-model isolation. // // First-chunk safety (ADR 0004 § Fallback safety): // collectAllChunks fully buffers the provider response before returning. // Therefore, if executeHopFn throws, zero bytes have been written to `res`. // The fallback engine safely advances the chain on hard triggers. // If executeHopFn returns successfully, the chunks are buffered and we write // them to `res` only AFTER executeWithFallback returns — ensuring no writes // occur during chain iteration. async function executeHopFn(hopProvider, hopModel, irReq) { const hopCacheKey = computeCacheKey(hopProvider, hopModel, irReq); const hopProviderPlugin = loadedProviders.get(hopProvider); if (!hopProviderPlugin) { // Provider in the chain is not loaded (config references a disabled provider) throw Object.assign( new Error(`Provider ${hopProvider} is not enabled`), { statusCode: 503 }, ); } // Collect all chunks from this provider, throwing on error chunks. // Error semantics: ProviderError thrown here propagates to executeWithFallback // which decides whether to advance the chain. async function collectAllChunks() { const chunks = []; for await (const irChunk of hopProviderPlugin.spawn(irReq, authContext)) { chunks.push(irChunk); if (irChunk.type === 'error') { throw new ProviderError( irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED', ); } if (irChunk.type === 'stop') break; } return chunks; } // D13: per-hop bypass evaluation (ADR 0005 § D2). // Bypass only when this hop's provider is Anthropic AND markers are present. const bypassCacheForThisHop = shouldBypassCacheForHop(hopProvider); if (bypassCacheForThisHop) { logEvent('debug', 'cache_bypass', { model: hopModel, provider: hopProvider, reason: 'cache_control_markers', }); return collectAllChunks(); } // D4 singleflight + D1 per-key isolation per ADR 0005. // Each hop has its own (provider, model) key — cross-provider contamination // is structurally impossible (ADR 0005 § Per-model isolation). return cacheStore.getOrCompute(keyId, hopCacheKey, collectAllChunks); } // ── Execute with fallback (ADR 0004) ──────────────────────────────────── // Pre-check for cache status reporting uses first hop's key (primary provider). // D13: preCheckHit is gated on whether the first hop would bypass — if it would // bypass (anthropic + markers), the cache is not consulted (preCheckHit=false). // If the first hop is non-Anthropic (or no markers), the cache peek proceeds normally. const bypassCacheForFirstHop = shouldBypassCacheForHop(chain[0].provider); const firstHopCacheKey = computeCacheKey(chain[0].provider, chain[0].model, ir); const preCheckHit = bypassCacheForFirstHop ? false : await cacheStore.peek(keyId, firstHopCacheKey); // ── P1.2: Real SSE streaming path (single-hop cache-miss) ────────────── // ADR 0003 entry adapter pattern: for await irChunk → res.write(irChunkToOpenAISSE). // Condition: streaming + single-hop + no bypass + no pre-check cache hit. // - stream===true → caller wants SSE // - chain.length===1 → no fallback needed; first-chunk rule allows streaming // - !bypassCacheForFirstHop + !preCheckHit → genuine cache miss (not hit/bypass) // // If any chunk has been written (firstChunkEmitted), fallback is impossible // per ADR 0004 § Fallback safety first-chunk rule. On error after first chunk: // truncate the response (end with no [DONE]). On error before any chunk: // throw so the outer handler can surface a clean error (no bytes written). // // On success: write chunks to res AND cache so subsequent identical requests // hit the burst-replay path. if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit) { const streamProvider = chain[0].provider; const streamModel = chain[0].model; const streamCacheKey = computeCacheKey(streamProvider, streamModel, ir); const streamPlugin = loadedProviders.get(streamProvider); if (!streamPlugin) { // Provider disappeared between chain build and here (edge case). return sendError(res, 503, `Provider ${streamProvider} is not enabled`, 'no_enabled_provider'); } const streamHeaders = olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0, }); // D14: writeHead is deferred until just before the first res.write so that // pre-first-chunk errors can still produce a JSON 502 (matching the buffered // path). Calling writeHead unconditionally here was the D14 defect. const streamedChunks = []; let firstChunkEmitted = false; try { for await (const irChunk of streamPlugin.spawn(ir, authContext)) { if (irChunk.type === 'error') { // Error chunk from provider if (firstChunkEmitted) { // Past first-chunk boundary — can't fallback; truncate stream. logEvent('warn', 'streaming_error_after_first_chunk', { provider: streamProvider, model: streamModel, error: irChunk.error, }); res.end(); return; } // No bytes written yet — throw to surface a clean error. throw new ProviderError(irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED'); } // Defer writeHead until the moment we are about to emit the first byte. // After this point firstChunkEmitted===true ↔ res.headersSent===true. if (!res.headersSent) { res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive', 'X-Accel-Buffering': 'no', ...streamHeaders, }); } streamedChunks.push(irChunk); res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model)); firstChunkEmitted = true; if (irChunk.type === 'stop') { res.write(SSE_DONE); res.end(); // Cache the buffered chunks for burst-replay on subsequent identical requests. await cacheStore.set(keyId, streamCacheKey, streamedChunks); logEvent('info', 'streaming_response_cached', { provider: streamProvider, model: streamModel, chunks: streamedChunks.length, }); return; } } // Generator exhausted without a stop chunk — emit [DONE] and cache. res.write(SSE_DONE); res.end(); if (streamedChunks.length > 0) { await cacheStore.set(keyId, streamCacheKey, streamedChunks); } } catch (e) { if (firstChunkEmitted) { // Past first-chunk boundary — truncate silently. logEvent('warn', 'streaming_error_after_first_chunk', { provider: streamProvider, model: streamModel, error: e.message, }); res.end(); } else { // No bytes written — surface a clean JSON error. logEvent('error', 'streaming_error_before_first_chunk', { provider: streamProvider, model: streamModel, error: e.message, }); if (!res.headersSent) { sendError(res, 502, e.message ?? 'Provider error', 'provider_error'); } else { res.end(); } } } return; } let fallbackResult; try { fallbackResult = await executeWithFallback(chain, ir, executeHopFn, { logEvent, }); } catch (e) { // executeWithFallback throws only on programming errors (empty chain). logEvent('error', 'fallback_engine_error', { error: e.message }); return sendError(res, 500, 'Internal server error', 'internal_error'); } const { chunks, providerUsed, modelUsed, fallbackHops, originalError, triedProviders, } = fallbackResult; // ── Chain exhausted or non-trigger error ───────────────────────────────── if (chunks === null) { logEvent('error', 'spawn_error', { model: ir.model, providerUsed, fallbackHops, triedProviders, error: originalError?.message, }); // Emit exhausted header if more than one provider was tried const exhaustedHeader = triedProviders.length > 1 ? { 'X-OLP-Fallback-Exhausted': triedProviders.join(',') } : {}; // Determine status: preserve client errors (400/401/403/404/422) as-is. // Otherwise map ProviderError → 502, unknown → 500. let errStatus = 502; if (originalError) { const httpStatus = originalError.statusCode ?? originalError.status ?? null; if (httpStatus !== null) { errStatus = httpStatus; } else if (!(originalError instanceof ProviderError)) { errStatus = 500; } } // Send error with exhausted header const payload = JSON.stringify({ error: { message: originalError?.message ?? 'Provider error', type: 'provider_error', }, }); res.writeHead(errStatus, { 'Content-Type': 'application/json', 'Content-Length': Buffer.byteLength(payload), ...exhaustedHeader, }); res.end(payload); return; } // ── Success: emit response ───────────────────────────────────────────── // Per ADR 0004 § Observability headers: X-OLP-Fallback-Hops reflects the // chain index of the serving hop; 0 = primary served, 1 = first fallback, etc. // D13: bypass status is per the SERVING hop's provider (providerUsed), not a // global flag. If the serving provider was Anthropic and markers were present, // the cache was bypassed; otherwise it was a hit (preCheckHit) or miss. const bypassCacheForServingHop = shouldBypassCacheForHop(providerUsed); const cacheStatus = bypassCacheForServingHop ? 'bypass' : (preCheckHit && fallbackHops === 0 ? 'hit' : 'miss'); const headers = olpHeaders({ providerUsed, modelUsed, startMs, cacheStatus, fallbackHops }); if (ir.stream) { // Streaming response path: burst replay from buffered chunks. // Reaches here only when: bypassCacheForFirstHop=true OR preCheckHit=true OR chain.length>1. // (Single-hop cache-miss streaming is handled by the real-streaming path above.) res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', Connection: 'keep-alive', 'X-Accel-Buffering': 'no', ...headers, }); for (const irChunk of chunks) { res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model)); if (irChunk.type === 'stop') break; } res.write(SSE_DONE); res.end(); } else { // Non-streaming response path const responseObj = irResponseToOpenAINonStream(chunks, requestId, ir.model); sendJSON(res, 200, responseObj, headers); } } // ── Request router ──────────────────────────────────────────────────────── /** * @param {import('node:http').IncomingMessage} req * @param {import('node:http').ServerResponse} res */ async function router(req, res) { const { method, url } = req; // Strip query string for routing const path = url?.split('?')[0] ?? '/'; try { if (method === 'GET' && path === '/health') { return handleHealth(req, res); } if (method === 'GET' && path === '/v1/models') { return handleModels(req, res); } if (method === 'POST' && path === '/v1/chat/completions') { return await handleChatCompletions(req, res); } // 404 for any unrecognised route sendError(res, 404, `Route ${method} ${path} not found`, 'not_found'); } catch (e) { logEvent('error', 'unhandled_request_error', { method, path, error: e?.message }); if (!res.headersSent) { sendError(res, 500, 'Internal server error', 'internal_error'); } } } // ── Server factory + main guard ─────────────────────────────────────────── // // Factory pattern: `createOlpServer()` returns an http.Server bound to the // shared router but NOT yet listening. Tests import this factory and call // .listen() on their own port. The main guard below only runs .listen() // when this file is invoked directly via `node server.mjs` — preventing // import-time side effects when tests pull in server.mjs. export function createOlpServer() { return createServer(router); } export { router, loadedProviders, VERSION }; // Main guard: only listen when invoked as the entrypoint. ESM equivalent of // `require.main === module` is comparing import.meta.url against argv[1]. const isMain = (() => { try { return import.meta.url === `file://${process.argv[1]}`; } catch { return false; } })(); if (isMain) { const server = createOlpServer(); server.listen(PORT, '127.0.0.1', () => { const enabledCount = loadedProviders.size; process.stdout.write( `OLP v${VERSION} listening on :${PORT} (${enabledCount} providers enabled — Phase 1 in progress)\n`, ); }); }