From e6701ff6985fcc685ab090a0fcc650d26d0be9ce Mon Sep 17 00:00:00 2001 From: dtzp555-max Date: Tue, 26 May 2026 08:23:17 +1000 Subject: [PATCH] =?UTF-8?q?feat+test:=20D61+D62+D63=20=E2=80=94=20SSE=20he?= =?UTF-8?q?artbeat=20+=20recentErrors[20]=20+=20/v0/management/status=20(#?= =?UTF-8?q?41)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat+test: D61+D62+D63 — SSE heartbeat + recentErrors[20] ring + /v0/management/status First substantive Phase 4 implementation. Bundle of 3 D-days per Iron Rule 11 IDR rationale: all three converge on the same observability surface (status endpoint reads recentErrors + provider stats + heartbeat-related counters; heartbeat shares the streaming branch with recentErrors emission; all live in server.mjs). ## D61 — SSE heartbeat Ported from OCP server.mjs:660-685 startHeartbeat() with the OCP db11105 "eager-headers-post-spawn" fix folded in from day one. - New config field streaming.heartbeat_interval_ms in ~/.olp/config.json (default 0 = disabled, matching OCP's safe default) - When enabled (>0), streaming branch emits `: keepalive\n\n` SSE comment every interval_ms ms during silent windows - Timer resets on every real chunk written - Cleanup on stream end / error / abort / client disconnect - SSE_DEFAULT_HEADERS constant centralizes Content-Type / Cache-Control / Connection / X-Accel-Buffering: no (the last was the missing OCP lesson that broke long streams behind nginx 60s idle) - Per-attached-client lifecycle (each tee output gets its own timer) - One heartbeat_active log per stream on first fire; no per-fire log noise Note: heartbeat NOT wired in the buffered-replay streaming branch because that branch writes the burst synchronously into the socket buffer — no silent windows exist there. Inline comment notes this. ## D62 — recentErrors[20] ring buffer Module-scope bounded ring, surfaced via /v0/management/status at D63. - _pushError({ error, provider, path, statusCode }) entry shape: { time (ISO8601), message (200-char cap), code, provider, path, status_code } - Filter: only ProviderError OR statusCode >= 500 (401/403 brute-force noise excluded — protects ring from auth-probe flooding) - Path sanitization via .replace(/\/[\w./-]+/g, '[path]') ported from OCP server.mjs:1395 — strips internal paths before they leave the proxy - Wired into 5 server-side error paths: chain-exhausted, pre-first-chunk streaming error, mid-stream IR error chunk, fallback-engine programming error, router-level unhandled error - In-memory only (not persisted across restart) per OCP precedent - Test seam __clearRecentErrors / __snapshotRecentErrors ## D63 — /v0/management/status combined endpoint OCP /status equivalent, OLP-namespaced per stricter discipline. - New route GET /v0/management/status, owner-only_block (matches ADR 0007 § 7 + ADR 0008 Phase 3 management endpoint gating pattern) - Returns { ok, version, uptime_ms, uptime_human, started_at, providers: {enabled, available, status}, stats: {total_requests, active_requests, cache: cacheStore.stats()}, recent_errors: [], generated_at } - _totalRequests + _activeRequests module-scope counters incremented at top of handleChatCompletions; _activeRequests decremented in res.on('close'/ 'finish') with idempotent guard - Counters NOT exposed via /health (owner-trim intentional there); only via /v0/management/status (owner-only_block) - Reuses _runOwnerOnlyManagementEndpoint helper from D50 Phase 3 work ## Test count 623 → 636 (+13 D61-D63 tests across Suites 29, 30, 31). All 636 pass locally. ## Scope discipline server.mjs + test-features.mjs + lib/fallback/engine.mjs only (engine.mjs touched only to extend loadFallbackConfigSync to surface the new streaming.heartbeat_interval_ms field; no engine behavior change). Untouched: provider plugins, IR, cache layer, dashboard.html, audit-query, README, CHANGELOG, package.json. /health payload unchanged. None of the existing 623 tests regressed. ## Authority - ADR 0010 § Phase 4 D-day plan (D61-D63 line) - OCP server.mjs:660-685 (startHeartbeat reference impl) - OCP commit db11105 (eager-headers-post-spawn fix) - OCP server.mjs:301, 354-358 (recentErrors ring pattern) - OCP server.mjs:1151-1188 (/status combined endpoint pattern) - OCP server.mjs:1395 (error path sanitization) - ADR 0007 § 7 (identity classes — owner-only_block gating) - ADR 0008 (management endpoints pattern reused) - 2026-05-26 brainstorm (Top 5 OCP inheritance candidates, items 1 + 4) Co-Authored-By: Claude Opus 4.7 * fix: D61-D63 reviewer P2 fold-in — explicit 401/403 filter + null status_code for post-headers Reviewer APPROVE WITH MINOR — 0 P0/P1, 2 P2 (both about _pushError filter clarity / defense-in-depth). P2-1 — explicit 401/403 reject at function level. The current call sites never invoke _pushError from authenticate() failures (call-site discipline), but a future contributor passing a ProviderError tagged statusCode=401 would slip past the isProviderError branch and flood the ring under brute force. Added explicit `if (statusCode === 401 || statusCode === 403) return;` as defense-in-depth. P2-2 — pass `statusCode: null` for the two streaming-error-after-first- chunk _pushError sites instead of `statusCode: 200`. Headers are already sent so any numeric status is misleading; null + record-by-error-code is the explicit intent. Avoids a future filter-refactor accidentally dropping these entries because they look like 200-OK. Test count unchanged 636/636 pass (filter behavior identical from call-site perspective; the changes are defensive + intent-clarifying). Co-Authored-By: Claude Opus 4.7 --------- Co-authored-by: dtzp555 Co-authored-by: Claude Opus 4.7 --- lib/fallback/engine.mjs | 16 +- server.mjs | 400 +++++++++++++++++++++++- test-features.mjs | 663 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 1065 insertions(+), 14 deletions(-) diff --git a/lib/fallback/engine.mjs b/lib/fallback/engine.mjs index 1327c56..e215901 100644 --- a/lib/fallback/engine.mjs +++ b/lib/fallback/engine.mjs @@ -667,24 +667,36 @@ function defaultConfigPath() { * Returns empty config (no chains, no soft triggers, no enabled providers) if the * file is absent, unreadable, or malformed. * + * D61 (ADR 0010 § Phase 4 D61-D63): adds `streaming` block. Currently + * exposes `heartbeat_interval_ms` (default 0 = heartbeat disabled). When + * heartbeat_interval_ms > 0, the streaming branch emits `: keepalive\n\n` + * SSE comment frames during silent windows of length >= the interval. Default + * 0 preserves backwards compat (no behavioural change). + * * @param {string} [configPath] — override path (for testing — do NOT write to ~/.olp/config.json in tests) - * @returns {{ chains: object, soft_triggers: object, providersEnabled: Record }} + * @returns {{ chains: object, soft_triggers: object, providersEnabled: Record, streaming: { heartbeat_interval_ms: number } }} */ export function loadFallbackConfigSync(configPath) { + const DEFAULT_STREAMING = { heartbeat_interval_ms: 0 }; try { const path = configPath ?? defaultConfigPath(); const raw = readFileSync(path, 'utf8'); const parsed = JSON.parse(raw); const routing = parsed?.routing ?? {}; const providers = parsed?.providers ?? {}; + const streaming = parsed?.streaming ?? {}; + const hb = Number(streaming.heartbeat_interval_ms); return { chains: routing.chains ?? {}, soft_triggers: routing.soft_triggers ?? {}, providersEnabled: providers.enabled ?? {}, + streaming: { + heartbeat_interval_ms: Number.isFinite(hb) && hb >= 0 ? hb : 0, + }, }; } catch { // File absent, unreadable, or malformed → no fallback config (single-hop mode) // Empty providersEnabled → all providers disabled → 503 per ALIGNMENT.md v0.1 posture. - return { chains: {}, soft_triggers: {}, providersEnabled: {} }; + return { chains: {}, soft_triggers: {}, providersEnabled: {}, streaming: { ...DEFAULT_STREAMING } }; } } diff --git a/server.mjs b/server.mjs index 7d9e74f..00ac237 100644 --- a/server.mjs +++ b/server.mjs @@ -90,6 +90,117 @@ function logEvent(level, event, data = {}) { } } +// ── 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; +} + // ── Startup config ──────────────────────────────────────────────────────── // Read ~/.olp/config.json once at startup. Provides: // - providers.enabled → which providers are loaded (ADR 0002 § Disable model) @@ -131,6 +242,76 @@ 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. @@ -705,6 +886,22 @@ function handleModels(req, 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 = { @@ -1216,6 +1413,12 @@ async function handleChatCompletions(req, res) { 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 })); } @@ -1250,8 +1453,16 @@ async function handleChatCompletions(req, res) { // 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); @@ -1282,6 +1493,37 @@ async function handleChatCompletions(req, res) { // 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; @@ -1297,6 +1539,12 @@ async function handleChatCompletions(req, res) { 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(); @@ -1307,15 +1555,16 @@ async function handleChatCompletions(req, res) { if (!res.headersSent) { res.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - 'X-Accel-Buffering': 'no', + ...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') { @@ -1352,10 +1601,7 @@ async function handleChatCompletions(req, res) { // yielded), emit headers + [DONE] so the response is still valid SSE. if (!res.headersSent) { res.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - 'X-Accel-Buffering': 'no', + ...SSE_DEFAULT_HEADERS, ...streamHeaders, }); } @@ -1376,6 +1622,12 @@ async function handleChatCompletions(req, res) { 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(); @@ -1387,6 +1639,12 @@ async function handleChatCompletions(req, res) { 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 })); @@ -1399,6 +1657,10 @@ async function handleChatCompletions(req, res) { // 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; } @@ -1414,6 +1676,12 @@ async function handleChatCompletions(req, res) { } 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 })); } @@ -1437,6 +1705,20 @@ async function handleChatCompletions(req, res) { 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 @@ -1543,11 +1825,13 @@ async function handleChatCompletions(req, res) { // 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, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - 'X-Accel-Buffering': 'no', + ...SSE_DEFAULT_HEADERS, ...headers, }); @@ -1718,6 +2002,89 @@ async function handleManagementQuota(req, res) { }); } +/** + * 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: , + * uptime_human: "Xh Ym Zs", + * started_at: , + * providers: { + * enabled: , + * available: , + * status: { [providerKey]: { ok, ...healthCheck, activeSpawns } } + * }, + * stats: { + * total_requests: , + * active_requests: , + * cache: { hits, misses, size, inflightCount, ... }, + * }, + * recent_errors: [{ time, message, code, provider, path, status_code }, ...], + * generated_at: , + * } + */ +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. @@ -1767,6 +2134,9 @@ async function router(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); } @@ -1775,6 +2145,12 @@ async function router(req, res) { 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'); } diff --git a/test-features.mjs b/test-features.mjs index 2642f61d..3558bb2 100644 --- a/test-features.mjs +++ b/test-features.mjs @@ -4341,6 +4341,11 @@ import { __clearCache, __setAuthConfig, __resetAuthConfig, + __setStreamingConfig, + __resetStreamingConfig, + __clearRecentErrors, + __snapshotRecentErrors, + __resetRequestCounters, } from './server.mjs'; // ── Phase 2 / D45+D46 server-side default override ──────────────────────── @@ -12928,3 +12933,661 @@ describe('Suite 28 — D58 streaming singleflight (server.mjs HTTP wiring, ADR 0 assert.equal(counter.count, 2, '28i r2: second request triggered a fresh spawn (no cache reuse for truncated entry)'); }); }); + +// ── Suite 29: D61 SSE heartbeat (ADR 0010 § Phase 4 D61-D63) ────────────── +// +// Tests the opt-in SSE heartbeat that emits `: keepalive\n\n` SSE comment +// frames during silent windows. Port of OCP `startHeartbeat` (server.mjs: +// 660-685) adapted to OLP's config (`streaming.heartbeat_interval_ms`), +// per-attached-client lifecycle, and eager-headers-post-spawn rule. +// +// Authority: ADR 0010 § Phase 4 D61-D63; OCP spec +// docs/superpowers/specs/2026-04-25-47-sse-heartbeat-design.md. + +describe('Suite 29 — D61 SSE heartbeat (ADR 0010 § Phase 4 D61-D63)', () => { + let server29; + let port29; + let savedToken29; + let lp29; + let savedAnthropic29; + + before(async () => { + savedToken29 = process.env.CLAUDE_CODE_OAUTH_TOKEN; + process.env.CLAUDE_CODE_OAUTH_TOKEN = 'fake-token-suite-29'; + __setProvidersEnabled({ anthropic: true }); + + const mod = await import('./server.mjs'); + lp29 = mod.loadedProviders; + savedAnthropic29 = lp29.get('anthropic'); + mod.__clearCache(); + + server29 = mod.createOlpServer(); + await new Promise((resolve, reject) => { + server29.listen(0, '127.0.0.1', resolve); + server29.once('error', reject); + }); + port29 = server29.address().port; + }); + + after(async () => { + if (savedAnthropic29 !== undefined) { + lp29.set('anthropic', savedAnthropic29); + } else { + lp29.delete('anthropic'); + } + __resetProvidersEnabled(); + __resetSpawnImpl(); + __resetStreamingConfig(); + if (savedToken29 !== undefined) { + process.env.CLAUDE_CODE_OAUTH_TOKEN = savedToken29; + } else { + delete process.env.CLAUDE_CODE_OAUTH_TOKEN; + } + if (!server29) return; + return new Promise(r => server29.close(r)); + }); + + beforeEach(async () => { + const mod = await import('./server.mjs'); + mod.__clearCache(); + __setStreamingConfig({ heartbeat_interval_ms: 0 }); + }); + + /** Install a fake provider whose async generator paces yields with `gapMs`. */ + function installPacedProvider(chunks, gapMs) { + const counter = { count: 0 }; + const fake = { + ...savedAnthropic29, + spawn: async function* (_ir, _ctx) { + counter.count++; + for (const c of chunks) { + if (gapMs > 0) await new Promise(r => setTimeout(r, gapMs)); + yield c; + } + }, + }; + lp29.set('anthropic', fake); + return counter; + } + + /** Make a streaming request and return body + arrival timestamps. */ + function makeTimedStreamRequest({ prompt }) { + return new Promise((resolve, reject) => { + const arrivals = []; + const req = httpRequest({ + hostname: '127.0.0.1', + port: port29, + method: 'POST', + path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + }, res => { + let data = ''; + res.on('data', d => { + const s = d.toString(); + arrivals.push({ ts: Date.now(), text: s }); + data += s; + }); + res.on('end', () => resolve({ + status: res.statusCode, + body: data, + headers: res.headers, + arrivals, + })); + res.on('error', reject); + }); + req.on('error', reject); + req.write(JSON.stringify({ + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: prompt }], + stream: true, + })); + req.end(); + }); + } + + it('29a — heartbeat ENABLED (interval=40ms) + slow provider → keepalive frame(s) appear in body', async () => { + // Provider stalls 200ms before first delta, then ~80ms between deltas. + // With heartbeat=40ms the silent windows produce multiple keepalive + // frames. Eager-headers must fire so the frames are flushed. + installPacedProvider([ + { type: 'delta', content: 'h29a-1' }, + { type: 'delta', content: 'h29a-2' }, + { type: 'stop', finish_reason: 'stop' }, + ], 100); + __setStreamingConfig({ heartbeat_interval_ms: 40 }); + + const r = await makeTimedStreamRequest({ prompt: 'h29a-prompt' }); + assert.equal(r.status, 200, `r status ${r.status}: ${r.body.slice(0, 200)}`); + // At least one keepalive comment frame must be present. + const keepaliveCount = (r.body.match(/: keepalive/g) ?? []).length; + assert.ok(keepaliveCount >= 1, `expected at least one keepalive frame, got ${keepaliveCount}; body=${r.body.slice(0, 400)}`); + // Content + [DONE] still present (heartbeat does not break the stream). + assert.ok(r.body.includes('h29a-1'), 'body must include h29a-1'); + assert.ok(r.body.includes('h29a-2'), 'body must include h29a-2'); + assert.ok(r.body.includes('[DONE]'), 'body must include [DONE]'); + }); + + it('29b — heartbeat DISABLED (default 0) → NO keepalive frame in body', async () => { + // Same provider pacing but no heartbeat config → no keepalive frames. + installPacedProvider([ + { type: 'delta', content: 'h29b-1' }, + { type: 'delta', content: 'h29b-2' }, + { type: 'stop', finish_reason: 'stop' }, + ], 100); + __setStreamingConfig({ heartbeat_interval_ms: 0 }); + + const r = await makeTimedStreamRequest({ prompt: 'h29b-prompt' }); + assert.equal(r.status, 200); + assert.equal( + (r.body.match(/: keepalive/g) ?? []).length, + 0, + `heartbeat disabled but found keepalive frame; body=${r.body.slice(0, 400)}`, + ); + assert.ok(r.body.includes('h29b-1')); + assert.ok(r.body.includes('[DONE]')); + }); + + it('29c — heartbeat timer resets on every real chunk (chunks at 30ms gaps with hb=50ms → 0 keepalives mid-stream)', async () => { + // With chunks arriving every 30ms and heartbeat=50ms, the timer resets + // before it can fire — there should be no keepalive frames between + // chunks (silent windows never exceed 50ms). Pacing the source by a + // long pre-first-chunk gap is avoided here: pre-first-chunk silence is + // covered by 29a. + installPacedProvider([ + { type: 'delta', content: 'r29c-1' }, + { type: 'delta', content: 'r29c-2' }, + { type: 'delta', content: 'r29c-3' }, + { type: 'delta', content: 'r29c-4' }, + { type: 'stop', finish_reason: 'stop' }, + ], 30); + __setStreamingConfig({ heartbeat_interval_ms: 50 }); + + const r = await makeTimedStreamRequest({ prompt: 'h29c-prompt' }); + assert.equal(r.status, 200); + // Each chunk fires within 30ms of the previous, so the heartbeat (50ms) + // is repeatedly reset before it can fire. There may still be ONE + // keepalive frame from the pre-first-chunk window (30ms wait before + // first delta is borderline), but we expect zero mid-stream frames. + const keepaliveCount = (r.body.match(/: keepalive/g) ?? []).length; + assert.ok(keepaliveCount <= 1, `expected <= 1 keepalive (reset working), got ${keepaliveCount}; body=${r.body.slice(0, 400)}`); + for (const tag of ['r29c-1', 'r29c-2', 'r29c-3', 'r29c-4', '[DONE]']) { + assert.ok(r.body.includes(tag), `body missing ${tag}`); + } + }); + + it('29d — heartbeat cancelled on client disconnect (no further keepalive after abort)', async () => { + // Source stalls forever. Heartbeat=20ms fires repeatedly while the + // client is connected. After abort, heartbeat.stop() must fire — the + // surface check we do here: the server cleanly closes (no uncaught + // exception on continued setTimeout firing). We can't observe "no more + // keepalive frames" directly from the aborted client, but we CAN + // observe that the test runner doesn't time out — the dead-socket + // res.write inside the heartbeat is swallowed (no throw bubbling up). + let yieldedCount = 0; + const fake = { + ...savedAnthropic29, + spawn: async function* (_ir, _ctx) { + // Hold open with a single delta then a long sleep. + yield { type: 'delta', content: 'first-delta' }; + yieldedCount++; + try { + // Long stall; should be interrupted by iterator.return() + await new Promise(r => setTimeout(r, 1000)); + yield { type: 'stop', finish_reason: 'stop' }; + } finally { + // intentional + } + }, + }; + lp29.set('anthropic', fake); + __setStreamingConfig({ heartbeat_interval_ms: 20 }); + + const result = await new Promise((resolve) => { + const req = httpRequest({ + hostname: '127.0.0.1', + port: port29, + method: 'POST', + path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + }, res => { + let body = ''; + res.on('data', d => { body += d.toString(); }); + res.on('end', () => resolve({ body, aborted: false })); + res.on('error', () => resolve({ body, aborted: true })); + }); + req.on('error', () => resolve({ body: '', aborted: true })); + req.write(JSON.stringify({ + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: 'h29d-abort' }], + stream: true, + })); + req.end(); + // Abort after 100ms (well into heartbeat firing window). + setTimeout(() => { try { req.destroy(); } catch { /* ignore */ } }, 100); + }); + // Sanity: client saw at least the first chunk + maybe keepalives. + assert.ok(yieldedCount === 1, 'first delta was yielded once'); + // Give the heartbeat loop time to fire post-abort — if heartbeat.stop() + // didn't fire, an uncaught exception or runaway timer would surface. + // Then assert that the server is still responsive to a new request, + // proving no fatal lingering state. + await new Promise(r => setTimeout(r, 60)); + + // Server health check: a fresh request must complete normally. + const probe = await fetch({ port: port29, method: 'GET', path: '/v1/models' }); + assert.equal(probe.status, 200, 'server remains responsive after disconnect; heartbeat timer did not crash the process'); + }); + + it('29e — eager-headers: heartbeat enabled → headers flushed BEFORE first content chunk', async () => { + // Pre-first-chunk silent window of 300ms; heartbeat=40ms ⇒ multiple + // keepalive frames must arrive BEFORE the first content delta. This + // is the OCP db11105 invariant: without eager-headers the keepalive + // writes would buffer (no headers sent yet) and the client wouldn't + // see anything until the first real chunk. + let firstDeltaIdx = -1; + let firstKeepaliveIdx = -1; + const fake = { + ...savedAnthropic29, + spawn: async function* (_ir, _ctx) { + // Long pre-first-chunk silence. + await new Promise(r => setTimeout(r, 250)); + yield { type: 'delta', content: 'eager-delta' }; + yield { type: 'stop', finish_reason: 'stop' }; + }, + }; + lp29.set('anthropic', fake); + __setStreamingConfig({ heartbeat_interval_ms: 40 }); + + const r = await makeTimedStreamRequest({ prompt: 'h29e-prompt' }); + assert.equal(r.status, 200); + // Headers received (status 200 was already received → headers flushed). + // Locate index of first arrival containing : keepalive and first arrival + // containing eager-delta. + for (let i = 0; i < r.arrivals.length; i++) { + if (firstKeepaliveIdx === -1 && r.arrivals[i].text.includes(': keepalive')) { + firstKeepaliveIdx = i; + } + if (firstDeltaIdx === -1 && r.arrivals[i].text.includes('eager-delta')) { + firstDeltaIdx = i; + } + } + assert.ok(firstKeepaliveIdx >= 0, `expected at least one keepalive arrival; arrivals=${JSON.stringify(r.arrivals.map(a => a.text.slice(0, 60)))}`); + assert.ok(firstDeltaIdx >= 0, 'expected the eager-delta to arrive'); + assert.ok(firstKeepaliveIdx < firstDeltaIdx, + `keepalive (idx ${firstKeepaliveIdx}) must arrive BEFORE first delta (idx ${firstDeltaIdx}); proves eager-headers`); + }); +}); + +// ── Suite 30: D62 recentErrors[20] ring buffer (ADR 0010 § Phase 4 D61-D63) ── +// +// Tests the in-memory ring of the last 20 server-side error events plus its +// filter (401/403 excluded so brute-force loops cannot flood the ring) and +// path-sanitization. Surfaced via /status (Suite 31). + +describe('Suite 30 — D62 recentErrors[20] ring buffer (ADR 0010 § Phase 4 D61-D63)', () => { + let server30; + let port30; + let savedToken30; + let lp30; + let savedAnthropic30; + + before(async () => { + savedToken30 = process.env.CLAUDE_CODE_OAUTH_TOKEN; + process.env.CLAUDE_CODE_OAUTH_TOKEN = 'fake-token-suite-30'; + __setProvidersEnabled({ anthropic: true }); + + const mod = await import('./server.mjs'); + lp30 = mod.loadedProviders; + savedAnthropic30 = lp30.get('anthropic'); + mod.__clearCache(); + + server30 = mod.createOlpServer(); + await new Promise((resolve, reject) => { + server30.listen(0, '127.0.0.1', resolve); + server30.once('error', reject); + }); + port30 = server30.address().port; + }); + + after(async () => { + if (savedAnthropic30 !== undefined) { + lp30.set('anthropic', savedAnthropic30); + } else { + lp30.delete('anthropic'); + } + __resetProvidersEnabled(); + __resetSpawnImpl(); + __clearRecentErrors(); + if (savedToken30 !== undefined) { + process.env.CLAUDE_CODE_OAUTH_TOKEN = savedToken30; + } else { + delete process.env.CLAUDE_CODE_OAUTH_TOKEN; + } + if (!server30) return; + return new Promise(r => server30.close(r)); + }); + + beforeEach(async () => { + const mod = await import('./server.mjs'); + mod.__clearCache(); + __clearRecentErrors(); + __setAuthConfig({ allow_anonymous: true, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + }); + + it('30a — provider error pushes an entry onto the recentErrors ring', async () => { + // Install a provider that throws SPAWN_FAILED before first chunk. + const fake = { + ...savedAnthropic30, + spawn: async function* (_ir, _ctx) { + // Throw before yielding any chunk → propagates through the streaming + // branch as a pre-first-chunk error → 502 + _pushError fires. + throw new (await import('./lib/providers/base.mjs')).ProviderError( + 'fake spawn failure for 30a', + 'SPAWN_FAILED', + ); + // eslint-disable-next-line no-unreachable + yield { type: 'stop', finish_reason: 'stop' }; + }, + }; + lp30.set('anthropic', fake); + + const before = __snapshotRecentErrors().length; + const r = await fetch({ + port: port30, method: 'POST', path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: '30a-trigger' }], + stream: true, + }, + }); + // The exact status varies (502 for pre-first-chunk error, or + // chain-exhausted 502 from the buffered path); both record _pushError. + assert.ok(r.status === 502 || r.status === 500, + `expected 5xx error, got ${r.status}: ${r.body.slice(0, 200)}`); + const after = __snapshotRecentErrors(); + assert.ok(after.length > before, `ring should have grown; before=${before} after=${after.length}`); + const last = after[after.length - 1]; + assert.ok(typeof last.time === 'string' && last.time.includes('T'), + 'entry must have ISO8601 time'); + assert.ok(typeof last.message === 'string' && last.message.length > 0, + 'entry must have a message'); + assert.ok(last.message.length <= 200, 'message must be capped at 200 chars'); + assert.equal(last.path, '/v1/chat/completions', 'entry path captured'); + assert.equal(typeof last.status_code, 'number', 'entry status_code captured'); + }); + + it('30b — ring caps at 20 entries (oldest evicted on push)', async () => { + __clearRecentErrors(); + // Push 25 synthetic errors via the public seam (we exercise this via a + // provider that throws repeatedly). Easier: call _pushError directly by + // hitting an endpoint that maps to a 5xx; install a fake that always + // throws and fire 25 requests. Each request adds one entry. + let pushCount = 0; + const fake = { + ...savedAnthropic30, + spawn: async function* (_ir, _ctx) { + pushCount++; + throw new (await import('./lib/providers/base.mjs')).ProviderError( + `30b-err-${pushCount}`, + 'SPAWN_FAILED', + ); + // eslint-disable-next-line no-unreachable + yield { type: 'stop', finish_reason: 'stop' }; + }, + }; + lp30.set('anthropic', fake); + + for (let i = 0; i < 25; i++) { + await fetch({ + port: port30, method: 'POST', path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: `30b-iter-${i}` }], + stream: false, + }, + }); + } + const snap = __snapshotRecentErrors(); + assert.equal(snap.length, 20, `ring must cap at 20, got ${snap.length}`); + // Oldest evicted → the first remaining entry's message should not be + // 30b-err-1 (which was pushed first and evicted). It should also not + // include `30b-err-2 .. 30b-err-5` (we pushed 25, last 20 retained → + // entries 6..25 remain). + const firstMsg = snap[0].message; + assert.ok(!firstMsg.includes('30b-err-1') || firstMsg.includes('30b-err-10') || firstMsg.includes('30b-err-11'), + `oldest 5 must be evicted; first message=${firstMsg}`); + }); + + it('30c — 401/403 auth failures are NOT pushed onto the ring (brute-force flood guard)', async () => { + __clearRecentErrors(); + __setAuthConfig({ allow_anonymous: false, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + try { + const before = __snapshotRecentErrors().length; + // Fire 5 requests with no auth header → all 401s. + for (let i = 0; i < 5; i++) { + const r = await fetch({ + port: port30, method: 'POST', path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + body: { model: 'claude-sonnet-4-6', messages: [{ role: 'user', content: `30c-${i}` }] }, + }); + assert.equal(r.status, 401, `expected 401 on no-auth request, got ${r.status}`); + } + const after = __snapshotRecentErrors().length; + assert.equal(after, before, '401 responses must NOT push onto recentErrors'); + } finally { + __setAuthConfig({ allow_anonymous: true, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + } + }); + + it('30d — path sanitization: filesystem-path-like tokens in message are replaced with [path]', async () => { + __clearRecentErrors(); + const fake = { + ...savedAnthropic30, + spawn: async function* (_ir, _ctx) { + // Error message embeds a filesystem path. + throw new (await import('./lib/providers/base.mjs')).ProviderError( + 'ENOENT: no such file or directory /Users/private/secret-file.json', + 'SPAWN_FAILED', + ); + // eslint-disable-next-line no-unreachable + yield { type: 'stop', finish_reason: 'stop' }; + }, + }; + lp30.set('anthropic', fake); + + await fetch({ + port: port30, method: 'POST', path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: '30d-path-sanitize' }], + stream: false, + }, + }); + const snap = __snapshotRecentErrors(); + assert.ok(snap.length >= 1, 'an entry must have been pushed'); + const last = snap[snap.length - 1]; + assert.ok(!last.message.includes('/Users/private/secret-file.json'), + `path leaked: ${last.message}`); + assert.ok(last.message.includes('[path]'), + `sanitization marker [path] expected; got ${last.message}`); + }); +}); + +// ── Suite 31: D63 /v0/management/status combined endpoint (ADR 0010 § Phase 4) ── +// +// Owner-only_block management endpoint that returns process/provider/cache +// stats + recentErrors. Tests gating + payload shape + recent_errors wiring. + +describe('Suite 31 — D63 /v0/management/status (ADR 0010 § Phase 4 D61-D63)', () => { + const GLOBAL_OLP_HOME_31 = process.env.OLP_HOME; + let TMP31, server31, port31; + let savedToken31; + let lp31; + let savedAnthropic31; + + before(async () => { + TMP31 = mkdtempSync(pathJoin(tmpdir(), 'olp-test-31-')); + process.env.OLP_HOME = TMP31; + savedToken31 = process.env.CLAUDE_CODE_OAUTH_TOKEN; + process.env.CLAUDE_CODE_OAUTH_TOKEN = 'fake-token-suite-31'; + __setProvidersEnabled({ anthropic: true }); + __setAuthConfig({ allow_anonymous: true, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + + const mod = await import('./server.mjs'); + lp31 = mod.loadedProviders; + savedAnthropic31 = lp31.get('anthropic'); + mod.__clearCache(); + mod.__clearRecentErrors(); + mod.__resetRequestCounters(); + + server31 = mod.createOlpServer(); + await new Promise((resolve, reject) => { + server31.listen(0, '127.0.0.1', resolve); + server31.once('error', reject); + }); + port31 = server31.address().port; + }); + + after(async () => { + if (savedAnthropic31 !== undefined) { + lp31.set('anthropic', savedAnthropic31); + } else { + lp31.delete('anthropic'); + } + __resetProvidersEnabled(); + __resetSpawnImpl(); + __clearRecentErrors(); + __resetRequestCounters(); + __setAuthConfig({ allow_anonymous: true, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + if (savedToken31 !== undefined) { + process.env.CLAUDE_CODE_OAUTH_TOKEN = savedToken31; + } else { + delete process.env.CLAUDE_CODE_OAUTH_TOKEN; + } + process.env.OLP_HOME = GLOBAL_OLP_HOME_31; + if (!server31) return; + await new Promise(r => server31.close(r)); + rmSync(TMP31, { recursive: true, force: true }); + }); + + beforeEach(async () => { + const mod = await import('./server.mjs'); + mod.__clearCache(); + }); + + it('31a — owner identity GETs /v0/management/status → 200 + full payload shape', async () => { + const { plaintext_token } = createKey({ + name: '31a-owner', owner_tier: 'owner', providers_enabled: '*', olpHome: TMP31, + }); + const r = await fetch({ + port: port31, method: 'GET', path: '/v0/management/status', + headers: { Authorization: `Bearer ${plaintext_token}` }, + }); + assert.equal(r.status, 200, `expected 200, got ${r.status}: ${r.body.slice(0, 200)}`); + const body = JSON.parse(r.body); + // Top-level shape per ADR 0010 § Phase 4 D63 spec. + assert.equal(body.ok, true); + assert.equal(typeof body.version, 'string'); + assert.equal(typeof body.uptime_ms, 'number'); + assert.ok(body.uptime_ms >= 0); + assert.equal(typeof body.uptime_human, 'string'); + assert.match(body.uptime_human, /^\d+h \d+m \d+s$/); + assert.equal(typeof body.started_at, 'string'); + assert.ok(body.started_at.includes('T')); + assert.ok(typeof body.providers === 'object' && body.providers !== null); + assert.equal(typeof body.providers.enabled, 'number'); + assert.equal(typeof body.providers.available, 'number'); + assert.ok(typeof body.providers.status === 'object'); + assert.ok(typeof body.stats === 'object'); + assert.equal(typeof body.stats.total_requests, 'number'); + assert.equal(typeof body.stats.active_requests, 'number'); + assert.ok(typeof body.stats.cache === 'object'); + assert.ok(Array.isArray(body.recent_errors)); + assert.equal(typeof body.generated_at, 'string'); + }); + + it('31b — non-owner (guest) → 401 owner_required', async () => { + const { plaintext_token } = createKey({ + name: '31b-guest', owner_tier: 'guest', providers_enabled: '*', olpHome: TMP31, + }); + const r = await fetch({ + port: port31, method: 'GET', path: '/v0/management/status', + headers: { Authorization: `Bearer ${plaintext_token}` }, + }); + assert.equal(r.status, 401); + const body = JSON.parse(r.body); + assert.equal(body.error.type, 'owner_required'); + }); + + it('31c — allow_anonymous=false + no header → 401 auth_required', async () => { + __setAuthConfig({ allow_anonymous: false, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + try { + const r = await fetch({ port: port31, method: 'GET', path: '/v0/management/status' }); + assert.equal(r.status, 401); + const body = JSON.parse(r.body); + assert.equal(body.error.type, 'auth_required'); + } finally { + __setAuthConfig({ allow_anonymous: true, owner_only_endpoints: [], fallback_detail_header_policy: 'all' }); + } + }); + + it('31d — payload recent_errors populated after a triggered chain-exhausted error', async () => { + __clearRecentErrors(); + // Install a provider that throws → triggers _pushError on the buffered path. + const fake = { + ...savedAnthropic31, + spawn: async function* (_ir, _ctx) { + throw new (await import('./lib/providers/base.mjs')).ProviderError( + 'fake spawn failure 31d', + 'SPAWN_FAILED', + ); + // eslint-disable-next-line no-unreachable + yield { type: 'stop', finish_reason: 'stop' }; + }, + }; + lp31.set('anthropic', fake); + + // Fire a non-streaming request → goes through executeWithFallback path + // → chain-exhausted → _pushError fires. + const trigger = await fetch({ + port: port31, method: 'POST', path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + body: { + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: '31d-trigger' }], + stream: false, + }, + }); + assert.ok(trigger.status === 502 || trigger.status === 500, + `expected 5xx error, got ${trigger.status}`); + + // Restore real provider so the /status healthCheck doesn't itself throw. + lp31.set('anthropic', savedAnthropic31); + + // Owner fetches /status — recent_errors must include our pushed entry. + const { plaintext_token } = createKey({ + name: '31d-owner', owner_tier: 'owner', providers_enabled: '*', olpHome: TMP31, + }); + const r = await fetch({ + port: port31, method: 'GET', path: '/v0/management/status', + headers: { Authorization: `Bearer ${plaintext_token}` }, + }); + assert.equal(r.status, 200); + const body = JSON.parse(r.body); + assert.ok(Array.isArray(body.recent_errors) && body.recent_errors.length >= 1, + `recent_errors expected to be populated; got ${JSON.stringify(body.recent_errors)}`); + const matchingEntry = body.recent_errors.find(e => + typeof e.message === 'string' && e.message.includes('fake spawn failure 31d'), + ); + assert.ok(matchingEntry, + `expected to find the triggered error in recent_errors; got ${JSON.stringify(body.recent_errors)}`); + assert.equal(matchingEntry.path, '/v1/chat/completions'); + assert.equal(matchingEntry.provider, 'anthropic'); + // total_requests must have advanced past 0 (we made at least 1 chat request). + assert.ok(body.stats.total_requests >= 1, + `total_requests expected >= 1, got ${body.stats.total_requests}`); + }); +});