Files
olp/server.mjs
T
taodengandClaude Opus 4.7 a7085d9718 fix(server): D14 — defer res.writeHead until first chunk in real-streaming branch
cold-audit catch from 2026-05-23

Cold-audit Finding 1 (P2 correctness). Pre-D14 the D10 real-streaming
branch called `res.writeHead(200, ...)` unconditionally before any
provider chunk arrived. If `streamPlugin.spawn(...)` threw or yielded
an error chunk before the first content chunk, the catch block found
`firstChunkEmitted === false` AND `res.headersSent === true` (because
writeHead had already fired), so the `!res.headersSent` branch was
dead code and the path fell through to bare `res.end()`. Result:
HTTP 200 + Content-Type: text/event-stream + zero bytes of body + no
`[DONE]` terminator. The client saw a "successful" empty response.

Compare the buffered path: identical upstream failure returns
HTTP 502 + Content-Type: application/json + `{error: {message, type}}`
body via `sendError(res, 502, ...)`. Inconsistent client-visible
outcomes across paths, with the streaming variant being silently lossy.

Changes (server.mjs +14/-7, test-features.mjs +66):

1. Removed the unconditional pre-loop `res.writeHead(200, ...)` block.
   Replaced with a D14 explanatory comment block.

2. Added a deferred writeHead inside the for-await loop, just before
   the first `res.write`, guarded by `if (!res.headersSent)`:
   ```
   if (!res.headersSent) {
     res.writeHead(200, { Content-Type, Cache-Control, Connection,
                          X-Accel-Buffering, ...streamHeaders });
   }
   streamedChunks.push(irChunk);
   res.write(irChunkToOpenAISSE(...));
   firstChunkEmitted = true;
   ```
   After the first iteration the guard becomes a no-op for subsequent
   chunks. Order inside the loop body is fully synchronous (no await
   between writeHead, write, and the flag assignment), so the
   `firstChunkEmitted` and `res.headersSent` predicates become
   equivalent in steady state.

3. The catch block was NOT touched. Its existing branching was correct
   but the `!res.headersSent → sendError(502)` arm was dead code
   pre-D14; post-D14 it becomes live. This makes the streaming path's
   pre-first-chunk error behavior identical to the buffered path's.

Tests: 291 → 292 (+1):

- Test 15d "streaming early error → 502 JSON not 200 empty body":
  Injects a mock anthropic provider via the `loadedProviders` test
  seam; mock's `spawn` is an async generator that throws ProviderError
  immediately (no yields). Asserts r.status === 502, Content-Type
  includes application/json, body parses to `{error: {message, type}}`
  with type === 'provider_error', and message contains the original
  error string. This test would FAIL on pre-D14 code (would have seen
  200 + empty SSE body).

Scope explicitly excludes:

- Finding #5 (streaming bypasses D4 singleflight) — deferred to D14.5
  with ADR 0005 § D4 amendment pinning the tee-vs-buffer-replay design
  choice. Verified no inflight map, no singleflight machinery, no
  Promise tracking added.

- No ADR change. ADR 0004 § Fallback safety first-chunk rule already
  governs the post-first-chunk behavior (truncated `res.end()`, no
  fallback, no in-band error); D14 preserves that semantics exactly.

Authority:
- OpenAI Chat Completions streaming spec: a failed completion before
  the first byte should be an HTTP error response, not a 200 with empty
  body
  https://platform.openai.com/docs/api-reference/chat/streaming
- ALIGNMENT.md Rule 2(b): no invented response shapes (200 with empty
  body when error occurred is an implicit invention since OpenAI spec
  does not define this as a success shape)
- ADR 0004 § Fallback safety — first-chunk rule (preserved unchanged)
  https://github.com/dtzp555-max/olp/blob/main/docs/adr/0004-fallback-engine.md

Reviewer (Iron Rule v1.6 § 10.x Mode A, fresh-context opus, independent
of drafter): APPROVE. Reviewer stashed the diff to trace pre-D14
control flow manually and confirmed the defect existed exactly as
described (writeHead unconditional → catch with firstChunkEmitted=false
and res.headersSent=true → bare res.end() → silent 200 + empty body).
Verified order inside loop body is correct, headers passed to writeHead
are unchanged, sendError signature matches, stop-as-first-chunk edge
case handled correctly, no singleflight introduced, hygiene clean.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-24 11:17:09 +10:00

697 lines
27 KiB
JavaScript

#!/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<string, boolean>} 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<any>}
*/
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<string,string>} [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<string,string>}
*/
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`,
);
});
}