olp/server.mjs

#!/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,
  listAllProviderNames,
  getAliasMap,
  getModelCreated,
  tryAcquireSpawn,
  releaseSpawn,
  getActiveSpawnCount,
  DEFAULT_MAX_CONCURRENT_SPAWNS,
} 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';
// Phase 2 / D45 — multi-key auth integration per ADR 0007.
import {
  validateKey,
  touchLastUsed,
  loadAuthConfigSync,
  ANONYMOUS_KEY_ID,
} from './lib/keys.mjs';
import { appendAuditEvent } from './lib/audit.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

// ── Logging ───────────────────────────────────────────────────────────────
// Defined early so it is available for startup-time warnings (e.g. F16
// soft-trigger deferred warning) before the provider registry is loaded.

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');
  }
}

// ── 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();

// D26 round-3 F16: soft triggers are deferred to v1.x per ADR 0004 Amendment 2.
// If the user has configured any, emit a startup warning so the inert state is
// observable rather than silently ignored. ADR 0004 Amendment 2 § Mitigations.
const _softTriggersConfigured = Object.keys(_startupConfig.soft_triggers ?? {}).length > 0;
if (_softTriggersConfigured) {
  logEvent('warn', 'soft_triggers_deferred_v1x', {
    configured_providers: Object.keys(_startupConfig.soft_triggers),
    message: 'routing.soft_triggers configured but soft triggers are deferred to v1.x; ' +
      'thresholds will not fire at v0.1 — see ADR 0004 Amendment 2',
  });
}

// ── Auth config (Phase 2 / D45, ADR 0007 § 7.2) ───────────────────────────
// auth.allow_anonymous default false (production-off). Test seam below.
let _authConfig = loadAuthConfigSync();
if (_authConfig.allow_anonymous === true) {
  logEvent('warn', 'auth_allow_anonymous_enabled', {
    message: 'config.json auth.allow_anonymous is true — all routes accept requests without an OLP API key; production deployments should set this to false (ADR 0007 § 7).',
  });
}

/** @internal — test seam: inject a synthetic auth config (no file I/O). */
export function __setAuthConfig(config) {
  _authConfig = config ?? { allow_anonymous: false, owner_only_endpoints: ['/health'], fallback_detail_header_policy: 'owner_only' };
}

/** @internal — reset auth config to file-loaded state. */
export function __resetAuthConfig() {
  _authConfig = loadAuthConfigSync();
}

// ── 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();

// ── 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
 * @param {Record<string,string>} [extraHeaders] — optional extra headers (e.g. X-OLP-Latency-Ms)
 */
function sendError(res, status, message, type, extraHeaders = {}) {
  sendJSON(res, status, { error: { message, type } }, extraHeaders);
}

// ── 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),
  };
}

/**
 * Returns the standard 5-header OLP diagnostic set for pre-chain error paths
 * (no provider was attempted yet). Used by sendError call sites that occur
 * before chain construction (415 wrong Content-Type, 400 bad JSON, 400 IR
 * validation failure, 503 no-enabled-providers).
 *
 * Per F8 (D32 round-4 cold-audit): ADR 0004 § Observability requires the full
 * 5-header set on every response; early error paths must emit canonical
 * "no provider attempted" defaults.
 *
 * @param {object} opts
 * @param {number} opts.startMs — request start timestamp (Date.now())
 * @param {string|null|undefined} [opts.model] — ir.model if IR was parsed; undefined/null otherwise
 * @returns {Record<string,string>}
 */
function olpErrorHeaders({ startMs, model }) {
  return {
    'X-OLP-Provider-Used': 'none',
    'X-OLP-Model-Used': model ?? 'unknown',
    'X-OLP-Fallback-Hops': '0',
    'X-OLP-Cache': 'bypass',
    'X-OLP-Latency-Ms': String(Date.now() - startMs),
  };
}

// ── X-OLP-Fallback-Detail (D40, issue #7) ─────────────────────────────────

/**
 * 4KB UTF-8 byte cap on the X-OLP-Fallback-Detail header value. Per RFC 7230 §3.2.5
 * intermediaries are not obligated to forward arbitrarily large header values;
 * 4KB matches the conservative upper bound (well below the 8KB total-header
 * default of common reverse proxies — nginx `large_client_header_buffers`,
 * Apache `LimitRequestFieldSize`). Tuples beyond the cap are dropped and
 * a sentinel { truncated:true, omitted_hops:N } is appended.
 */
export const FALLBACK_DETAIL_BYTE_CAP = 4096;

/**
 * Serialises per-hop fallback failure tuples into the X-OLP-Fallback-Detail
 * header value. Returns null if the input array is empty/missing (header
 * should not be emitted in that case).
 *
 * D40 (issue #7) — see ADR 0004 § Observability headers.
 *
 * Cap behaviour:
 *   - JSON.stringify the tuples; if Buffer.byteLength <= 4096, return as-is.
 *   - Otherwise, drop tuples from the tail one at a time until the array PLUS
 *     a trailing { truncated:true, omitted_hops:N } sentinel fits under the cap.
 *   - If even a single tuple + sentinel cannot fit (extremely long error_message
 *     beyond engine truncation, e.g. very long provider/model names), return
 *     just the sentinel { truncated:true, omitted_hops:<all> } — never produce
 *     a value > 4096 bytes.
 *
 * RFC 7230 hygiene: JSON.stringify already escapes raw newlines (\n → \\n),
 * carriage returns, and other control characters. In addition, we escape all
 * non-ASCII code points to \uXXXX sequences because Node's HTTP header
 * validator rejects multi-byte UTF-8 in field values (and RFC 7230 §3.2.6
 * limits `field-vchar` to ASCII VCHAR / obs-text). Without this step, an
 * em dash (U+2014) in a synthesised error message — e.g. the CONCURRENCY_LIMIT
 * message produced in server.mjs `collectAllChunks` — would trigger
 * `Invalid character in header content` from `res.writeHead`.
 *
 * @param {Array<object>|null|undefined} fallbackDetail — from FallbackResult.fallbackDetail
 * @returns {string|null} — header value, or null to skip emission
 */
export function serializeFallbackDetailHeader(fallbackDetail) {
  if (!Array.isArray(fallbackDetail) || fallbackDetail.length === 0) {
    return null;
  }
  const full = jsonStringifyAscii(fallbackDetail);
  if (Buffer.byteLength(full, 'utf8') <= FALLBACK_DETAIL_BYTE_CAP) {
    return full;
  }

  // Cap exceeded — drop tail tuples until [...kept, sentinel] fits.
  // Linear scan from the full array down to 0 kept tuples. Worst case O(n^2)
  // on serialisation length, but n is bounded by chain length (small) so this
  // is fine in practice.
  for (let kept = fallbackDetail.length - 1; kept >= 0; kept--) {
    const omitted = fallbackDetail.length - kept;
    const sentinel = { truncated: true, omitted_hops: omitted };
    const candidate = jsonStringifyAscii([...fallbackDetail.slice(0, kept), sentinel]);
    if (Buffer.byteLength(candidate, 'utf8') <= FALLBACK_DETAIL_BYTE_CAP) {
      return candidate;
    }
  }

  // Even an array containing only the sentinel exceeds the cap — produce the
  // shortest possible valid sentinel value. This branch should be unreachable
  // for any realistic chain (the sentinel itself is ~45 bytes for omitted_hops
  // up to 9999).
  return jsonStringifyAscii([{ truncated: true, omitted_hops: fallbackDetail.length }]);
}

/**
 * JSON.stringify wrapper that escapes every non-ASCII code point as \uXXXX
 * so the result is safe to embed in an HTTP header value (RFC 7230 §3.2.6
 * field-vchar). JSON itself accepts both literal Unicode and \uXXXX escapes,
 * so JSON.parse round-trips correctly.
 *
 * Surrogate-pair handling: characters above U+FFFF (emoji etc.) are already
 * emitted as JS surrogate pairs by the string iterator; each surrogate is
 * a code unit in range 0xD800–0xDFFF, which our >= 0x80 guard catches.
 *
 * @param {unknown} value
 * @returns {string}
 */
function jsonStringifyAscii(value) {
  // The replace pattern is the UTF-8 literal byte range for code points
  // U+0080..U+FFFF (every non-ASCII BMP character). U+0080 is non-printable,
  // so the source line can render as the empty character class "[-...]" in
  // editors that hide it — the range is intentional and load-bearing for
  // RFC 7230 §3.2.6 compliance.
  return JSON.stringify(value).replace(/[€-￿]/g, (ch) => {
    return '\\u' + ch.charCodeAt(0).toString(16).padStart(4, '0');
  });
}

/**
 * Merges X-OLP-Fallback-Detail into a base header object when the per-hop
 * failure tuples are non-empty. Returns the base object unchanged otherwise.
 *
 * D40 (issue #7).
 *
 * @param {Record<string,string>} baseHeaders
 * @param {Array<object>|null|undefined} fallbackDetail
 * @returns {Record<string,string>}
 */
function withFallbackDetailHeader(baseHeaders, fallbackDetail) {
  const value = serializeFallbackDetailHeader(fallbackDetail);
  if (value === null) return baseHeaders;
  return { ...baseHeaders, 'X-OLP-Fallback-Detail': value };
}

// ── Route handlers ────────────────────────────────────────────────────────

// ── Auth middleware (Phase 2 / D45, ADR 0007 § 5 + § 7) ───────────────────

/**
 * Extract the plaintext OLP token from request headers.
 * Tries `Authorization: Bearer <token>` first, then `x-api-key: <token>`.
 * Returns the token string or null.
 *
 * @param {import('node:http').IncomingMessage} req
 * @returns {string|null}
 */
function extractToken(req) {
  const auth = req.headers['authorization'];
  if (typeof auth === 'string') {
    const match = /^Bearer\s+(\S+)$/i.exec(auth);
    if (match) return match[1];
  }
  const xKey = req.headers['x-api-key'];
  if (typeof xKey === 'string' && xKey.length > 0) return xKey;
  return null;
}

/**
 * Authenticate the request per ADR 0007 §§ 5 / 7 / 9.4.
 *   - Try env-owner override first (OLP_OWNER_TOKEN).
 *   - Then filesystem manifest lookup by SHA-256 hash.
 *   - Else, if auth.allow_anonymous is true → anonymous identity.
 *   - Else → 401 auth_required.
 *
 * Returns { ok: true, authContext } on success or
 * { ok: false, status, code, message } on failure (401).
 *
 * @param {import('node:http').IncomingMessage} req
 * @returns {{ ok: true, authContext: { keyId: string, owner_tier: 'owner'|'guest'|'anonymous', providers_enabled: string[]|'*', source: 'env'|'filesystem'|'anonymous' } } | { ok: false, status: number, code: string, message: string }}
 */
function authenticate(req) {
  const token = extractToken(req);
  const identity = validateKey(token, { allowAnonymous: _authConfig.allow_anonymous });
  if (identity === null) {
    // Distinguish "no token presented" from "token presented but invalid/revoked".
    // Both surface as 401 to the client (don't leak which case it was), but the
    // server-side audit + log captures the source for operator diagnosis.
    return {
      ok: false,
      status: 401,
      code: token ? 'invalid_or_revoked_key' : 'auth_required',
      message: token
        ? 'OLP API key is invalid or has been revoked.'
        : 'OLP API key required. Pass via "Authorization: Bearer <token>" or "x-api-key: <token>".',
    };
  }
  return {
    ok: true,
    authContext: {
      keyId: identity.id,
      owner_tier: identity.owner_tier,
      providers_enabled: identity.providers_enabled,
      source: identity.source,
    },
  };
}

/**
 * Check whether the given provider key is permitted for this identity.
 * `providers_enabled: '*'` grants all; an array is a whitelist.
 *
 * @param {{ providers_enabled: string[]|'*' }} authContext
 * @param {string} providerKey
 * @returns {boolean}
 */
function isProviderEnabled(authContext, providerKey) {
  if (authContext.providers_enabled === '*') return true;
  return Array.isArray(authContext.providers_enabled) && authContext.providers_enabled.includes(providerKey);
}

/**
 * GET /health
 * Returns server health including count of loaded providers and per-provider
 * healthCheck() snapshots (ADR 0002 § Provider contract: healthCheck is used
 * by startup and /health endpoint per ADR 0002 § Provider contract description).
 *
 * Phase 2 / D45: no auth gate at this endpoint yet. Owner-vs-guest /health
 * payload trimming per ADR 0007 § 7.1 lands at D46.
 */
async function handleHealth(req, res) {
  const enabled = loadedProviders.size;
  const available = listAllProviderNames().length;
  const providerStatuses = {};
  for (const [name, provider] of loadedProviders) {
    try {
      providerStatuses[name] = await provider.healthCheck();
    } catch (e) {
      providerStatuses[name] = { ok: false, error: e.message };
    }
  }
  sendJSON(res, 200, {
    ok: true,
    version: VERSION,
    providers: { enabled, available, status: providerStatuses },
  });
}

/**
 * GET /v1/models
 * Returns the list of models served by all currently loaded (enabled) providers.
 * Per ADR 0002 § "Loading model" + OpenAI spec /v1/models:
 *   Each entry: { id, object: 'model', created, owned_by }
 *   - id:        canonical model ID (or alias string for alias entries)
 *   - object:    literal 'model' (OpenAI spec)
 *   - created:   Unix epoch seconds (stable per request; computed once from Date.now())
 *   - owned_by:  provider.name (e.g. 'anthropic', 'openai', 'mistral')
 * Order: canonical models first (insertion order of loadedProviders, then models[]),
 *        then alias entries (for aliases whose target provider is currently loaded).
 * Authority: OpenAI /v1/models spec (https://platform.openai.com/docs/api-reference/models);
 *            models-registry.json alias map (SPOT per D17); F15 round-3 adds alias surfacing.
 * Empty case: if no providers are enabled, data: [] is returned naturally.
 */
function handleModels(req, res) {
  const startMs = Date.now();

  // Audit + auth for /v1/models per Phase 2 / D45 (ADR 0007 § 7).
  const auditCtx = {
    ts: new Date().toISOString(),
    key_id: ANONYMOUS_KEY_ID,
    owner_tier: 'anonymous',
    method: 'GET',
    path: '/v1/models',
    provider: null,
    model: null,
    status_code: 0,
    latency_ms: 0,
    cache_status: null,
    fallback_hops: 0,
    tried_providers: [],
    error_code: null,
    ir_request_hash: null,
    chain_id: null,
  };
  let _authedKeyId = null;
  res.on('finish', () => {
    auditCtx.status_code = res.statusCode;
    auditCtx.latency_ms = Date.now() - startMs;
    try { appendAuditEvent(auditCtx); } catch { /* best-effort */ }
    if (_authedKeyId && _authedKeyId !== ANONYMOUS_KEY_ID) {
      touchLastUsed(_authedKeyId).catch(() => {});
    }
  });

  const authResult = authenticate(req);
  if (!authResult.ok) {
    auditCtx.error_code = authResult.code;
    return sendError(res, authResult.status, authResult.message, authResult.code,
      olpErrorHeaders({ startMs }));
  }
  auditCtx.key_id = authResult.authContext.keyId;
  auditCtx.owner_tier = authResult.authContext.owner_tier;
  _authedKeyId = authResult.authContext.keyId;

  const data = [];

  // Canonical entries first
  for (const [providerName, provider] of loadedProviders) {
    for (const modelId of provider.models) {
      data.push({
        id: modelId,
        object: 'model',
        // F12 (round-5 cold-audit): use stable per-model timestamp from
        // models-registry.json rather than Date.now() on each request.
        // OpenAI spec treats `created` as a stable per-model attribute.
        created: getModelCreated(modelId),
        owned_by: providerName,
      });
    }
  }

  // Alias entries for loaded (enabled) providers, canonical-first ordering preserved
  for (const [alias, { providerName, canonicalModel }] of getAliasMap()) {
    if (loadedProviders.has(providerName)) {
      data.push({
        id: alias,
        object: 'model',
        // Alias entries use the same stable timestamp as their canonical model.
        created: getModelCreated(canonicalModel),
        owned_by: providerName,
      });
    }
  }

  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();

  // Audit context — fields populated as the request proceeds; § 8 schema.
  // Fired on res.on('finish') below regardless of success / error path.
  const auditCtx = {
    ts: new Date().toISOString(),
    key_id: ANONYMOUS_KEY_ID,        // updated post-auth
    owner_tier: 'anonymous',         // updated post-auth
    method: 'POST',
    path: '/v1/chat/completions',
    provider: null,
    model: null,
    status_code: 0,                  // updated on finish
    latency_ms: 0,                   // updated on finish
    cache_status: null,
    fallback_hops: 0,
    tried_providers: [],
    error_code: null,
    ir_request_hash: null,
    chain_id: null,
  };
  // Wire post-response audit + lazy last_used_at update once, at the top, so
  // every code path below (401, 400, 500, 200, streaming) emits an audit row.
  // touchLastUsed is best-effort and never throws (§ 6.3).
  let _authedKeyId = null;
  res.on('finish', () => {
    auditCtx.status_code = res.statusCode;
    auditCtx.latency_ms = Date.now() - startMs;
    try { appendAuditEvent(auditCtx); } catch { /* audit is best-effort */ }
    if (_authedKeyId && _authedKeyId !== ANONYMOUS_KEY_ID) {
      // Anonymous + __env_owner__ have no on-disk manifest to touch; touchLastUsed
      // itself early-returns for those identities (§ 6.3 wrapper). For
      // filesystem-stored keys, fire-and-forget.
      touchLastUsed(_authedKeyId).catch(() => { /* warned internally */ });
    }
  });

  // ── Authentication (Phase 2 / D45, ADR 0007 § 5 + § 7) ───────────────────
  // olpIdentity carries the OLP-side identity (keyId / owner_tier /
  // providers_enabled). It is SEPARATE from `authContext` which is the
  // per-provider OAuth/credential artifact passed to provider.spawn().
  // Provider plugins treat `authContext === null` as "fall back to your
  // own credential discovery (env / keychain / file)" — that contract is
  // preserved at D45. Per-provider per-key credential mapping is a Phase
  // 3+ concern (ADR 0007 § 12 Out of scope; v0.1 spec § 4.5 anticipated).
  const authResult = authenticate(req);
  if (!authResult.ok) {
    auditCtx.error_code = authResult.code;
    return sendError(res, authResult.status, authResult.message, authResult.code,
      olpErrorHeaders({ startMs }));
  }
  const olpIdentity = authResult.authContext;
  auditCtx.key_id = olpIdentity.keyId;
  auditCtx.owner_tier = olpIdentity.owner_tier;
  _authedKeyId = olpIdentity.keyId;

  // Require JSON content-type
  const ct = req.headers['content-type'] ?? '';
  if (!ct.includes('application/json')) {
    auditCtx.error_code = 'invalid_content_type';
    return sendError(res, 415, 'Content-Type must be application/json', 'invalid_request_error',
      olpErrorHeaders({ startMs }));
  }

  let body;
  try {
    body = await readJSON(req);
  } catch (e) {
    auditCtx.error_code = 'invalid_request_body';
    return sendError(res, e.statusCode ?? 400, e.message, 'invalid_request_error',
      olpErrorHeaders({ startMs }));
  }

  // Translate OpenAI → IR (ADR 0003)
  let ir;
  try {
    ir = openAIToIR(body);
  } catch (e) {
    if (e instanceof BadRequestError) {
      auditCtx.error_code = 'invalid_ir';
      return sendError(res, 400, e.message, 'invalid_request_error',
        olpErrorHeaders({ startMs }));
    }
    throw e;
  }
  auditCtx.model = ir.model;

  // ── 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.
  //
  // `let` (not `const`) because the chain may be filtered below per the
  // authenticated key's providers_enabled allowlist (ADR 0007 § 10 #11).
  let chain = buildDefaultChain(
    ir.model,
    loadedProviders,
    _fallbackConfig.chains,
    _fallbackConfig.soft_triggers,
  );

  if (!chain) {
    // ALIGNMENT.md: 0 Enabled Providers at v0.1 → 503 per spec
    auditCtx.error_code = 'no_enabled_provider';
    return sendError(
      res, 503,
      `No enabled providers for model ${ir.model}. See README § Supported Providers.`,
      'no_enabled_provider',
      olpErrorHeaders({ startMs, model: ir.model }),
    );
  }

  // ── providers_enabled scope enforcement (Phase 2 / D45, ADR 0007 § 10 #11) ─
  // Filter the chain to providers this key is authorized for. If the resulting
  // chain is empty, return 403 — the model exists in the registry but no hop
  // is reachable from this identity's allowlist.
  const _originalChainProviders = chain.map(hop => hop.provider);
  chain = chain.filter(hop => isProviderEnabled(olpIdentity, hop.provider));
  if (chain.length === 0) {
    auditCtx.error_code = 'key_no_provider_access';
    auditCtx.tried_providers = _originalChainProviders;
    const allowed = olpIdentity.providers_enabled === '*' ? '*' : (olpIdentity.providers_enabled ?? []).join(', ') || '(none)';
    return sendError(
      res, 403,
      `This OLP key does not have access to any provider serving model "${ir.model}". Key's providers_enabled: [${allowed}]. Chain providers: [${_originalChainProviders.join(', ')}].`,
      'key_no_provider_access',
      olpErrorHeaders({ startMs, model: ir.model }),
    );
  }

  // Auth context is null at D45 — providers fall back to their own credential
  // discovery (env var, keychain, credentials file). The OLP-side identity
  // (olpIdentity above) is consumed for cache namespacing + providers_enabled
  // gating + audit attribution. Per-provider per-key credential mapping is a
  // Phase 3+ deferral (ADR 0007 § 12).
  const authContext = null;

  const requestId = generateRequestId();

  // ── Cache layer (ADR 0005, namespaced per OLP key per ADR 0007 § 7) ─────
  // keyId is the authenticated identity's namespace token. For anonymous
  // (when auth.allow_anonymous=true) this is the legacy '__anonymous__'
  // shared namespace; for filesystem keys it is the per-key id; for the
  // OLP_OWNER_TOKEN env override it is the synthetic '__env_owner__'.
  const keyId = olpIdentity.keyId;

  // 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;

  // D36 #2 (ADR 0005 § D2): when cache_control markers are present AND at least one
  // hop in the chain is non-Anthropic, the markers are noop'd for those hops.
  // Per ADR 0005 § Context: "for non-Anthropic targets, the bypass markers are
  // noop'd (logged once per request at debug level so users can see they were
  // ignored)." Fires at most once per request, gated on (markers AND mixed/non-anthropic
  // chain). No log when no markers, or when every chain hop is Anthropic.
  if (hasCacheControlMarkers && chain.some(hop => hop.provider !== 'anthropic')) {
    // marker_count sums body-side and IR-side markers. At v0.1 the IR term is
    // structurally 0 (openAIToIR strips cache_control). When a future ADR 0003
    // amendment activates cache_control in the IR whitelist, both terms will be
    // non-zero for the same logical marker set → revisit to avoid 2× counting.
    const markerCount =
      extractCacheControlMarkers(body?.messages ?? []).length +
      extractCacheControlMarkers(ir.messages ?? []).length;
    logEvent('debug', 'cache_control_partial_noop', {
      chain: chain.map(hop => hop.provider),
      marker_count: markerCount,
    });
  }

  /**
   * 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.
  //
  // F8 (round-5 cold-audit): track whether the SERVING hop's response came from
  // cache. preCheckHit only covered the PRIMARY hop's key; when a fallback hop
  // serves from its own cache, the header must report 'hit', not 'miss'.
  // lastHopWasCached is set by executeHopFn before returning so the cacheStatus
  // computation below can consume it after executeWithFallback completes.
  let lastHopWasCached = false;

  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.
    //
    // D16 (ADR 0004 Amendment 1): if the generator throws ProviderError SPAWN_FAILED
    // after yielding one or more chunks, those chunks are usable content the provider
    // committed — dropping them and falling back to the next hop is strict waste.
    // In that case: synthesize a stop chunk with finish_reason='length', return the
    // partial chunks, and do NOT re-throw (fallback engine sees a success, hops=0).
    // If chunks.length===0 on SPAWN_FAILED, re-throw as before (hard trigger fires).
    // Any other error code always re-throws unchanged.
    //
    // D16 cache note: truncated responses are NOT cached (ADR 0005 § cache write
    // conditions item 1 requires no truncation). A non-enumerable __truncated marker
    // is set on the returned array so executeHopFn can evict the cache entry below.
    async function collectAllChunks() {
      // D38 (issue #1): maxConcurrent runtime enforcement.
      // Authority: ADR 0002 Amendment 6 + ADR 0004 Amendment 4.
      //
      // Try to acquire a spawn slot for this provider BEFORE invoking spawn().
      // If at limit, synthesise ProviderError(CONCURRENCY_LIMIT) which the
      // fallback engine treats as a hard trigger — the chain advances to the
      // next hop. validateProvider guarantees hints.maxConcurrent is present;
      // DEFAULT_MAX_CONCURRENT_SPAWNS is a defense-in-depth fallback.
      const maxConcurrent = hopProviderPlugin.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS;
      if (!tryAcquireSpawn(hopProvider, maxConcurrent)) {
        const concurrencyErr = new ProviderError(
          `provider ${hopProvider} at maxConcurrent (${maxConcurrent}) — advancing to next hop`,
          'CONCURRENCY_LIMIT',
        );
        concurrencyErr.providerName = hopProvider;
        concurrencyErr.maxConcurrent = maxConcurrent;
        // activeSpawns reflects the live counter at the rejection moment,
        // queried directly. Since tryAcquireSpawn returned false the value
        // equals maxConcurrent — read it explicitly for diagnostic clarity
        // rather than echoing the limit (avoids future-reader confusion).
        concurrencyErr.activeSpawns = getActiveSpawnCount(hopProvider);
        throw concurrencyErr;
      }

      const chunks = [];
      // try/finally: releaseSpawn MUST fire on every exit path — success
      // (return at end), spawn throw (caught and re-thrown below), or the
      // D16 truncation-salvage return. The finally is the only mechanism
      // that guarantees release across all three.
      try {
        try {
          for await (const irChunk of hopProviderPlugin.spawn(irReq, authContext)) {
            // D16: check error chunks BEFORE pushing — preserves the invariant that
            // chunks array contains only delta/stop chunks. Without this, the catch
            // block's `chunks.length > 0` would mistake a single error chunk for
            // "usable content streamed" (Case B) and synthesize a stop + return,
            // sending an empty body to the client when the correct behavior is to
            // re-throw and let the fallback engine advance the chain.
            if (irChunk.type === 'error') {
              throw new ProviderError(
                irChunk.error ?? 'Provider emitted error chunk',
                'SPAWN_FAILED',
              );
            }
            chunks.push(irChunk);
            if (irChunk.type === 'stop') break;
          }
        } catch (spawnErr) {
          if (spawnErr instanceof ProviderError && spawnErr.code === 'SPAWN_FAILED' && chunks.length > 0) {
            // Case B (ADR 0004 Amendment 1): provider emitted usable chunks then exited
            // non-zero. Synthesize a truncated stop and surface the partial response.
            chunks.push({ type: 'stop', finish_reason: 'length' });
            logEvent('warn', 'spawn_failed_after_usable_chunks', {
              chunks_count: chunks.length - 1, // exclude the synthesized stop
              provider: hopProvider,
              model: hopModel,
            });
            // Mark as truncated so the caller can evict this entry from cache.
            Object.defineProperty(chunks, '__truncated', { value: true, enumerable: false });
            return chunks;
          }
          // Case A (SPAWN_FAILED with no chunks) or any other error: re-throw.
          // Fallback engine fires hard trigger and advances chain as before.
          throw spawnErr;
        }
        return chunks;
      } finally {
        // D38: spawn lifecycle ended (drain completed, stop chunk received,
        // SPAWN_FAILED salvage returned, or any unexpected throw). Release
        // the slot so the next caller can acquire it. Single-threaded JS
        // guarantees no other caller has incremented this provider's count
        // between our tryAcquireSpawn() above and this releaseSpawn().
        releaseSpawn(hopProvider);
      }
    }

    // D23: cacheable opt-out check (ADR 0002 Amendment 3 + ADR 0005 Amendment 3).
    // If a provider explicitly sets hints.cacheable === false, skip the cache entirely
    // for every request to this provider. collectAllChunks is called directly; neither
    // cacheStore.get nor cacheStore.set is invoked. This is upstream of D13's
    // cache_control bypass — both are "skip the cache" paths, but for different reasons.
    if (hopProviderPlugin.hints?.cacheable === false) {
      logEvent('debug', 'cache_opted_out', {
        provider: hopProvider,
        model: hopModel,
      });
      return collectAllChunks();
    }

    // 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).
    //
    // D16: after getOrCompute returns, check if collectAllChunks used the salvage
    // path (chunks.__truncated===true). If so, evict the just-written cache entry
    // to satisfy ADR 0005 § "Cache write conditions" item 1 (no truncation).
    // D4 singleflight is preserved: getOrCompute still deduplicates concurrent
    // requests during the spawn; the eviction only affects persistent caching.
    //
    // F8 (round-5 cold-audit): peek before getOrCompute so we know whether the
    // value comes from cache or is freshly computed. peek() is stats-neutral per
    // its contract (no hit/miss counter side-effect), so it does not distort the
    // stats that getOrCompute will update on the canonical miss path.
    // This is option (b) of the F8 spec: check cache BEFORE calling getOrCompute.
    const hopWasCached = await cacheStore.peek(keyId, hopCacheKey);
    const result = await cacheStore.getOrCompute(keyId, hopCacheKey, collectAllChunks);
    // Record for use in cacheStatus computation after executeWithFallback returns.
    lastHopWasCached = hopWasCached;
    if (result.__truncated) {
      // Evict the truncated entry so future requests get a fresh spawn.
      // ADR 0005 § "Cache write conditions" item 1: truncated responses must
      // not persist in cache.
      //
      // D39 (issue #3 Part 1): use explicit cacheStore.delete() rather than
      // the prior set(..., ttlMs=0) tombstone. delete() removes the entry
      // from the namespace Map immediately (and removes the empty namespace
      // entry from the outer Map if applicable), instead of waiting for the
      // next get/peek to lazily purge a TTL=0 entry.
      // Capture the boolean — false indicates a race (concurrent eviction or
      // TTL purge already removed the entry). Surfaces in the log so the
      // dashboard can distinguish "we evicted" from "we tried but it was gone."
      const evicted = cacheStore.delete(keyId, hopCacheKey);
      // D39 (issue #3 Part 2): observability — surface salvage frequency to
      // dashboards. Provider + model identify which hop's truncated entry was
      // evicted. cache_eviction_hit distinguishes actual-evict vs already-gone.
      logEvent('info', 'cache_evicted_truncated', {
        provider: hopProvider,
        model: hopModel,
        cache_eviction_hit: evicted,
      });
    }
    return result;
  }

  // ── 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);
  // D23: ADR 0002 Amendment 3 — cacheable: false providers skip cache entirely.
  // Compute once here so both the peek guard and the streaming-branch entry condition
  // can consult the same flag without re-reading the plugin map.
  const firstHopProvider = loadedProviders.get(chain[0].provider);
  const cacheableForFirstHop = firstHopProvider?.hints?.cacheable !== false;
  const firstHopCacheKey = computeCacheKey(chain[0].provider, chain[0].model, ir);
  const preCheckHit = (bypassCacheForFirstHop || !cacheableForFirstHop) ? 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)
  //   - cacheableForFirstHop → provider allows caching (D23: cacheable: false falls
  //     through to the buffered executeHopFn path which already respects the opt-out)
  //
  // 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.
  // D38 (issue #1): pre-acquire a concurrency slot for the streaming path so
  // saturation here behaves identically to saturation on the buffered path.
  // If acquire fails, fall through (skip this branch) — the buffered path
  // below also gates via tryAcquireSpawn and will surface a chain-exhausted
  // error through executeWithFallback (correct behaviour: a single-hop chain
  // at maxConcurrent has no other hop to advance to).
  //
  // Acquired here, released in the `finally` below. The gate intentionally
  // lives BEFORE the streaming-branch entry check so the buffered fallthrough
  // can re-attempt acquire from a clean slate.
  // TODO(v1.x — ADR 0005 Amendment 8 / issue #16): replace this peek+spawn
  // pattern with cacheStore.getOrComputeStreaming(...) to close the TOCTOU
  // window between line ~782 peek and the spawn at line ~846, and to make N
  // concurrent identical streaming requests share one spawn (tee-streaming).
  // Design ratified in D42; see docs/v1x-roadmap.md #1 for the trigger and
  // acceptance criteria. DO NOT remove this comment until the v1.x impl lands.
  let streamingAcquired = false;
  if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop) {
    const candidatePlugin = loadedProviders.get(chain[0].provider);
    const candidateMax = candidatePlugin?.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS;
    streamingAcquired = candidatePlugin ? tryAcquireSpawn(chain[0].provider, candidateMax) : false;
  }

  if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop && streamingAcquired) {
    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).
      // Release the slot we acquired above so the counter stays balanced.
      auditCtx.provider = streamProvider;
      auditCtx.error_code = 'no_enabled_provider';
      releaseSpawn(streamProvider);
      return sendError(res, 503, `Provider ${streamProvider} is not enabled`, 'no_enabled_provider',
        olpErrorHeaders({ startMs, model: ir.model }));
    }

    // D45 fold-in P1: populate audit ctx for the real-streaming path. Each
    // exit below (success, error-after-first-chunk, pre-first-chunk-error,
    // 503 above) leaves these fields representing the streaming attempt.
    // Error paths amend `error_code`; success leaves it null.
    auditCtx.provider = streamProvider;
    auditCtx.tried_providers = [streamProvider];
    auditCtx.cache_status = 'miss';

    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; emit truncation marker + [DONE]
            // so clients can detect the incomplete response in-band (aligns D26 F19
            // stop-less exhaustion behaviour and the catch-block fix in D35 #10).
            logEvent('warn', 'streaming_error_after_first_chunk', {
              provider: streamProvider,
              model: streamModel,
              error: irChunk.error,
            });
            auditCtx.error_code = 'streaming_error_after_first_chunk';
            res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
            res.write(SSE_DONE);
            res.end();
            return;
          }
          // No bytes written yet — throw to surface a clean error.
          // auditCtx.error_code is set by the downstream catch handler (the
          // outer streaming-path catch block fills it from the thrown 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.
          // D23 defense-in-depth: cacheableForFirstHop is true here (cacheable: false
          // falls through to the buffered path, never enters this block), but the guard
          // makes the intent explicit and survives future refactors.
          if (cacheableForFirstHop) {
            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] but do NOT cache.
      // A stop-less exhaustion means the response is truncated (the generator
      // ended without the model signalling completion). Caching a truncated
      // response would serve wrong answers to future identical requests.
      // Compare: D16's buffered-path truncation eviction explicitly avoids
      // persisting truncated entries for the same reason.
      //
      // D26 round-3 F19: emit a synthetic truncation marker BEFORE [DONE] so
      // clients can detect the incomplete response in-band. Only emit when there
      // is actual partial content (streamedChunks.length > 0) — emitting a
      // truncation marker on an empty response is misleading.
      // The buffered D16 path synthesizes {type:'stop', finish_reason:'length'}
      // before returning; this aligns the streaming branch with that behaviour.
      if (streamedChunks.length > 0) {
        const truncMarker = { type: 'stop', finish_reason: 'length' };
        res.write(irChunkToOpenAISSE(truncMarker, requestId, ir.model));
      }

      // D35 #9: Zero-chunk empty-stream path — writeHead is still deferred
      // (firstChunkEmitted===false) when the generator yields no chunks at all and
      // exits cleanly. Without an explicit writeHead Node auto-emits 200 with the
      // default Content-Type and none of the X-OLP-* headers.
      // A provider that yielded nothing still constitutes an attempted call, so we
      // emit the full olpHeaders (provider WAS attempted, just yielded zero chunks).
      if (!res.headersSent) {
        res.writeHead(200, {
          'Content-Type': 'text/event-stream',
          'Cache-Control': 'no-cache',
          Connection: 'keep-alive',
          'X-Accel-Buffering': 'no',
          ...streamHeaders,
        });
      }

      res.write(SSE_DONE);
      res.end();
      // Loop exhausted without stop chunk = truncation. The stop-chunk completion
      // path returns earlier (above, inside the for-await loop); reaching here means
      // the generator returned without emitting stop. Never cache (per ADR 0005 cache
      // write conditions item 1: truncated responses must not persist in cache).
      if (streamedChunks.length > 0 && cacheableForFirstHop) {
        logEvent('warn', 'streaming_no_stop_chunk', {
          chunks_count: streamedChunks.length,
          provider: streamProvider,
          model: streamModel,
        });
      }
    } catch (e) {
      if (firstChunkEmitted) {
        // Past first-chunk boundary — can't fallback; emit truncation marker + [DONE]
        // so clients can detect the incomplete response in-band (aligns with D26 F19
        // stop-less exhaustion behaviour). ADR 0004 § Fallback safety: no fallback
        // after first-chunk boundary; truncation is the correct recovery.
        logEvent('warn', 'streaming_error_after_first_chunk', {
          provider: streamProvider,
          model: streamModel,
          error: e.message,
        });
        res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
        res.write(SSE_DONE);
        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,
        });
        auditCtx.error_code = e?.code ?? 'provider_error';
        if (!res.headersSent) {
          sendError(res, 502, e.message ?? 'Provider error', 'provider_error',
            olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 }));
        } else {
          res.end();
        }
      }
    } finally {
      // D38 (issue #1): streaming spawn lifecycle ended — drain completed,
      // stop chunk seen, generator exhausted without stop, or any catch
      // path returned via res.end(). Release the slot acquired before
      // entering the streaming branch. The finally fires on every JS exit
      // path including the `return;` statements inside the try/catch body.
      releaseSpawn(streamProvider);
    }
    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',
      olpErrorHeaders({ startMs, model: ir.model }));
  }

  const {
    chunks,
    providerUsed,
    modelUsed,
    fallbackHops,
    originalError,
    triedProviders,
    fallbackDetail,  // D40 (issue #7): per-hop failure tuples for X-OLP-Fallback-Detail
  } = 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;
      }
    }

    // Per ADR 0004 § Observability headers: all responses (including errors) carry
    // the standard 5-header set. On the exhausted/error path the engine returns
    // values per ADR 0004 step 4 ("preserve A's identity — return the FIRST hop's
    // provider/model and original error to the user"):
    //   - providerUsed: chain[0].provider on chain-exhausted (the primary that
    //                   first failed); set to 'none' only if engine somehow returned null
    //   - modelUsed:    chain[0].model on chain-exhausted, or the original request
    //                   model if engine state is unknown
    //   - cacheStatus:  'miss' — all hops were attempted (bypass is per-hop, not relevant
    //                   when the whole chain exhausted)
    //   - fallbackHops: number of hops actually attempted before exhaustion
    // X-OLP-Fallback-Exhausted is preserved as an additional flag on top of these.
    const errorOlpHeaders = olpHeaders({
      providerUsed: providerUsed ?? 'none',
      modelUsed: modelUsed ?? ir.model,
      startMs,
      cacheStatus: 'miss',
      fallbackHops: fallbackHops ?? 0,
    });

    // Audit ctx capture for chain-exhausted / provider-error path (audit fires
    // on res.on('finish'); fields populated here so the row reflects what we
    // know at exhaustion time — provider is the chain[0] primary per ADR 0004
    // step 4, tried_providers is the failed-hop trail from fallbackDetail).
    auditCtx.provider = providerUsed ?? 'none';
    auditCtx.fallback_hops = fallbackHops ?? 0;
    auditCtx.tried_providers = Array.isArray(fallbackDetail)
      ? fallbackDetail.map(t => t?.provider).filter(Boolean)
      : [];
    auditCtx.cache_status = 'miss';
    auditCtx.error_code = originalError?.code ?? 'provider_error';

    // Send error with standard OLP headers + optional exhausted header +
    // D40 X-OLP-Fallback-Detail (when any hop attempted to spawn failed).
    // D40 (issue #7) — ungated v0.1 per maintainer decision; owner-vs-non-owner
    // gating planned for Phase 2 with lib/keys.mjs.
    const payload = JSON.stringify({
      error: {
        message: originalError?.message ?? 'Provider error',
        type: 'provider_error',
      },
    });
    const detailHeader = withFallbackDetailHeader({}, fallbackDetail);
    res.writeHead(errStatus, {
      'Content-Type': 'application/json',
      'Content-Length': Buffer.byteLength(payload),
      ...errorOlpHeaders,
      ...exhaustedHeader,
      ...detailHeader,
    });
    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);
  // F8 (round-5 cold-audit): cacheStatus accounts for fallback-hop cache hits.
  // preCheckHit covers the primary-hop case (fallbackHops===0); lastHopWasCached
  // covers any hop (including fallback hops that served from their own cache).
  // The two flags combine: if either signals a cache hit AND no bypass, report 'hit'.
  const cacheStatus = bypassCacheForServingHop ? 'bypass'
    : (lastHopWasCached || (preCheckHit && fallbackHops === 0)) ? 'hit'
    : 'miss';
  // D40 (issue #7): when at least one prior hop failed before this success,
  // surface X-OLP-Fallback-Detail with the failure trail. Header is omitted
  // when fallbackDetail is empty (single-hop success).
  const headers = withFallbackDetailHeader(
    olpHeaders({ providerUsed, modelUsed, startMs, cacheStatus, fallbackHops }),
    fallbackDetail,
  );

  // Audit ctx capture for success path (audit fires on res.on('finish');
  // status_code + latency_ms populated then).
  auditCtx.provider = providerUsed;
  auditCtx.fallback_hops = fallbackHops;
  auditCtx.tried_providers = (Array.isArray(fallbackDetail) ? fallbackDetail.map(t => t?.provider).filter(Boolean) : []).concat([providerUsed]);
  auditCtx.cache_status = cacheStatus;

  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 await 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`,
    );
  });
}