mirror of
https://github.com/dtzp555-max/olp.git
synced 2026-07-19 09:45:07 +00:00
cold-audit catch from 2026-05-23
Cold-audit Finding 2 (P2 cache correctness). Pre-D13 server.mjs:325
computed `bypassCache` once per request globally and passed it unchanged
into every chain hop. Per ADR 0005 § D2, the bypass must only fire when
the active hop's provider is Anthropic. Pre-D13 Codex/Mistral hops in
fallback chains wrongly bypassed OLP's response cache whenever the
original body had a cache_control marker, defeating per-model cache.
Changes (server.mjs +54/-12, test-features.mjs +218):
1. Replaced the global `bypassCache` const with:
- `hasCacheControlMarkers` (request-level boolean, computed once)
- `shouldBypassCacheForHop(hopProviderName)` (helper, returns
hasCacheControlMarkers && hopProviderName === 'anthropic')
2. Refactored 4 call sites to use the helper:
- executeHopFn: bypassCacheForThisHop = shouldBypassCacheForHop(hopProvider)
— log now includes provider field for observability
- preCheckHit: gated on shouldBypassCacheForHop(chain[0].provider)
— first-hop scope (correct for the pre-loop peek)
- Real-streaming gate (D10): same bypassCacheForFirstHop — single-hop
streaming so first === serving
- cacheStatus header: shouldBypassCacheForHop(providerUsed) — reports
SERVING hop's bypass status. Semantic improvement: a fallback chain
anthropic→openai where openai serves now correctly reports
`X-OLP-Cache: miss` rather than the pre-D13 global `bypass`
No engine.mjs change required — executeWithFallback already passes
hopProvider as a string into executeHopFn.
No marker-strip step needed — verified that openai-to-ir.mjs's
translateMessage drops cache_control from the IR object (copies only
role/content/name/tool_call_id/tool_calls/function_call). Non-Anthropic
plugins never see the markers. Cache key over the IR is therefore
identical for "same prompt with markers" and "same prompt without
markers" on non-Anthropic hops — caching is safe and beneficial.
Tests: 288 → 291 (+3 in new Suite 9d):
- Test 31: openai + cache_control → X-OLP-Cache: miss (the fix's core
assertion)
- Test 32: anthropic + cache_control → X-OLP-Cache: bypass (regression
guard, preserves Anthropic behavior)
- Test 33: 2-hop chain anthropic→openai, anthropic hard-fails, openai
serves → X-OLP-Cache: miss + X-OLP-Provider-Used: openai
(per-hop correctness in fallback)
Authority:
- ADR 0005 § D2: "If the IR request contains Anthropic cache_control
markers AND the active provider in the current chain hop is Anthropic,
the OLP response cache is bypassed... If the active provider is not
Anthropic, the cache_control markers are stripped from the IR before
provider translation"
https://github.com/dtzp555-max/olp/blob/main/docs/adr/0005-cache-cross-provider.md
- ADR 0004 § Observability headers (X-OLP-Cache semantics for the
serving hop)
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.
Key verification: reviewer read openai-to-ir.mjs::translateMessage
end-to-end to verify cache_control markers never propagate into IR
(the load-bearing claim — if markers DID propagate, D13 would have
been incomplete and needed a strip step). Confirmed: markers are
dropped at IR translation, no additional strip needed in D13.
Reviewer also walked 4 scenarios (anthropic-only no markers,
anthropic-only with markers, mixed chain anthropic-fail openai-serves,
mixed chain openai-first) against the post-D13 code; all match the
expected per-hop semantics.
Follow-up tracked separately: ADR 0005 § D2 mentions "logged once per
request at debug level" for non-Anthropic noop case; post-D13 the
log fires only for actual bypass (Anthropic + markers). Reviewer's
non-blocking suggestion to either add the log or amend the ADR will
be tracked as a GitHub issue.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
690 lines
26 KiB
JavaScript
690 lines
26 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,
|
|
});
|
|
|
|
res.writeHead(200, {
|
|
'Content-Type': 'text/event-stream',
|
|
'Cache-Control': 'no-cache',
|
|
Connection: 'keep-alive',
|
|
'X-Accel-Buffering': 'no',
|
|
...streamHeaders,
|
|
});
|
|
|
|
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');
|
|
}
|
|
|
|
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`,
|
|
);
|
|
});
|
|
}
|