mirror of
https://github.com/dtzp555-max/olp.git
synced 2026-07-21 21:15:10 +00:00
Wraps the claude CLI spawn in @anthropic-ai/sandbox-runtime per ADR 0014 § PR-B. Achieves multi-tenant filesystem + network isolation: the spawned claude subprocess can no longer read ~/.olp/keys.json, ~/.claude/.credentials.json, ~/.ssh/, or any other per-client OAuth material. Only api.anthropic.com and statsig.anthropic.com are reachable; only /tmp/olp-spawn/<uuid>/ is writable per spawn (ephemeral, UUID-scoped to prevent cross-request contamination). ## Authority citations - @anthropic-ai/sandbox-runtime v0.0.52 https://github.com/anthropic-experimental/sandbox-runtime dist/sandbox/sandbox-manager.js — SandboxManager.initialize(), wrapWithSandbox() dist/sandbox/sandbox-utils.js — getDefaultWritePaths() - 2026-05-28 spike report at /tmp/sandbox-spike/report.md on PI231: spike-anthropic.mjs (wrapWithSandbox call signature + NDJSON proof), spike-deny.mjs (cat ~/.olp/keys.json MUST fail) - OLP ADR 0014 § 2.1 PR-B, § 4.1 PR-B acceptance criteria - OLP ADR 0009 Amendment 1 § Caveats #3 (sandbox is cloud prerequisite) - OLP ALIGNMENT.md Rule 1 — provider plugin authority citation - cc-mem incident 2026-05-27 § 3 (multi-tenant OAuth token exposure via prompt injection) ## Files changed - lib/sandbox/manager.mjs (new): bootstrap + spawn-wrap layer. Exports: bootstrapSandbox(), isSandboxActive(), wrapSpawn(), __resetSandboxManagerForTests(). Config-at-boot model (one SandboxManager.initialize() at server start; per-request wrapWithSandbox() reads from already-initialized state). Transparent pass-through when inactive. - lib/providers/anthropic.mjs: spawn site wrapped via wrapSpawn({ bin, args, env, allowedDomains: ['api.anthropic.com','statsig.anthropic.com'] }). ADR 0009 Amendment 1 spawn args unchanged — only execution is wrapped. - server.mjs: bootstrapSandbox() called before server.listen(); startup banner logs sandbox.active state. /health.sandbox now includes active:boolean field (distinguishes "deps present" from "SandboxManager initialized and wrapping"). - test-features.mjs: Suite 43 (8 tests — manager unit: bootstrap state, idempotency, isSandboxActive, wrapSpawn passthrough, /health.active field) + Suite 44 (2 PI231-gated tests: 44a security negative test + 44b positive echo test, skipped by default, run with OLP_E2E_SANDBOX=1 npm test). - CHANGELOG.md, docs/adr/0014-sandbox-runtime-integration.md: PR-B status updated; ADR table + /health JSON example updated with active field. ## Test count 805 (pre-PR-B) → 813 (+8 Suite 43; Suite 44 skipped on macOS, runs on PI231) All 813 pass on macOS dev machine. 0 regressions. ## PI231 validation (required before merge — Suite 44) After apt-get install bubblewrap socat (ripgrep already present) + server restart: 1. curl /health → confirm sandbox.available=true AND sandbox.active=true 2. Real anthropic request → confirm stream-json still works end-to-end 3. OLP_E2E_SANDBOX=1 npm test → confirm Suite 44a (cat ~/.olp/keys.json MUST fail) and Suite 44b (echo SANDBOX_PROOF succeeds) Reviewer: must SSH PI231, run Suite 44, and confirm the ADR 0014 § 4.1 criteria. This commit is ready to push; do NOT push before Suite 44 transcript is captured. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2412 lines
108 KiB
JavaScript
2412 lines
108 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: 4567 since D60 / v0.4.0 — moved off 3456
|
||
* so OLP can co-host with OCP for migration windows. ADR 0010 §
|
||
* Default port. Set OLP_PORT=3456 explicitly to restore the
|
||
* pre-D60 default when not co-hosting with OCP.)
|
||
* OLP_BIND — listen address (default: 127.0.0.1 since D76 / v0.4.3). Set
|
||
* to 0.0.0.0 (or a specific interface IP) to accept connections
|
||
* from LAN clients (required for olp-connect <ip> to actually
|
||
* reach the server). Server emits a startup warn if BIND
|
||
* resolves to a non-loopback address AND auth.allow_anonymous
|
||
* is true (anonymous-key over LAN may be acceptable; anonymous-
|
||
* key over public internet is not — see ADR 0011 § Deployment
|
||
* configurations).
|
||
*/
|
||
|
||
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.
|
||
// D69 (ADR 0011): findAdvertisedKey for /health.anonymousKey opt-in surface.
|
||
import {
|
||
validateKey,
|
||
touchLastUsed,
|
||
loadAuthConfigSync,
|
||
findAdvertisedKey,
|
||
ANONYMOUS_KEY_ID,
|
||
ENV_OWNER_KEY_ID,
|
||
} from './lib/keys.mjs';
|
||
import { appendAuditEvent } from './lib/audit.mjs';
|
||
// Phase 7 / PR-A — sandbox availability preflight module (ADR 0014).
|
||
// checkSandboxAvailability is called lazily at first /health hit and memoized
|
||
// process-wide (bwrap/socat install state does not change at runtime; we don't
|
||
// want a child_process.execFileSync per /health call).
|
||
import { checkSandboxAvailability } from './lib/sandbox/doctor.mjs';
|
||
// Phase 7 / PR-B — sandbox manager bootstrap + spawn-wrap (ADR 0014 § PR-B).
|
||
// bootstrapSandbox() is called at server startup (before listen) and sets up
|
||
// the process-wide SandboxManager singleton. isSandboxActive() is used by
|
||
// /health to report sandbox.active.
|
||
import { bootstrapSandbox, isSandboxActive, __resetSandboxManagerForTests } from './lib/sandbox/manager.mjs';
|
||
// Phase 3 / D50 — management endpoints consume the audit aggregate query layer.
|
||
// D81 (Phase 5) — adds aggregateProviderQuota for quota_v2 shape.
|
||
import {
|
||
aggregateRequests as auditAggregateRequests,
|
||
topFallbackChains as auditTopFallbackChains,
|
||
spendTrendDaily as auditSpendTrendDaily,
|
||
cacheHitRateWindow as auditCacheHitRateWindow,
|
||
aggregateProviderQuota as auditAggregateProviderQuota,
|
||
} from './lib/audit-query.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 ?? '4567', 10);
|
||
// F5 / D76: OLP_BIND env. Defaults to 127.0.0.1 (loopback only — secure
|
||
// default). Operators expose LAN by setting OLP_BIND=0.0.0.0 (or a specific
|
||
// interface). Per ADR 0011 § Deployment configurations:
|
||
// - 127.0.0.1: trusted single-machine; safe with any auth posture
|
||
// - RFC1918 / tailnet / specific LAN IP: trusted-LAN — anonymous_key OK
|
||
// - 0.0.0.0: ALL interfaces — operator MUST ensure auth posture matches the
|
||
// network reachability (e.g. no advertise_anonymous_key on public IP)
|
||
const BIND = process.env.OLP_BIND ?? '127.0.0.1';
|
||
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');
|
||
}
|
||
}
|
||
|
||
// ── SSE response headers (D61, ADR 0010 § Phase 4 D61-D63) ────────────────
|
||
// Canonical SSE response-header set used by every streaming response. Pulled
|
||
// out as a top-level constant so all streaming write sites share a single
|
||
// source of truth (per OCP db11105 lesson: drift between branches caused the
|
||
// "headers sent but no Buffering directive" 502 regression).
|
||
//
|
||
// Authority: RFC 8895 (text/event-stream); nginx `X-Accel-Buffering: no`
|
||
// disables the upstream proxy buffer (load-bearing for real-time delivery
|
||
// behind nginx / Cloudflare / Tailscale Funnel).
|
||
const SSE_DEFAULT_HEADERS = Object.freeze({
|
||
'Content-Type': 'text/event-stream',
|
||
'Cache-Control': 'no-cache',
|
||
Connection: 'keep-alive',
|
||
'X-Accel-Buffering': 'no',
|
||
});
|
||
|
||
// ── D62 recentErrors[20] ring buffer (ADR 0010 § Phase 4 D61-D63) ─────────
|
||
// Last 20 server-side error events. Surfaced via /status (owner-only). Ported
|
||
// from OCP server.mjs:301 + 354-358 — same shape adapted to OLP's identity-
|
||
// aware error vocabulary (key_id present when known; never client-tier-only
|
||
// errors like 401/403 which would let brute-force loops fill the ring).
|
||
//
|
||
// Lifecycle: module-scope, in-memory only. NOT persisted across restart
|
||
// (OCP precedent: same; acceptable per the spec).
|
||
//
|
||
// Sanitization: filesystem-path-like tokens in `message` are replaced with
|
||
// `[path]` so an unintentionally logged absolute path doesn't surface in a
|
||
// `/status` payload an owner copy-pastes elsewhere. Port of OCP server.mjs:1395.
|
||
//
|
||
// 200-char message cap: matches OCP precedent; keeps the ring bounded even
|
||
// for verbose stack traces (the dashboard / `olp` CLI surface a one-line
|
||
// preview, not the full trace).
|
||
const RECENT_ERRORS_MAX = 20;
|
||
const recentErrors = [];
|
||
|
||
/**
|
||
* Push a server-side error event onto the recentErrors ring. Filters out
|
||
* client-tier auth rejections (401/403) so a brute-force loop cannot flood
|
||
* the ring. ProviderError instances are always recorded; other errors are
|
||
* recorded when `statusCode >= 500`.
|
||
*
|
||
* @param {object} opts
|
||
* @param {Error|null|undefined} opts.error
|
||
* @param {string|null} [opts.provider]
|
||
* @param {string|null} [opts.path]
|
||
* @param {number|null} [opts.statusCode]
|
||
*/
|
||
function _pushError({ error, provider = null, path = null, statusCode = null }) {
|
||
// D61-D63 reviewer P2-1 (defense-in-depth): explicitly reject 401/403 at the
|
||
// function level even if a caller mistakenly passes one. The current call
|
||
// sites all use the "no auth" path → never invoke _pushError, but a future
|
||
// contributor passing a ProviderError tagged statusCode=401 would otherwise
|
||
// slip past the isProviderError branch and flood the ring under brute force.
|
||
if (statusCode === 401 || statusCode === 403) return;
|
||
// Filter: any other status is recorded when (a) statusCode >= 500 OR (b) the
|
||
// error is a ProviderError (provider-shape failure, always interesting for
|
||
// operators).
|
||
const isProviderError = error?.code === 'PROVIDER_ERROR'
|
||
|| error?.code === 'SPAWN_FAILED'
|
||
|| error?.code === 'CONCURRENCY_LIMIT'
|
||
|| error?.constructor?.name === 'ProviderError';
|
||
const looksServerSide = statusCode !== null && statusCode >= 500;
|
||
if (!isProviderError && !looksServerSide) return;
|
||
|
||
const rawMessage = String(error?.message ?? error ?? 'unknown error');
|
||
// Strip filesystem-path-like tokens. Port of OCP server.mjs:1395 — the
|
||
// regex catches `/foo/bar` plus dotted/dashed variants. Replaces every
|
||
// match with `[path]`. Truncate to 200 chars after sanitization.
|
||
const sanitized = rawMessage
|
||
.replace(/\/[\w./-]+/g, '[path]')
|
||
.slice(0, 200);
|
||
|
||
recentErrors.push({
|
||
time: new Date().toISOString(),
|
||
message: sanitized,
|
||
code: error?.code ?? null,
|
||
provider,
|
||
path,
|
||
status_code: statusCode,
|
||
});
|
||
while (recentErrors.length > RECENT_ERRORS_MAX) recentErrors.shift();
|
||
}
|
||
|
||
/** @internal — test seam: clear the recentErrors ring. */
|
||
export function __clearRecentErrors() {
|
||
recentErrors.length = 0;
|
||
}
|
||
|
||
/** @internal — test seam: snapshot the recentErrors ring (copy). */
|
||
export function __snapshotRecentErrors() {
|
||
return recentErrors.slice();
|
||
}
|
||
|
||
// ── D63 request counters (ADR 0010 § Phase 4 D61-D63) ─────────────────────
|
||
// Module-scope counters surfaced via /status (owner-only). _totalRequests is
|
||
// the cumulative count of /v1/chat/completions invocations since process
|
||
// start; _activeRequests is the live concurrent count.
|
||
//
|
||
// NOT exposed via /health (already owner-trimmed per Phase 2 D46); only
|
||
// /status (owner-only_block) reads them. Ported from OCP `stats.totalRequests`
|
||
// + `stats.activeRequests` (server.mjs:288-300).
|
||
let _totalRequests = 0;
|
||
let _activeRequests = 0;
|
||
const _serverStartMs = Date.now();
|
||
|
||
/** @internal — test seam: reset request counters. */
|
||
export function __resetRequestCounters() {
|
||
_totalRequests = 0;
|
||
_activeRequests = 0;
|
||
}
|
||
|
||
// ── Phase 7 PR-A: sandbox availability cache ──────────────────────────────
|
||
// checkSandboxAvailability() forks `which bwrap` / `which socat` / `which rg`
|
||
// and imports @anthropic-ai/sandbox-runtime. Neither can change at runtime —
|
||
// bwrap is either installed or it isn't. Memoize the first result to avoid
|
||
// repeated child_process.execFileSync calls on every /health hit.
|
||
//
|
||
// _sandboxStatusCache: null → not yet fetched
|
||
// object → memoized result from checkSandboxAvailability()
|
||
let _sandboxStatusCache = null;
|
||
/** @internal — test seam: reset sandbox cache between tests. */
|
||
export function __resetSandboxStatusCache() {
|
||
_sandboxStatusCache = null;
|
||
}
|
||
|
||
// ── 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).',
|
||
});
|
||
}
|
||
// D69 (ADR 0011): startup-time prerequisite validation for /health.anonymousKey
|
||
// advertisement. The field is emitted ONLY when all three prerequisites hold:
|
||
// (1) auth.advertise_anonymous_key === true (config opt-in)
|
||
// (2) auth.allow_anonymous === true (anonymous tier must be reachable
|
||
// for the advertised key to be
|
||
// meaningful to zero-config callers)
|
||
// (3) at least one advertised key exists on disk (manifest with
|
||
// plaintext_advertise field, not revoked)
|
||
// Violations log warn at startup but the server still boots; the runtime
|
||
// handleHealth checks the same conditions and simply omits the field.
|
||
if (_authConfig.advertise_anonymous_key === true) {
|
||
if (_authConfig.allow_anonymous !== true) {
|
||
logEvent('warn', 'anonymous_key_advertised_but_denied', {
|
||
message: 'auth.advertise_anonymous_key=true but auth.allow_anonymous=false — the advertised key would not be usable anonymously. /health.anonymousKey will NOT be emitted until allow_anonymous=true. See ADR 0011.',
|
||
});
|
||
} else if (findAdvertisedKey() === null) {
|
||
logEvent('warn', 'anonymous_key_advertised_but_no_anonymous_key_exists', {
|
||
message: 'auth.advertise_anonymous_key=true but no active key with plaintext_advertise exists. Run `olp-keys keygen --anonymous --advertise` to create one. /health.anonymousKey will NOT be emitted until then. See ADR 0011.',
|
||
});
|
||
}
|
||
// F5 / D76 (ADR 0011 Deployment configurations): publishing the anonymous
|
||
// key via /health is only safe when /health is reachable ONLY from a
|
||
// trusted network. If OLP_BIND is set to a non-loopback address AND
|
||
// advertise_anonymous_key is true, warn the operator. We can't tell from
|
||
// here whether the non-loopback bind is "trusted LAN" (RFC1918 / tailnet)
|
||
// or "public internet" — that's the operator's responsibility. The warn is
|
||
// a checkpoint, not a hard gate.
|
||
if (BIND !== '127.0.0.1' && BIND !== 'localhost' && BIND !== '::1') {
|
||
logEvent('warn', 'anonymous_key_advertised_with_lan_bind', {
|
||
message: `auth.advertise_anonymous_key=true with OLP_BIND=${BIND} — /health.anonymousKey will be reachable from any host that can connect to ${BIND}:${PORT}. Confirm this address is on a trusted LAN (RFC1918 / tailnet) — never a public IP. See ADR 0011 § Deployment configurations.`,
|
||
bind: BIND,
|
||
});
|
||
}
|
||
}
|
||
|
||
/** @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', advertise_anonymous_key: false };
|
||
// D69: defensively normalize so tests passing partial configs still get a
|
||
// boolean advertise_anonymous_key value (downstream code treats undefined
|
||
// as falsy but explicit normalisation matches loadAuthConfigSync's contract).
|
||
if (typeof _authConfig.advertise_anonymous_key !== 'boolean') {
|
||
_authConfig.advertise_anonymous_key = false;
|
||
}
|
||
}
|
||
|
||
/** @internal — reset auth config to file-loaded state. */
|
||
export function __resetAuthConfig() {
|
||
_authConfig = loadAuthConfigSync();
|
||
}
|
||
|
||
// ── Streaming config (D61, ADR 0010 § Phase 4 D61-D63) ────────────────────
|
||
// Reads `streaming.heartbeat_interval_ms` from ~/.olp/config.json via the
|
||
// same loader used by routing/providers config. Default 0 = disabled;
|
||
// behaviour is opt-in. Tests inject synthetic configs via __setStreamingConfig.
|
||
let _streamingConfig = _startupConfig.streaming ?? { heartbeat_interval_ms: 0 };
|
||
|
||
/** @internal — test seam: inject a synthetic streaming config (no file I/O). */
|
||
export function __setStreamingConfig(config) {
|
||
_streamingConfig = config ?? { heartbeat_interval_ms: 0 };
|
||
}
|
||
|
||
/** @internal — reset streaming config to file-loaded state. */
|
||
export function __resetStreamingConfig() {
|
||
_streamingConfig = loadFallbackConfigSync().streaming ?? { heartbeat_interval_ms: 0 };
|
||
}
|
||
|
||
// ── SSE heartbeat (D61, ADR 0010 § Phase 4 D61-D63) ───────────────────────
|
||
// Port of OCP `startHeartbeat` (ocp/server.mjs:660-685). Emits `: keepalive\n\n`
|
||
// SSE comment frames on `res` every `intervalMs` ms during silent windows.
|
||
// The timer is reset on each real chunk write (`reset()`); cancelled on
|
||
// stream end / error / client disconnect (`stop()`).
|
||
//
|
||
// Discipline (load-bearing): this is a downstream liveness hint only — it
|
||
// MUST NOT abort or time out a request. The OCP timeout-tier regressions
|
||
// (v2.2-v2.5; see OCP spec) showed that any auto-cancellation primitive in
|
||
// the streaming branch is hostile to long-reasoning prompts.
|
||
//
|
||
// Per-attached-client: each call to startHeartbeat creates its own timer
|
||
// state. The streaming-singleflight `attached` client and the `source`
|
||
// client each get their own heartbeat instance (they share an underlying
|
||
// spawn but write to different `res` objects). The `res.on('close')` handler
|
||
// in the streaming branch calls `stop()` to cancel each client's timer
|
||
// independently.
|
||
//
|
||
// @param {import('node:http').ServerResponse} res
|
||
// @param {number} intervalMs — 0 or negative ⇒ no-op (timer never starts)
|
||
// @param {string} requestId — for the one-time `heartbeat_active` log event
|
||
// @returns {{ reset: () => void, stop: () => void }}
|
||
function startHeartbeat(res, intervalMs, requestId) {
|
||
if (!intervalMs || intervalMs <= 0) {
|
||
return { reset: () => {}, stop: () => {} };
|
||
}
|
||
let handle = null;
|
||
let hasFired = false;
|
||
const onFire = () => {
|
||
if (res.writableEnded || res.destroyed) return;
|
||
try {
|
||
res.write(': keepalive\n\n');
|
||
} catch {
|
||
// res may have been destroyed between the writableEnded check and the
|
||
// write — swallow so the heartbeat is a no-op on a dead socket.
|
||
return;
|
||
}
|
||
if (!hasFired) {
|
||
hasFired = true;
|
||
logEvent('info', 'heartbeat_active', { interval_ms: intervalMs, request_id: requestId });
|
||
}
|
||
handle = setTimeout(onFire, intervalMs);
|
||
};
|
||
handle = setTimeout(onFire, intervalMs);
|
||
return {
|
||
reset: () => {
|
||
if (handle) { clearTimeout(handle); handle = setTimeout(onFire, intervalMs); }
|
||
},
|
||
stop: () => {
|
||
if (handle) { clearTimeout(handle); handle = null; }
|
||
},
|
||
};
|
||
}
|
||
|
||
// ── 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');
|
||
});
|
||
}
|
||
|
||
/**
|
||
* Identity-aware gate for X-OLP-Fallback-Detail emission per ADR 0007 § 7.2.
|
||
* Reads `_authConfig.fallback_detail_header_policy`:
|
||
* - 'owner_only' (default) → emit only when olpIdentity.owner_tier === 'owner'
|
||
* - 'all' → emit unconditionally (v0.1.1 behaviour, opt-in)
|
||
* - 'none' → suppress unconditionally
|
||
* When olpIdentity is null (early-error paths before auth completed),
|
||
* defaults to 'all' → emit (preserves the v0.1.1 ungated behaviour for
|
||
* pre-auth errors where we don't yet know identity).
|
||
*
|
||
* @param {{owner_tier?: string}|null|undefined} olpIdentity
|
||
* @returns {boolean}
|
||
*/
|
||
function shouldEmitFallbackDetailHeader(olpIdentity) {
|
||
const policy = _authConfig?.fallback_detail_header_policy ?? 'owner_only';
|
||
if (policy === 'none') return false;
|
||
if (policy === 'all') return true;
|
||
// 'owner_only' — gate by identity tier
|
||
if (!olpIdentity) return true; // pre-auth path: don't suppress diagnostic info
|
||
return olpIdentity.owner_tier === 'owner';
|
||
}
|
||
|
||
/**
|
||
* Merges X-OLP-Fallback-Detail into a base header object when the per-hop
|
||
* failure tuples are non-empty AND the per-request identity is permitted
|
||
* to see the header per the policy (ADR 0007 § 7.2). Returns the base
|
||
* object unchanged otherwise.
|
||
*
|
||
* D40 (issue #7) — gating added at D46 per ADR 0004 Amendment 5 ratification.
|
||
*
|
||
* @param {Record<string,string>} baseHeaders
|
||
* @param {Array<object>|null|undefined} fallbackDetail
|
||
* @param {{owner_tier?: string}|null|undefined} olpIdentity - request identity; null on pre-auth paths
|
||
* @returns {Record<string,string>}
|
||
*/
|
||
function withFallbackDetailHeader(baseHeaders, fallbackDetail, olpIdentity) {
|
||
if (!shouldEmitFallbackDetailHeader(olpIdentity)) return baseHeaders;
|
||
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
|
||
*
|
||
* Phase 2 / D46 (ADR 0007 § 7.1): identity-aware payload.
|
||
* - owner identity → full per-provider details (existing payload)
|
||
* - guest / anonymous → trimmed { ok, version } only
|
||
* - no auth, allow_anonymous=false → 401 (consistent with /v1/* routes)
|
||
*
|
||
* The trimming behavior is gated on `_authConfig.owner_only_endpoints` —
|
||
* the entry `/health` lives there by default. Removing `/health` from
|
||
* the list reverts to v0.1.1 full-payload-to-everyone behaviour.
|
||
*
|
||
* Authority: ADR 0007 § 7.1 (Identity-class table) + § 7.2 (owner_only_endpoints).
|
||
* Closes acceptance criterion #4.
|
||
*/
|
||
async function handleHealth(req, res) {
|
||
const startMs = Date.now();
|
||
|
||
// D46: audit on /health is intentionally NOT enabled at Phase 2.
|
||
// /health is a high-volume monitoring endpoint; per-call audit rows would
|
||
// generate operational noise that has no observability value until a
|
||
// Phase 3+ Dashboard aggregates /health stats. Deferred to Phase 3.
|
||
|
||
const authResult = authenticate(req);
|
||
if (!authResult.ok) {
|
||
return sendError(res, authResult.status, authResult.message, authResult.code);
|
||
}
|
||
const olpIdentity = authResult.authContext;
|
||
|
||
// Per § 7.1: owner sees full payload; guest + anonymous see trimmed.
|
||
// Per § 7.2: gating is opt-out via `owner_only_endpoints` config; if
|
||
// `/health` is removed from the list, all identities see the full
|
||
// payload (operators wanting v0.1.1 behaviour have this knob).
|
||
const gatedEndpoints = Array.isArray(_authConfig?.owner_only_endpoints)
|
||
? _authConfig.owner_only_endpoints
|
||
: ['/health'];
|
||
const isGated = gatedEndpoints.includes('/health');
|
||
const isOwner = olpIdentity.owner_tier === 'owner';
|
||
|
||
// Touch last_used_at for filesystem identities post-response. The callee
|
||
// also early-returns on ANONYMOUS / ENV_OWNER keyIds (lib/keys.mjs § 6.3
|
||
// wrapper) — this guard is defense-in-depth + skip the async call entirely
|
||
// for non-filesystem identities.
|
||
if (olpIdentity.keyId !== ANONYMOUS_KEY_ID && olpIdentity.keyId !== ENV_OWNER_KEY_ID) {
|
||
res.on('finish', () => {
|
||
touchLastUsed(olpIdentity.keyId).catch(() => {});
|
||
});
|
||
}
|
||
|
||
// D69 (ADR 0011): compute opt-in /health.anonymousKey. The field is added
|
||
// to BOTH trimmed and full payloads (the trimmed payload's whole purpose
|
||
// is to be readable by anonymous clients so they can self-bootstrap into
|
||
// a real key). Three prerequisites must hold (see startup warns above):
|
||
// (1) auth.advertise_anonymous_key === true
|
||
// (2) auth.allow_anonymous === true
|
||
// (3) findAdvertisedKey() returns a non-null active manifest
|
||
// When any fails, the field is absent (NOT null) — preserves the v0.3.x
|
||
// /health payload shape for clients that strict-validate keys.
|
||
let anonymousKey = null;
|
||
if (_authConfig.advertise_anonymous_key === true && _authConfig.allow_anonymous === true) {
|
||
try {
|
||
const adv = findAdvertisedKey();
|
||
if (adv && typeof adv.plaintext_advertise === 'string' && adv.plaintext_advertise.length > 0) {
|
||
anonymousKey = adv.plaintext_advertise;
|
||
}
|
||
} catch {
|
||
// Filesystem error reading manifests is non-fatal; just omit the field.
|
||
anonymousKey = null;
|
||
}
|
||
}
|
||
|
||
if (isGated && !isOwner) {
|
||
// Trimmed payload per § 7.1. D69: include anonymousKey here too — zero-
|
||
// config olp-connect callers read it BEFORE they have a key.
|
||
const trimmed = { ok: true, version: VERSION };
|
||
if (anonymousKey !== null) trimmed.anonymousKey = anonymousKey;
|
||
return sendJSON(res, 200, trimmed);
|
||
}
|
||
|
||
// Full payload (owner OR /health removed from owner_only_endpoints).
|
||
const enabled = loadedProviders.size;
|
||
const available = listAllProviderNames().length;
|
||
const providerStatuses = {};
|
||
for (const [name, provider] of loadedProviders) {
|
||
// D56 / v1.x roadmap #4 (ADR 0002 Amendment 6 forward note): surface
|
||
// per-provider active spawn count for capacity-planning observability.
|
||
// D38 shipped getActiveSpawnCount; this is the /health integration.
|
||
// The field is set BEFORE healthCheck() in case healthCheck throws —
|
||
// activeSpawns is cheap (in-memory counter read) and useful even
|
||
// when healthCheck fails.
|
||
const activeSpawns = getActiveSpawnCount(name);
|
||
try {
|
||
providerStatuses[name] = { ...(await provider.healthCheck()), activeSpawns };
|
||
} catch (e) {
|
||
providerStatuses[name] = { ok: false, error: e.message, activeSpawns };
|
||
}
|
||
}
|
||
// Phase 7 PR-A (ADR 0014): sandbox availability field.
|
||
// Result is memoized process-wide in _sandboxStatusCache — bwrap/socat
|
||
// install state does not change at runtime. If the library call throws for
|
||
// any reason, the field is still included with available: false + error
|
||
// (don't crash /health).
|
||
if (_sandboxStatusCache === null) {
|
||
try {
|
||
_sandboxStatusCache = await checkSandboxAvailability();
|
||
} catch (e) {
|
||
_sandboxStatusCache = {
|
||
available: false,
|
||
missing: [],
|
||
details: { platform: process.platform, error: String(e?.message ?? e) },
|
||
};
|
||
}
|
||
}
|
||
const sandboxField = {
|
||
available: _sandboxStatusCache.available,
|
||
// Phase 7 PR-B: active = sandbox was bootstrapped and SandboxManager is
|
||
// ready to wrap spawns. available=true + active=true means every provider
|
||
// spawn is actually sandboxed. available=true + active=false means deps
|
||
// present but bootstrap failed at runtime (see server startup log).
|
||
active: isSandboxActive(),
|
||
missing: _sandboxStatusCache.missing ?? [],
|
||
platform: _sandboxStatusCache.details?.platform ?? process.platform,
|
||
};
|
||
if (!_sandboxStatusCache.available) {
|
||
// Include human-readable install hint for owner-tier callers.
|
||
const missingDeps = (_sandboxStatusCache.missing ?? []).filter(
|
||
m => m === 'bubblewrap' || m === 'socat' || m === 'ripgrep',
|
||
);
|
||
if (missingDeps.length > 0) {
|
||
sandboxField.message =
|
||
`Sandbox dependencies not available: ${missingDeps.map(m => `${m} not installed`).join(', ')}.` +
|
||
` Install: sudo apt-get install -y ${missingDeps.join(' ')}`;
|
||
} else if (_sandboxStatusCache.details?.error) {
|
||
sandboxField.message = `Sandbox check error: ${_sandboxStatusCache.details.error}`;
|
||
} else if (_sandboxStatusCache.details?.libError) {
|
||
sandboxField.message = `Sandbox library error: ${_sandboxStatusCache.details.libError}`;
|
||
}
|
||
}
|
||
|
||
const fullPayload = {
|
||
ok: true,
|
||
version: VERSION,
|
||
providers: { enabled, available, status: providerStatuses },
|
||
sandbox: sandboxField,
|
||
};
|
||
if (anonymousKey !== null) fullPayload.anonymousKey = anonymousKey;
|
||
sendJSON(res, 200, fullPayload);
|
||
}
|
||
|
||
/**
|
||
* 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();
|
||
|
||
// D63 (ADR 0010 § Phase 4 D61-D63): request counters for /status. Increment
|
||
// at the top (every request counts, even authentication failures); decrement
|
||
// in a single `res.on('close')` listener that fires on both normal end and
|
||
// client disconnect, ensuring _activeRequests cannot leak. `res.on('finish')`
|
||
// alone would miss client-disconnect-before-finish cases.
|
||
_totalRequests++;
|
||
_activeRequests++;
|
||
let _activeDecremented = false;
|
||
const _decrementActive = () => {
|
||
if (_activeDecremented) return;
|
||
_activeDecremented = true;
|
||
_activeRequests = Math.max(0, _activeRequests - 1);
|
||
};
|
||
res.on('close', _decrementActive);
|
||
res.on('finish', _decrementActive);
|
||
|
||
// 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';
|
||
// D53 (D45 P2 deferral fix): tried_providers semantic per ADR 0007 § 8 is
|
||
// "providers the server actually dispatched a spawn against". On 403 the
|
||
// filtered chain is empty — no provider was dispatched, so the audit
|
||
// value is []. The configured-but-blocked chain providers go into the
|
||
// human-readable error message + a separate _diagnostic field we don't
|
||
// surface in audit (would distort downstream queries like "which
|
||
// providers did key X actually call"). ADR 0007 § 8 amended in this
|
||
// D-day's CHANGELOG entry to spell the semantic.
|
||
auditCtx.tried_providers = [];
|
||
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 = [];
|
||
// D75 (v0.4.2) F7 — per-hop model override.
|
||
// The chain config (routing.chains[<requested-model>][<hop>].model) is the
|
||
// model name to pass to THIS hop's provider plugin — NOT the model the
|
||
// user originally requested. Pre-D75, executeHopFn used hopModel for the
|
||
// cache key + audit ctx but passed the ORIGINAL irReq (with irReq.model
|
||
// still set to the user's request) into hopProviderPlugin.spawn(). Result:
|
||
// a 2-hop chain like [{anthropic, claude-sonnet-4-6}, {openai, gpt-5.5}]
|
||
// would spawn codex with --model claude-sonnet-4-6 on hop 1 — openai rejects
|
||
// the unknown model and the chain dies. This broke the core OLP value prop
|
||
// (cross-provider fallback with provider-appropriate model substitution).
|
||
//
|
||
// Fix: build a per-hop IR variant with the hop's model substituted. Skip
|
||
// the clone when hopModel === irReq.model (single-provider chains and
|
||
// chain hops whose model matches the request). Authority: ADR 0004 §
|
||
// Chain advancement step 1 (per-hop config supplies provider AND model).
|
||
const hopIrReq = irReq.model === hopModel ? irReq : { ...irReq, model: hopModel };
|
||
try {
|
||
try {
|
||
for await (const irChunk of hopProviderPlugin.spawn(hopIrReq, 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: chunks are written to res, and the cache layer (via
|
||
// getOrComputeStreaming) writes accumulated chunks on source completion.
|
||
//
|
||
// D58 — ADR 0005 Amendment 8 §§7, 11, 12: streaming singleflight via
|
||
// cacheStore.getOrComputeStreaming(...). The peek+spawn TOCTOU window
|
||
// (server.mjs:1104 peek vs. spawn) is collapsed by the synchronous Map
|
||
// check+insert in the cache layer (D57). The D38 tryAcquireSpawn gate now
|
||
// lives INSIDE the sourceFactory closure: only the first caller acquires a
|
||
// slot; attached joiners share it. On factory-throw CONCURRENCY_LIMIT, the
|
||
// cache layer rejects → we fall through to the buffered path (same outcome
|
||
// as today). preCheckHit gates entry exactly as before — the cache_hit
|
||
// outcome from getOrComputeStreaming is the TTL-race safety net; in
|
||
// practice it should not fire because preCheckHit excludes alive entries.
|
||
// See docs/adr/0005-cache-cross-provider.md Amendment 8 + issue #16.
|
||
if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop) {
|
||
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).
|
||
auditCtx.provider = streamProvider;
|
||
auditCtx.error_code = 'no_enabled_provider';
|
||
return sendError(res, 503, `Provider ${streamProvider} is not enabled`, 'no_enabled_provider',
|
||
olpErrorHeaders({ startMs, model: ir.model }));
|
||
}
|
||
|
||
// D58 — ADR 0005 Amendment 8 §7: sourceFactory wraps acquire + spawn.
|
||
// Acquire fires ONLY for the first caller; attached clients share the
|
||
// slot. Release fires EXACTLY ONCE when the source iterator's
|
||
// try/finally executes (normal exhaustion, error, or iterator.return()
|
||
// propagated from sourceAbortController via the cache layer).
|
||
const candidateMax = streamPlugin.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS;
|
||
const sourceFactory = () => {
|
||
if (!tryAcquireSpawn(streamProvider, candidateMax)) {
|
||
const err = new ProviderError(
|
||
`provider ${streamProvider} at maxConcurrent (${candidateMax})`,
|
||
'CONCURRENCY_LIMIT',
|
||
);
|
||
err.providerName = streamProvider;
|
||
err.maxConcurrent = candidateMax;
|
||
err.activeSpawns = getActiveSpawnCount(streamProvider);
|
||
throw err;
|
||
}
|
||
// Wrap streamPlugin.spawn in an async generator that guarantees
|
||
// releaseSpawn fires exactly once in finally — regardless of normal
|
||
// exhaustion, mid-stream throw, or iterator.return() from cache-layer
|
||
// sourceAbortController propagation (§9).
|
||
//
|
||
// D75 (v0.4.2) F7 — per-hop model override (streaming path).
|
||
// Mirror the buffered-path fix: pass the hop's configured model (streamModel)
|
||
// into streamPlugin.spawn(), not the user's original ir.model. Skip the
|
||
// clone when ir.model === streamModel. See executeHopFn() above for the
|
||
// full F7 rationale + authority citation.
|
||
const streamIr = ir.model === streamModel ? ir : { ...ir, model: streamModel };
|
||
return (async function* sourceWithRelease() {
|
||
try {
|
||
for await (const irChunk of streamPlugin.spawn(streamIr, authContext)) {
|
||
yield irChunk;
|
||
}
|
||
} finally {
|
||
releaseSpawn(streamProvider);
|
||
}
|
||
})();
|
||
};
|
||
|
||
// D58 — invoke the cache-layer coordinator. CONCURRENCY_LIMIT from the
|
||
// factory falls through to the buffered path (the buffered fallback
|
||
// engine re-attempts acquire). Any other pre-stream error surfaces 502.
|
||
let stream;
|
||
let role;
|
||
try {
|
||
const result = await cacheStore.getOrComputeStreaming(
|
||
keyId,
|
||
streamCacheKey,
|
||
sourceFactory,
|
||
{ clientId: requestId },
|
||
);
|
||
stream = result.stream;
|
||
role = result.role;
|
||
} catch (e) {
|
||
if (e instanceof ProviderError && e.code === 'CONCURRENCY_LIMIT') {
|
||
// Fall through to the buffered path — the existing executeHopFn /
|
||
// executeWithFallback machinery re-attempts acquire and surfaces a
|
||
// chain-exhausted 502 for a single-hop chain at maxConcurrent, which
|
||
// matches today's pre-D58 behaviour for this saturation scenario.
|
||
logEvent('debug', 'streaming_concurrency_limit_fallthrough', {
|
||
provider: streamProvider,
|
||
model: streamModel,
|
||
});
|
||
// (proceed past this block)
|
||
} else {
|
||
// Any other pre-first-chunk error — surface a clean 502.
|
||
auditCtx.provider = streamProvider;
|
||
auditCtx.tried_providers = [streamProvider];
|
||
auditCtx.cache_status = 'miss';
|
||
auditCtx.error_code = e?.code ?? 'provider_error';
|
||
logEvent('error', 'streaming_error_before_first_chunk', {
|
||
provider: streamProvider,
|
||
model: streamModel,
|
||
error: e.message,
|
||
});
|
||
_pushError({
|
||
error: e,
|
||
provider: streamProvider,
|
||
path: '/v1/chat/completions',
|
||
statusCode: 502,
|
||
});
|
||
return sendError(res, 502, e.message ?? 'Provider error', 'provider_error',
|
||
olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 }));
|
||
}
|
||
}
|
||
|
||
if (stream) {
|
||
// D58 — ADR 0005 Amendment 8 §11: cache_status reflects role-of-this-
|
||
// client at attach-time. `source` did real work (miss); `attached`
|
||
// shared a live spawn (streaming_attached); `cache_hit` served from
|
||
// cache (hit). See lib/audit.mjs cache_status JSDoc for the enum.
|
||
auditCtx.provider = streamProvider;
|
||
auditCtx.tried_providers = [streamProvider];
|
||
if (role === 'cache_hit') {
|
||
auditCtx.cache_status = 'hit';
|
||
} else if (role === 'attached') {
|
||
auditCtx.cache_status = 'streaming_attached';
|
||
} else {
|
||
auditCtx.cache_status = 'miss';
|
||
}
|
||
|
||
// D58 — ADR 0005 Amendment 8 §9: wire HTTP req close → iterator
|
||
// return() so a client disconnect propagates into the cache layer.
|
||
// Without this the cache layer never learns of disconnect and the
|
||
// source spawn may orphan (last-client-gone abort would not fire).
|
||
// try/catch guards against concurrent return() throws (Node async
|
||
// generators can throw if return() is invoked while iteration is
|
||
// pending). The cache layer's iterator return() is idempotent.
|
||
// D58 — ADR 0005 Amendment 8 §9: detect client disconnect via res
|
||
// 'close' event (not req 'close'). For SSE responses Node emits
|
||
// res.on('close') when the underlying socket goes away while the
|
||
// response is still streaming. req.on('close') in Node 18+ fires only
|
||
// after the response is fully sent, which is too late for our abort
|
||
// propagation needs. Without this wire-up the cache layer would never
|
||
// learn of disconnect → potential orphan spawns.
|
||
// Forward declaration of the heartbeat handle so the close-listener
|
||
// installed here can stop the timer when the client disconnects.
|
||
// (The handle itself is assigned a few lines below — startHeartbeat
|
||
// is invoked after eager-headers run, but the close listener must
|
||
// already be wired so a disconnect during the source-factory await
|
||
// path still propagates iterator return + heartbeat stop.)
|
||
let heartbeatRef = null;
|
||
const onClose = () => {
|
||
try { stream.return?.(); } catch { /* best-effort */ }
|
||
try { heartbeatRef?.stop?.(); } catch { /* best-effort */ }
|
||
};
|
||
res.on('close', onClose);
|
||
|
||
// D58 — ADR 0005 Amendment 8 §11: X-OLP-Streaming-Inflight header.
|
||
// role === 'source' → 'source' (first caller; may upgrade to
|
||
// 'solo' post-stream if no joiners ever
|
||
// attached — see deferral note below).
|
||
// role === 'attached' → 'attached'
|
||
// role === 'cache_hit'→ omitted; the existing X-OLP-Cache: hit
|
||
// already signals that path.
|
||
// 'solo' (no joiners ever attached) is observable only post-stream —
|
||
// emitted via streaming_inflight_source_done log event with
|
||
// attached_count=0; the wire header reports `source` initially and
|
||
// does NOT flip mid-stream. Future ADR may expose it on a trailer.
|
||
const cacheStatusForHeader = (role === 'cache_hit') ? 'hit' : 'miss';
|
||
const streamHeaders = olpHeaders({
|
||
providerUsed: streamProvider,
|
||
modelUsed: streamModel,
|
||
startMs,
|
||
cacheStatus: cacheStatusForHeader,
|
||
fallbackHops: 0,
|
||
});
|
||
if (role === 'source' || role === 'attached') {
|
||
streamHeaders['X-OLP-Streaming-Inflight'] = role;
|
||
}
|
||
|
||
// 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 was the D14
|
||
// defect.
|
||
//
|
||
// D61 (ADR 0010 § Phase 4 D61-D63): when streaming heartbeat is
|
||
// enabled (streaming.heartbeat_interval_ms > 0), headers must be
|
||
// flushed BEFORE the first chunk so the `: keepalive\n\n` SSE comment
|
||
// frame has somewhere to land. Without eager-headers the heartbeat
|
||
// would write into a buffer that the client never sees (OCP db11105
|
||
// lesson — the bug was that `ensureHeaders()` returned `false` after
|
||
// headers were sent for the "connection-dead" case, which made all
|
||
// post-headers chunks no-ops). We send eager-headers only when
|
||
// heartbeat is enabled — preserving D14's lazy-headers default so
|
||
// pre-first-chunk errors still surface as JSON 502s when heartbeat
|
||
// is off (the legacy / current path).
|
||
const heartbeatIntervalMs = _streamingConfig.heartbeat_interval_ms ?? 0;
|
||
const eagerHeaders = heartbeatIntervalMs > 0;
|
||
if (eagerHeaders && !res.headersSent) {
|
||
res.writeHead(200, {
|
||
...SSE_DEFAULT_HEADERS,
|
||
...streamHeaders,
|
||
});
|
||
}
|
||
|
||
// Heartbeat instance is per-attached-client (each call to
|
||
// startHeartbeat creates its own timer state; source and attached
|
||
// clients write to different `res` objects and therefore each get
|
||
// their own timer). Heartbeat fires only AFTER headers are sent
|
||
// (eager-headers gate above); when intervalMs is 0 the returned
|
||
// object is a no-op stub so `reset()` / `stop()` calls are cheap.
|
||
const heartbeat = startHeartbeat(res, heartbeatIntervalMs, requestId);
|
||
// Expose to the onClose listener so a client disconnect cancels the
|
||
// timer immediately rather than waiting for the finally block.
|
||
heartbeatRef = heartbeat;
|
||
|
||
const streamedChunks = [];
|
||
let firstChunkEmitted = false;
|
||
try {
|
||
for await (const irChunk of stream) {
|
||
if (irChunk.type === 'error') {
|
||
// Error chunk from provider
|
||
if (firstChunkEmitted) {
|
||
// Past first-chunk boundary — emit truncation marker + [DONE].
|
||
logEvent('warn', 'streaming_error_after_first_chunk', {
|
||
provider: streamProvider,
|
||
model: streamModel,
|
||
error: irChunk.error,
|
||
});
|
||
auditCtx.error_code = 'streaming_error_after_first_chunk';
|
||
_pushError({
|
||
error: { message: irChunk.error ?? 'streaming_error_after_first_chunk', code: 'SPAWN_FAILED' },
|
||
provider: streamProvider,
|
||
path: '/v1/chat/completions',
|
||
statusCode: null, // headers already sent; status_code intentionally null — record by error code only (D61-D63 reviewer P2-2: explicit intent over numeric-but-meaningless 200)
|
||
});
|
||
res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model));
|
||
res.write(SSE_DONE);
|
||
res.end();
|
||
return;
|
||
}
|
||
throw new ProviderError(irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED');
|
||
}
|
||
|
||
if (!res.headersSent) {
|
||
res.writeHead(200, {
|
||
...SSE_DEFAULT_HEADERS,
|
||
...streamHeaders,
|
||
});
|
||
}
|
||
streamedChunks.push(irChunk);
|
||
res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model));
|
||
// D61: reset the heartbeat timer on every real chunk so the next
|
||
// keepalive frame fires only after another silent window of
|
||
// intervalMs ms (rather than rapidly chasing the data stream).
|
||
heartbeat.reset();
|
||
firstChunkEmitted = true;
|
||
|
||
if (irChunk.type === 'stop') {
|
||
res.write(SSE_DONE);
|
||
res.end();
|
||
// D58 — ADR 0005 Amendment 8 §4: cache write is performed by the
|
||
// cache layer's tee task on source completion. Server no longer
|
||
// calls cacheStore.set from this branch; the cache layer applies
|
||
// the same write conditions (truncated-not-cached, cacheable:false
|
||
// opt-out, replay-cap, maxEntryBytes cap) internally.
|
||
return;
|
||
}
|
||
}
|
||
|
||
// Generator exhausted without a stop chunk — emit truncation marker
|
||
// + [DONE]. D58 — the cache layer (per ADR 0005 Amendment 8 §4) writes
|
||
// accumulatedChunks on normal source exhaustion regardless of stop-
|
||
// chunk semantics; IR-level stop-chunk inspection is the server's
|
||
// responsibility. Mirror D16 truncation-eviction: the source role
|
||
// explicitly deletes the just-written entry so subsequent identical
|
||
// requests respawn (matches D16 buffered-path truncation behaviour
|
||
// and ADR 0005 § "Cache write conditions" item 1). Only the source
|
||
// role performs the delete; attached clients did not trigger the
|
||
// write. Idempotent across concurrent source-truncation observers.
|
||
if (streamedChunks.length > 0) {
|
||
const truncMarker = { type: 'stop', finish_reason: 'length' };
|
||
res.write(irChunkToOpenAISSE(truncMarker, requestId, ir.model));
|
||
}
|
||
if (role === 'source' && cacheableForFirstHop) {
|
||
cacheStore.delete(keyId, streamCacheKey);
|
||
}
|
||
|
||
// D35 #9: zero-chunk empty-stream — writeHead deferred (no chunks
|
||
// yielded), emit headers + [DONE] so the response is still valid SSE.
|
||
if (!res.headersSent) {
|
||
res.writeHead(200, {
|
||
...SSE_DEFAULT_HEADERS,
|
||
...streamHeaders,
|
||
});
|
||
}
|
||
|
||
res.write(SSE_DONE);
|
||
res.end();
|
||
if (streamedChunks.length > 0 && cacheableForFirstHop) {
|
||
logEvent('warn', 'streaming_no_stop_chunk', {
|
||
chunks_count: streamedChunks.length,
|
||
provider: streamProvider,
|
||
model: streamModel,
|
||
});
|
||
}
|
||
} catch (e) {
|
||
if (firstChunkEmitted) {
|
||
logEvent('warn', 'streaming_error_after_first_chunk', {
|
||
provider: streamProvider,
|
||
model: streamModel,
|
||
error: e.message,
|
||
});
|
||
_pushError({
|
||
error: e,
|
||
provider: streamProvider,
|
||
path: '/v1/chat/completions',
|
||
statusCode: null, // headers already sent; status_code intentionally null — record by error code only (D61-D63 reviewer P2-2)
|
||
});
|
||
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';
|
||
_pushError({
|
||
error: e,
|
||
provider: streamProvider,
|
||
path: '/v1/chat/completions',
|
||
statusCode: 502,
|
||
});
|
||
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 {
|
||
// D58 — releaseSpawn lives inside sourceFactory's finally (only the
|
||
// source caller acquired). The req-close listener is detached here
|
||
// so it doesn't leak past the response lifetime.
|
||
res.removeListener('close', onClose);
|
||
// D61: cancel the heartbeat timer on every exit path (normal stop,
|
||
// error, generator-exhausted, client-disconnect). The no-op stub
|
||
// returned when intervalMs<=0 makes this cheap when disabled.
|
||
heartbeat.stop();
|
||
}
|
||
return;
|
||
}
|
||
// CONCURRENCY_LIMIT fallthrough — flow continues into the buffered path
|
||
// below.
|
||
}
|
||
|
||
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 });
|
||
_pushError({
|
||
error: e,
|
||
provider: null,
|
||
path: '/v1/chat/completions',
|
||
statusCode: 500,
|
||
});
|
||
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,
|
||
});
|
||
// D62: surface the chain-exhausted / provider-error onto the recentErrors
|
||
// ring so /status (D63) shows it to owner identities. The ring filter
|
||
// requires either a ProviderError code or statusCode >= 500; spawn errors
|
||
// typically map to 502 below — fall back to 500 if no statusCode is
|
||
// present so a programming-error throw still records.
|
||
{
|
||
const httpStatus = originalError?.statusCode ?? originalError?.status ?? 502;
|
||
_pushError({
|
||
error: originalError ?? new Error('unknown chain-exhausted error'),
|
||
provider: providerUsed ?? (chain[0]?.provider ?? null),
|
||
path: '/v1/chat/completions',
|
||
statusCode: httpStatus,
|
||
});
|
||
}
|
||
|
||
// 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, olpIdentity);
|
||
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,
|
||
olpIdentity,
|
||
);
|
||
|
||
// 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.)
|
||
// D61: SSE headers are emitted from the shared SSE_DEFAULT_HEADERS
|
||
// constant so all streaming paths agree on `X-Accel-Buffering: no`.
|
||
// Heartbeat is intentionally not wired in the buffered replay path: the
|
||
// burst write completes synchronously into the socket buffer, so silent
|
||
// windows are bounded by the chunk count, not by provider think-time.
|
||
res.writeHead(200, {
|
||
...SSE_DEFAULT_HEADERS,
|
||
...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);
|
||
}
|
||
}
|
||
|
||
// ── Phase 3 / D50 — Management endpoints (owner_only_block per ADR 0008 §8) ──
|
||
|
||
// Read dashboard.html once at startup; cached in memory thereafter. If the
|
||
// file is absent (e.g. test that imports server.mjs from a non-repo cwd), we
|
||
// still serve a minimal in-memory fallback. The file lives at the repo root.
|
||
let _dashboardHtmlCache = null;
|
||
function _loadDashboardHtml() {
|
||
if (_dashboardHtmlCache !== null) return _dashboardHtmlCache;
|
||
try {
|
||
const path = join(__dirname, 'dashboard.html');
|
||
_dashboardHtmlCache = readFileSync(path, 'utf-8');
|
||
} catch {
|
||
_dashboardHtmlCache = '<!DOCTYPE html><title>OLP Dashboard</title><p>dashboard.html not on disk; server has fallen back to this in-memory stub. Check the OLP installation.</p>';
|
||
}
|
||
return _dashboardHtmlCache;
|
||
}
|
||
|
||
/**
|
||
* Gated owner-only_block handler factory per ADR 0008 §8 + §7.5. Builds the
|
||
* audit ctx, runs auth, blocks non-owner with 401, wires res.on('finish') for
|
||
* audit + touchLastUsed, then calls inner(req, res, olpIdentity, auditCtx).
|
||
*
|
||
* inner should be async (or sync); errors propagate to the router's catch.
|
||
*/
|
||
async function _runOwnerOnlyManagementEndpoint(req, res, method, path, inner) {
|
||
const startMs = Date.now();
|
||
const auditCtx = {
|
||
ts: new Date().toISOString(),
|
||
key_id: ANONYMOUS_KEY_ID,
|
||
owner_tier: 'anonymous',
|
||
method,
|
||
path,
|
||
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 && _authedKeyId !== ENV_OWNER_KEY_ID) {
|
||
touchLastUsed(_authedKeyId).catch(() => {});
|
||
}
|
||
});
|
||
|
||
// Step 1: authenticate.
|
||
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;
|
||
|
||
// Step 2: owner-only_block. ADR 0008 §8 says management endpoints REJECT
|
||
// non-owner identities outright (vs /health's trim model). Anonymous +
|
||
// guest both produce 401 here.
|
||
if (olpIdentity.owner_tier !== 'owner') {
|
||
auditCtx.error_code = 'owner_required';
|
||
return sendError(res, 401,
|
||
'OLP owner-tier identity required for this management endpoint.',
|
||
'owner_required',
|
||
olpErrorHeaders({ startMs }));
|
||
}
|
||
|
||
// Step 3: delegate.
|
||
return await inner(req, res, olpIdentity, auditCtx);
|
||
}
|
||
|
||
/**
|
||
* GET /dashboard
|
||
* Serves dashboard.html to owner identities. Owner-only_block (per ADR 0008 §8).
|
||
* Cached in memory at startup; D51 ships the real multi-panel UI; D50 ships a
|
||
* placeholder.
|
||
*/
|
||
async function handleDashboard(req, res) {
|
||
return _runOwnerOnlyManagementEndpoint(req, res, 'GET', '/dashboard',
|
||
async (_req, res2, _identity, _auditCtx) => {
|
||
const html = _loadDashboardHtml();
|
||
res2.writeHead(200, {
|
||
'Content-Type': 'text/html; charset=utf-8',
|
||
'Content-Length': Buffer.byteLength(html, 'utf-8'),
|
||
'Cache-Control': 'no-cache',
|
||
});
|
||
res2.end(html);
|
||
});
|
||
}
|
||
|
||
/**
|
||
* GET /v0/management/dashboard-data
|
||
* Full JSON aggregate per ADR 0008 § 7.2. The dashboard 30s poll consumes
|
||
* this. Owner-only_block.
|
||
*/
|
||
async function handleManagementDashboardData(req, res) {
|
||
return _runOwnerOnlyManagementEndpoint(req, res, 'GET', '/v0/management/dashboard-data',
|
||
async (_req, res2, _identity, _auditCtx) => {
|
||
// Quota panel (legacy): collect quotaStatus from each loaded provider; null on
|
||
// throw or null return → "unavailable" indicator.
|
||
// DEPRECATED: kept for backwards compat with current dashboard.html (D82 will
|
||
// switch consumers to quota_v2; legacy 'quota' key removed at v1.0.0 or earlier).
|
||
const quota = [];
|
||
for (const [name, provider] of loadedProviders) {
|
||
try {
|
||
const q = await provider.quotaStatus(null);
|
||
quota.push({ provider: name, ...(q ?? {}), available: q?.available ?? null });
|
||
} catch (err) {
|
||
quota.push({ provider: name, error: err?.message ?? String(err), available: null });
|
||
}
|
||
}
|
||
|
||
// quota_v2 (D81 / Phase 5): normalized per-provider quota shape for enriched
|
||
// dashboard rendering. Built from aggregateProviderQuota() in lib/audit-query.mjs.
|
||
// Each entry contains: provider, status, schema_version, last_fresh_at,
|
||
// utilization, reset, representative_claim, fallback_percentage, overage, raw_available.
|
||
// Providers with null quotaStatus() return { status: 'unavailable', reason: ... }.
|
||
// Authority: ADR 0008 Amendment (D81) + ADR 0012 D81 + ADR 0013 Rule 5.
|
||
let quota_v2 = [];
|
||
try {
|
||
quota_v2 = await auditAggregateProviderQuota({ providers: loadedProviders });
|
||
} catch (err) {
|
||
// Graceful degradation: quota_v2 is optional enrichment; don't fail entire payload.
|
||
logEvent('warn', 'dashboard_data_quota_v2_failed', { error: err?.message ?? String(err) });
|
||
}
|
||
|
||
const WINDOW_24H = 24 * 60 * 60 * 1000;
|
||
const payload = {
|
||
generated_at: new Date().toISOString(),
|
||
window_24h: auditAggregateRequests({ windowMs: WINDOW_24H, logEvent }),
|
||
cache_hit_24h: auditCacheHitRateWindow({ windowMs: WINDOW_24H, logEvent }),
|
||
quota,
|
||
quota_v2,
|
||
spend_trend_30d: auditSpendTrendDaily({ days: 30, logEvent }),
|
||
top_fallback_chains_24h: auditTopFallbackChains({ windowMs: WINDOW_24H, limit: 10, logEvent }),
|
||
cache_stats: cacheStore.stats(),
|
||
};
|
||
sendJSON(res2, 200, payload);
|
||
});
|
||
}
|
||
|
||
/**
|
||
* GET /v0/management/quota
|
||
* Per-provider quota snapshot only (subset of dashboard-data). Useful for
|
||
* scripted monitoring. Owner-only_block.
|
||
*/
|
||
async function handleManagementQuota(req, res) {
|
||
return _runOwnerOnlyManagementEndpoint(req, res, 'GET', '/v0/management/quota',
|
||
async (_req, res2, _identity, _auditCtx) => {
|
||
// Legacy quota array (backwards compat).
|
||
const quota = [];
|
||
for (const [name, provider] of loadedProviders) {
|
||
try {
|
||
const q = await provider.quotaStatus(null);
|
||
quota.push({ provider: name, ...(q ?? {}), available: q?.available ?? null });
|
||
} catch (err) {
|
||
quota.push({ provider: name, error: err?.message ?? String(err), available: null });
|
||
}
|
||
}
|
||
// quota_v2 (D81): normalized per-provider quota shape per ADR 0008 Amendment (D81).
|
||
let quota_v2 = [];
|
||
try {
|
||
quota_v2 = await auditAggregateProviderQuota({ providers: loadedProviders });
|
||
} catch (err) {
|
||
logEvent('warn', 'management_quota_v2_failed', { error: err?.message ?? String(err) });
|
||
}
|
||
sendJSON(res2, 200, { generated_at: new Date().toISOString(), quota, quota_v2 });
|
||
});
|
||
}
|
||
|
||
/**
|
||
* GET /v0/management/status
|
||
*
|
||
* D63 (ADR 0010 § Phase 4 D61-D63). Combined snapshot of process lifecycle,
|
||
* per-provider health, request counters, cache stats, and recentErrors[20].
|
||
* Owner-only_block per ADR 0008 §8 (matches the rest of the management
|
||
* namespace).
|
||
*
|
||
* Port of OCP server.mjs:1151-1188 adapted to OLP's namespace + identity model:
|
||
* - Path is /v0/management/status (not root /status — OLP's namespace
|
||
* discipline is stricter than OCP's).
|
||
* - No "plan" field (OCP's plan came from the Anthropic Pro/Max API which
|
||
* OLP does not call directly).
|
||
* - Adds provider-quota subset + cache stats + cumulative request counters.
|
||
* - recentErrors is filtered upstream (no 401/403 entries) and capped at 20.
|
||
*
|
||
* Response shape:
|
||
* {
|
||
* ok: true,
|
||
* version: "0.4.0-phase4",
|
||
* uptime_ms: <number>,
|
||
* uptime_human: "Xh Ym Zs",
|
||
* started_at: <ISO8601>,
|
||
* providers: {
|
||
* enabled: <number>,
|
||
* available: <number>,
|
||
* status: { [providerKey]: { ok, ...healthCheck, activeSpawns } }
|
||
* },
|
||
* stats: {
|
||
* total_requests: <number>,
|
||
* active_requests: <number>,
|
||
* cache: { hits, misses, size, inflightCount, ... },
|
||
* },
|
||
* recent_errors: [{ time, message, code, provider, path, status_code }, ...],
|
||
* generated_at: <ISO8601>,
|
||
* }
|
||
*/
|
||
async function handleManagementStatus(req, res) {
|
||
return _runOwnerOnlyManagementEndpoint(req, res, 'GET', '/v0/management/status',
|
||
async (_req, res2, _identity, _auditCtx) => {
|
||
const now = Date.now();
|
||
const uptimeMs = now - _serverStartMs;
|
||
const hours = Math.floor(uptimeMs / 3600000);
|
||
const minutes = Math.floor((uptimeMs % 3600000) / 60000);
|
||
const seconds = Math.floor((uptimeMs % 60000) / 1000);
|
||
|
||
// Provider status subset: ok + activeSpawns + a single optional error
|
||
// string (mirrors the /health full payload shape but trimmed — /status
|
||
// is for owner monitoring, not external probes).
|
||
const providerStatus = {};
|
||
for (const [name, provider] of loadedProviders) {
|
||
const activeSpawns = getActiveSpawnCount(name);
|
||
try {
|
||
const hc = await provider.healthCheck();
|
||
providerStatus[name] = { ...hc, activeSpawns };
|
||
} catch (err) {
|
||
providerStatus[name] = { ok: false, error: err?.message ?? String(err), activeSpawns };
|
||
}
|
||
}
|
||
|
||
const payload = {
|
||
ok: true,
|
||
version: VERSION,
|
||
uptime_ms: uptimeMs,
|
||
uptime_human: `${hours}h ${minutes}m ${seconds}s`,
|
||
started_at: new Date(_serverStartMs).toISOString(),
|
||
providers: {
|
||
enabled: loadedProviders.size,
|
||
available: listAllProviderNames().length,
|
||
status: providerStatus,
|
||
},
|
||
stats: {
|
||
total_requests: _totalRequests,
|
||
active_requests: _activeRequests,
|
||
cache: cacheStore.stats(),
|
||
},
|
||
recent_errors: recentErrors.slice(),
|
||
generated_at: new Date().toISOString(),
|
||
};
|
||
sendJSON(res2, 200, payload);
|
||
});
|
||
}
|
||
|
||
/**
|
||
* GET /cache/stats
|
||
* Live in-memory CacheStore stats. Owner-only_block.
|
||
* Per ADR 0008 § 7.4: returns the current cacheStore.stats() shape
|
||
* ({ hits, misses, size, inflightCount }). Per-(provider, model) breakdown
|
||
* is a Phase 4+ amendment trigger.
|
||
*/
|
||
async function handleCacheStats(req, res) {
|
||
return _runOwnerOnlyManagementEndpoint(req, res, 'GET', '/cache/stats',
|
||
async (_req, res2, _identity, _auditCtx) => {
|
||
sendJSON(res2, 200, { generated_at: new Date().toISOString(), ...cacheStore.stats() });
|
||
});
|
||
}
|
||
|
||
// ── 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);
|
||
}
|
||
|
||
// Phase 3 / D50 — management endpoints (owner-only_block per ADR 0008 § 8)
|
||
if (method === 'GET' && path === '/dashboard') {
|
||
return await handleDashboard(req, res);
|
||
}
|
||
if (method === 'GET' && path === '/v0/management/dashboard-data') {
|
||
return await handleManagementDashboardData(req, res);
|
||
}
|
||
if (method === 'GET' && path === '/v0/management/quota') {
|
||
return await handleManagementQuota(req, res);
|
||
}
|
||
if (method === 'GET' && path === '/v0/management/status') {
|
||
return await handleManagementStatus(req, res);
|
||
}
|
||
if (method === 'GET' && path === '/cache/stats') {
|
||
return await handleCacheStats(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 });
|
||
_pushError({
|
||
error: e,
|
||
provider: null,
|
||
path,
|
||
statusCode: 500,
|
||
});
|
||
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 };
|
||
// Phase 7 PR-B: re-export sandbox manager test seam so tests can reset state.
|
||
export { __resetSandboxManagerForTests };
|
||
|
||
// 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) {
|
||
// Phase 7 PR-B (ADR 0014 § PR-B): bootstrap sandbox before listening.
|
||
// bootstrapSandbox() is idempotent + error-safe — server always starts even
|
||
// if sandbox initialization fails (degrades to unsandboxed, logs a warning).
|
||
// The /health.sandbox.active field reflects the result.
|
||
const sandboxBoot = await bootstrapSandbox();
|
||
if (sandboxBoot.active) {
|
||
process.stdout.write(
|
||
`OLP sandbox active (config-at-boot): ${sandboxBoot.summary}\n`,
|
||
);
|
||
} else {
|
||
process.stderr.write(
|
||
`OLP sandbox NOT active: ${sandboxBoot.reason} — ` +
|
||
`provider spawns will run UNSANDBOXED (test/dev only; not safe for cloud)\n`,
|
||
);
|
||
}
|
||
|
||
const server = createOlpServer();
|
||
server.listen(PORT, BIND, () => {
|
||
const enabledCount = loadedProviders.size;
|
||
// D74 P3-5: banner no longer hardcodes the phase. Derives from VERSION
|
||
// (which advances at every Phase close) so banner stays accurate
|
||
// without future-maintenance touch-ups at every Phase boundary.
|
||
process.stdout.write(
|
||
`OLP v${VERSION} listening on :${PORT} (${enabledCount} providers enabled)\n`,
|
||
);
|
||
});
|
||
}
|