mirror of
https://github.com/dtzp555-max/olp.git
synced 2026-07-19 09:45:07 +00:00
feat+test: D61+D62+D63 — SSE heartbeat + recentErrors[20] + /v0/management/status (#41)
* 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: [<ring>], 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 <noreply@anthropic.com>
* 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 <noreply@anthropic.com>
---------
Co-authored-by: dtzp555 <dtzp555@gmail.com>
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
dtzp555-max
+14
-2
@@ -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<string, boolean> }}
|
||||
* @returns {{ chains: object, soft_triggers: object, providersEnabled: Record<string, boolean>, 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 } };
|
||||
}
|
||||
}
|
||||
|
||||
+388
-12
@@ -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: <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.
|
||||
@@ -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');
|
||||
}
|
||||
|
||||
@@ -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}`);
|
||||
});
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user