feat+test: D58 — server.mjs streaming singleflight wiring (ADR 0005 Amendment 8, issue #16) (#37)

* feat+test: D58 — server.mjs streaming singleflight wiring (ADR 0005 Amendment 8 §§7,11,12, issue #16) Second of three D-days for v1.x roadmap #1. D57 landed the cache-layer primitive (cacheStore.getOrComputeStreaming); D58 wires it into the streaming branch of server.mjs and adds the X-OLP-Streaming-Inflight header. D59 closes issue #16 + polishes README known-limitations. ## What server.mjs — replaced the streaming branch (formerly the peek+spawn pattern at lines 1138-1327) with cacheStore.getOrComputeStreaming(...): - tryAcquireSpawn moves INSIDE sourceFactory closure (§7). Only first caller acquires; attached joiners share the slot. releaseSpawn lives in the source generator try/finally so it fires once on source completion / error / abort. - CONCURRENCY_LIMIT thrown by the factory triggers fallthrough to the buffered path (preserving today's behaviour); any other pre-stream factory error surfaces a 502. - X-OLP-Streaming-Inflight: source | attached header per §11 (cache_hit role omits — X-OLP-Cache: hit already says it). The §11 'solo' value is deferred to a future amendment — observable only post-stream via the streaming_inflight_source_done log event's attached_count. - auditCtx.cache_status: 'miss' for source, 'streaming_attached' for joiners, 'hit' for the TTL-race cache_hit branch. - res.on('close', () => stream.return?.()) propagates client disconnect into the tee's attachedClients accounting (§9). Note: Node 25 emits 'close' on ServerResponse, NOT on IncomingMessage — empirically verified in test 28g. - Cache writes now happen inside the cache layer's tee task on source completion (§4). Server still issues cacheStore.delete on stop-less exhaustion to preserve D16 truncated-not-cached invariant — the cache layer is IR-agnostic and writes accumulatedChunks unconditionally; the IR-aware server deletes the entry if no stop chunk was observed. - The pre-cache-store-acquire and matching releaseSpawn-on-503 branches are gone — they were vestigial once the factory owns acquire+release. lib/audit.mjs — JSDoc cache_status enum extended with 'streaming_attached'. Free-form string at the wire (no schema validator on append); the JSDoc is the source of truth for the consumer enum. test-features.mjs Suite 28 — 7 HTTP integration tests: - 28a single SSE request (source role + cache populated + second request → X-OLP-Cache: hit) - 28b 2 concurrent identical SSE → one spawn, source + attached roles, identical chunk sequences delivered - 28c TOCTOU regression — pre-populated cache + 2 concurrent → both hit buffered replay path, no streaming branch entry, no X-OLP-Streaming-Inflight header - 28d mid-stream join — late joiner receives accumulated burst + live tail - 28f one-of-N disconnect — source NOT aborted; survivor completes - 28g ALL clients disconnect → source aborted; no cache write; subsequent request gets fresh source spawn - 28h CONCURRENCY_LIMIT fallthrough — factory throws at maxConcurrent=1; buffered path's chain-exhausted 502 surfaces (28e backpressure deferred to Suite 27g unit-level coverage.) ## Scope server.mjs + lib/audit.mjs + test-features.mjs. Untouched: cache/store.mjs (D57 frozen), provider plugins, fallback engine, IR. CHANGELOG and package.json bump fires at D59 close. ## Authority - docs/adr/0005-cache-cross-provider.md Amendment 8 §§7, 8, 9, 11, 12 - docs/v1x-roadmap.md #1 - GitHub issue #16 (TOCTOU window; closed in D59) - ADR 0002 Amendment 6 (D38 tryAcquireSpawn/releaseSpawn semantics that §7 now invokes inside the sourceFactory closure) ## Test count 615 → 622 (+7 D58 integration tests). Local: 622/622 pass. ## Iron Rule 10 follow-up notes for the reviewer - res.on('close') vs req.on('close'): switched to res after empirical verification (28g fails on req under Node 25). Comment in code. - Cache-layer write + server-layer delete for stop-less exhaustion is cosmetically inconsistent with streaming_inflight_source_done's cache_written: true log. Functionally correct; flagged for future amendment. - 'solo' header value deferred — would need trailer mechanics or post-stream emission. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com> * fix: D58 reviewer follow-ups — P2-1 (audit-query gauge drift) + P2-2 (stop-less HTTP test) Fold-in for D58 PR #37 fresh-context opus reviewer findings (APPROVE WITH MINOR — 0 P0/P1, 4 P2). P2-3 (`isFirst` unused) and P2-4 (`solo` not emitted) are design-acceptable per ADR 0005 Amendment 8 §11 and left as-is. P2-1 — `lib/audit-query.mjs` gauge reconciliation. `aggregateRequests` and `cacheHitRateWindow` previously counted `streaming_attached` rows in `total` / `pe.total` without contributing to hit/miss/bypass numerators, breaking the invariant that the cache_status breakdown should sum to the total. Added an explicit `streaming_attached` field to both the global return shape and the `by_provider` shape; the counter is excluded from `hit_rate` numerator AND denominator (joiners did not hit a literal cache so they don't belong in either side of the ratio). Test count is unchanged for D49 suites — they only assert presence + non-negative + reconciliation invariants that the new field preserves; if a test asserted exact value equality on a fixture with NO streaming_attached rows, the new field defaults to 0 and the assertion still passes. P2-2 — Suite 28 stop-less HTTP coverage gap. Test 28i fires an SSE request to a fake provider whose source generator returns WITHOUT a {type:"stop"} chunk; asserts (a) the synthetic truncation marker appears in the body (D26 F19 in-band signal), (b) [DONE] terminator follows, (c) a subsequent identical request triggers a fresh spawn (cache was NOT populated by the truncated stream). Pins the D58 `cacheStore.delete` path at server.mjs:1344-1346 end-to-end. 622 → 623 (+1 D58 follow-up test). 623/623 pass locally. 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>
2026-07-21 21:15:10 +00:00 · 2026-05-25 20:59:54 +10:00
parent 1062e88e77
commit 9b66326e72
4 changed files with 750 additions and 175 deletions
@@ -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,
  };
 }
@@ -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]
@@ -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
+  // On success: chunks are written to res, and the cache layer (via
-  // hit the burst-replay path.
+  // getOrComputeStreaming) writes accumulated chunks on source completion.
  // 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).
  //
-  // Acquired here, released in the `finally` below. The gate intentionally
+  // D58 — ADR 0005 Amendment 8 §§7, 11, 12: streaming singleflight via
-  // lives BEFORE the streaming-branch entry check so the buffered fallthrough
+  // cacheStore.getOrComputeStreaming(...). The peek+spawn TOCTOU window
-  // can re-attempt acquire from a clean slate.
+  // (server.mjs:1104 peek vs. spawn) is collapsed by the synchronous Map
-  // TODO(v1.x — ADR 0005 Amendment 8 / issue #16): replace this peek+spawn
+  // check+insert in the cache layer (D57). The D38 tryAcquireSpawn gate now
-  // pattern with cacheStore.getOrComputeStreaming(...) to close the TOCTOU
+  // lives INSIDE the sourceFactory closure: only the first caller acquires a
-  // window between line ~782 peek and the spawn at line ~846, and to make N
+  // slot; attached joiners share it. On factory-throw CONCURRENCY_LIMIT, the
-  // concurrent identical streaming requests share one spawn (tee-streaming).
+  // cache layer rejects → we fall through to the buffered path (same outcome
-  // Design ratified in D42; see docs/v1x-roadmap.md #1 for the trigger and
+  // as today). preCheckHit gates entry exactly as before — the cache_hit
-  // acceptance criteria. DO NOT remove this comment until the v1.x impl lands.
+  // outcome from getOrComputeStreaming is the TTL-race safety net; in
-  let streamingAcquired = false;
+  // 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
+    // D58 — ADR 0005 Amendment 8 §7: sourceFactory wraps acquire + spawn.
-    // exit below (success, error-after-first-chunk, pre-first-chunk-error,
+    // Acquire fires ONLY for the first caller; attached clients share the
-    // 503 above) leaves these fields representing the streaming attempt.
+    // slot. Release fires EXACTLY ONCE when the source iterator's
-    // Error paths amend `error_code`; success leaves it null.
+    // try/finally executes (normal exhaustion, error, or iterator.return()
-    auditCtx.provider = streamProvider;
+    // propagated from sourceAbortController via the cache layer).
-    auditCtx.tried_providers = [streamProvider];
+    const candidateMax = streamPlugin.hints?.maxConcurrent ?? DEFAULT_MAX_CONCURRENT_SPAWNS;
-    auditCtx.cache_status = 'miss';
+    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({
+    // D58 — invoke the cache-layer coordinator. CONCURRENCY_LIMIT from the
-      providerUsed: streamProvider,
+    // factory falls through to the buffered path (the buffered fallback
-      modelUsed: streamModel,
+    // engine re-attempts acquire). Any other pre-stream error surfaces 502.
-      startMs,
+    let stream;
-      cacheStatus: 'miss',
+    let role;
      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;
    try {
-      for await (const irChunk of streamPlugin.spawn(ir, authContext)) {
+      const result = await cacheStore.getOrComputeStreaming(
-        if (irChunk.type === 'error') {
+        keyId,
-          // Error chunk from provider
+        streamCacheKey,
-          if (firstChunkEmitted) {
+        sourceFactory,
-            // Past first-chunk boundary — can't fallback; emit truncation marker + [DONE]
+        { clientId: requestId },
-            // so clients can detect the incomplete response in-band (aligns D26 F19
+      );
-            // stop-less exhaustion behaviour and the catch-block fix in D35 #10).
+      stream = result.stream;
-            logEvent('warn', 'streaming_error_after_first_chunk', {
+      role = result.role;
-              provider: streamProvider,
+    } catch (e) {
-              model: streamModel,
+      if (e instanceof ProviderError && e.code === 'CONCURRENCY_LIMIT') {
-              error: irChunk.error,
+        // 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.
+        // Generator exhausted without a stop chunk — emit truncation marker
-        // After this point firstChunkEmitted===true ↔ res.headersSent===true.
+        // + [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 {
+        if (streamedChunks.length > 0 && cacheableForFirstHop) {
-        // No bytes written — surface a clean JSON error.
+          logEvent('warn', 'streaming_no_stop_chunk', {
-        logEvent('error', 'streaming_error_before_first_chunk', {
+            chunks_count: streamedChunks.length,
-          provider: streamProvider,
+            provider: streamProvider,
-          model: streamModel,
+            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();
        }
      } 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 {
+      return;
      // 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;
+    // CONCURRENCY_LIMIT fallthrough — flow continues into the buffered path
    // below.
  }
  let fallbackResult;
@@ -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)');
  });
 });