diff --git a/lib/audit-query.mjs b/lib/audit-query.mjs index cae4703..0c7e8de 100644 --- a/lib/audit-query.mjs +++ b/lib/audit-query.mjs @@ -227,7 +227,7 @@ export function* readAuditWindow({ startMs, endMs, olpHome, logEvent } = {}) { * { * window: { startMs, endMs }, * request_count, status_2xx, status_4xx, status_5xx, - * by_provider: { [providerKey]: { count, cache_hit, cache_miss, cache_bypass, fallback_count } }, + * by_provider: { [providerKey]: { count, cache_hit, cache_miss, cache_bypass, cache_streaming_attached, fallback_count } }, * by_owner_tier: { owner: N, guest: N, anonymous: N }, * by_path: { '/v1/chat/completions': N, '/v1/models': N, ... }, * median_latency_ms, p95_latency_ms, @@ -276,12 +276,17 @@ export function aggregateRequests({ windowMs, olpHome, logEvent, _nowFn } = {}) // By provider if (typeof ev.provider === 'string' && ev.provider.length > 0) { const p = result.by_provider[ev.provider] ??= { - count: 0, cache_hit: 0, cache_miss: 0, cache_bypass: 0, fallback_count: 0, + count: 0, cache_hit: 0, cache_miss: 0, cache_bypass: 0, cache_streaming_attached: 0, fallback_count: 0, }; p.count++; if (ev.cache_status === 'hit') p.cache_hit++; else if (ev.cache_status === 'miss') p.cache_miss++; else if (ev.cache_status === 'bypass') p.cache_bypass++; + // D58 — ADR 0005 Amendment 8 §11 + lib/audit.mjs cache_status enum: streaming + // singleflight joiners (attached) share the source spawn but did not hit a + // cache. Tracked separately so `count` and `cache_hit + cache_miss + + // cache_bypass + cache_streaming_attached` reconcile. + else if (ev.cache_status === 'streaming_attached') p.cache_streaming_attached++; if (typeof ev.fallback_hops === 'number' && ev.fallback_hops > 0) p.fallback_count++; } @@ -443,7 +448,12 @@ export function spendTrendDaily({ days, olpHome, logEvent, _nowFn } = {}) { * `cacheStore.stats()` in server.mjs: that is the live in-process counter; * this is the audit-side rate scoped to the rolling window. * - * { window: { startMs, endMs }, total, hit, miss, bypass, hit_rate, by_provider } + * { window: { startMs, endMs }, total, hit, miss, bypass, streaming_attached, hit_rate, by_provider } + * + * `streaming_attached` (D58, ADR 0005 Amendment 8 §11): D58 streaming + * singleflight joiners did not hit a literal cache, so they are excluded + * from both numerator AND denominator of `hit_rate`. Tracked separately + * so the count reconciles with `total = hit + miss + bypass + streaming_attached`. * * @param {object} args * @param {number} args.windowMs @@ -459,18 +469,22 @@ export function cacheHitRateWindow({ windowMs, olpHome, logEvent, _nowFn } = {}) const startMs = now - windowMs; const endMs = now; - let total = 0, hit = 0, miss = 0, bypass = 0; + let total = 0, hit = 0, miss = 0, bypass = 0, streaming_attached = 0; const by_provider = {}; for (const ev of readAuditWindow({ startMs, endMs, olpHome, logEvent })) { if (ev.cache_status === null || ev.cache_status === undefined) continue; total++; const p = typeof ev.provider === 'string' && ev.provider.length > 0 ? ev.provider : '__unknown__'; - const pe = by_provider[p] ??= { total: 0, hit: 0, miss: 0, bypass: 0, hit_rate: 0 }; + const pe = by_provider[p] ??= { total: 0, hit: 0, miss: 0, bypass: 0, streaming_attached: 0, hit_rate: 0 }; pe.total++; if (ev.cache_status === 'hit') { hit++; pe.hit++; } else if (ev.cache_status === 'miss') { miss++; pe.miss++; } else if (ev.cache_status === 'bypass') { bypass++; pe.bypass++; } + // D58 — ADR 0005 Amendment 8 §11: streaming singleflight joiners. + // Excluded from hit_rate numerator + denominator (they did not hit a + // literal cache); tracked so `total` reconciles. + else if (ev.cache_status === 'streaming_attached') { streaming_attached++; pe.streaming_attached++; } } // Compute hit_rate per provider + overall (excludes bypass from denominator @@ -484,6 +498,6 @@ export function cacheHitRateWindow({ windowMs, olpHome, logEvent, _nowFn } = {}) return { window: { startMs, endMs }, - total, hit, miss, bypass, hit_rate, by_provider, + total, hit, miss, bypass, streaming_attached, hit_rate, by_provider, }; } diff --git a/lib/audit.mjs b/lib/audit.mjs index ea961e2..793f0d5 100644 --- a/lib/audit.mjs +++ b/lib/audit.mjs @@ -174,6 +174,20 @@ export function _maybeRotateAudit(args = {}) { * fallback_hops, tried_providers, error_code, ir_request_hash, chain_id). * Caller is responsible for populating fields; missing fields are * serialized as undefined → omitted by JSON.stringify. + * + * cache_status enum (free-form string; not schema-validated at append): + * 'hit' — served from cache (buffered-replay or streaming + * cache_hit role from ADR 0005 Amendment 8 §1). + * 'miss' — buffered or streaming source path; provider + * spawn fired for this request. + * 'bypass' — D2 cache_control bypass (no cache read/write). + * 'streaming_attached' — D58 / ADR 0005 Amendment 8 §11: client joined + * an in-flight streaming source spawn from + * another caller; this request did NOT spawn a + * provider but also did NOT hit the cache (the + * cache was empty at the inflight Map check). + * null — pre-chain error paths (the cache layer was + * never consulted; e.g. 401, 415, 400 IR). * @param {object} [opts] * @param {string} [opts.olpHome] - test override; defaults to ~/.olp * @param {(level: string, event: string, data?: object) => void} [opts.logEvent] diff --git a/server.mjs b/server.mjs index a71c570..ffc1b7b 100644 --- a/server.mjs +++ b/server.mjs @@ -1117,32 +1117,21 @@ async function handleChatCompletions(req, res) { // truncate the response (end with no [DONE]). On error before any chunk: // throw so the outer handler can surface a clean error (no bytes written). // - // On success: write chunks to res AND cache so subsequent identical requests - // hit the burst-replay path. - // D38 (issue #1): pre-acquire a concurrency slot for the streaming path so - // saturation here behaves identically to saturation on the buffered path. - // If acquire fails, fall through (skip this branch) — the buffered path - // below also gates via tryAcquireSpawn and will surface a chain-exhausted - // error through executeWithFallback (correct behaviour: a single-hop chain - // at maxConcurrent has no other hop to advance to). + // On success: chunks are written to res, and the cache layer (via + // getOrComputeStreaming) writes accumulated chunks on source completion. // - // Acquired here, released in the `finally` below. The gate intentionally - // lives BEFORE the streaming-branch entry check so the buffered fallthrough - // can re-attempt acquire from a clean slate. - // TODO(v1.x — ADR 0005 Amendment 8 / issue #16): replace this peek+spawn - // pattern with cacheStore.getOrComputeStreaming(...) to close the TOCTOU - // window between line ~782 peek and the spawn at line ~846, and to make N - // concurrent identical streaming requests share one spawn (tee-streaming). - // Design ratified in D42; see docs/v1x-roadmap.md #1 for the trigger and - // acceptance criteria. DO NOT remove this comment until the v1.x impl lands. - let streamingAcquired = false; + // D58 — ADR 0005 Amendment 8 §§7, 11, 12: streaming singleflight via + // cacheStore.getOrComputeStreaming(...). The peek+spawn TOCTOU window + // (server.mjs:1104 peek vs. spawn) is collapsed by the synchronous Map + // check+insert in the cache layer (D57). The D38 tryAcquireSpawn gate now + // lives INSIDE the sourceFactory closure: only the first caller acquires a + // slot; attached joiners share it. On factory-throw CONCURRENCY_LIMIT, the + // cache layer rejects → we fall through to the buffered path (same outcome + // as today). preCheckHit gates entry exactly as before — the cache_hit + // outcome from getOrComputeStreaming is the TTL-race safety net; in + // practice it should not fire because preCheckHit excludes alive entries. + // See docs/adr/0005-cache-cross-provider.md Amendment 8 + issue #16. if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop) { - const candidatePlugin = loadedProviders.get(chain[0].provider); - const candidateMax = candidatePlugin?.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS; - streamingAcquired = candidatePlugin ? tryAcquireSpawn(chain[0].provider, candidateMax) : false; - } - - if (ir.stream && chain.length === 1 && !bypassCacheForFirstHop && !preCheckHit && cacheableForFirstHop && streamingAcquired) { const streamProvider = chain[0].provider; const streamModel = chain[0].model; const streamCacheKey = computeCacheKey(streamProvider, streamModel, ir); @@ -1150,63 +1139,214 @@ async function handleChatCompletions(req, res) { if (!streamPlugin) { // Provider disappeared between chain build and here (edge case). - // Release the slot we acquired above so the counter stays balanced. auditCtx.provider = streamProvider; auditCtx.error_code = 'no_enabled_provider'; - releaseSpawn(streamProvider); return sendError(res, 503, `Provider ${streamProvider} is not enabled`, 'no_enabled_provider', olpErrorHeaders({ startMs, model: ir.model })); } - // D45 fold-in P1: populate audit ctx for the real-streaming path. Each - // exit below (success, error-after-first-chunk, pre-first-chunk-error, - // 503 above) leaves these fields representing the streaming attempt. - // Error paths amend `error_code`; success leaves it null. - auditCtx.provider = streamProvider; - auditCtx.tried_providers = [streamProvider]; - auditCtx.cache_status = 'miss'; + // D58 — ADR 0005 Amendment 8 §7: sourceFactory wraps acquire + spawn. + // Acquire fires ONLY for the first caller; attached clients share the + // slot. Release fires EXACTLY ONCE when the source iterator's + // try/finally executes (normal exhaustion, error, or iterator.return() + // propagated from sourceAbortController via the cache layer). + const candidateMax = streamPlugin.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS; + const sourceFactory = () => { + if (!tryAcquireSpawn(streamProvider, candidateMax)) { + const err = new ProviderError( + `provider ${streamProvider} at maxConcurrent (${candidateMax})`, + 'CONCURRENCY_LIMIT', + ); + err.providerName = streamProvider; + err.maxConcurrent = candidateMax; + err.activeSpawns = getActiveSpawnCount(streamProvider); + throw err; + } + // Wrap streamPlugin.spawn in an async generator that guarantees + // releaseSpawn fires exactly once in finally — regardless of normal + // exhaustion, mid-stream throw, or iterator.return() from cache-layer + // sourceAbortController propagation (§9). + return (async function* sourceWithRelease() { + try { + for await (const irChunk of streamPlugin.spawn(ir, authContext)) { + yield irChunk; + } + } finally { + releaseSpawn(streamProvider); + } + })(); + }; - const streamHeaders = olpHeaders({ - providerUsed: streamProvider, - modelUsed: streamModel, - startMs, - cacheStatus: 'miss', - fallbackHops: 0, - }); - - // D14: writeHead is deferred until just before the first res.write so that - // pre-first-chunk errors can still produce a JSON 502 (matching the buffered - // path). Calling writeHead unconditionally here was the D14 defect. - - const streamedChunks = []; - let firstChunkEmitted = false; + // D58 — invoke the cache-layer coordinator. CONCURRENCY_LIMIT from the + // factory falls through to the buffered path (the buffered fallback + // engine re-attempts acquire). Any other pre-stream error surfaces 502. + let stream; + let role; try { - for await (const irChunk of streamPlugin.spawn(ir, authContext)) { - if (irChunk.type === 'error') { - // Error chunk from provider - if (firstChunkEmitted) { - // Past first-chunk boundary — can't fallback; emit truncation marker + [DONE] - // so clients can detect the incomplete response in-band (aligns D26 F19 - // stop-less exhaustion behaviour and the catch-block fix in D35 #10). - logEvent('warn', 'streaming_error_after_first_chunk', { - provider: streamProvider, - model: streamModel, - error: irChunk.error, + const result = await cacheStore.getOrComputeStreaming( + keyId, + streamCacheKey, + sourceFactory, + { clientId: requestId }, + ); + stream = result.stream; + role = result.role; + } catch (e) { + if (e instanceof ProviderError && e.code === 'CONCURRENCY_LIMIT') { + // Fall through to the buffered path — the existing executeHopFn / + // executeWithFallback machinery re-attempts acquire and surfaces a + // chain-exhausted 502 for a single-hop chain at maxConcurrent, which + // matches today's pre-D58 behaviour for this saturation scenario. + logEvent('debug', 'streaming_concurrency_limit_fallthrough', { + provider: streamProvider, + model: streamModel, + }); + // (proceed past this block) + } else { + // Any other pre-first-chunk error — surface a clean 502. + auditCtx.provider = streamProvider; + auditCtx.tried_providers = [streamProvider]; + auditCtx.cache_status = 'miss'; + auditCtx.error_code = e?.code ?? 'provider_error'; + logEvent('error', 'streaming_error_before_first_chunk', { + provider: streamProvider, + model: streamModel, + error: e.message, + }); + return sendError(res, 502, e.message ?? 'Provider error', 'provider_error', + olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 })); + } + } + + if (stream) { + // D58 — ADR 0005 Amendment 8 §11: cache_status reflects role-of-this- + // client at attach-time. `source` did real work (miss); `attached` + // shared a live spawn (streaming_attached); `cache_hit` served from + // cache (hit). See lib/audit.mjs cache_status JSDoc for the enum. + auditCtx.provider = streamProvider; + auditCtx.tried_providers = [streamProvider]; + if (role === 'cache_hit') { + auditCtx.cache_status = 'hit'; + } else if (role === 'attached') { + auditCtx.cache_status = 'streaming_attached'; + } else { + auditCtx.cache_status = 'miss'; + } + + // D58 — ADR 0005 Amendment 8 §9: wire HTTP req close → iterator + // return() so a client disconnect propagates into the cache layer. + // Without this the cache layer never learns of disconnect and the + // source spawn may orphan (last-client-gone abort would not fire). + // try/catch guards against concurrent return() throws (Node async + // generators can throw if return() is invoked while iteration is + // pending). The cache layer's iterator return() is idempotent. + // D58 — ADR 0005 Amendment 8 §9: detect client disconnect via res + // 'close' event (not req 'close'). For SSE responses Node emits + // res.on('close') when the underlying socket goes away while the + // response is still streaming. req.on('close') in Node 18+ fires only + // after the response is fully sent, which is too late for our abort + // propagation needs. Without this wire-up the cache layer would never + // learn of disconnect → potential orphan spawns. + const onClose = () => { + try { stream.return?.(); } catch { /* best-effort */ } + }; + res.on('close', onClose); + + // D58 — ADR 0005 Amendment 8 §11: X-OLP-Streaming-Inflight header. + // role === 'source' → 'source' (first caller; may upgrade to + // 'solo' post-stream if no joiners ever + // attached — see deferral note below). + // role === 'attached' → 'attached' + // role === 'cache_hit'→ omitted; the existing X-OLP-Cache: hit + // already signals that path. + // 'solo' (no joiners ever attached) is observable only post-stream — + // emitted via streaming_inflight_source_done log event with + // attached_count=0; the wire header reports `source` initially and + // does NOT flip mid-stream. Future ADR may expose it on a trailer. + const cacheStatusForHeader = (role === 'cache_hit') ? 'hit' : 'miss'; + const streamHeaders = olpHeaders({ + providerUsed: streamProvider, + modelUsed: streamModel, + startMs, + cacheStatus: cacheStatusForHeader, + fallbackHops: 0, + }); + if (role === 'source' || role === 'attached') { + streamHeaders['X-OLP-Streaming-Inflight'] = role; + } + + // D14: writeHead is deferred until just before the first res.write so + // that pre-first-chunk errors can still produce a JSON 502 (matching + // the buffered path). Calling writeHead unconditionally was the D14 + // defect. + + const streamedChunks = []; + let firstChunkEmitted = false; + try { + for await (const irChunk of stream) { + if (irChunk.type === 'error') { + // Error chunk from provider + if (firstChunkEmitted) { + // Past first-chunk boundary — emit truncation marker + [DONE]. + logEvent('warn', 'streaming_error_after_first_chunk', { + provider: streamProvider, + model: streamModel, + error: irChunk.error, + }); + auditCtx.error_code = 'streaming_error_after_first_chunk'; + res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model)); + res.write(SSE_DONE); + res.end(); + return; + } + throw new ProviderError(irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED'); + } + + if (!res.headersSent) { + res.writeHead(200, { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + ...streamHeaders, }); - auditCtx.error_code = 'streaming_error_after_first_chunk'; - res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model)); + } + streamedChunks.push(irChunk); + res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model)); + firstChunkEmitted = true; + + if (irChunk.type === 'stop') { res.write(SSE_DONE); res.end(); + // D58 — ADR 0005 Amendment 8 §4: cache write is performed by the + // cache layer's tee task on source completion. Server no longer + // calls cacheStore.set from this branch; the cache layer applies + // the same write conditions (truncated-not-cached, cacheable:false + // opt-out, replay-cap, maxEntryBytes cap) internally. return; } - // No bytes written yet — throw to surface a clean error. - // auditCtx.error_code is set by the downstream catch handler (the - // outer streaming-path catch block fills it from the thrown error). - throw new ProviderError(irChunk.error ?? 'Provider emitted error chunk', 'SPAWN_FAILED'); } - // Defer writeHead until the moment we are about to emit the first byte. - // After this point firstChunkEmitted===true ↔ res.headersSent===true. + // Generator exhausted without a stop chunk — emit truncation marker + // + [DONE]. D58 — the cache layer (per ADR 0005 Amendment 8 §4) writes + // accumulatedChunks on normal source exhaustion regardless of stop- + // chunk semantics; IR-level stop-chunk inspection is the server's + // responsibility. Mirror D16 truncation-eviction: the source role + // explicitly deletes the just-written entry so subsequent identical + // requests respawn (matches D16 buffered-path truncation behaviour + // and ADR 0005 § "Cache write conditions" item 1). Only the source + // role performs the delete; attached clients did not trigger the + // write. Idempotent across concurrent source-truncation observers. + if (streamedChunks.length > 0) { + const truncMarker = { type: 'stop', finish_reason: 'length' }; + res.write(irChunkToOpenAISSE(truncMarker, requestId, ir.model)); + } + if (role === 'source' && cacheableForFirstHop) { + cacheStore.delete(keyId, streamCacheKey); + } + + // D35 #9: zero-chunk empty-stream — writeHead deferred (no chunks + // yielded), emit headers + [DONE] so the response is still valid SSE. if (!res.headersSent) { res.writeHead(200, { 'Content-Type': 'text/event-stream', @@ -1216,114 +1356,51 @@ async function handleChatCompletions(req, res) { ...streamHeaders, }); } - streamedChunks.push(irChunk); - res.write(irChunkToOpenAISSE(irChunk, requestId, ir.model)); - firstChunkEmitted = true; - if (irChunk.type === 'stop') { - res.write(SSE_DONE); - res.end(); - // Cache the buffered chunks for burst-replay on subsequent identical requests. - // D23 defense-in-depth: cacheableForFirstHop is true here (cacheable: false - // falls through to the buffered path, never enters this block), but the guard - // makes the intent explicit and survives future refactors. - if (cacheableForFirstHop) { - await cacheStore.set(keyId, streamCacheKey, streamedChunks); - logEvent('info', 'streaming_response_cached', { - provider: streamProvider, - model: streamModel, - chunks: streamedChunks.length, - }); - } - return; - } - } - - // Generator exhausted without a stop chunk — emit [DONE] but do NOT cache. - // A stop-less exhaustion means the response is truncated (the generator - // ended without the model signalling completion). Caching a truncated - // response would serve wrong answers to future identical requests. - // Compare: D16's buffered-path truncation eviction explicitly avoids - // persisting truncated entries for the same reason. - // - // D26 round-3 F19: emit a synthetic truncation marker BEFORE [DONE] so - // clients can detect the incomplete response in-band. Only emit when there - // is actual partial content (streamedChunks.length > 0) — emitting a - // truncation marker on an empty response is misleading. - // The buffered D16 path synthesizes {type:'stop', finish_reason:'length'} - // before returning; this aligns the streaming branch with that behaviour. - if (streamedChunks.length > 0) { - const truncMarker = { type: 'stop', finish_reason: 'length' }; - res.write(irChunkToOpenAISSE(truncMarker, requestId, ir.model)); - } - - // D35 #9: Zero-chunk empty-stream path — writeHead is still deferred - // (firstChunkEmitted===false) when the generator yields no chunks at all and - // exits cleanly. Without an explicit writeHead Node auto-emits 200 with the - // default Content-Type and none of the X-OLP-* headers. - // A provider that yielded nothing still constitutes an attempted call, so we - // emit the full olpHeaders (provider WAS attempted, just yielded zero chunks). - if (!res.headersSent) { - res.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - Connection: 'keep-alive', - 'X-Accel-Buffering': 'no', - ...streamHeaders, - }); - } - - res.write(SSE_DONE); - res.end(); - // Loop exhausted without stop chunk = truncation. The stop-chunk completion - // path returns earlier (above, inside the for-await loop); reaching here means - // the generator returned without emitting stop. Never cache (per ADR 0005 cache - // write conditions item 1: truncated responses must not persist in cache). - if (streamedChunks.length > 0 && cacheableForFirstHop) { - logEvent('warn', 'streaming_no_stop_chunk', { - chunks_count: streamedChunks.length, - provider: streamProvider, - model: streamModel, - }); - } - } catch (e) { - if (firstChunkEmitted) { - // Past first-chunk boundary — can't fallback; emit truncation marker + [DONE] - // so clients can detect the incomplete response in-band (aligns with D26 F19 - // stop-less exhaustion behaviour). ADR 0004 § Fallback safety: no fallback - // after first-chunk boundary; truncation is the correct recovery. - logEvent('warn', 'streaming_error_after_first_chunk', { - provider: streamProvider, - model: streamModel, - error: e.message, - }); - res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model)); res.write(SSE_DONE); res.end(); - } else { - // No bytes written — surface a clean JSON error. - logEvent('error', 'streaming_error_before_first_chunk', { - provider: streamProvider, - model: streamModel, - error: e.message, - }); - auditCtx.error_code = e?.code ?? 'provider_error'; - if (!res.headersSent) { - sendError(res, 502, e.message ?? 'Provider error', 'provider_error', - olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 })); - } else { - res.end(); + if (streamedChunks.length > 0 && cacheableForFirstHop) { + logEvent('warn', 'streaming_no_stop_chunk', { + chunks_count: streamedChunks.length, + provider: streamProvider, + model: streamModel, + }); } + } catch (e) { + if (firstChunkEmitted) { + logEvent('warn', 'streaming_error_after_first_chunk', { + provider: streamProvider, + model: streamModel, + error: e.message, + }); + res.write(irChunkToOpenAISSE({ type: 'stop', finish_reason: 'length' }, requestId, ir.model)); + res.write(SSE_DONE); + res.end(); + } else { + // No bytes written — surface a clean JSON error. + logEvent('error', 'streaming_error_before_first_chunk', { + provider: streamProvider, + model: streamModel, + error: e.message, + }); + auditCtx.error_code = e?.code ?? 'provider_error'; + if (!res.headersSent) { + sendError(res, 502, e.message ?? 'Provider error', 'provider_error', + olpHeaders({ providerUsed: streamProvider, modelUsed: streamModel, startMs, cacheStatus: 'miss', fallbackHops: 0 })); + } else { + res.end(); + } + } + } finally { + // D58 — releaseSpawn lives inside sourceFactory's finally (only the + // source caller acquired). The req-close listener is detached here + // so it doesn't leak past the response lifetime. + res.removeListener('close', onClose); } - } finally { - // D38 (issue #1): streaming spawn lifecycle ended — drain completed, - // stop chunk seen, generator exhausted without stop, or any catch - // path returned via res.end(). Release the slot acquired before - // entering the streaming branch. The finally fires on every JS exit - // path including the `return;` statements inside the try/catch body. - releaseSpawn(streamProvider); + return; } - return; + // CONCURRENCY_LIMIT fallthrough — flow continues into the buffered path + // below. } let fallbackResult; diff --git a/test-features.mjs b/test-features.mjs index 4a2dd6e..2642f61d 100644 --- a/test-features.mjs +++ b/test-features.mjs @@ -9,7 +9,7 @@ * D5 adds: Suite 9 (cache layer unit + HTTP integration), Suite 10 (Anthropic E2E gated) */ -import { describe, it, before, after } from 'node:test'; +import { describe, it, before, after, beforeEach } from 'node:test'; import assert from 'node:assert/strict'; import { request as httpRequest } from 'node:http'; import { EventEmitter } from 'node:events'; @@ -12458,3 +12458,473 @@ describe('Suite 27 — D57 streaming singleflight (cache layer)', () => { assert.equal(store.stats('key-B').misses, 1); }); }); + +// ── Suite 28: D58 streaming singleflight (server.mjs HTTP wiring) ────────── +// +// Authority: ADR 0005 Amendment 8 §§ 7, 8, 9, 11, 12 (issue #16, D58). +// +// HTTP-layer integration tests for the server.mjs streaming branch wired to +// cacheStore.getOrComputeStreaming(). D57 (cache layer) tested singleflight +// in unit form; D58 verifies the server actually plumbs it through and emits +// the new X-OLP-Streaming-Inflight header. Provider spawn is mocked at the +// plugin level via lp.set('anthropic', { ...real, spawn: fake }) — the +// pattern used by Suites 15d/15e — so no real CLI is invoked. +// +// Test count: 8 (28a single, 28b 2-concurrent join, 28c TOCTOU pre-cache, +// 28d mid-stream join behaviour-validated via parallel HTTP, 28e omitted in +// favour of Suite 27g unit coverage per D58 prompt, 28f one-of-N disconnect, +// 28g all-disconnect, 28h CONCURRENCY_LIMIT fallthrough). + +describe('Suite 28 — D58 streaming singleflight (server.mjs HTTP wiring, ADR 0005 Amendment 8)', () => { + let server28; + let port28; + let savedToken28; + let lp28; + let savedAnthropic28; + + before(async () => { + savedToken28 = process.env.CLAUDE_CODE_OAUTH_TOKEN; + process.env.CLAUDE_CODE_OAUTH_TOKEN = 'fake-token-suite-28'; + __setProvidersEnabled({ anthropic: true }); + + const mod = await import('./server.mjs'); + lp28 = mod.loadedProviders; + savedAnthropic28 = lp28.get('anthropic'); + mod.__clearCache(); + + server28 = mod.createOlpServer(); + await new Promise((resolve, reject) => { + server28.listen(0, '127.0.0.1', resolve); + server28.once('error', reject); + }); + port28 = server28.address().port; + }); + + after(async () => { + // Restore original anthropic provider so other suites are unaffected. + if (savedAnthropic28 !== undefined) { + lp28.set('anthropic', savedAnthropic28); + } else { + lp28.delete('anthropic'); + } + __resetProvidersEnabled(); + __resetSpawnImpl(); + if (savedToken28 !== undefined) { + process.env.CLAUDE_CODE_OAUTH_TOKEN = savedToken28; + } else { + delete process.env.CLAUDE_CODE_OAUTH_TOKEN; + } + if (!server28) return; + return new Promise(r => server28.close(r)); + }); + + /** + * Install a custom spawn async generator on the anthropic plugin for one + * test. `factory` is a function returning the async generator each spawn. + * Tracks invocation count via spawnCount. + */ + function installFakeStreamProvider(spawnImpl) { + const counter = { count: 0 }; + const fake = { + ...savedAnthropic28, + spawn: async function* (ir, authContext) { + counter.count++; + yield* spawnImpl(ir, authContext); + }, + }; + lp28.set('anthropic', fake); + return counter; + } + + /** + * Fire an SSE request and collect the full body + headers. The promise + * resolves when the server ends the response (res 'end' event). + */ + function makeStreamRequest(extra = {}) { + return new Promise((resolve, reject) => { + const req = httpRequest({ + hostname: '127.0.0.1', + port: port28, + method: 'POST', + path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + }, res => { + let data = ''; + res.on('data', d => { data += d.toString(); }); + res.on('end', () => resolve({ + status: res.statusCode, + body: data, + headers: res.headers, + })); + res.on('error', reject); + }); + req.on('error', reject); + req.write(JSON.stringify({ + model: extra.model ?? 'claude-sonnet-4-6', + messages: [{ role: 'user', content: extra.prompt ?? 'd58-default' }], + stream: true, + })); + req.end(); + }); + } + + /** + * Fire an SSE request and abort it after `abortAfterMs` ms. Returns the + * partial body collected up to abort. Resolves on the abort event. + */ + function makeAbortableStreamRequest({ prompt, abortAfterMs }) { + return new Promise((resolve, reject) => { + const req = httpRequest({ + hostname: '127.0.0.1', + port: port28, + method: 'POST', + path: '/v1/chat/completions', + headers: { 'Content-Type': 'application/json' }, + }, res => { + let data = ''; + res.on('data', d => { data += d.toString(); }); + res.on('end', () => resolve({ + status: res.statusCode, + body: data, + headers: res.headers, + aborted: false, + })); + res.on('error', () => resolve({ + status: res.statusCode, + body: data, + headers: res.headers, + aborted: true, + })); + }); + req.on('error', () => resolve({ aborted: true })); + req.write(JSON.stringify({ + model: 'claude-sonnet-4-6', + messages: [{ role: 'user', content: prompt }], + stream: true, + })); + req.end(); + setTimeout(() => { + try { req.destroy(); } catch { /* ignore */ } + }, abortAfterMs); + }); + } + + beforeEach(async () => { + const mod = await import('./server.mjs'); + mod.__clearCache(); + }); + + it('28a — single SSE request: X-OLP-Streaming-Inflight: source; cache populated; identical re-request → cache hit', async () => { + // D58 — ADR 0005 Amendment 8 §11: single client gets `source` role, + // X-OLP-Cache: miss. Subsequent identical request hits the cache. + const counter = installFakeStreamProvider(async function* (_ir) { + yield { type: 'delta', content: 'chunk-A' }; + yield { type: 'delta', content: 'chunk-B' }; + yield { type: 'stop', finish_reason: 'stop' }; + }); + + const r1 = await makeStreamRequest({ prompt: 'd58-28a' }); + assert.equal(r1.status, 200, `r1 status ${r1.status}: ${r1.body.slice(0, 200)}`); + assert.equal(r1.headers['x-olp-streaming-inflight'], 'source', + 'r1 must emit X-OLP-Streaming-Inflight: source'); + assert.equal(r1.headers['x-olp-cache'], 'miss', + 'r1 must be X-OLP-Cache: miss'); + assert.ok(r1.body.includes('chunk-A'), 'r1 body must include chunk-A'); + assert.ok(r1.body.includes('chunk-B'), 'r1 body must include chunk-B'); + assert.ok(r1.body.includes('[DONE]'), 'r1 body must end with [DONE]'); + assert.equal(counter.count, 1, 'exactly one spawn for r1'); + + // Second identical request — cache hit, no spawn. + const r2 = await makeStreamRequest({ prompt: 'd58-28a' }); + assert.equal(r2.status, 200); + assert.equal(r2.headers['x-olp-cache'], 'hit', 'r2 must be cache hit'); + // X-OLP-Streaming-Inflight is omitted on cache_hit (X-OLP-Cache: hit + // already signals that path per D58 design). + assert.equal(r2.headers['x-olp-streaming-inflight'], undefined, + 'cache_hit path must omit X-OLP-Streaming-Inflight header'); + assert.equal(counter.count, 1, 'no additional spawn for r2'); + }); + + it('28b — 2 concurrent identical SSE: one spawn; first=source second=attached; identical chunks', async () => { + // D58 — ADR 0005 Amendment 8 §1, §11: two concurrent identical streams + // share one underlying spawn. First gets X-OLP-Streaming-Inflight: + // source; second gets attached. Both bodies contain the same chunks. + // + // Pacing: source yields chunks with a setTimeout gap so the second + // request has time to fire and attach mid-stream. + const counter = installFakeStreamProvider(async function* (_ir) { + // Slow pacing so the second request can attach before completion. + for (const t of ['c0', 'c1', 'c2', 'c3']) { + yield { type: 'delta', content: t }; + await new Promise(r => setTimeout(r, 20)); + } + yield { type: 'stop', finish_reason: 'stop' }; + }); + + // Fire request 1, then a few ms later fire request 2; both run to + // completion in parallel. + const p1 = makeStreamRequest({ prompt: 'd58-28b' }); + // Give p1 a head start to register the inflight entry. + await new Promise(r => setTimeout(r, 10)); + const p2 = makeStreamRequest({ prompt: 'd58-28b' }); + + const [r1, r2] = await Promise.all([p1, p2]); + assert.equal(r1.status, 200); + assert.equal(r2.status, 200); + assert.equal(counter.count, 1, 'only one underlying spawn for both requests'); + + // First caller (registered the inflight entry) gets source; the other + // gets attached. The order is set by who hit getOrComputeStreaming + // first — we asserted that with the 10ms head start above. + assert.equal(r1.headers['x-olp-streaming-inflight'], 'source', + 'r1 must be source role'); + assert.equal(r2.headers['x-olp-streaming-inflight'], 'attached', + 'r2 must be attached role'); + + // Both responses must contain all 4 delta chunks + [DONE]. + for (const tag of ['c0', 'c1', 'c2', 'c3', '[DONE]']) { + assert.ok(r1.body.includes(tag), `r1 missing ${tag}`); + assert.ok(r2.body.includes(tag), `r2 missing ${tag}`); + } + + // Cache_status header: source=miss, attached=hit (cache_hit role)? + // No — attached client did NOT hit the cache (cache was empty). The + // server distinguishes by setting cache_status header to 'miss' for + // attached and 'miss' for source. Verify both are 'miss' on the wire + // (X-OLP-Cache header). cache_status='streaming_attached' is the AUDIT + // value, not the wire header value. + assert.equal(r1.headers['x-olp-cache'], 'miss'); + assert.equal(r2.headers['x-olp-cache'], 'miss'); + }); + + it('28c — TOCTOU regression: pre-populated cache + 2 concurrent identical streams → both cache_hit, no spawn', async () => { + // D58 — ADR 0005 Amendment 8 §6 / §11: when the cache is already + // populated, preCheckHit gates entry to the streaming-singleflight + // branch. The streaming-singleflight branch should not run at all. + // 2 concurrent identical requests both replay from cache; spawn count + // stays at 0. + const counter = installFakeStreamProvider(async function* (_ir) { + yield { type: 'delta', content: 'should-not-fire' }; + yield { type: 'stop', finish_reason: 'stop' }; + }); + + // Populate the cache via a first request. + const r0 = await makeStreamRequest({ prompt: 'd58-28c' }); + assert.equal(r0.status, 200); + assert.equal(counter.count, 1, 'r0 spawned once to populate cache'); + + // Now fire 2 concurrent identical requests — both should be cache hits. + const [r1, r2] = await Promise.all([ + makeStreamRequest({ prompt: 'd58-28c' }), + makeStreamRequest({ prompt: 'd58-28c' }), + ]); + assert.equal(r1.status, 200); + assert.equal(r2.status, 200); + assert.equal(r1.headers['x-olp-cache'], 'hit', 'r1 must be cache hit'); + assert.equal(r2.headers['x-olp-cache'], 'hit', 'r2 must be cache hit'); + // Neither should have entered the streaming-singleflight branch. + assert.equal(r1.headers['x-olp-streaming-inflight'], undefined, + 'r1 must not emit X-OLP-Streaming-Inflight (served from buffered cache replay)'); + assert.equal(r2.headers['x-olp-streaming-inflight'], undefined, + 'r2 must not emit X-OLP-Streaming-Inflight (served from buffered cache replay)'); + assert.equal(counter.count, 1, 'no additional spawns'); + }); + + it('28d — mid-stream join (HTTP-level): 2 concurrent SSE requests share one spawn; both bodies identical', async () => { + // D58 — ADR 0005 Amendment 8 §5: late joiner gets accumulated burst + + // live tail. At the HTTP level we can't observe the burst-vs-live split + // precisely (it's internal to the cache layer), but we can verify the + // end-to-end invariant: both clients receive the same chunk sequence + // even when one joins mid-source. Suite 27c covers the burst-vs-live + // split at the cache-layer unit level. + const counter = installFakeStreamProvider(async function* (_ir) { + // Slower pacing so the second client clearly joins mid-stream. + for (const t of ['m0', 'm1', 'm2', 'm3', 'm4']) { + yield { type: 'delta', content: t }; + await new Promise(r => setTimeout(r, 25)); + } + yield { type: 'stop', finish_reason: 'stop' }; + }); + + const p1 = makeStreamRequest({ prompt: 'd58-28d' }); + // p2 joins ~40ms in — at least 2 chunks have been accumulated by then. + await new Promise(r => setTimeout(r, 40)); + const p2 = makeStreamRequest({ prompt: 'd58-28d' }); + + const [r1, r2] = await Promise.all([p1, p2]); + assert.equal(r1.status, 200); + assert.equal(r2.status, 200); + assert.equal(counter.count, 1, 'exactly one underlying spawn'); + assert.equal(r1.headers['x-olp-streaming-inflight'], 'source'); + assert.equal(r2.headers['x-olp-streaming-inflight'], 'attached'); + for (const tag of ['m0', 'm1', 'm2', 'm3', 'm4']) { + assert.ok(r1.body.includes(tag), `r1 missing ${tag}`); + assert.ok(r2.body.includes(tag), `r2 missing ${tag} (late-joiner replay must include burst)`); + } + }); + + it('28f — one of N clients disconnects mid-stream: source NOT aborted; other client completes; cache populated', async () => { + // D58 — ADR 0005 Amendment 8 §9: when one of N attached clients drops, + // the source continues for the remaining clients. The cache write + // happens because the source completed normally for the surviving + // client. + const counter = installFakeStreamProvider(async function* (_ir) { + for (const t of ['s0', 's1', 's2', 's3', 's4', 's5']) { + yield { type: 'delta', content: t }; + await new Promise(r => setTimeout(r, 25)); + } + yield { type: 'stop', finish_reason: 'stop' }; + }); + + // r1 starts and registers the inflight entry. + const p1 = makeStreamRequest({ prompt: 'd58-28f' }); + await new Promise(r => setTimeout(r, 15)); + // r2 attaches but aborts after ~40ms. + const p2 = makeAbortableStreamRequest({ prompt: 'd58-28f', abortAfterMs: 40 }); + + const [r1, r2] = await Promise.all([p1, p2]); + // r1 (the source) completes normally with the full stream. + assert.equal(r1.status, 200); + assert.equal(r1.headers['x-olp-streaming-inflight'], 'source'); + for (const tag of ['s0', 's1', 's2', 's3', 's4', 's5', '[DONE]']) { + assert.ok(r1.body.includes(tag), `r1 missing ${tag}`); + } + assert.equal(counter.count, 1, 'source spawn was NOT re-fired by the disconnect'); + + // r2 was aborted — partial body is okay; what matters is the source + // wasn't killed (verified above by r1 completing). + // Subsequent identical request must be a cache hit (cache was populated + // by the source on normal completion). + const r3 = await makeStreamRequest({ prompt: 'd58-28f' }); + assert.equal(r3.status, 200); + assert.equal(r3.headers['x-olp-cache'], 'hit', 'cache populated after source completion'); + assert.equal(counter.count, 1, 'no additional spawn for r3'); + }); + + it('28g — ALL clients disconnect mid-stream: source aborted; no cache write; subsequent request respawns', async () => { + // D58 — ADR 0005 Amendment 8 §9 + §4: when all attached clients + // disconnect, the cache layer fires sourceAbortController.abort() and + // does NOT write the cache. A subsequent identical request must spawn + // afresh (no inflight entry left dangling). + // + // Implementation detail: the fake spawn's async generator must respect + // the abort signal (or simply have its iterator.return() called when + // the spawn is iterated by the tee task — see Suite 27 unit tests). + // For an async generator with `await new Promise(setTimeout)` between + // yields, calling .return() naturally propagates because the for-await + // exits cleanly via the try/finally. + let sourceFinished = false; + const counter = installFakeStreamProvider(async function* (_ir) { + try { + for (let i = 0; i < 30; i++) { + yield { type: 'delta', content: `g${i}` }; + await new Promise(r => setTimeout(r, 20)); + } + yield { type: 'stop', finish_reason: 'stop' }; + sourceFinished = true; + } finally { + // intentionally no-op; the "finished without abort" signal lives in + // sourceFinished above. + } + }); + + // Only one client; abort it after ~40ms (well before completion). + const r1 = await makeAbortableStreamRequest({ prompt: 'd58-28g', abortAfterMs: 40 }); + assert.equal(counter.count, 1, 'spawn fired once'); + // Wait for the cache layer to observe attachedClients.size === 0 and abort. + await new Promise(r => setTimeout(r, 100)); + assert.equal(sourceFinished, false, 'source was aborted, not completed'); + + // Subsequent identical request must respawn (no cache, no inflight). + const r2 = await makeStreamRequest({ prompt: 'd58-28g' }); + assert.equal(r2.status, 200); + assert.equal(r2.headers['x-olp-cache'], 'miss', 'no cache write after abort'); + assert.equal(r2.headers['x-olp-streaming-inflight'], 'source', 'fresh source role'); + assert.equal(counter.count, 2, 'r2 triggered a fresh spawn'); + }); + + it('28h — CONCURRENCY_LIMIT fallthrough: different cacheKeys at maxConcurrent=1 → first succeeds; second falls through to buffered path', async () => { + // D58 — ADR 0005 Amendment 8 §7: CONCURRENCY_LIMIT thrown by + // sourceFactory falls through to the buffered path. Different + // cacheKeys do NOT share an inflight entry (they're not singleflight + // candidates), so request 2's sourceFactory throws and the streaming + // branch is bypassed for that request. The buffered path then + // re-attempts acquire; since maxConcurrent=1 and request 1 still has + // the slot, it surfaces a chain-exhausted 502 (single-hop saturation). + // + // We install a custom provider with maxConcurrent=1 + slow stream to + // hold the slot while request 2 fires. + let releaseSig; + const releaseGate = new Promise(r => { releaseSig = r; }); + const counter = { count: 0 }; + const fake = { + ...savedAnthropic28, + hints: { ...savedAnthropic28.hints, maxConcurrent: 1 }, + spawn: async function* (_ir, _ctx) { + counter.count++; + // First spawn holds the slot until releaseGate is signalled. + yield { type: 'delta', content: `h${counter.count}-0` }; + await releaseGate; + yield { type: 'stop', finish_reason: 'stop' }; + }, + }; + lp28.set('anthropic', fake); + + try { + // Fire request 1 with prompt-A; it acquires the slot and stalls. + const p1 = makeStreamRequest({ prompt: 'd58-28h-A' }); + await new Promise(r => setTimeout(r, 30)); + // Fire request 2 with prompt-B (different cache key) — should + // CONCURRENCY_LIMIT in factory, fall through to buffered path, + // which will also fail acquire → chain-exhausted 502. + const r2 = await makeStreamRequest({ prompt: 'd58-28h-B' }); + // The buffered fallback engine surfaces a chain-exhausted error as + // 502 for single-hop saturation per pre-D58 behaviour. + assert.ok(r2.status === 502 || r2.status === 503, + `expected 502/503 chain-exhausted, got ${r2.status}: ${r2.body.slice(0, 200)}`); + + // Release request 1's stall. + releaseSig(); + const r1 = await p1; + assert.equal(r1.status, 200, 'r1 must complete normally'); + // Exactly one spawn — request 2 never spawned (CONCURRENCY_LIMIT). + assert.equal(counter.count, 1, 'only one underlying spawn'); + } finally { + // In case of an early failure, release the gate so promises settle. + try { releaseSig(); } catch { /* already released */ } + } + }); + + it('28i — stop-less exhaustion: source generator returns without {type:"stop"} → cache NOT populated (D58 follow-up to D58 reviewer P2-2)', async () => { + // ADR 0005 § "Cache write conditions" item 1 (D16 truncated-not-cached + // invariant): if the source generator exhausts without emitting a stop + // chunk, the response is treated as truncated and MUST NOT persist in + // cache. D57's cache layer is IR-agnostic and writes accumulatedChunks + // on any source exhaustion; D58's server.mjs handles the IR semantics by + // calling cacheStore.delete(...) immediately after on the no-stop path. + // This test pins the end-to-end behaviour from the HTTP layer. + const counter = { count: 0 }; + const fake = { + ...savedAnthropic28, + spawn: async function* (_ir, _ctx) { + counter.count++; + yield { type: 'delta', content: 'no-stop-1' }; + yield { type: 'delta', content: 'no-stop-2' }; + // Generator returns WITHOUT a stop chunk — truncation path. + }, + }; + lp28.set('anthropic', fake); + const r1 = await makeStreamRequest({ prompt: 'd58-28i' }); + assert.equal(r1.status, 200, '28i r1: SSE response 200'); + assert.equal(counter.count, 1, '28i r1: spawned once'); + // The synthetic truncation marker {type:"stop", finish_reason:"length"} + // should appear (D26 F19 in-band signal); [DONE] terminator follows. + assert.ok(r1.body.includes('"finish_reason":"length"'), + `28i r1: truncation marker expected; body=${r1.body.slice(0, 300)}`); + assert.ok(r1.body.includes('[DONE]'), '28i r1: [DONE] terminator'); + // Subsequent identical request must respawn (cache was NOT populated). + const r2 = await makeStreamRequest({ prompt: 'd58-28i' }); + assert.equal(r2.status, 200, '28i r2: SSE response 200'); + assert.equal(counter.count, 2, '28i r2: second request triggered a fresh spawn (no cache reuse for truncated entry)'); + }); +});