release(v0.3.2): patch release — post-Phase-3 cleanup batch #2 (streaming-path singleflight + TOCTOU close, D57+D58+D59) (#39)

Patch release closing v1.x roadmap #1 end-to-end. ADR 0005 Amendment 8
§§1-14 implemented across 3 D-days (PR #36, #37, #38, all merged):

- D57 (PR #36): cache layer — cacheStore.getOrComputeStreaming(...) +
  tee fan-out + late-joiner replay + per-client backpressure +
  AbortController propagation. New STREAM_BACKPRESSURE error code (not
  a hard trigger). Suite 27 = 12 unit tests.
- D58 (PR #37): server wiring — streaming branch swap; tryAcquireSpawn
  moved into sourceFactory; X-OLP-Streaming-Inflight: source|attached
  header; cache_status: 'streaming_attached' audit value;
  audit-query gauge reconciliation. Suite 28 = 8 HTTP integration tests.
- D59 (PR #38): docs polish — README § Known limitations inverted;
  v1.x roadmap #1 closed; issue #16 closed.

Test count: 603 (v0.3.1) → 623 (v0.3.2). +20 tests across the SF arc.

Patch-release classification per release_kit.phase_rolling_mode +
maintainer release-cut decision (this session, 2026-05-25): the new
wire surface (X-OLP-Streaming-Inflight header + streaming_attached
cache_status) is semver-wise a minor bump, but this is roadmap-cleanup
work — NOT Phase 4 product scope. The reserved 0.4.0 identifier stays
for the formal Phase 4 close. v0.3.2 ships as a patch under the Phase 4
pre-release banner.

Authority: ADR 0005 Amendment 8 (design ratified at D42 2026-05-25;
implementation gated on maintainer "go" — fired 2026-05-25 post-v0.3.1).
docs/v1x-roadmap.md #1 (closed). GitHub issue #16 (closed). ADR 0002
Amendment 6 (D38 tryAcquireSpawn/releaseSpawn semantics, now invoked
from sourceFactory closure).

Co-authored-by: dtzp555 <dtzp555@gmail.com>
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>

CHANGELOG.md

@@ -6,6 +6,34 @@ All notable changes to OLP land here. Per `CLAUDE.md` release_kit overlay, this
 
 (empty — Phase 4 entries land here once Phase 4 opens)
 
+## v0.3.2 — 2026-05-25
+
+### Post-Phase-3 cleanup batch #2 — streaming-path singleflight + TOCTOU close (D57 + D58 + D59)
+
+Patch release closing v1.x roadmap #1 end-to-end. The cache layer's D4 singleflight (one spawn per identical concurrent request) was fully wired on the buffered path since v0.1 but NOT on the streaming path — N concurrent identical streaming requests each spawned their own CLI process. v0.3.2 ships the streaming sibling: tee fan-out, late-joiner replay, per-client backpressure, AbortController propagation, and TOCTOU close. 3 D-day commits (D57 + D58 + D59); ADR 0005 Amendment 8 §§1–14 implemented.
+
+- **D57** (PR #36) — **cache layer.** New `cacheStore.getOrComputeStreaming(keyId, cacheKey, sourceFactory, opts) → { stream, isFirst, role }` mirroring `getOrCompute` on the streaming side. Internals: `_streamingInflight: Map<compositeKey, StreamingInflightEntry>` (composite key `keyId + '\0' + cacheKey`) with synchronous check+insert atomicity (closes TOCTOU per ADR 0005 Amendment 8 §1, §6); single-reader tee fan-out across all attached clients; late-joiner replay buffer (synchronous drain on attach; `STREAM_BACKPRESSURE` terminator if drain or replay-truncation would corrupt); per-client backpressure (`PER_CLIENT_QUEUE_CAP = 1 MB`, overridable via opts); accumulated replay cap (`ACCUMULATED_REPLAY_CAP = 10 MB`, mirrors D23 cache-entry cap); AbortController fires source-iterator return when all clients disconnect. New `'STREAM_BACKPRESSURE'` entry in `PROVIDER_ERROR_CODES` — NOT a hard trigger (whitelist-only `HARD_TRIGGER_CODES`). Suite 27 = 12 unit tests.
+- **D58** (PR #37) — **server wiring.** Streaming branch in `server.mjs` swapped from the peek+spawn pattern to `cacheStore.getOrComputeStreaming(...)`. `tryAcquireSpawn`/`releaseSpawn` moved INSIDE the `sourceFactory` closure per ADR 0005 Amendment 8 §7 (only the first caller acquires; attached joiners share the slot; release fires exactly once on source completion/error/abort). `CONCURRENCY_LIMIT` thrown by the factory triggers fallthrough to the buffered path (preserves today's behaviour). New `X-OLP-Streaming-Inflight: source | attached` HTTP header annotates per-response role (§11). New `cache_status: 'streaming_attached'` audit value tracks the singleflight win. `lib/audit-query.mjs` aggregate APIs (`aggregateRequests`, `cacheHitRateWindow`) extended with `cache_streaming_attached` / `streaming_attached` fields so the cache_status breakdown reconciles. `res.on('close')` propagates client disconnect into the tee's `attachedClients` accounting (§9). D16 truncated-not-cached invariant preserved via server-layer `cacheStore.delete` on stop-less exhaustion (the cache layer is IR-agnostic and writes accumulatedChunks on any source exhaustion; the IR-aware server deletes the entry when the source returned without a `{type:'stop'}` chunk). Suite 28 = 8 HTTP integration tests.
+- **D59** (PR #38) — **docs polish.** README § Known limitations bullet inverted to ✅ shipped marker. `docs/v1x-roadmap.md` #1 rewritten to closed state with 3-D-day breakdown. #6 (streaming SPAWN_FAILED salvage) unbundled from #1 because the tee architecture as implemented does not carry salvage semantics. Issue #16 closed with refs to PRs #36 / #37 / #38.
+- **Test count:** 603 (v0.3.1) → 623 (v0.3.2). +20 streaming-SF tests (Suite 27 = 12 unit, Suite 28 = 8 HTTP integration).
+- **Deferred sub-items (not blocking #1 closure):** (a) `X-OLP-Streaming-Inflight: solo` wire value not emitted — observable only post-stream via `streaming_inflight_source_done` log event's `attached_count: 0`. Future ADR amendment may expose via HTTP trailer. (b) `streaming_inflight_join` log event not emitted from the cache-layer `_attachClient` path because provider/model context lives in the sourceFactory closure (server-layer concern). (c) `isFirst` field returned by `getOrComputeStreaming` is unused by server.mjs (`role` supersedes); could be removed in a future cache-layer API cleanup.
+- **Authority:** ADR 0005 Amendment 8 (design ratified at D42 2026-05-25; implementation gated on maintainer "go" — fired 2026-05-25 post-v0.3.1). `docs/v1x-roadmap.md` #1 (closed). GitHub issue #16 (closed). ADR 0002 Amendment 6 (D38 `tryAcquireSpawn`/`releaseSpawn` semantics, now invoked from sourceFactory closure).
+
+**Patch-release classification.** Per `release_kit.phase_rolling_mode` cross-Phase discipline + maintainer release-cut decision (this session, 2026-05-25): the new wire surface (`X-OLP-Streaming-Inflight` header + `streaming_attached` cache_status) is semver-wise a minor bump, but this is roadmap-cleanup work — NOT Phase 4 product scope. The reserved `0.4.0` identifier stays for the formal Phase 4 close. v0.3.2 ships as a patch under the Phase 4 pre-release banner. Tag push triggers `release.yml`.
+
+## v0.3.1 — 2026-05-25
+
+### Post-Phase-3 cleanup batch #1 (D56)
+
+Patch release closing two XS v1.x-roadmap deferrals (`docs/v1x-roadmap.md` #4 + #7) that became actionable now that Phase 3 management endpoints exist. No new feature surface; pins existing behaviour into tests + finally wires the ADR-documented `activeSpawns` field on `/health`.
+
+- **AUTH_MISSING tuple test** (v1.x roadmap #7 / D45 reviewer P3 deferral). New engine-level test in Suite D40 asserts that an `AUTH_MISSING` hop produces a `fallbackDetail` tuple with `trigger_type: 'auth_missing'` AND that the engine does NOT advance past the AUTH_MISSING hop (per ADR 0004 § Decision — `HARD_TRIGGER_CODES[AUTH_MISSING] = false`). Pre-D56 the behaviour was implicit through other engine-path tests; this commit makes it explicit so a future refactor that moves the tuple-push past the auth_missing branch fails this test directly.
+- **`/health` `activeSpawns` integration** (v1.x roadmap #4 / ADR 0002 Amendment 6 forward note). `handleHealth` now surfaces `providers.status.<name>.activeSpawns` (sourced from D38 `getActiveSpawnCount(name)`). The field is computed BEFORE `healthCheck()` is awaited so it remains present even when `healthCheck()` throws (cheap in-memory counter read). New Suite 21c-extra test pins the field presence + non-negative value for every enabled provider. With no requests in flight: 0; under saturation: equals `hints.maxConcurrent`.
+- **Test count:** 601 (v0.3.0) → 603 (v0.3.1). +2 D56 tests.
+- **Authority:** `docs/v1x-roadmap.md` #4 + #7; ADR 0002 Amendment 6 (concurrency observability forward note); ADR 0004 § Decision + Amendment 5 (X-OLP-Fallback-Detail tuple shape).
+
+**Patch-release classification.** Per `release_kit.phase_rolling_mode` cross-Phase discipline: D56 landed on main after v0.3.0 was tagged, so this is a hotfix-class patch — bump patch, tag, release before next push. Tag push triggers `release.yml`.
+
 ## v0.3.0 — 2026-05-25
 
 ### Phase 3 — Dashboard + audit query layer + daily audit rotation (D48 → D54)

README.md

@@ -197,7 +197,7 @@ Phase 1 closed at v0.1.1 (multi-provider proxy core + pre-Phase-2 cleanup). Phas
 
 Behaviors that work correctly at personal/family scale but have ratified follow-ups for a v1.x sprint. Single landing page: [`docs/v1x-roadmap.md`](./docs/v1x-roadmap.md).
 
-- **Streaming-path singleflight not implemented.** The cache layer's D4 singleflight (one spawn per identical concurrent request) is fully wired on the buffered path but NOT on the streaming path. N concurrent identical streaming requests at v0.1 will each spawn their own CLI process. Design ratified in [ADR 0005 Amendment 8](./docs/adr/0005-cache-cross-provider.md); implementation tracked via [issue #16](https://github.com/dtzp555-max/olp/issues/16) and [v1.x roadmap #1](./docs/v1x-roadmap.md). At family scale this is observably fine — every caller still receives the correct response; the cost is N CLI processes instead of one.
+- **Streaming-path singleflight ✅ shipped (D57 + D58, 2026-05-25).** `cacheStore.getOrComputeStreaming(...)` mirrors the buffered-path `getOrCompute` and resolves the TOCTOU window between peek and spawn ([issue #16](https://github.com/dtzp555-max/olp/issues/16)). Two concurrent identical streaming requests share one CLI spawn via tee fan-out; late joiners receive accumulated replay + the live tail; per-client backpressure (`PER_CLIENT_QUEUE_CAP=1MB`) protects against slow consumers; full-disconnect aborts the source CLI via AbortController. New `X-OLP-Streaming-Inflight: source | attached` header annotates the role. New `cache_status: 'streaming_attached'` audit value tracks the singleflight win. Authority: [ADR 0005 Amendment 8](./docs/adr/0005-cache-cross-provider.md), v1.x roadmap #1.
 - **Soft triggers configured but inert.** `routing.soft_triggers` in `~/.olp/config.json` is honored by the engine's evaluation logic but `quotaStatus()` polling is not wired (ADR 0004 Amendment 2). A startup warning fires if the field is non-empty so the inert state is visible.
 - **Multi-key auth + owner gating + keygen CLI shipped at v0.2.0 (D44 + D45 + D46 + D47).** `lib/keys.mjs` (core), `lib/audit.mjs` (audit), owner-vs-guest `/health` payload trimming + `X-OLP-Fallback-Detail` policy gating, `bin/olp-keys.mjs` (keygen CLI). All 11 ADR 0007 § 10 acceptance criteria covered. v0.2.0 maintainer-merged 2026-05-25.
 

docs/adr/0009-interactive-mode-path-placeholder.md

@@ -0,0 +1,197 @@
+# ADR 0009 — Anthropic Interactive-Mode Path (Placeholder)
+
+- **Date:** 2026-05-25
+- **Status:** Draft (Placeholder — blocked on OCP ADR 0007 P0 experiment outcome; no implementation D-day scheduled until P0 lands)
+- **Authors:** project maintainer (with AI advisory drafting)
+- **Related:**
+  - **OCP ADR 0007** (Interactive-Mode Execution Pool, stream-json) — at `~/ocp/docs/adr/0007-interactive-mode-pool.md` on the maintainer's workstation. Pin reference at the time of this writing: OCP ADR 0007 is Draft status pending the same P0 outcome.
+  - OLP ADR 0001 (Project Founding) — establishes the 2026-06-15 Anthropic billing-split trigger that motivated OLP's multi-provider posture in the first place.
+  - OLP ADR 0006 (Provider Inclusion / Risk Tier Framework) — anthropic is currently a Tier-D Candidate; this ADR amends the operational shape of that plugin if/when P0 succeeds.
+  - OLP ADR 0007 (Multi-Key Auth) — Phase 2 design; per-key cache + audit layer that any future interactive-mode implementation must continue to satisfy.
+  - OLP ADR 0008 (Dashboard + Audit Query) — Phase 3 design; any interactive-mode change must not regress the Dashboard's per-provider observability fields.
+- **Standing autopilot grant note:** Phase 4 is a "new authorization required" scope per `~/.cc-rules/memory/auto/standing_autopilot_phase_2.md`. This placeholder ADR is recorded NOT as implementation work but as the maintainer's "do not forget this when planning Phase 4" anchor. No implementation D-day is scheduled until P0 lands AND maintainer issues a Phase 4 "go" specific to this ADR.
+
+---
+
+## 1. Context
+
+### 1.1 The triggering external work
+
+OCP shipped ADR 0007 (`docs/adr/0007-interactive-mode-pool.md` in `dtzp555-max/ocp`) on 2026-05-25, draft status. The ADR designs a dual-path execution model for the post-2026-06-15 Anthropic billing split:
+
+- **Current `claude -p` path**: programmatic billing → Agent SDK $100/month credit pool (~20–50 heavy coding sessions/month).
+- **Proposed interactive-mode path**: spawn Claude without `-p`, communicate via either (Transport A) piped NDJSON over stdio or (Transport B) `node-pty` PTY — possibly classified as interactive billing → subscription pool.
+
+The OCP team accepted the *concept* but rejected an external contributor's PR #101 implementation (tmux + hook-file polling + `--dangerously-skip-permissions`) on alignment + security grounds, then drafted ADR 0007 as the clean redesign.
+
+### 1.2 Why this is OLP's concern
+
+OLP's founding premise (ADR 0001) was that OCP would become uneconomical post-2026-06-15, motivating a multi-provider hedge. OLP's `lib/providers/anthropic.mjs` today uses the same `claude -p` invocation OCP uses → same billing consequence post-2026-06-15.
+
+If OCP's ADR 0007 P0 experiment confirms that an interactive-mode spawn (Transport A or B) bills against subscription rather than Agent SDK credit, the implementation pattern is directly portable to OLP's anthropic provider plugin. OLP would inherit the same billing benefit without needing to commission an independent P0.
+
+If P0 fails for both transports, OLP's anthropic provider remains stuck on `-p` post-June-15; the multi-provider routing (OpenAI Codex, Mistral) becomes the operational mitigation, exactly per OLP's original founding logic.
+
+### 1.3 The unverified premise (binding caveat)
+
+Per OCP ADR 0007 § "Unverified Premise":
+
+> "TTY detection matters. Local testing (Claude Code 2.1.150) shows: in a real TTY, `claude` without `-p` enters the TUI and does not emit NDJSON. With piped stdin/stdout (`child_process.spawn`), it emits NDJSON even without `-p`. This means Anthropic could use `isTTY` as the billing signal, not the `-p` flag."
+
+OLP cannot independently confirm or refute this prior to 2026-06-15 — Anthropic's billing pool signaling is not exposed on any observable surface until the billing split goes live. **Any OLP implementation that bets on Transport A (stdio pipe) without P0 confirmation risks burning real Agent SDK credit on every Anthropic request.**
+
+---
+
+## 2. Decision
+
+**Option 3 — Wait for OCP ADR 0007 P0 experiment outcome, then port the validated approach.**
+
+Rationale:
+
+- **Avoid duplicated P0 risk.** Both OCP and OLP would run the same experiment against the same Anthropic billing pool. OCP is already designed to run it; OLP riding their result is a free observation.
+- **Lower decision-tree noise.** P0 has three outcomes (Transport A wins / Transport B wins / both fail). OLP's right answer differs per outcome; making the decision before the data is speculation.
+- **No code is wasted.** OLP currently routes anthropic via `claude -p`, which works (just expensive post-June-15). The cost during the wait window is bounded by the Agent SDK $100/month credit + OLP's family-scale request volume.
+
+What "wait" means concretely:
+
+1. **No code change to `lib/providers/anthropic.mjs`** until OCP ADR 0007 transitions from Draft → Accepted (which requires P0 success per OCP ADR 0007 § "Status").
+2. **No Phase 4 D-day scheduled for this scope** until that transition AND maintainer issues an explicit Phase 4 "go" naming this ADR.
+3. **This placeholder ADR stays Draft** until either OCP P0 lands AND OLP decides to port, OR OCP P0 fails decisively AND OLP marks this ADR Rejected with a "shelved per upstream P0 failure" note.
+
+---
+
+## 3. P0 outcome → OLP action decision tree
+
+Recorded here so a future Phase 4 brief can act mechanically once OCP P0 lands.
+
+```
+OCP ADR 0007 P0 result
+  │
+  ├─ Transport A (stdio pipe) confirmed interactive-billing
+  │   → OLP Option 1 OR Option 2 (see § 4); maintainer decides.
+  │     Likely Option 1: port the lib/interactive-pool.mjs +
+  │     billing-router.mjs pattern into lib/providers/anthropic.mjs.
+  │     Updates ADR 0006 to spell out the new spawn mode.
+  │
+  ├─ Transport B (PTY) confirmed; Transport A fails
+  │   → OLP Option 1 with PTY adapter; node-pty dependency added
+  │     under engines-bump scrutiny (this triggers a separate prior
+  │     PR per ADR 0007 § 11-like discipline: native addon adds
+  │     CI matrix work).
+  │
+  ├─ Both transports billed as programmatic
+  │   → OLP marks this ADR Rejected. Anthropic provider stays on
+  │     `claude -p`. Operational mitigation: family-scale users
+  │     either (a) accept the Agent SDK $100 cap, (b) bring their
+  │     own API key (BYOK env path is already there), or (c) shift
+  │     volume to other providers (Codex / Mistral) via OLP's
+  │     existing fallback chain.
+  │
+  └─ P0 results unobservable (billing signals not exposed)
+      → Continue waiting. Re-evaluate one billing cycle (30 days)
+        post-2026-06-15. OLP Anthropic provider remains on `-p`
+        path during the wait; users see Agent SDK credit consumption
+        as the cost signal.
+```
+
+---
+
+## 4. Implementation lanes (to be selected when P0 lands)
+
+This section is informational only. No lane is selected at placeholder time.
+
+### Option 1 — OLP parallel implementation
+
+OLP's `lib/providers/anthropic.mjs` reimplements OCP's interactive pool natively. Replaces the current `claude -p` spawn with a pool-managed warm process + adapter selected per P0 outcome.
+
+- **Pros:** OLP self-contained; no runtime dependency on OCP being installed.
+- **Cons:** Duplicates substantial logic (pool lifecycle, transport adapter, crash backoff DEGRADED, permission auto-response). Two codebases drift over time.
+
+### Option 2 — OLP chains OCP as backend
+
+OLP's `lib/providers/anthropic.mjs` invokes OCP (via its existing HTTP entry surface, or via a future direct-spawn API) rather than spawning `claude` directly. OLP becomes a multi-provider layer ON TOP OF OCP for the Anthropic provider; other providers (Codex, Mistral) continue to be direct.
+
+- **Pros:** Zero duplication; OLP benefits from OCP's P0-validated work automatically. Architectural separation: OCP owns Claude execution, OLP owns multi-provider routing.
+- **Cons:** OLP gains a runtime dependency on OCP being installed + running. Double caching (OCP cache + OLP cache; cache key composition needs to avoid stampede). OCP's HTTP shim is OpenAI-spec-compatible but adds an extra hop's latency. OCP failure modes propagate.
+
+### Option 3 — Both (default to OCP backend if available, fallback to local pool)
+
+A hybrid: OLP detects OCP installed locally, prefers chaining; otherwise falls back to the parallel implementation. Most defensive but most complex.
+
+**Default at placeholder time:** Option 1 is the simpler ship if P0 transports prove out. Option 2 is the cleaner architecture but adds operational coupling. Maintainer decides at P0-resolution time.
+
+---
+
+## 5. Risk assessment (placeholder snapshot)
+
+| Risk | Likelihood (now) | Impact | Mitigation pending P0 |
+|---|---|---|---|
+| OLP forgets this ADR exists | Medium (multi-month wait) | High (would mean OLP misses the post-June-15 window) | This ADR + cc-rules memory `~/.cc-rules/memory/learnings/ocp_adr_0007_interactive_mode_pool.md` |
+| OCP ADR 0007 P0 fails entirely | Medium | High for OLP Anthropic users (Agent SDK $100 cap binds) | OLP's multi-provider fallback (Codex/Mistral) already shipped as Phase 1 work; users have a path |
+| OCP ADR 0007 ships incomplete (interim) | Medium | Medium (OLP can't reliably port) | Wait for OCP to mark Accepted; don't port from Draft |
+| Anthropic policy changes mid-wait | Medium | Depends — could obsolete the whole approach | Re-read OCP ADR 0007 + this ADR before any Phase 4 anthropic work; no decisions on stale information |
+
+---
+
+## 6. Out of scope (explicitly NOT in this ADR)
+
+- **Any code change.** This is a placeholder + decision-tree record only.
+- **OCP P0 design.** That work is OCP's responsibility per OCP ADR 0007 § "Implementation phases".
+- **A new OCP-OLP integration protocol.** If Option 2 is selected at P0-resolution time, the integration shape is a separate ADR.
+- **Engines-bump for node-pty.** Only relevant if P0 picks Transport B and Option 1 is selected. Then it lands as a separate prior PR per the ADR 0007 § 11 pattern.
+
+---
+
+## 7. Phase 4 priority interaction
+
+This ADR is recorded BEFORE Phase 4 implementation scope is finalized. Phase 4 currently lists (per OLP v0.3.0 CHANGELOG):
+
+- Per-key per-provider auth artifact mapping (ADR 0007 § 12 deferral)
+- Audit retention policies (ADR 0008 § 11 deferral)
+- SQLite hybrid migration (ADR 0007 § 13 trigger)
+- Provider-cost weights for spend trend (ADR 0008 § 11 deferral)
+
+If OCP P0 succeeds, **the interactive-mode port likely jumps to the top of Phase 4** (highest user impact: keeps OLP Anthropic users on subscription billing). The other items remain Phase 4 but reorder downstream.
+
+If OCP P0 fails, **this ADR is shelved** and Phase 4 ordering is unchanged.
+
+---
+
+## Consequences
+
+**Positive:**
+
+- OLP retains a documented anchor for the interactive-mode option without committing implementation effort prematurely.
+- Future Phase 4 planning has a structured decision tree, not a vague "we should look at OCP someday".
+- Cross-machine + cross-session memory (cc-rules) ensures the dependency is visible to any future session reading the OLP project context.
+
+**Negative:**
+
+- During the wait window (now → P0 outcome ≥ 2026-07-15), OLP Anthropic users consume Agent SDK credit post-2026-06-15. Bounded by family-scale request volume but a real operational cost.
+- Some risk that "wait" turns into "forget" if multiple unrelated Phase 4 priorities crowd the agenda. Mitigated by this ADR + the cc-rules memory pointer.
+
+**Reversibility:**
+
+- This is a placeholder. Either P0 result transitions it (Accepted → port, Rejected → shelve) cleanly. The placeholder itself doesn't lock OLP into anything.
+
+---
+
+## Authority citations
+
+- **OCP ADR 0007** at `~/ocp/docs/adr/0007-interactive-mode-pool.md` (maintainer workstation). Project repo: `dtzp555-max/ocp`.
+- **PR #101** to `dtzp555-max/ocp` — external contributor's tmux-based prototype that triggered the OCP ADR 0007 redesign.
+- **Anthropic 2026-06-15 billing-split announcement** — see `~/.cc-rules/memory/learnings/anthropic_claude_code_billing_split_2026_06_15.md`.
+- **OLP ADR 0001** (Project Founding) — establishes the original billing-split → multi-provider motivation.
+- **OLP ADR 0006** (Provider Inclusion + Risk Tier Framework) — the anthropic plugin's tier classification + the surface this ADR would amend.
+- **OLP ADR 0007** (Multi-Key Auth, Phase 2) — per-key audit + cache layer that must continue to work across any anthropic execution-mode change.
+- **OLP ADR 0008** (Dashboard + Audit Query, Phase 3) — Dashboard per-provider fields must continue to populate.
+- **CLAUDE.md `release_kit.phase_rolling_mode`** — current_phase: Phase 4 (post-v0.3.0); this ADR explicitly does NOT consume a Phase 4 D-day until P0 lands.
+- **Standing autopilot grant** (`~/.cc-rules/memory/auto/standing_autopilot_phase_2.md` in cc-rules `bf0ed9a`) — Phase 4+ requires new authorization; this placeholder is a decision-tree pre-record, not Phase 4 implementation.
+
+---
+
+## Status transitions (recorded for clarity)
+
+- 2026-05-25 — Created as Draft (Placeholder). OCP ADR 0007 also Draft.
+- _(future)_ — If OCP ADR 0007 → Accepted with a confirmed transport: this ADR moves to "Pending Phase 4 implementation D-day", maintainer decides Option 1 / 2 / 3 + lane.
+- _(future)_ — If OCP ADR 0007 → Rejected: this ADR moves to "Shelved (upstream P0 failure)" with a note explaining the fallback (multi-provider routing already covers).

docs/adr/README.md

@@ -22,6 +22,7 @@ New ADRs increment from the highest existing number. Filenames are `NNNN-<short-
 | [0006](0006-provider-inclusion.md) | Provider Inclusion / Exclusion + Risk-Tier Framework | The 4-tier classification (A excluded by default / B explicit consent / C opt-in / D eligible-for-default-enabled), Candidate-vs-Enabled distinction, current v0.1 candidate inventory (0 Enabled), Antigravity exclusion rationale (named prohibition + no cost advantage + reinstatement friction; pending primary-source pin), consent UX, future provider addition procedure. |
 | [0007](0007-multi-key-auth.md) | Multi-Key Auth (`lib/keys.mjs`) | Phase 2 design ADR (D43-B, 2026-05-25). Option 2 (filesystem manifest at `~/.olp/keys/<key-id>/manifest.json`) + opaque `olp_<32-byte>` token + SHA-256 hash. Owner / guest / anonymous tier gating with explicit `config.json auth.allow_anonymous` (default false). Bootstrap keygen command surface + `OLP_OWNER_TOKEN` env override with stable synthetic `key_id`. Audit ndjson append-only at `~/.olp/logs/audit.ndjson`, warn+1-retry on append failure. Rejects direct SQLite port at v0.2.0 due to Node baseline (`engines >=18` + CI 20/24 vs `node:sqlite` added 22.5.0 / RC); Option 3 hybrid documented as forward path when Phase 3+ Dashboard / SQL-aggregate quota arrives. |
 | [0008](0008-dashboard-and-audit-query.md) | Dashboard + Audit Query Layer | Phase 3 design ADR (D48, 2026-05-25). Static HTML dashboard + vanilla JS + fetch (no build step). In-memory ndjson scan for aggregate queries (O(N) per call; family-scale acceptable; defers SQLite migration to Option 3 hybrid trigger). Daily audit rotation `audit-YYYY-MM-DD.ndjson` on first append after UTC midnight; cross-file query layer for rolling 30-day windows. Owner-only gating on `/dashboard` + 3 `/v0/management/*` JSON endpoints reusing ADR 0007 § 7 auth model. 30s page poll (no SSE infra). Panels: per-provider quota / 24h request+cache+fallback / 30d spend trend / top-N fallback chains per spec § 4.6. Opens ADR 0007 § 12 Phase 3 deferral (Dashboard + audit query + rotation). |
+| [0009](0009-interactive-mode-path-placeholder.md) | Anthropic Interactive-Mode Path (Placeholder) | Placeholder ADR (2026-05-25, Draft) — blocked on OCP ADR 0007 P0 experiment outcome. Records the maintainer's "wait + port" decision: do NOT independently implement; ride OCP's P0 result. If P0 confirms Transport A (stdio NDJSON) or B (PTY) bills as subscription rather than Agent SDK credit, port to OLP `lib/providers/anthropic.mjs` (Option 1 parallel impl, or Option 2 OCP-as-backend; decision deferred to P0-resolution time). If P0 fails on both, shelve. No Phase 4 D-day scheduled until P0 lands AND maintainer issues explicit "go" naming this ADR. |
 
 ## When to write a new ADR
 

docs/v1x-roadmap.md

@@ -8,21 +8,23 @@
 3. **Where** does the work live in the tree today (file + anchor).
 4. **When** does it need to land (trigger: load profile, security event, governance amendment).
 
-**Reading order for a v1.x sprint kickoff.** Items #1–#3 are the most architecturally consequential and should be designed in dependency order: #2 (multi-key auth) blocks header gating in #1 and observability ownership in #4. #1 (streaming SF) blocks #5 (soft trigger reactivation) only if soft triggers are wired on streaming requests.
+**Reading order for a v1.x sprint kickoff.** As of 2026-05-25, #1 (streaming SF, D57+D58) and #2 (multi-key auth, Phase 2) are CLOSED, and #4 and #7 closed in D56. Remaining v1.x scope: #3 (soft trigger reactivation), #5 (provider cacheKeyFields mask), #6 (streaming SPAWN_FAILED salvage — unbundled from #1 at #1 close). All three remaining items have explicit "trigger to start" gates that have not fired.
 
 ---
 
-## #1 — Streaming-path singleflight + TOCTOU close
+## #1 — Streaming-path singleflight + TOCTOU close — ✅ **SHIPPED (D57 + D58, 2026-05-25)**
 
-- **What.** `cacheStore.getOrComputeStreaming(keyId, cacheKey, sourceFactory)` API replacing the current `peek + spawn` pattern in `server.mjs`. Per-(keyId, cacheKey) inflight Map with tee fan-out, bounded per-client backpressure queues, late-joiner replay buffer, AbortController propagation on all-disconnect.
-- **Why deferred.** Personal/family-scale single-tenant load — N concurrent identical streaming requests is an edge case that has not been reported. Each concurrent caller receives the correct response; the waste is N CLI processes instead of one.
-- **Design ADR (ratified).** [`docs/adr/0005-cache-cross-provider.md` Amendment 8](./adr/0005-cache-cross-provider.md) — full design including the inflight Map shape, tee policy, late-joiner replay, backpressure cap, D38 semaphore coordination, abort policy, cache TTL race handling, observability event set, and X-OLP-Streaming-Inflight header. Implementation acceptance criteria are in Amendment 8 §13.
-- **Tracking issue.** GitHub issue [#16](https://github.com/dtzp555-max/olp/issues/16) — STAYS OPEN as v1.x tracker. Sibling: the closed-but-not-implemented Amendment 6 deferral (D34 F1).
-- **Code anchors today.**
-  - `server.mjs` lines ~782 (`preCheckHit = await cacheStore.peek(...)`) and ~811–817 (streaming branch entry) — these are the lines the new API replaces.
-  - `lib/cache/store.mjs` `getOrCompute` — sibling API; the new one mirrors its shape on the streaming path.
-- **Trigger to start.** Any of: (a) report of N>1 concurrent identical streaming requests in the wild, (b) v1.x sprint planning kickoff with the maintainer explicitly opening this scope, (c) downstream feature requiring tee-streaming primitive (e.g., browser-side observer attaching to an existing stream).
-- **Estimated effort.** Design ADR ratified (Amendment 8) = 30 min done. Implementation = 200–400 lines + 15-20 tests + fresh-context reviewer pass. ~3-4 hours of subagent runtime with full Iron Rule 10 discipline.
+- **Status.** Closed. Trigger (b) fired 2026-05-25 — maintainer "go" after v0.3.1. Shipped across three D-days:
+  - **D57** (PR #36) — cache layer: `cacheStore.getOrComputeStreaming(keyId, cacheKey, sourceFactory, opts) → { stream, isFirst, role }` with `_streamingInflight` Map, tee fan-out, late-joiner replay buffer, per-client backpressure (`PER_CLIENT_QUEUE_CAP=1MB`), replay cap (`ACCUMULATED_REPLAY_CAP=10MB`), AbortController propagation, synchronous Map check+insert (closes TOCTOU). Suite 27 = 12 unit tests.
+  - **D58** (PR #37) — server.mjs wiring: streaming branch swap; `tryAcquireSpawn`/`releaseSpawn` moved inside `sourceFactory` closure (D38 §7 coordination); `CONCURRENCY_LIMIT` fallthrough preserved; `X-OLP-Streaming-Inflight: source | attached` header; `cache_status: 'streaming_attached'` audit value + audit-query gauge reconciliation; `res.on('close') → stream.return()` for client disconnect; D16 truncated-not-cached invariant preserved via `cacheStore.delete` on stop-less exhaustion. Suite 28 = 8 HTTP integration tests.
+  - **D59** (this commit) — docs polish: README known-limitations entry inverted; this roadmap entry closed; issue #16 closed.
+- **Design authority.** [`docs/adr/0005-cache-cross-provider.md` Amendment 8](./adr/0005-cache-cross-provider.md) — implemented per spec §§1–14 across D57+D58.
+- **Tracking issue.** GitHub issue [#16](https://github.com/dtzp555-max/olp/issues/16) — CLOSED at D59 with refs to D57+D58 PRs.
+- **Final test count delta.** 603 (v0.3.1) → 623 (v0.3.2/v0.4.0). +20 tests across the SF arc.
+- **Deferred sub-items** (left here as future-work pointers, NOT blocking #1 closure):
+  - `X-OLP-Streaming-Inflight: solo` value not emitted on the wire (Amendment 8 §11). It's observable only post-stream via the `streaming_inflight_source_done` log event's `attached_count: 0`. Future ADR amendment may expose via HTTP trailer.
+  - `streaming_inflight_join` log event from `_attachClient` cache-layer path (carries no provider/model context). D58 emits the event from the server-layer wrapper instead; cache-layer emission would need a provider/model plumb (TODO marker at `lib/cache/store.mjs:~620`).
+  - `isFirst` field returned by `getOrComputeStreaming` is currently unused by server.mjs (`role` supersedes). Could be removed in a future cache-layer API cleanup.
 
 ## #2 — Multi-key auth (`lib/keys.mjs`) — **PHASE 2 ACTIVE (no longer deferred)**
 
@@ -81,9 +83,10 @@
 
 - **What.** Currently the streaming branch does NOT participate in D16 salvage (the salvage-on-SPAWN_FAILED + chunks pattern that the buffered path uses). Streaming SPAWN_FAILED mid-stream → the truncation marker (D35 #10) fires, but no salvage logic captures partial chunks for downstream cache reuse.
 - **Why deferred.** Less impactful than #1 — at most one client benefits per spawn event, and the buffered path already provides salvage for the bulk of requests. Streaming is the minority path.
-- **Design ADR.** Not yet ratified. Coordinated with #1 because the tee architecture changes the salvage semantics (multiple clients may want different finish_reason interpretations on source-mid-stream-failure).
+- **Status update post-#1 close (2026-05-25).** #1 was originally bundled with #6 in the design ADR (Amendment 8). The tee architecture as implemented does NOT carry salvage semantics — D57's tee writes `accumulatedChunks` to cache only on normal source completion (stop chunk seen); on SPAWN_FAILED mid-stream the cache layer rejects all clients with the error and does NOT persist partial chunks. D58 preserves D16's truncated-not-cached invariant via server-layer `cacheStore.delete` on stop-less exhaustion. #6 therefore remains independently deferrable.
+- **Design ADR.** Not yet ratified. The unbundling from #1 means #6 now needs its own ADR amendment when triggered.
 - **Tracking.** Not a GitHub issue. Tracked here.
-- **Trigger to start.** Bundled with #1 implementation work (the inflight tee architecture changes the salvage semantics, so designing them together is cheaper than serializing).
+- **Trigger to start.** First report of streaming-path SPAWN_FAILED mid-stream where partial-chunk salvage would have helped a downstream caller. Practically unlikely at family scale.
 
 ## #7 — AUTH_MISSING tuple path test coverage (D40 follow-up)
 

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,
   };
 }

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]

lib/cache/store.mjs

@@ -51,6 +51,75 @@
  * @property {number} inflightCount
  */
 
+// ── D57 — ADR 0005 Amendment 8 streaming-singleflight shapes ──────────────
+
+/**
+ * @typedef {Object} StreamingInflightEntry
+ * @property {string} compositeKey            - `${keyId}\0${cacheKey}`
+ * @property {AsyncIterator<*>|null} source   - underlying source iterator (null while factory pending)
+ * @property {AbortController} sourceAbortController - propagates "all clients gone" to the source
+ * @property {Array<*>} accumulatedChunks     - late-joiner replay buffer (bounded by §10)
+ * @property {number} accumulatedByteSize     - running byte size of accumulatedChunks
+ * @property {boolean} accumulatedReplayCapExceeded - true once §10 cap hit; cache write will skip
+ * @property {Set<AttachedClient>} attachedClients - all live clients tee'ing this source
+ * @property {boolean} factoryPending         - true while sourceFactory() is awaited
+ * @property {Array<{ resolve: function, reject: function }>} pendingJoiners - late joiners arriving during factoryPending
+ * @property {boolean} sourceDone             - source iterator exhausted normally
+ * @property {Error|null} sourceError         - non-null if source threw
+ * @property {boolean} sourceAborted          - true if AbortController fired
+ * @property {number} ttlMs                   - TTL to use when writing the completed accumulated chunks to cache
+ */
+
+/**
+ * @typedef {Object} AttachedClient
+ * @property {string} id                      - request id / correlator
+ * @property {Array<*>} queue                 - per-client tee buffer
+ * @property {number} queueByteSize           - running byte size sum
+ * @property {boolean} yieldedAccumulated     - true after late-joiner replay drained
+ * @property {boolean} done                   - terminal sentinel hit
+ * @property {boolean} backpressured          - true once STREAM_BACKPRESSURE terminator scheduled
+ * @property {Error|null} error               - non-null if source threw or replay-drain over-cap
+ * @property {((chunk: { value: *, done: boolean }) => void)|null} resolveNext - pending pull promise resolver
+ * @property {((err: Error) => void)|null} rejectNext - pending pull promise rejecter
+ */
+
+// ADR 0005 Amendment 8 §14 — implementation defaults. Per-call overrides flow
+// via the `opts` argument to getOrComputeStreaming (used by tests to exercise
+// caps cheaply without producing megabytes of fixture data).
+export const PER_CLIENT_QUEUE_CAP_DEFAULT = 1 * 1024 * 1024;
+export const ACCUMULATED_REPLAY_CAP_DEFAULT = 10 * 1024 * 1024;
+
+/**
+ * Returns an approximate byte size for a chunk. The tee + cap math uses
+ * JSON.stringify length as the serialization yardstick (matches the cache
+ * store's existing size accounting via Buffer.byteLength(JSON.stringify(...))).
+ * Non-stringifiable values (circular, etc.) fall back to 0 — the tee continues
+ * but the size accounting under-estimates that chunk. In practice IR chunks
+ * are always JSON-safe.
+ *
+ * @param {*} chunk
+ * @returns {number}
+ */
+function _chunkByteSize(chunk) {
+  try {
+    return Buffer.byteLength(JSON.stringify(chunk) ?? '', 'utf8');
+  } catch {
+    return 0;
+  }
+}
+
+/**
+ * Synthesises a STREAM_BACKPRESSURE terminator stream (ADR 0005 Amendment 8 §8):
+ * yields `{ type: 'stop', finish_reason: 'length' }` then a `[DONE]` sentinel
+ * then ends. Used both for late-joiner-too-late and per-client overflow paths.
+ *
+ * @returns {AsyncGenerator<*>}
+ */
+async function* _backpressureTerminator() {
+  yield { type: 'stop', finish_reason: 'length' };
+  yield '[DONE]';
+}
+
 // ── CacheStore ────────────────────────────────────────────────────────────
 
 export class CacheStore {
@@ -82,6 +151,14 @@ export class CacheStore {
     /** @type {Map<string, Promise<*>>} */
     this._inflight = new Map();
 
+    // D57 — ADR 0005 Amendment 8: streaming singleflight per-(keyId,cacheKey)
+    // inflight Map. Composite key uses `\0` separator per §2 (avoids colliding
+    // with keyId or cacheKey content). The check + insert against this Map is
+    // synchronous in getOrComputeStreaming (no `await` between read & write),
+    // mirroring D38 `tryAcquireSpawn` atomicity.
+    /** @type {Map<string, StreamingInflightEntry>} */
+    this._streamingInflight = new Map();
+
     // Stats per keyId
     /** @type {Map<string, { hits: number, misses: number }>} */
     this._stats = new Map();
@@ -266,13 +343,10 @@ export class CacheStore {
    * @param {number} [ttlMs]
    * @returns {Promise<*>}
    *
-   * TODO(v1.x — ADR 0005 Amendment 8 / issue #16): add a sibling
-   * `getOrComputeStreaming(keyId, cacheKey, sourceFactory)` for the streaming
-   * path. This API handles buffered responses only; the streaming branch in
-   * server.mjs currently uses a peek+spawn pattern with a TOCTOU window.
-   * The streaming sibling will mirror this method's shape but with a tee
-   * fan-out and per-client backpressure queues. See docs/v1x-roadmap.md #1
-   * for the design contract and acceptance criteria.
+   * D57 — ADR 0005 Amendment 8 / issue #16 resolved at the cache layer:
+   * the sibling `getOrComputeStreaming` method below provides tee-fan-out +
+   * per-client backpressure for the streaming path. Server-side wiring lands
+   * separately in D58.
    */
   async getOrCompute(keyId, cacheKey, computeFn, ttlMs) {
     // 1. Cache hit — return immediately, no singleflight overhead
@@ -311,6 +385,493 @@ export class CacheStore {
     return computePromise;
   }
 
+  /**
+   * D57 — ADR 0005 Amendment 8 (issue #16): streaming singleflight + tee-fan-out.
+   *
+   * Streaming-path sibling of `getOrCompute`. Coordinates concurrent identical
+   * streaming requests so only one source spawn occurs, all attached clients
+   * receive identical chunk sequences in order, late joiners are replayed from
+   * an accumulated buffer, slow clients are disconnected with a synthetic
+   * STREAM_BACKPRESSURE terminator instead of stalling the source, and the
+   * source is aborted when all clients disconnect.
+   *
+   * The Map check + insert against `_streamingInflight` is synchronous — no
+   * `await` between read and write — matching the D38 `tryAcquireSpawn`
+   * atomicity invariant and collapsing the original TOCTOU window in
+   * `server.mjs:782` (peek + spawn). See §1 + §6.
+   *
+   * Three outcomes (Amendment 8 §1):
+   *   - **cache_hit**: cached entry exists and is alive → returns
+   *     `{ stream: <async iterator over cached chunks>, isFirst: false,
+   *        role: 'cache_hit' }`. No spawn. `hits` incremented.
+   *   - **attached**: inflight entry exists in `_streamingInflight` → attaches
+   *     a new AttachedClient. Late-joiner replay (§5) drains accumulated
+   *     chunks synchronously; if drain would exceed `perClientQueueCap` the
+   *     client receives a STREAM_BACKPRESSURE synthesised stream instead.
+   *     `isFirst: false`, `role: 'attached'`. `hits` incremented (sharing a
+   *     spawn is functionally a "cache-like" benefit; documented choice).
+   *   - **source**: no cache hit, no inflight entry → the cache layer takes
+   *     the inflight lock synchronously (placeholder entry inserted), then
+   *     `await sourceFactory()`. If the factory throws (e.g.
+   *     CONCURRENCY_LIMIT) the placeholder is removed and the error
+   *     propagates. On success the source is wired up, the tee task starts,
+   *     `isFirst: true`, `role: 'source'`. `misses` incremented.
+   *
+   * **`role` enum** (returned at attach-time):
+   *   - `source` — first caller; entry created. Lifetime-end may upgrade to
+   *     `solo` (no joiners ever attached) at the server (D58); the cache
+   *     layer reports `source` at attach-time and does not flip mid-stream.
+   *   - `attached` — joined an existing inflight entry.
+   *   - `cache_hit` — served from cache; no entry created.
+   *
+   * **Amendment 8 §11 header values**: the X-OLP-Streaming-Inflight header
+   * (D58) uses `source | attached | solo`. The cache layer's `cache_hit` is
+   * the "served from cache without inflight" case; D58 chooses whether to
+   * emit `solo` or omit the header for that path. The cache layer reports
+   * `cache_hit` as a distinct role so the server can disambiguate.
+   *
+   * **§14 defaults**: `PER_CLIENT_QUEUE_CAP = 1 MB`, `ACCUMULATED_REPLAY_CAP
+   * = 10 MB`. Both overridable via `opts` for cheap test exercise of the cap
+   * paths.
+   *
+   * @param {string} keyId
+   * @param {string} cacheKey
+   * @param {() => Promise<AsyncIterator<*>>|AsyncIterator<*>} sourceFactory
+   *   Invoked exactly once per inflight lifetime (only on first caller).
+   *   Returns the source async iterator. May throw (e.g. CONCURRENCY_LIMIT
+   *   from `tryAcquireSpawn`); on throw, the inflight entry is removed and
+   *   the error propagates to the first caller. Late joiners that attached
+   *   while the factory was pending are rejected with the same error.
+   * @param {object} [opts]
+   * @param {string} [opts.clientId]              - correlator id (defaults to incrementing counter)
+   * @param {number} [opts.ttlMs]                 - TTL for completed accumulated-chunks cache write
+   * @param {number} [opts.perClientQueueCap]     - override §14 default (1 MB)
+   * @param {number} [opts.accumulatedReplayCap]  - override §14 default (10 MB)
+   * @returns {Promise<{ stream: AsyncGenerator<*>, isFirst: boolean, role: 'source'|'attached'|'cache_hit' }>}
+   */
+  async getOrComputeStreaming(keyId, cacheKey, sourceFactory, opts = {}) {
+    const compositeKey = `${keyId}\0${cacheKey}`;
+    const perClientQueueCap = opts.perClientQueueCap ?? PER_CLIENT_QUEUE_CAP_DEFAULT;
+    const accumulatedReplayCap = opts.accumulatedReplayCap ?? ACCUMULATED_REPLAY_CAP_DEFAULT;
+    const ttlMs = opts.ttlMs;
+    const clientId = opts.clientId ?? `c-${this._nowFn()}-${Math.floor(Math.random() * 1e9).toString(36)}`;
+
+    // ── (A) Inflight Map check FIRST (Amendment 8 §6 — TTL race) ───────────
+    // Late joiners that arrive after a cache entry has expired but during an
+    // active source spawn must still attach via the inflight Map rather than
+    // re-spawn. The synchronous Map.get + (if hit) Map preservation here
+    // satisfies the no-`await`-between-check-and-decision invariant.
+    const inflightEntry = this._streamingInflight.get(compositeKey);
+    if (inflightEntry) {
+      // Hits-as-share decision (documented at method header): sharing a spawn
+      // is a cache-like benefit; increment hits for consistency with
+      // cache_hit accounting and to expose the singleflight win in stats().
+      this._getStats(keyId).hits++;
+      const stream = this._attachClient(inflightEntry, {
+        clientId,
+        perClientQueueCap,
+      });
+      return { stream, isFirst: false, role: 'attached' };
+    }
+
+    // ── (B) Cache hit check (no inflight) ──────────────────────────────────
+    // Replays cached chunks via a synthetic async iterator. No source spawn.
+    const ns = this._getNamespace(keyId);
+    const existing = ns.get(cacheKey);
+    if (existing && this._isAlive(existing)) {
+      this._getStats(keyId).hits++;
+      const cachedChunks = Array.isArray(existing.value) ? existing.value : [existing.value];
+      const stream = (async function* cacheReplay() {
+        for (const chunk of cachedChunks) {
+          yield chunk;
+        }
+      })();
+      return { stream, isFirst: false, role: 'cache_hit' };
+    }
+
+    // ── (C) Miss + no inflight: take the lock synchronously, then await ───
+    // The placeholder entry is inserted BEFORE invoking sourceFactory so that
+    // late joiners arriving while the factory is awaited see the inflight
+    // entry and attach (they're parked in `pendingJoiners` until the factory
+    // resolves or rejects). If the factory throws, the placeholder is
+    // removed and the error propagates to the first caller AND all parked
+    // joiners. This preserves the §1 invariant: the Map insert is atomic
+    // from later joiners' perspective.
+    const entry = /** @type {StreamingInflightEntry} */ ({
+      compositeKey,
+      source: null,
+      sourceAbortController: new AbortController(),
+      accumulatedChunks: [],
+      accumulatedByteSize: 0,
+      accumulatedReplayCapExceeded: false,
+      attachedClients: new Set(),
+      factoryPending: true,
+      pendingJoiners: [],
+      sourceDone: false,
+      sourceError: null,
+      sourceAborted: false,
+      ttlMs,
+    });
+    this._streamingInflight.set(compositeKey, entry);
+    this._getStats(keyId).misses++;
+
+    // Attach the first caller synchronously so any subsequent joiners during
+    // the factory await see the same set/topology as the first caller.
+    const firstStream = this._attachClient(entry, {
+      clientId,
+      perClientQueueCap,
+    });
+
+    let sourceIter;
+    try {
+      const factoryResult = sourceFactory();
+      sourceIter = factoryResult && typeof factoryResult.then === 'function'
+        ? await factoryResult
+        : factoryResult;
+    } catch (err) {
+      // Factory rejected — remove placeholder and reject first caller + any
+      // late joiners that arrived during the await.
+      this._streamingInflight.delete(compositeKey);
+      for (const client of entry.attachedClients) {
+        if (client.rejectNext) {
+          client.rejectNext(err);
+          client.resolveNext = null;
+          client.rejectNext = null;
+        }
+        client.error = err;
+        client.done = true;
+      }
+      throw err;
+    }
+
+    entry.source = sourceIter;
+    entry.factoryPending = false;
+
+    // Kick off the tee task. It runs detached; lifetime is bounded by the
+    // source iterator's completion / error / abort.
+    this._teeStreamingSource(keyId, cacheKey, entry, {
+      accumulatedReplayCap,
+    });
+
+    return { stream: firstStream, isFirst: true, role: 'source' };
+  }
+
+  /**
+   * D57 — ADR 0005 Amendment 8 §3 + §5: attach a new client to an inflight
+   * entry. Synchronously drains the accumulated replay buffer into the
+   * client's queue (§5). If the drain would exceed `perClientQueueCap`, the
+   * client receives a STREAM_BACKPRESSURE synthesised stream INSTEAD of the
+   * normal tee — the source continues for the other clients.
+   *
+   * Returns the async iterator the caller will consume.
+   *
+   * @private
+   */
+  _attachClient(entry, { clientId, perClientQueueCap }) {
+    // Late-joiner replay drain cap check (§5 + §10): a late joiner cannot
+    // catch up if either
+    //   (a) the accumulated buffer alone would overflow the per-client cap
+    //       (burst > PER_CLIENT_QUEUE_CAP), or
+    //   (b) the replay buffer is already truncated (§10 cap was hit and
+    //       further source chunks were not appended to accumulatedChunks),
+    //       so even a successful drain would give the joiner a partial view
+    //       that disagrees with later live chunks.
+    // Either condition → STREAM_BACKPRESSURE synthetic terminator. The
+    // source / other clients are unaffected.
+    if (
+      entry.accumulatedByteSize > perClientQueueCap
+      || entry.accumulatedReplayCapExceeded
+    ) {
+      this._warnFn('stream_backpressure_disconnect', {
+        client_id: clientId,
+        queue_byte_size: entry.accumulatedByteSize,
+        per_client_cap: perClientQueueCap,
+        composite_key: entry.compositeKey,
+        reason: entry.accumulatedReplayCapExceeded
+          ? 'replay_cap_truncated'
+          : 'replay_drain_over_cap',
+      });
+      return _backpressureTerminator();
+    }
+
+    /** @type {AttachedClient} */
+    const client = {
+      id: clientId,
+      queue: [],
+      queueByteSize: 0,
+      yieldedAccumulated: false,
+      done: false,
+      backpressured: false,
+      error: null,
+      resolveNext: null,
+      rejectNext: null,
+      // Per-client cap is captured here so the tee task can apply it
+      // without re-plumbing opts; documented as an internal field.
+      __perClientQueueCap__: perClientQueueCap,
+    };
+
+    // Synchronous replay drain — push every accumulated chunk into the
+    // client's queue at attach-time. From this point on the tee task pushes
+    // live chunks.
+    for (const chunk of entry.accumulatedChunks) {
+      client.queue.push(chunk);
+      client.queueByteSize += _chunkByteSize(chunk);
+    }
+    client.yieldedAccumulated = true;
+    entry.attachedClients.add(client);
+
+    // TODO(D58 — ADR 0005 Amendment 8 §11): emit `streaming_inflight_join`
+    // event from the server-layer wrapper, which has provider/model context.
+    // Cache layer alone does not have provider/model identity (sourceFactory
+    // is a closure), so the join event lives at the consumer of `role:
+    // 'attached'` in server.mjs. D57 reviewer P2-3 follow-up.
+
+    // If source already completed before this attach (last-second join)
+    // mark the client as terminal-after-drain so the iterator returns
+    // cleanly once the replay queue is drained.
+    if (entry.sourceDone) {
+      client.done = true;
+    } else if (entry.sourceError) {
+      client.error = entry.sourceError;
+      client.done = true;
+    }
+
+    // Per-client AbortController for client-side cancellation (HTTP close).
+    // The async iterator's return() removes the client from attachedClients;
+    // if the entry's attachedClients size hits zero, the tee task aborts the
+    // source. The teardown logic lives in the iterator below.
+    const store = this;
+    const iterator = (async function* clientStream() {
+      try {
+        while (true) {
+          // Drain queue chunks first.
+          if (client.queue.length > 0) {
+            const next = client.queue.shift();
+            client.queueByteSize -= _chunkByteSize(next);
+            yield next;
+            continue;
+          }
+          // Backpressure-terminated client: yield the synthetic terminator.
+          if (client.backpressured) {
+            yield { type: 'stop', finish_reason: 'length' };
+            yield '[DONE]';
+            client.done = true;
+            return;
+          }
+          // Source already errored.
+          if (client.error) {
+            throw client.error;
+          }
+          // Source already completed and no more queued chunks.
+          if (client.done) {
+            return;
+          }
+          // Block until the tee task pushes the next chunk (or signals
+          // source-done / source-error / backpressure).
+          await new Promise((resolve, reject) => {
+            client.resolveNext = resolve;
+            client.rejectNext = reject;
+          });
+          client.resolveNext = null;
+          client.rejectNext = null;
+        }
+      } finally {
+        // Iterator return() fired (HTTP close, break, or normal return) —
+        // remove client and possibly trigger source abort.
+        if (entry.attachedClients.has(client)) {
+          entry.attachedClients.delete(client);
+        }
+        // If we're the last client AND the source is still running, fire
+        // the AbortController so the source iterator's return() / cleanup
+        // can reap any underlying resources.
+        if (
+          entry.attachedClients.size === 0
+          && !entry.sourceDone
+          && !entry.sourceError
+          && !entry.sourceAborted
+          && !entry.factoryPending
+        ) {
+          entry.sourceAborted = true;
+          try {
+            entry.sourceAbortController.abort();
+          } catch {
+            // best-effort
+          }
+          // Tee task observes attachedClients.size === 0 + sourceAborted on
+          // its next loop iteration and exits without a cache write.
+          store._streamingInflight.delete(entry.compositeKey);
+          store._warnFn('streaming_inflight_abort', {
+            composite_key: entry.compositeKey,
+            accumulated_chunk_count: entry.accumulatedChunks.length,
+          });
+        }
+      }
+    })();
+
+    return iterator;
+  }
+
+  /**
+   * D57 — ADR 0005 Amendment 8 §4: tee fan-out task. One reader pulls from
+   * `entry.source`; on each chunk, pushes to `accumulatedChunks` (bounded by
+   * §10) and to every attached client's queue (per-client cap from §8).
+   *
+   * On source completion: writes accumulated chunks to cache if (a) cap not
+   * exceeded and (b) `set()`'s own `maxEntryBytes` cap admits it. Resolves
+   * all clients to drain-out state. Removes entry.
+   *
+   * On source error: rejects all clients via their `rejectNext`. No cache
+   * write. Removes entry.
+   *
+   * Source-abort short-circuit: if `attachedClients.size === 0` after a push,
+   * fires `sourceAbortController.abort()`, exits without cache write.
+   *
+   * @private
+   */
+  _teeStreamingSource(keyId, cacheKey, entry, { accumulatedReplayCap }) {
+    const store = this;
+    (async () => {
+      try {
+        for (;;) {
+          // Pre-check: if all clients have already gone away before we even
+          // pull the next chunk, abort the source and bail out. (The
+          // per-client iterator's finally-block sets sourceAborted=true and
+          // removes the entry; we just need to stop pulling.)
+          if (entry.attachedClients.size === 0 && !entry.factoryPending) {
+            if (!entry.sourceAborted) {
+              entry.sourceAborted = true;
+              try { entry.sourceAbortController.abort(); } catch { /* best-effort */ }
+            }
+            // Try to call return() on the source so the underlying generator
+            // cleans up. Best-effort; not all iterators implement it.
+            try {
+              if (entry.source && typeof entry.source.return === 'function') {
+                await entry.source.return();
+              }
+            } catch { /* best-effort */ }
+            return;
+          }
+
+          const result = await entry.source.next();
+          if (result.done) break;
+          const chunk = result.value;
+          const size = _chunkByteSize(chunk);
+
+          // §10 — replay buffer cap. Past the cap we stop accumulating (so
+          // future late joiners can still see the chunks they need to
+          // catch up to live), but we mark the entry not-cacheable so the
+          // §4 completion path skips the cache write.
+          if (!entry.accumulatedReplayCapExceeded) {
+            if (entry.accumulatedByteSize + size > accumulatedReplayCap) {
+              entry.accumulatedReplayCapExceeded = true;
+              store._warnFn('streaming_inflight_replay_cap_exceeded', {
+                composite_key: entry.compositeKey,
+                accumulated_byte_size: entry.accumulatedByteSize,
+                chunk_size: size,
+                accumulated_replay_cap: accumulatedReplayCap,
+              });
+              // Continue accumulating up to this chunk so existing late
+              // joiners' drain decision was based on the size they saw at
+              // attach-time. We do NOT push this chunk to accumulatedChunks
+              // (it would corrupt the "<= cap at attach-time" invariant for
+              // future joiners). Future joiners arriving past this point
+              // see accumulatedByteSize already > cap and get the
+              // backpressure terminator at attach-time per §5.
+            } else {
+              entry.accumulatedChunks.push(chunk);
+              entry.accumulatedByteSize += size;
+            }
+          }
+
+          // Fan out to each client synchronously (no await inside the for-
+          // each-client loop). Disconnecting a client is mutation-during-
+          // iteration; we snapshot the set first.
+          const clientsSnapshot = [...entry.attachedClients];
+          for (const client of clientsSnapshot) {
+            // Per-client backpressure (§8). If the push would exceed the
+            // per-client cap, disconnect this client only.
+            if (client.queueByteSize + size > client.__perClientQueueCap__) {
+              client.backpressured = true;
+              store._warnFn('stream_backpressure_disconnect', {
+                client_id: client.id,
+                queue_byte_size: client.queueByteSize,
+                per_client_cap: client.__perClientQueueCap__,
+                composite_key: entry.compositeKey,
+                reason: 'queue_overflow',
+              });
+              // Remove from set so future fan-out skips this client.
+              entry.attachedClients.delete(client);
+              // Wake the client's pull-promise so it can yield the synthetic
+              // STREAM_BACKPRESSURE terminator.
+              if (client.resolveNext) {
+                const r = client.resolveNext;
+                client.resolveNext = null;
+                client.rejectNext = null;
+                r({ value: undefined, done: false });
+              }
+              continue;
+            }
+            client.queue.push(chunk);
+            client.queueByteSize += size;
+            if (client.resolveNext) {
+              const r = client.resolveNext;
+              client.resolveNext = null;
+              client.rejectNext = null;
+              r({ value: undefined, done: false });
+            }
+          }
+
+          // If the fan-out emptied the attached set (all over-cap), the
+          // top-of-loop pre-check will fire on the next iteration and abort.
+        }
+
+        // Source iterator returned normally.
+        entry.sourceDone = true;
+        const cacheWritten =
+          !entry.accumulatedReplayCapExceeded
+          && entry.accumulatedChunks.length > 0;
+        if (cacheWritten) {
+          // ADR 0005 Amendment 8 §4: write accumulated chunks to cache via
+          // the standard set() path (which itself applies the D23
+          // maxEntryBytes cap — separate from the §10 replay cap).
+          await store.set(keyId, cacheKey, entry.accumulatedChunks, entry.ttlMs);
+        }
+        store._warnFn('streaming_inflight_source_done', {
+          composite_key: entry.compositeKey,
+          attached_count: entry.attachedClients.size,
+          accumulated_chunk_count: entry.accumulatedChunks.length,
+          cache_written: cacheWritten,
+        });
+        // Wake every remaining attached client so they drain their queue and
+        // observe `done = true`.
+        for (const client of [...entry.attachedClients]) {
+          client.done = true;
+          if (client.resolveNext) {
+            const r = client.resolveNext;
+            client.resolveNext = null;
+            client.rejectNext = null;
+            r({ value: undefined, done: true });
+          }
+        }
+        store._streamingInflight.delete(entry.compositeKey);
+      } catch (err) {
+        // Source threw mid-stream. Reject every attached client.
+        entry.sourceError = err;
+        for (const client of [...entry.attachedClients]) {
+          client.error = err;
+          client.done = true;
+          if (client.rejectNext) {
+            const rej = client.rejectNext;
+            client.resolveNext = null;
+            client.rejectNext = null;
+            rej(err);
+          }
+        }
+        store._streamingInflight.delete(entry.compositeKey);
+      }
+    })();
+  }
+
   /**
    * Returns stats for a specific keyId, or aggregate stats across all keyIds.
    *
@@ -322,11 +883,14 @@ export class CacheStore {
       const s = this._stats.get(keyId) ?? { hits: 0, misses: 0 };
       const ns = this._store.get(keyId);
       const size = ns ? ns.size : 0;
+      // D57 — ADR 0005 Amendment 8 §1: inflightCount aggregates both the
+      // buffered-path singleflight Map and the streaming-path inflight Map
+      // so stats() reflects all active dedup-coordination entries.
       return {
         hits: s.hits,
         misses: s.misses,
         size,
-        inflightCount: this._inflight.size,
+        inflightCount: this._inflight.size + this._streamingInflight.size,
       };
     }
 
@@ -345,7 +909,8 @@ export class CacheStore {
       hits: totalHits,
       misses: totalMisses,
       size: totalSize,
-      inflightCount: this._inflight.size,
+      // D57 — see per-keyId branch above; streaming entries counted alongside buffered.
+      inflightCount: this._inflight.size + this._streamingInflight.size,
     };
   }
 
@@ -397,10 +962,18 @@ export class CacheStore {
           this._inflight.delete(k);
         }
       }
+      // D57 — also clear streaming inflight entries scoped to this keyId.
+      // Composite key uses `\0` separator (see _streamingInflight init).
+      for (const k of this._streamingInflight.keys()) {
+        if (k.startsWith(`${keyId}\0`)) {
+          this._streamingInflight.delete(k);
+        }
+      }
     } else {
       this._store.clear();
       this._stats.clear();
       this._inflight.clear();
+      this._streamingInflight.clear();
     }
   }
 }

lib/providers/base.mjs

@@ -162,6 +162,15 @@ export const PROVIDER_ERROR_CODES = /** @type {const} */ ([
   'SPAWN_FAILED',
   'SPAWN_TIMEOUT',     // ADR 0004 § Trigger taxonomy bullet 4: spawn timeout is a hard trigger
   'CONCURRENCY_LIMIT', // ADR 0002 Amendment 6 / ADR 0004 Amendment 4 (D38, issue #1)
+  /* ADR 0005 Amendment 8 §8 (D57): per-client streaming queue overflow.
+   * NOT a hard trigger — the source spawned successfully and other attached
+   * clients continue to receive chunks; only this client's queue exceeded
+   * PER_CLIENT_QUEUE_CAP. The affected client receives a synthetic
+   * { type: 'stop', finish_reason: 'length' } + [DONE] terminator.
+   * D58 wires server-side header/log surface; HARD_TRIGGER_CODES in
+   * lib/fallback/engine.mjs is a whitelist so absence here gives the
+   * correct default (no fallback advancement). */
+  'STREAM_BACKPRESSURE',
 ]);
 
 export class ProviderError extends Error {

package.json

@@ -1,6 +1,6 @@
 {
   "name": "olp",
-  "version": "0.3.0",
+  "version": "0.3.2",
   "description": "Personal multi-provider LLM proxy. Successor to OCP. One HTTP endpoint, multiple subscriptions behind it, automatic routing + fallback + caching.",
   "type": "module",
   "main": "server.mjs",

server.mjs

@@ -577,10 +577,17 @@ async function handleHealth(req, res) {
   const available = listAllProviderNames().length;
   const providerStatuses = {};
   for (const [name, provider] of loadedProviders) {
+    // D56 / v1.x roadmap #4 (ADR 0002 Amendment 6 forward note): surface
+    // per-provider active spawn count for capacity-planning observability.
+    // D38 shipped getActiveSpawnCount; this is the /health integration.
+    // The field is set BEFORE healthCheck() in case healthCheck throws —
+    // activeSpawns is cheap (in-memory counter read) and useful even
+    // when healthCheck fails.
+    const activeSpawns = getActiveSpawnCount(name);
     try {
-      providerStatuses[name] = await provider.healthCheck();
+      providerStatuses[name] = { ...(await provider.healthCheck()), activeSpawns };
     } catch (e) {
-      providerStatuses[name] = { ok: false, error: e.message };
+      providerStatuses[name] = { ok: false, error: e.message, activeSpawns };
     }
   }
   sendJSON(res, 200, {
@@ -1110,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);
@@ -1143,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',
@@ -1209,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;

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';
@@ -6247,6 +6247,30 @@ describe('D40 — X-OLP-Fallback-Detail header (issue #7)', () => {
     assert.equal(result.fallbackDetail[0].code, 'SPAWN_FAILED');
   });
 
+  it('engine: AUTH_MISSING terminates chain, fallbackDetail tuple records trigger_type:"auth_missing" (D56, v1.x roadmap #7)', async () => {
+    // v1.x roadmap #7 (D40 follow-up): explicit pin that the AUTH_MISSING
+    // path produces a fallbackDetail tuple with trigger_type:"auth_missing"
+    // BEFORE the engine's early-return at engine.mjs:486. Was implicit via
+    // other engine-path tests; D56 makes it explicit so any future refactor
+    // that moves the tuple-push past the auth_missing branch fails this.
+    const err = new ProviderError('No OAuth token found', 'AUTH_MISSING');
+    const chain = [
+      { provider: 'anthropic', model: 'claude-sonnet-4-6' },
+      { provider: 'openai', model: 'gpt-5.5' }, // present to verify AUTH_MISSING does NOT advance
+    ];
+    const hopFn = async () => { throw err; };
+    const result = await executeWithFallback(chain, makeIR({ model: 'claude-sonnet-4-6' }), hopFn);
+    // AUTH_MISSING is HARD_TRIGGER_CODES[AUTH_MISSING]=false (engine.mjs L52);
+    // chain stops at hop 0 instead of advancing to openai.
+    assert.equal(result.chunks, null, 'AUTH_MISSING terminates chain');
+    assert.equal(result.fallbackHops, 0, 'AUTH_MISSING does NOT advance — stays at hop 0');
+    assert.equal(result.fallbackDetail.length, 1, 'fallbackDetail has exactly 1 tuple (the AUTH_MISSING hop)');
+    assert.equal(result.fallbackDetail[0].code, 'AUTH_MISSING');
+    assert.equal(result.fallbackDetail[0].trigger_type, 'auth_missing');
+    assert.equal(result.fallbackDetail[0].provider, 'anthropic');
+    assert.equal(result.fallbackDetail[0].hop, 0);
+  });
+
   it('engine: non-ProviderError exception → tuple code is "UNKNOWN"', async () => {
     const err = new Error('Something unexpected'); // no .code field
     const chain = [{ provider: 'anthropic', model: 'claude-sonnet-4-6' }];
@@ -10522,6 +10546,29 @@ describe('Suite 21 — D46 owner-vs-guest gating (ADR 0007 §§ 7.1, 7.2)', () =
       assert.ok('status' in body.providers, 'owner /health providers.status must be present');
     });
 
+    it('21c-extra: owner /health each provider status carries activeSpawns field (D56, v1.x #4 / ADR 0002 Amendment 6)', async () => {
+      // ADR 0002 Amendment 6 forward note: when surfaced on /health, the
+      // per-provider concurrency counter lives at providers.status.<name>.
+      // activeSpawns. D56 wires it. With no requests in flight, the value
+      // is 0; under saturation it equals hints.maxConcurrent.
+      const { plaintext_token } = createKey({ name: '21c-extra', owner_tier: 'owner', providers_enabled: '*', olpHome: TMP });
+      const r = await fetch({
+        port, method: 'GET', path: '/health',
+        headers: { Authorization: `Bearer ${plaintext_token}` },
+      });
+      assert.equal(r.status, 200);
+      const body = JSON.parse(r.body);
+      // At least one provider must be enabled in the fixture
+      const statusEntries = Object.entries(body.providers.status);
+      assert.ok(statusEntries.length >= 1, 'fixture has at least one enabled provider');
+      for (const [name, status] of statusEntries) {
+        assert.ok('activeSpawns' in status,
+          `providers.status.${name}.activeSpawns MUST be present (ADR 0002 Amendment 6)`);
+        assert.equal(typeof status.activeSpawns, 'number');
+        assert.ok(status.activeSpawns >= 0, 'activeSpawns >= 0');
+      }
+    });
+
     it('21d: owner_only_endpoints config opt-out — empty list → guest gets full payload', async () => {
       __setAuthConfig({
         allow_anonymous: true,
@@ -11988,3 +12035,896 @@ describe('Suite 26 — D52 audit rotation (Phase 3, ADR 0008 § 5)', () => {
     });
   });
 });
+
+// ── Suite 27: D57 streaming singleflight (cache layer) ─────────────────────
+//
+// Authority: ADR 0005 Amendment 8 §§1-11, 14 (issue #16, D57).
+//
+// Cache-layer unit tests for the streaming-singleflight tee fan-out. Each
+// test constructs a fake sourceFactory that returns an async generator
+// producing a fixed chunk sequence; the cache store coordinates dedup +
+// late-joiner replay + per-client backpressure. No real provider CLIs are
+// spawned; no HTTP requests issued. D58 will wire this into server.mjs in a
+// separate PR (Iron Rule 11).
+//
+// Test count: 12 (one per Amendment 8 §§1-11 + §14 fixture + composite-key
+// isolation). Aim for +12 tests minimum per D57 prompt.
+
+describe('Suite 27 — D57 streaming singleflight (cache layer)', () => {
+  // Helper: build a deterministic async source. Each yielded chunk waits a
+  // microtask so the tee/queue dynamics are observable across concurrent
+  // attached clients.
+  function makeChunkSequence(chunks, opts = {}) {
+    let returnedCount = 0;
+    const gen = (async function* fakeStream() {
+      try {
+        for (const c of chunks) {
+          if (opts.signal && opts.signal.aborted) return;
+          yield c;
+          await new Promise(r => setImmediate(r));
+        }
+      } finally {
+        returnedCount++;
+        if (opts.onReturn) opts.onReturn(returnedCount);
+      }
+    })();
+    return gen;
+  }
+
+  // D57 — ADR 0005 Amendment 8 §1: cache-hit + single client + source-mode
+  it('27a — single client streaming: behaviour identical to today (source role, full sequence)', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    let spawns = 0;
+    const factory = () => {
+      spawns++;
+      return makeChunkSequence(['a', 'b', 'c']);
+    };
+    const r = await store.getOrComputeStreaming('k1', 'ck-27a', factory);
+    assert.equal(r.isFirst, true);
+    assert.equal(r.role, 'source');
+    const out = [];
+    for await (const c of r.stream) out.push(c);
+    assert.deepEqual(out, ['a', 'b', 'c']);
+    assert.equal(spawns, 1);
+    // After completion, the cache should have an entry for replay.
+    const cached = await store.get('k1', 'ck-27a');
+    assert.ok(cached, 'cache populated post-completion');
+    assert.deepEqual(cached.value, ['a', 'b', 'c']);
+  });
+
+  // D57 — ADR 0005 Amendment 8 §1, §4: 2 concurrent identical streams
+  it('27b — 2 concurrent identical streams: only 1 sourceFactory call; identical chunks in order', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    let spawns = 0;
+    const factory = () => {
+      spawns++;
+      return makeChunkSequence(['x', 'y', 'z']);
+    };
+    const p1 = store.getOrComputeStreaming('k', 'ck-27b', factory, { clientId: 'A' });
+    const p2 = store.getOrComputeStreaming('k', 'ck-27b', factory, { clientId: 'B' });
+    const [r1, r2] = await Promise.all([p1, p2]);
+    assert.equal(r1.isFirst, true);
+    assert.equal(r1.role, 'source');
+    assert.equal(r2.isFirst, false);
+    assert.equal(r2.role, 'attached');
+    const out1 = [];
+    const out2 = [];
+    const c1 = (async () => { for await (const c of r1.stream) out1.push(c); })();
+    const c2 = (async () => { for await (const c of r2.stream) out2.push(c); })();
+    await Promise.all([c1, c2]);
+    assert.equal(spawns, 1, 'sourceFactory invoked exactly once');
+    assert.deepEqual(out1, ['x', 'y', 'z']);
+    assert.deepEqual(out2, ['x', 'y', 'z']);
+  });
+
+  // D57 — ADR 0005 Amendment 8 §5: mid-stream join (replay burst + live tail)
+  //  + post-completion join (cache_hit)
+  it('27c — 3 concurrent, mid-stream join: A=source, B=attached (burst + tail), C=cache_hit', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    let spawns = 0;
+    // Six chunks. A iterates fast; B attaches after A has consumed ~3.
+    const factory = () => {
+      spawns++;
+      return makeChunkSequence(['1', '2', '3', '4', '5', '6']);
+    };
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27c', factory, { clientId: 'A' });
+    assert.equal(r1.role, 'source');
+
+    // A iterates the first 3 chunks before B attaches.
+    const itA = r1.stream;
+    const outA = [];
+    const n1 = await itA.next(); outA.push(n1.value);
+    const n2 = await itA.next(); outA.push(n2.value);
+    const n3 = await itA.next(); outA.push(n3.value);
+
+    // Now B attaches mid-stream. Its replay drain picks up everything in
+    // accumulatedChunks at attach-time + the live tail thereafter.
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27c', factory, { clientId: 'B' });
+    assert.equal(r2.role, 'attached');
+
+    // A finishes consuming.
+    const outA2 = [];
+    for await (const c of itA) outA2.push(c);
+    const outB = [];
+    for await (const c of r2.stream) outB.push(c);
+
+    assert.deepEqual([...outA, ...outA2], ['1', '2', '3', '4', '5', '6']);
+    // B receives the full sequence too (burst replays earlier chunks + live tail).
+    assert.deepEqual(outB, ['1', '2', '3', '4', '5', '6']);
+    assert.equal(spawns, 1, 'still only 1 spawn');
+
+    // C attaches AFTER source completion → cache_hit (not inflight, not respawn).
+    const r3 = await store.getOrComputeStreaming('k', 'ck-27c', factory, { clientId: 'C' });
+    assert.equal(r3.role, 'cache_hit');
+    assert.equal(r3.isFirst, false);
+    const outC = [];
+    for await (const c of r3.stream) outC.push(c);
+    assert.deepEqual(outC, ['1', '2', '3', '4', '5', '6']);
+    assert.equal(spawns, 1, 'no additional spawns');
+  });
+
+  // D57 — ADR 0005 Amendment 8 §9: first client early-return, others continue
+  it('27d — first client iterator early-return mid-stream: others continue; source NOT aborted; cache written', async () => {
+    const warnings = [];
+    const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) });
+    let sourceReturned = false;
+    const factory = () => (async function* () {
+      try {
+        for (let i = 0; i < 6; i++) {
+          yield `chunk-${i}`;
+          await new Promise(r => setImmediate(r));
+        }
+      } finally { sourceReturned = true; }
+    })();
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27d', factory, { clientId: 'A' });
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27d', factory, { clientId: 'B' });
+    // A early-returns after 2 chunks; B keeps consuming.
+    const itA = r1.stream;
+    const outA = [];
+    outA.push((await itA.next()).value);
+    outA.push((await itA.next()).value);
+    await itA.return(); // simulates HTTP client close
+    const outB = [];
+    for await (const c of r2.stream) outB.push(c);
+    // Source should have completed normally (not aborted) because B was still attached.
+    assert.equal(sourceReturned, true);
+    assert.deepEqual(outB, ['chunk-0', 'chunk-1', 'chunk-2', 'chunk-3', 'chunk-4', 'chunk-5']);
+    // Cache should be populated post-completion (B was a live client to the end).
+    const cached = await store.get('k', 'ck-27d');
+    assert.ok(cached, 'cache populated despite A\'s mid-stream return');
+    // No abort warning emitted.
+    assert.equal(warnings.filter(w => w.msg === 'streaming_inflight_abort').length, 0);
+  });
+
+  // D57 — ADR 0005 Amendment 8 §9: all clients disconnect → source aborted, no cache
+  it('27e — all clients disconnect mid-stream: source aborted via AbortController; no cache write; next call respawns', async () => {
+    const warnings = [];
+    const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) });
+    let spawns = 0;
+    let sourceReturned = false;
+    const factory = () => {
+      spawns++;
+      return (async function* () {
+        try {
+          for (let i = 0; i < 20; i++) {
+            yield `c-${i}`;
+            await new Promise(r => setImmediate(r));
+          }
+        } finally { sourceReturned = true; }
+      })();
+    };
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27e', factory, { clientId: 'A' });
+    const itA = r1.stream;
+    await itA.next();
+    await itA.next();
+    await itA.return();
+    // Wait microtasks for the tee task to observe attachedClients.size === 0 and abort.
+    await new Promise(r => setTimeout(r, 30));
+    assert.equal(sourceReturned, true);
+    assert.equal(warnings.filter(w => w.msg === 'streaming_inflight_abort').length, 1);
+    // No cache entry — subsequent call respawns (no inflight, no cache hit).
+    const cached = await store.get('k', 'ck-27e');
+    assert.equal(cached, null, 'no cache write on abort');
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27e', factory, { clientId: 'B' });
+    assert.equal(r2.role, 'source', 'subsequent call gets fresh source spawn');
+    assert.equal(spawns, 2);
+    // Drain r2 so it doesn't dangle.
+    for await (const _c of r2.stream) { /* drain */ }
+  });
+
+  // D57 — ADR 0005 Amendment 8 §4: source throws mid-stream
+  it('27f — source throws mid-stream: all attached clients receive the error; no cache write; entry removed', async () => {
+    const warnings = [];
+    const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) });
+    const err = new Error('synthetic source failure');
+    const factory = () => (async function* () {
+      yield 'a';
+      await new Promise(r => setImmediate(r));
+      yield 'b';
+      await new Promise(r => setImmediate(r));
+      throw err;
+    })();
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27f', factory, { clientId: 'A' });
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27f', factory, { clientId: 'B' });
+    // Both clients should reject when the source throws.
+    let e1, e2;
+    const c1 = (async () => { try { for await (const _c of r1.stream) {} } catch (e) { e1 = e; } })();
+    const c2 = (async () => { try { for await (const _c of r2.stream) {} } catch (e) { e2 = e; } })();
+    await Promise.all([c1, c2]);
+    assert.ok(e1, 'client A received error');
+    assert.ok(e2, 'client B received error');
+    assert.equal(e1.message, 'synthetic source failure');
+    assert.equal(e2.message, 'synthetic source failure');
+    // No cache write.
+    const cached = await store.get('k', 'ck-27f');
+    assert.equal(cached, null);
+    // Inflight entry removed (next call would respawn).
+    assert.equal(store.stats('k').inflightCount, 0);
+  });
+
+  // D57 — ADR 0005 Amendment 8 §8: backpressure — slow client overflows queue
+  it('27g — backpressure: slow client queue overflow → STREAM_BACKPRESSURE terminator; fast client unaffected', async () => {
+    const warnings = [];
+    const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) });
+    // Inject perClientQueueCap=1024 bytes; chunks ~256B each.
+    const bigText = 'x'.repeat(250);
+    const factory = () => makeChunkSequence(
+      [{ idx: 0, t: bigText }, { idx: 1, t: bigText }, { idx: 2, t: bigText }, { idx: 3, t: bigText }, { idx: 4, t: bigText }, { idx: 5, t: bigText }, { type: 'stop', finish_reason: 'stop' }]
+    );
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27g', factory, { clientId: 'fast', perClientQueueCap: 1024 });
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27g', factory, { clientId: 'slow', perClientQueueCap: 1024 });
+    // Fast drains immediately; slow defers consumption.
+    const fastOut = [];
+    const fast = (async () => { for await (const c of r1.stream) fastOut.push(c); })();
+    await fast;
+    // Now consume slow's stream; should hit backpressure terminator.
+    const slowOut = [];
+    for await (const c of r2.stream) slowOut.push(c);
+    // Fast client received the full sequence.
+    assert.equal(fastOut.length, 7);
+    assert.deepEqual(fastOut[6], { type: 'stop', finish_reason: 'stop' });
+    // Slow client received some prefix + the STREAM_BACKPRESSURE terminator.
+    assert.deepEqual(slowOut[slowOut.length - 2], { type: 'stop', finish_reason: 'length' });
+    assert.equal(slowOut[slowOut.length - 1], '[DONE]');
+    assert.ok(slowOut.length < 7, 'slow client cut short before reaching natural end');
+    // Backpressure warning emitted for slow client.
+    const bp = warnings.filter(w => w.msg === 'stream_backpressure_disconnect');
+    assert.ok(bp.length >= 1, 'at least one stream_backpressure_disconnect emitted');
+    assert.ok(bp.some(w => w.meta.client_id === 'slow'), 'slow client identified in warning');
+  });
+
+  // D57 — ADR 0005 Amendment 8 §10: replay buffer cap — cache write skipped
+  it('27h — replay cap exceeded: cache write skipped; first caller still receives full stream; late joiner past cap gets STREAM_BACKPRESSURE', async () => {
+    const warnings = [];
+    const store = new CacheStore({ _warnFn: (msg, meta) => warnings.push({ msg, meta }) });
+    const big = 'y'.repeat(400);
+    // Source emits 6 chunks at ~400B each → ~2400B total, exceeds replay cap 1024.
+    const factory = () => makeChunkSequence(
+      [{ t: big }, { t: big }, { t: big }, { t: big }, { t: big }, { t: big }]
+    );
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27h', factory, {
+      clientId: 'first',
+      accumulatedReplayCap: 1024,
+      perClientQueueCap: 1024 * 1024, // huge per-client cap so first caller never overflows
+    });
+    // First caller iterates through.
+    const itA = r1.stream;
+    const outA = [];
+    // Pull 2 chunks then attempt late join while past cap.
+    outA.push((await itA.next()).value);
+    outA.push((await itA.next()).value);
+    outA.push((await itA.next()).value); // now well past 1024B accumulated
+    // Late joiner attaches past replay cap → gets STREAM_BACKPRESSURE.
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27h', factory, {
+      clientId: 'late',
+      accumulatedReplayCap: 1024,
+      perClientQueueCap: 1024,
+    });
+    assert.equal(r2.role, 'attached');
+    const outLate = [];
+    for await (const c of r2.stream) outLate.push(c);
+    // Late joiner's stream is just the backpressure terminator (drain over cap).
+    assert.deepEqual(outLate, [{ type: 'stop', finish_reason: 'length' }, '[DONE]']);
+    // Drain first caller fully.
+    for await (const c of itA) outA.push(c);
+    assert.equal(outA.length, 6, 'first caller receives full source stream');
+    // Cache write skipped.
+    const cached = await store.get('k', 'ck-27h');
+    assert.equal(cached, null, 'cache NOT written when replay cap exceeded');
+    // Replay-cap-exceeded warning emitted.
+    assert.ok(
+      warnings.some(w => w.msg === 'streaming_inflight_replay_cap_exceeded'),
+      'replay-cap-exceeded warning fired'
+    );
+  });
+
+  // D57 — ADR 0005 Amendment 8 §6: cache TTL race — late joiner attaches via inflight
+  it('27i — cache TTL race: cached entry expires during inflight; late joiner attaches via inflight Map', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    // Pre-populate cache with TTL=10ms — will expire shortly.
+    await store.set('k', 'ck-27i', ['cached-a', 'cached-b'], 10);
+    // Wait so the cached entry expires.
+    await new Promise(r => setTimeout(r, 20));
+    // Now a streaming request comes in: cache is expired → entry is recomputed.
+    let spawns = 0;
+    const factory = () => {
+      spawns++;
+      return makeChunkSequence(['live-1', 'live-2', 'live-3']);
+    };
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27i', factory, { clientId: 'A' });
+    assert.equal(r1.role, 'source', 'expired cache → fresh source spawn');
+    // While the source is running, a late joiner arrives — must attach via inflight Map.
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27i', factory, { clientId: 'B' });
+    assert.equal(r2.role, 'attached', 'late joiner attaches via inflight Map even after cache expiry');
+    assert.equal(spawns, 1, 'no respawn');
+    // Drain both.
+    const o1 = []; for await (const c of r1.stream) o1.push(c);
+    const o2 = []; for await (const c of r2.stream) o2.push(c);
+    assert.deepEqual(o1, ['live-1', 'live-2', 'live-3']);
+    assert.deepEqual(o2, ['live-1', 'live-2', 'live-3']);
+    // Inflight completion overwrites the expired slot.
+    const cached = await store.get('k', 'ck-27i');
+    assert.ok(cached);
+    assert.deepEqual(cached.value, ['live-1', 'live-2', 'live-3']);
+  });
+
+  // D57 — ADR 0005 Amendment 8 §7: sourceFactory throws → first caller errors; no zombie state
+  it('27j — sourceFactory throws (e.g. CONCURRENCY_LIMIT): first caller errors; subsequent call retries; no zombie inflight', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    let attempts = 0;
+    const factory = () => {
+      attempts++;
+      if (attempts === 1) {
+        throw new Error('CONCURRENCY_LIMIT');
+      }
+      return makeChunkSequence(['a', 'b']);
+    };
+    let firstErr;
+    try {
+      await store.getOrComputeStreaming('k', 'ck-27j', factory, { clientId: 'A' });
+    } catch (e) {
+      firstErr = e;
+    }
+    assert.ok(firstErr, 'first call rejected with factory error');
+    assert.equal(firstErr.message, 'CONCURRENCY_LIMIT');
+    // No zombie inflight entry left dangling.
+    assert.equal(store.stats('k').inflightCount, 0, 'no stale streaming-inflight entry');
+    // Subsequent call uses the factory again (it returns a real iterator now).
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27j', factory, { clientId: 'B' });
+    assert.equal(r2.role, 'source');
+    const out = [];
+    for await (const c of r2.stream) out.push(c);
+    assert.deepEqual(out, ['a', 'b']);
+    assert.equal(attempts, 2);
+  });
+
+  // D57 — ADR 0005 Amendment 8 §1: stats accounting (hits / misses / inflightCount)
+  it('27k — stats: hits incremented for cache_hit + attached; misses for source; inflightCount reflects active streaming entries', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    const factory = () => makeChunkSequence(['p', 'q', 'r']);
+
+    // First caller — miss.
+    const r1 = await store.getOrComputeStreaming('k', 'ck-27k', factory, { clientId: 'A' });
+    let stats = store.stats('k');
+    assert.equal(stats.misses, 1, 'first caller increments misses');
+    assert.equal(stats.hits, 0);
+    // Inflight entry alive during source phase.
+    assert.equal(stats.inflightCount, 1, 'streaming entry counted in inflightCount');
+
+    // Concurrent joiner — attached → hit.
+    const r2 = await store.getOrComputeStreaming('k', 'ck-27k', factory, { clientId: 'B' });
+    stats = store.stats('k');
+    assert.equal(stats.hits, 1, 'attached client increments hits');
+    // Drain both.
+    await Promise.all([
+      (async () => { for await (const _c of r1.stream) {} })(),
+      (async () => { for await (const _c of r2.stream) {} })(),
+    ]);
+    // After completion, entry removed from inflight.
+    stats = store.stats('k');
+    assert.equal(stats.inflightCount, 0, 'streaming entry removed post-completion');
+
+    // Third call after completion — cache_hit (also a hit).
+    const r3 = await store.getOrComputeStreaming('k', 'ck-27k', factory, { clientId: 'C' });
+    assert.equal(r3.role, 'cache_hit');
+    for await (const _c of r3.stream) { /* drain */ }
+    stats = store.stats('k');
+    assert.equal(stats.hits, 2, 'cache_hit also increments hits');
+  });
+
+  // D57 — ADR 0005 Amendment 8 §2: composite key isolation (keyId\0cacheKey)
+  it('27l — composite key isolation: same cacheKey + different keyId → two independent inflight entries + two spawns', async () => {
+    const store = new CacheStore({ _warnFn: () => {} });
+    let spawns = 0;
+    const factory = () => {
+      spawns++;
+      return makeChunkSequence(['n1', 'n2']);
+    };
+    // Same cacheKey, different keyId.
+    const p1 = store.getOrComputeStreaming('key-A', 'shared-cache-key', factory, { clientId: 'A' });
+    const p2 = store.getOrComputeStreaming('key-B', 'shared-cache-key', factory, { clientId: 'B' });
+    const [r1, r2] = await Promise.all([p1, p2]);
+    // Both should be `source` — no cross-keyId sharing.
+    assert.equal(r1.role, 'source');
+    assert.equal(r2.role, 'source');
+    assert.equal(spawns, 2, 'two spawns because two distinct (keyId,cacheKey) composites');
+    // Drain.
+    const out1 = []; for await (const c of r1.stream) out1.push(c);
+    const out2 = []; for await (const c of r2.stream) out2.push(c);
+    assert.deepEqual(out1, ['n1', 'n2']);
+    assert.deepEqual(out2, ['n1', 'n2']);
+    // Stats: each keyId has its own miss counter.
+    assert.equal(store.stats('key-A').misses, 1);
+    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)');
+  });
+});