From bc6f10289340e9628ecd9f4a59816e2a10f0754f Mon Sep 17 00:00:00 2001
From: Epinephrine <luvs01@hanmail.net>
Date: Fri, 2 Oct 2026 05:36:30 -0700
Subject: [PATCH 1/4] fix(zed): bound buffered translation and cancel delegated
 readers

---
 scripts/test-layout/layout.json               |   1 +
 src/adapters/openai-responses/passthrough.ts  | 205 +++++++++---------
 src/adapters/zed.ts                           |  29 ++-
 structure/providers-and-adapters.md           |   2 +-
 structure/transports/byte-accounting.md       |   4 +
 tests/fixtures/test-layout-expected.json      |   1 +
 tests/providers/zed-retained-budget.test.ts   | 123 +++++++++++
 .../openai-responses-passthrough.test.ts      |   6 +-
 8 files changed, 264 insertions(+), 107 deletions(-)
 create mode 100644 tests/providers/zed-retained-budget.test.ts

diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json
index 646700d05c6..ff974ce9a72 100644
--- a/scripts/test-layout/layout.json
+++ b/scripts/test-layout/layout.json
@@ -2,6 +2,7 @@
   "version": 1,
   "root": "tests",
   "explicit": {
+    "zed-retained-budget.test.ts": "providers",
     "claude-intercept-on-demand.test.ts": "claude-integration",
     "main-account-hard-lock-thresholds.test.ts": "codex-integration",
     "main-account-external-usage.test.ts": "codex-integration",
diff --git a/src/adapters/openai-responses/passthrough.ts b/src/adapters/openai-responses/passthrough.ts
index 5aa3474942a..bd7da14ab14 100644
--- a/src/adapters/openai-responses/passthrough.ts
+++ b/src/adapters/openai-responses/passthrough.ts
@@ -590,112 +590,119 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig):
       let compactionEncryptedContent: string | undefined;
       let compactionEncryptedContentBytes = 0;
       let completedSeen = false;
-      for await (const event of decodeServerSentEvents(response.body, { translatorBudget: budget })) {
-        let payload: unknown;
-        try { payload = JSON.parse(event.data); } catch { continue; }
-        if (!isPlainObject(payload)) continue;
-        switch (payload.type) {
-          case "response.output_text.delta":
-            if (typeof payload.delta === "string") {
-              const next = deltas + payload.delta;
-              const nextBytes = appendedUtf8Bytes(deltasBytes, deltasLastCodeUnit, payload.delta);
-              const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" });
-              deltas = next;
-              reservation.commitRetained();
-              budget.releaseRetained(deltasBytes, { kind: "retained_collectors" });
-              deltasBytes = nextBytes;
-              if (payload.delta.length > 0) deltasLastCodeUnit = payload.delta.charCodeAt(payload.delta.length - 1);
-            }
-            break;
-          case "response.output_text.done":
-            if (typeof payload.text === "string") {
-              const next = doneText + payload.text;
-              const nextBytes = appendedUtf8Bytes(doneTextBytes, doneTextLastCodeUnit, payload.text);
-              const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" });
-              doneText = next;
-              reservation.commitRetained();
-              budget.releaseRetained(doneTextBytes, { kind: "retained_collectors" });
-              doneTextBytes = nextBytes;
-              if (payload.text.length > 0) doneTextLastCodeUnit = payload.text.charCodeAt(payload.text.length - 1);
-            }
-            break;
-          case "response.failed":
-          case "error":
-            yield { type: "error", message: responsesErrorMessage(payload.response ?? payload) };
-            return;
-          case "response.incomplete":
-            yield { type: "incomplete", reason: responsesErrorMessage(payload.response ?? payload) };
-            return;
-          case "response.completed":
-            {
-              completedSeen = true;
-              const responsePayload = isPlainObject(payload.response) ? payload.response : undefined;
-              const output = Array.isArray(responsePayload?.output) ? responsePayload.output : [];
-              const compaction = output.find(item => isPlainObject(item) && item.type === "compaction");
-              if (isPlainObject(compaction) && typeof compaction.encrypted_content === "string") {
-                const nextEncryptedContent = compaction.encrypted_content;
-                const nextEncryptedContentBytes = Buffer.byteLength(nextEncryptedContent, "utf8");
-                const reservation = budget.reserveTransient(nextEncryptedContentBytes, { kind: "retained_collectors" });
-                compactionEncryptedContent = nextEncryptedContent;
+      let ciphertextTransferred = false;
+      try {
+        for await (const event of decodeServerSentEvents(response.body, { translatorBudget: budget })) {
+          let payload: unknown;
+          try { payload = JSON.parse(event.data); } catch { continue; }
+          if (!isPlainObject(payload)) continue;
+          switch (payload.type) {
+            case "response.output_text.delta":
+              if (typeof payload.delta === "string") {
+                const next = deltas + payload.delta;
+                const nextBytes = appendedUtf8Bytes(deltasBytes, deltasLastCodeUnit, payload.delta);
+                const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" });
+                deltas = next;
                 reservation.commitRetained();
-                budget.releaseRetained(compactionEncryptedContentBytes, { kind: "retained_collectors" });
-                compactionEncryptedContentBytes = nextEncryptedContentBytes;
+                budget.releaseRetained(deltasBytes, { kind: "retained_collectors" });
+                deltasBytes = nextBytes;
+                if (payload.delta.length > 0) deltasLastCodeUnit = payload.delta.charCodeAt(payload.delta.length - 1);
               }
-              const next = responsesPayloadText(payload.response);
-              const nextBytes = Buffer.byteLength(next, "utf8");
-              const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" });
-              snapshot = next;
-              reservation.commitRetained();
-              budget.releaseRetained(snapshotBytes, { kind: "retained_collectors" });
-              snapshotBytes = nextBytes;
-            }
-            {
-              const nextUsage = usageFromResponsesPayload(payload.response);
-              // The attached raw usage object can be event-sized (unknown keys carry arbitrary
-              // values); it stays reachable until the terminal yields, so charge it like the
-              // adjacent retained collectors or it would defeat the per-request memory cap.
-              const nextRawBytes = nextUsage?.rawUsage === undefined ? 0
-                : Buffer.byteLength(JSON.stringify(nextUsage.rawUsage), "utf8");
-              if (nextRawBytes > 0) {
-                const reservation = budget.reserveTransient(nextRawBytes, { kind: "retained_collectors" });
-                usage = nextUsage;
+              break;
+            case "response.output_text.done":
+              if (typeof payload.text === "string") {
+                const next = doneText + payload.text;
+                const nextBytes = appendedUtf8Bytes(doneTextBytes, doneTextLastCodeUnit, payload.text);
+                const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" });
+                doneText = next;
                 reservation.commitRetained();
-              } else {
-                usage = nextUsage;
+                budget.releaseRetained(doneTextBytes, { kind: "retained_collectors" });
+                doneTextBytes = nextBytes;
+                if (payload.text.length > 0) doneTextLastCodeUnit = payload.text.charCodeAt(payload.text.length - 1);
               }
-              if (usageRawBytes > 0) {
-                budget.releaseRetained(usageRawBytes, { kind: "retained_collectors" });
+              break;
+            case "response.failed":
+            case "error":
+              yield { type: "error", message: responsesErrorMessage(payload.response ?? payload) };
+              return;
+            case "response.incomplete":
+              yield { type: "incomplete", reason: responsesErrorMessage(payload.response ?? payload) };
+              return;
+            case "response.completed":
+              {
+                completedSeen = true;
+                const responsePayload = isPlainObject(payload.response) ? payload.response : undefined;
+                const output = Array.isArray(responsePayload?.output) ? responsePayload.output : [];
+                const compaction = output.find(item => isPlainObject(item) && item.type === "compaction");
+                if (isPlainObject(compaction) && typeof compaction.encrypted_content === "string") {
+                  const nextEncryptedContent = compaction.encrypted_content;
+                  const nextEncryptedContentBytes = Buffer.byteLength(nextEncryptedContent, "utf8");
+                  const reservation = budget.reserveTransient(nextEncryptedContentBytes, { kind: "retained_collectors" });
+                  compactionEncryptedContent = nextEncryptedContent;
+                  reservation.commitRetained();
+                  budget.releaseRetained(compactionEncryptedContentBytes, { kind: "retained_collectors" });
+                  compactionEncryptedContentBytes = nextEncryptedContentBytes;
+                }
+                const next = responsesPayloadText(payload.response);
+                const nextBytes = Buffer.byteLength(next, "utf8");
+                const reservation = budget.reserveTransient(nextBytes, { kind: "retained_collectors" });
+                snapshot = next;
+                reservation.commitRetained();
+                budget.releaseRetained(snapshotBytes, { kind: "retained_collectors" });
+                snapshotBytes = nextBytes;
               }
-              usageRawBytes = nextRawBytes;
-            }
-            break;
-        }
-        // Buffered text is still upstream progress, but gateway keepalives are not.
-        // Yield after accounting, directly to the consumer: no progress queue or content leak.
-        if (
-          !completedSeen
-          && (payload.type === "response.output_text.delta"
-            || payload.type === "response.reasoning_summary_text.delta"
-            || payload.type === "response.reasoning_text.delta")
-          && typeof payload.delta === "string"
-          && payload.delta.length > 0
-        ) {
-          yield { type: "heartbeat" };
+              {
+                const nextUsage = usageFromResponsesPayload(payload.response);
+                // The attached raw usage object can be event-sized (unknown keys carry arbitrary
+                // values); it stays reachable until the terminal yields, so charge it like the
+                // adjacent retained collectors or it would defeat the per-request memory cap.
+                const nextRawBytes = nextUsage?.rawUsage === undefined ? 0
+                  : Buffer.byteLength(JSON.stringify(nextUsage.rawUsage), "utf8");
+                if (nextRawBytes > 0) {
+                  const reservation = budget.reserveTransient(nextRawBytes, { kind: "retained_collectors" });
+                  usage = nextUsage;
+                  reservation.commitRetained();
+                } else {
+                  usage = nextUsage;
+                }
+                if (usageRawBytes > 0) {
+                  budget.releaseRetained(usageRawBytes, { kind: "retained_collectors" });
+                }
+                usageRawBytes = nextRawBytes;
+              }
+              break;
+          }
+          // Buffered text is still upstream progress, but gateway keepalives are not.
+          // Yield after accounting, directly to the consumer: no progress queue or content leak.
+          if (
+            !completedSeen
+            && (payload.type === "response.output_text.delta"
+              || payload.type === "response.reasoning_summary_text.delta"
+              || payload.type === "response.reasoning_text.delta")
+            && typeof payload.delta === "string"
+            && payload.delta.length > 0
+          ) {
+            yield { type: "heartbeat" };
+          }
         }
+        // Gateways differ in which of these they emit; prefer the authoritative
+        // completed snapshot so text is never double-counted.
+        const text = snapshot || doneText || deltas;
+        if (text) yield { type: "text_delta", text };
+        // The bridge owns the raw ciphertext lease once it receives done.
+        ciphertextTransferred = true;
+        yield {
+          type: "done",
+          ...(usage ? { usage } : {}),
+          ...(compactionEncryptedContent ? { compactionEncryptedContent } : {}),
+        };
+      } finally {
+        budget.releaseRetained(
+          deltasBytes + doneTextBytes + snapshotBytes + usageRawBytes
+            + (ciphertextTransferred ? 0 : compactionEncryptedContentBytes),
+          { kind: "retained_collectors" },
+        );
       }
-      // Gateways differ in which of these they emit; prefer the authoritative
-      // completed snapshot so text is never double-counted.
-      const text = snapshot || doneText || deltas;
-      if (text) yield { type: "text_delta", text };
-      budget.releaseRetained(
-        deltasBytes + doneTextBytes + snapshotBytes + usageRawBytes,
-        { kind: "retained_collectors" },
-      );
-      yield {
-        type: "done",
-        ...(usage ? { usage } : {}),
-        ...(compactionEncryptedContent ? { compactionEncryptedContent } : {}),
-      };
     },
 
     async parseResponse(response: Response, budget: TranslatorBudget): Promise<AdapterEvent[]> {
diff --git a/src/adapters/zed.ts b/src/adapters/zed.ts
index 6dad922cfd7..4d43aeda350 100644
--- a/src/adapters/zed.ts
+++ b/src/adapters/zed.ts
@@ -5,7 +5,7 @@ import { createGoogleAdapter } from "./google";
 import { createOpenAIChatAdapter } from "./openai-chat";
 import { createResponsesPassthroughAdapter } from "./openai-responses";
 import type { AdapterEvent, OcxParsedRequest, OcxProviderConfig } from "../types";
-import { createTranslatorBudget, type TranslatorBudget } from "../lib/translator-budget";
+import { createTranslatorBudget, releaseTranslatedEvent, retainTranslatedEvent, type TranslatorBudget } from "../lib/translator-budget";
 import { redactSecretString } from "../lib/redact";
 import {
   normalizeZedProvider,
@@ -365,12 +365,33 @@ export function createZedAdapter(provider: OcxProviderConfig): ProviderAdapter {
         status: response.status,
         headers: { "Content-Type": "text/event-stream" },
       });
-      yield* delegate.adapter.parseStream(translated, budget);
+      try {
+        yield* delegate.adapter.parseStream(translated, budget);
+      } finally {
+        // Some delegates only unlock their reader when a collector stops early.
+        try { await translated.body?.cancel(); } catch { /* already closed/errored */ }
+      }
     },
     async parseResponse(response: Response, budget: TranslatorBudget): Promise<AdapterEvent[]> {
       const events: AdapterEvent[] = [];
-      for await (const event of adapter.parseStream(response, budget)) events.push(event);
-      return events;
+      try {
+        for await (const event of adapter.parseStream(response, budget)) {
+          try {
+            retainTranslatedEvent(event, budget, events.at(-1));
+          } catch (error) {
+            // A refused Responses done event never transfers its source ciphertext lease.
+            if (delegate?.provider === "open_ai" && event.type === "done" && event.compactionEncryptedContent) {
+              budget.releaseRetained(Buffer.byteLength(event.compactionEncryptedContent), { kind: "retained_collectors" });
+            }
+            throw error;
+          }
+          events.push(event);
+        }
+        return events;
+      } catch (error) {
+        for (const event of events) releaseTranslatedEvent(event, budget);
+        throw error;
+      }
     },
   };
   return adapter;
diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md
index 8104b16cd03..d4331c9495c 100644
--- a/structure/providers-and-adapters.md
+++ b/structure/providers-and-adapters.md
@@ -185,7 +185,7 @@ rewrite rules and the routed-id settlement.
 | `src/adapters/declaration-carrier.ts`, `src/adapters/input-media-guard.ts` | Default-deny allowlists for constraints the normalized request carries but a wire may not be able to express: `tools[*].allowed_callers`, which fences a tool off from callers, and inline document bytes. Both are refused with a 400 at the single guard every registered adapter passes through, rather than left to each adapter, because an adapter that never learned about the carrier rebuilds without it and answers normally. `allowed_callers` reaches the `anthropic` wire; document bytes reach `anthropic`, `openai-chat` and `google`; the `openai-responses` wire is exempt from the whole guard because it forwards the original body. Adding an `AdapterWire` member makes the omission visible in these lists instead of at a customer's upstream. The unrestricted `["direct"]` caller default is not a restriction. |
 | `src/adapters/azure.ts` | Azure OpenAI bridge. |
 | `src/adapters/cursor.ts`, `src/adapters/cursor/` | Cursor protobuf transport: discovery, request builder, event decoding, MCP, thread continuity, native-exec policy. |
-| `src/adapters/zed.ts`, `src/providers/zed.ts`, `src/oauth/zed.ts` | Zed Hosted AI native-app RSA login, account-scoped short-lived LLM-token exchange, bounded live model discovery, and provider-envelope translation through the existing Anthropic, Google, Responses, and Chat builders. Experimental and unofficial; the Zed terms risk is intentionally surfaced in user docs and release review. |
+| `src/adapters/zed.ts`, `src/providers/zed.ts`, `src/oauth/zed.ts` | Zed Hosted AI native-app RSA login, account-scoped short-lived LLM-token exchange, bounded live model discovery, and provider-envelope translation through the existing Anthropic, Google, Responses, and Chat builders. Buffered delivery retains events under the shared translator budget and cancels delegated streams on early exit. Experimental and unofficial; the Zed terms risk is intentionally surfaced in user docs and release review. |
 | `src/adapters/devin.ts`, `src/adapters/devin/cloud-direct/` | Devin runTurn transport over Cognition Connect-RPC. Assistant reasoning replays as ChatMessagePrompt #11 thinking, #12 signature and #18 signature type (`src/adapters/devin/reasoning-signature.ts`): the #10/#21 pair arrives after the visible answer and becomes its own signature-only reasoning item, so a single unsigned thinking block plus exactly one signature-only block is replayed as one signed prompt, and a signature-only turn (GPT, Gemini) is replayed rather than dropped. An Anthropic signature is replayed, but because the streamed thinking is a summary the signature may not cover, a turn Cognition refuses with `invalid_argument` before any visible output (reasoning alone does not count) is retried once with Anthropic signatures withheld and the thinking text kept. `GetChatMessage` uses the Responses provider executor and shared physical-send budget; catalog, JWT, and `src/web-search/devin-executor.ts` native search support RPCs remain outside inference-send accounting. Provider-stated pre-output 429 reset delays are surfaced immediately by default, releasing shared active-turn capacity. A positive `OPENCODEX_DEVIN_STATED_RESET_WAIT_MS` explicitly enables bounded waiting and up to two replays on standalone turns, which hold that capacity until completion or cancellation. Combo children bypass that wait and surface a pre-output 429 so the next target can run. During an opted-in standalone wait, safe heartbeats commit the response preflight and keep the stream's stall watchdog fed. Invalid values fail closed to the immediate-refusal behavior. A recorded tenant host is used only for the stored account whose credential owns the transmitted key, searched in the configured provider id and then its deprecated alias; a configured, forwarded, or unmatched key uses the configured base URL or the US default. Native search previews the current route by effective adapter without mutating combo selection state, pins one admitted active-account snapshot for the request, and calls `GetWebSearchResults`, so it starts no CLI or second model. The wire model UID comes from the catalog's family metadata (`ClientModelConfig` #23/#30/#31): a family id with no effort anchors on the family's default member and selects the nearest enabled rung, rounding up first; an effort moves only the effort axis, to the lowest rung at or above it (else the highest below) while Fast Mode, 1M Context and the other axes stay at the anchor's values unless the caller asked for `fast` or a `1m` value; not lowering a requested effort outranks keeping those axes (an unranked member counts as lower), a disabled row the caller named is kept when the request still selects it so the preflight names that refusal, and resolution never leaves the family. Rows without family metadata fall back to suffix resolution, whose variant scan matches the collapsed base rather than a string prefix. With no caller or configured output cap, the selected row's catalog `maxOutputTokens` fills CompletionConfiguration #2; #3 is `max_newlines` and is sent at a fixed value, never a context window. The leading system text is sent as `GetChatMessage` #2, a failed tool result sets ChatMessagePrompt #9 and keeps an in-band `ERROR:` marker, and Gemini uids have JSON-Schema type arrays expressed as type-only `anyOf` branches in tool parameters. Shared constraints remain on the surrounding node, so nested schemas have one wire representation while outer enum/const, boolean constraints (including `not`, `oneOf`, and `allOf`) on null, and existing null-branch restrictions retain their conjunctive meaning. A null-only type keeps its null constraint even when an outer enum/const excludes null, preserving an unsatisfiable schema without producing an empty anyOf. Schema resource identifiers, anchors, definition maps, and existing anyOf branches remain on their original node. The type constraint is appended to existing allOf entries without shifting their indices, so JSON Pointer targets and reference scopes remain unchanged. A pre-output `invalid_argument` on a history whose word-piece estimate, using the sanitized and truncated tool descriptions actually sent on the wire, reaches 95% of the selected UID's catalog input window, capped by configured provider and model limits (512 KiB of text only when no window is known) is surfaced as `context_length_exceeded` so Codex compacts; a small request with the same code stays a plain 400. Known limit: a malformed schema on a history already at or above that 95% threshold is also reported as overflow because the upstream returns the same `invalid_argument`. The `COGNITION_BLOCKLIST_REWRITES` sanitizer rewrites exact phrases Cognition refuses with `permission_denied`; it runs only on instruction surfaces — tool descriptions and the #2 system prompt — while message text, replayed thinking (#11), and tool-call arguments stay byte-exact, because a rewrite there would silently change literal content such as patches or quoted file bytes; a trigger inside a data field surfaces as the upstream `permission_denied` instead (the Codex `<permissions instructions>` escalation boilerplate is the recorded #2 case). |
 | `src/adapters/kiro.ts` and `src/adapters/kiro/` | Kiro event/tool/thinking/truncation/retry handling, including an egress-aware completion fallback, fixed public HTTP 5xx text, and closed-set status/code diagnostics. The original path is a facade over leaves for wire identity, reasoning, conversation state, token estimation, payload assembly, streaming, and the adapter. |
 | `src/adapters/mimo-free.ts` | Mimo Free transport (client identity + JWT). Concurrent requests share one JWT bootstrap bound only to its timeout; each request stops waiting on its own abort without cancelling the others. |
diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md
index 160b25b1426..e2da4c9e575 100644
--- a/structure/transports/byte-accounting.md
+++ b/structure/transports/byte-accounting.md
@@ -103,6 +103,10 @@ truncates diagnostic text at UTF-8 code-point boundaries without allocating arra
 byte sizing retains TextEncoder's coercion behavior for legacy non-string runtime callers.
 These optimizations do not add request queues, retry policies, or RSS-based admission gates.
 
+Zed's buffered response collector charges each delegated event before retaining it and releases
+owned events when collection fails. Its translated stream is cancelled on early exit for every
+provider family; the Responses delegate releases partial text and usage collectors on every exit.
+
 Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate.
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
 
diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json
index f25fa0b2e99..564d82a3c54 100644
--- a/tests/fixtures/test-layout-expected.json
+++ b/tests/fixtures/test-layout-expected.json
@@ -1,4 +1,5 @@
 {
+  "zed-retained-budget.test.ts": "providers",
   "claude-intercept-on-demand.test.ts": "claude-integration",
   "main-account-hard-lock-thresholds.test.ts": "codex-integration",
   "main-account-external-usage.test.ts": "codex-integration",
diff --git a/tests/providers/zed-retained-budget.test.ts b/tests/providers/zed-retained-budget.test.ts
new file mode 100644
index 00000000000..03fcaa46ed8
--- /dev/null
+++ b/tests/providers/zed-retained-budget.test.ts
@@ -0,0 +1,123 @@
+import { afterEach, describe, expect, spyOn, test } from "bun:test";
+import { createZedAdapter, type ZedProvider } from "../../src/adapters/zed";
+import { createTranslatorBudget, releaseTranslatedEvent, TranslatorBudgetExceededError } from "../../src/lib/translator-budget";
+import { clearZedCaches } from "../../src/providers/zed";
+import { parseRequest } from "../../src/responses/parser";
+
+const families: ZedProvider[] = ["anthropic", "google", "open_ai", "x_ai"];
+afterEach(clearZedCaches);
+
+async function configuredAdapter(family: ZedProvider) {
+  const adapter = createZedAdapter({ adapter: "zed", baseUrl: "https://cloud.zed.dev", apiKey: "fixture-token" });
+  const parsed = parseRequest({ model: "fixture-model", input: "hello", stream: false });
+  parsed._zedAuthContext = { userId: "fixture-user" };
+  const setupBudget = createTranslatorBudget();
+  const replies = [
+    { default_organization_id: "fixture-org" }, { token: "fixture-llm-token" },
+    { models: [{ id: "fixture-model", provider: family }] },
+  ];
+  try {
+    await adapter.buildRequest(parsed, {
+      headers: new Headers(), translatorBudget: setupBudget,
+      providerFetch: (async () => Response.json(replies.shift())) as typeof fetch,
+    });
+  } finally { setupBudget.dispose(); }
+  return adapter;
+}
+
+function textFrame(family: ZedProvider, text: string) {
+  if (family === "anthropic") return { type: "content_block_delta", index: 0, delta: { type: "text_delta", text } };
+  if (family === "google") return { candidates: [{ content: { parts: [{ text }] } }] };
+  if (family === "open_ai") return { type: "response.output_text.delta", item_id: "msg_fixture", output_index: 0, content_index: 0, delta: text };
+  return { choices: [{ index: 0, delta: { content: text }, finish_reason: null }] };
+}
+
+function completion(family: ZedProvider, count: number, text = "x".repeat(1024)) {
+  let pulled = 0;
+  let cancelled = false;
+  const body = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (pulled++ < count) {
+        controller.enqueue(new TextEncoder().encode(JSON.stringify({ event: textFrame(family, text) }) + "\n"));
+      } else {
+        controller.enqueue(new TextEncoder().encode(JSON.stringify({ status: "stream_ended" }) + "\n"));
+        controller.close();
+      }
+    },
+    cancel() { cancelled = true; },
+  }, { highWaterMark: 0 });
+  return { response: new Response(body), pulled: () => pulled, cancelled: () => cancelled };
+}
+
+describe("Zed buffered event ownership", () => {
+  for (const family of families) {
+    test(`${family}: aggregate overflow cancels upstream and releases retained events`, async () => {
+      const adapter = await configuredAdapter(family);
+      const budget = createTranslatorBudget({ maxTurnBytes: 16 * 1024 });
+      const source = completion(family, 128);
+      try {
+        try {
+          const events = await adapter.parseResponse!(source.response, budget);
+          expect(events.at(-1)).toMatchObject({ type: "error", code: "translation_buffer_limit" });
+          for (const event of events) releaseTranslatedEvent(event, budget);
+        } catch (error) {
+          expect(error).toMatchObject({ code: "translation_buffer_limit" });
+        }
+        expect(source.cancelled()).toBe(true);
+        expect(source.pulled()).toBeLessThan(32);
+        expect(budget.snapshot().currentBytes).toBe(0);
+        expect(budget.snapshot().overflows).toBeGreaterThan(0);
+      } finally { budget.dispose(); }
+    });
+    test(`${family}: normal text remains intact and transfers each retained lease`, async () => {
+      const adapter = await configuredAdapter(family);
+      const budget = createTranslatorBudget({ maxTurnBytes: 16 * 1024 });
+      try {
+        const events = await adapter.parseResponse!(completion(family, 2, "hello 🌍").response, budget);
+        expect(events.filter(event => event.type === "text_delta").map(event => event.text).join(""))
+          .toBe("hello 🌍hello 🌍");
+        expect(events.at(-1)?.type).toBe("done");
+        expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(JSON.stringify(events)));
+        for (const event of events) releaseTranslatedEvent(event, budget);
+        expect(budget.snapshot().currentBytes).toBe(0);
+      } finally { budget.dispose(); }
+    });
+  }
+});
+
+
+describe("Zed Responses ciphertext failure ownership", () => {
+  const completed = JSON.stringify({ event: { type: "response.completed", response: {
+    output: [{ type: "compaction", encrypted_content: "fixture-ciphertext" }],
+  } } }) + "\n";
+  test("read failure after a completed snapshot releases ciphertext with no done owner", async () => {
+    const adapter = await configuredAdapter("open_ai");
+    const budget = createTranslatorBudget();
+    const failure = new Error("fixture read failure");
+    let sent = false;
+    const body = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (!sent) { sent = true; controller.enqueue(new TextEncoder().encode(completed)); }
+        else controller.error(failure);
+      },
+    }, { highWaterMark: 0 });
+    try {
+      await expect(adapter.parseResponse!(new Response(body), budget)).rejects.toThrow("fixture read failure");
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { budget.dispose(); }
+  });
+  test("a refused done-event lease releases its untransferred ciphertext source", async () => {
+    const adapter = await configuredAdapter("open_ai");
+    const budget = createTranslatorBudget();
+    const originalCharge = budget.chargeRetained.bind(budget);
+    const charge = spyOn(budget, "chargeRetained").mockImplementation((bytes, scope) => {
+      if (scope.kind === "retained_collectors") throw new TranslatorBudgetExceededError("retained_collectors", 1);
+      originalCharge(bytes, scope);
+    });
+    try {
+      await expect(adapter.parseResponse!(new Response(completed), budget)).rejects.toMatchObject({ code: "translation_buffer_limit" });
+      expect(charge.mock.calls.filter(([, scope]) => scope.kind === "retained_collectors")).toHaveLength(1);
+      expect(budget.snapshot().currentBytes).toBe(0);
+    } finally { charge.mockRestore(); budget.dispose(); }
+  });
+});
diff --git a/tests/responses/openai-responses-passthrough.test.ts b/tests/responses/openai-responses-passthrough.test.ts
index f9c2029f38a..736f715c23b 100644
--- a/tests/responses/openai-responses-passthrough.test.ts
+++ b/tests/responses/openai-responses-passthrough.test.ts
@@ -156,7 +156,7 @@ describe("Responses request and compaction byte accounting", () => {
       expect(await iterator.next()).toEqual({ done: false, value: { type: "heartbeat" } });
       await expect(iterator.next()).rejects.toMatchObject({ code: "translation_buffer_limit" });
       expect(cancelled).toBe(true);
-      expect(budget.snapshot()).toMatchObject({ currentBytes: fragment.length, overflows: 1 });
+      expect(budget.snapshot()).toMatchObject({ currentBytes: 0, overflows: 1 });
     } finally { await iterator.return(undefined); budget.dispose(); }
     expect(budget.snapshot().currentBytes).toBe(0);
   });
@@ -200,7 +200,7 @@ describe("Responses request and compaction byte accounting", () => {
   });
 
   test.each(["response.failed", "response.incomplete", "consumer return"])(
-    "%s leaves collector cleanup with the owning budget and cancels upstream", async ending => {
+    "%s releases partial collectors and cancels upstream", async ending => {
       const partial = "中\ud800";
       const endingPayload = ending === "response.failed"
         ? { type: ending, response: { error: { message: "stopped" } } }
@@ -224,7 +224,7 @@ describe("Responses request and compaction byte accounting", () => {
           await iterator.return(undefined);
         }
         expect(cancelled).toBe(true);
-        expect(budget.snapshot().currentBytes).toBe(encoder.encode(partial).byteLength);
+        expect(budget.snapshot().currentBytes).toBe(0);
       } finally { await iterator.return(undefined); budget.dispose(); }
       expect(budget.snapshot().currentBytes).toBe(0);
     },

From 6485b1503a0a07b0184694741296fc51b4e0285c Mon Sep 17 00:00:00 2001
From: Epinephrine <luvs01@hanmail.net>
Date: Fri, 2 Oct 2026 06:23:50 -0700
Subject: [PATCH 2/4] docs(zed): specify translated event and ciphertext
 ownership

---
 structure/transports/byte-accounting.md | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md
index e2da4c9e575..b7ebb27522b 100644
--- a/structure/transports/byte-accounting.md
+++ b/structure/transports/byte-accounting.md
@@ -103,9 +103,11 @@ truncates diagnostic text at UTF-8 code-point boundaries without allocating arra
 byte sizing retains TextEncoder's coercion behavior for legacy non-string runtime callers.
 These optimizations do not add request queues, retry policies, or RSS-based admission gates.
 
-Zed's buffered response collector charges each delegated event before retaining it and releases
-owned events when collection fails. Its translated stream is cancelled on early exit for every
-provider family; the Responses delegate releases partial text and usage collectors on every exit.
+The buffered collector in `src/adapters/zed.ts` charges each delegated event before retaining it
+and releases owned events when collection fails. Its translated stream is cancelled on early exit
+for every provider family. `src/adapters/openai-responses/passthrough.ts` releases partial text and
+usage collectors on every exit, and compaction ciphertext unless its lease transfers with the
+`done` event. Zed releases that source ciphertext lease if the collector cannot retain `done`.
 
 Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate.
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.

From 79c88e9c8d8c370f562f64173eb28874329f00e3 Mon Sep 17 00:00:00 2001
From: Epinephrine <luvs01@hanmail.net>
Date: Fri, 2 Oct 2026 07:45:40 -0700
Subject: [PATCH 3/4] fix(bridge): release transferred event leases on every
 exit

---
 scripts/test-layout/layout.json               |   1 +
 src/adapters/openai-responses/passthrough.ts  |  10 +-
 src/adapters/zed.ts                           |  11 +-
 src/bridge/response-json.ts                   |  28 +--
 src/bridge/sse.ts                             |  19 +-
 src/responses/compaction.ts                   |  19 ++
 structure/transports/byte-accounting.md       |   8 +-
 structure/transports/responses.md             |   2 +-
 tests/fixtures/test-layout-expected.json      |   1 +
 tests/providers/zed-retained-budget.test.ts   |  13 ++
 .../compaction-event-ownership.test.ts        | 211 ++++++++++++++++++
 11 files changed, 293 insertions(+), 30 deletions(-)
 create mode 100644 tests/responses/compaction-event-ownership.test.ts

diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json
index ff974ce9a72..01ceaae5364 100644
--- a/scripts/test-layout/layout.json
+++ b/scripts/test-layout/layout.json
@@ -2,6 +2,7 @@
   "version": 1,
   "root": "tests",
   "explicit": {
+    "compaction-event-ownership.test.ts": "responses",
     "zed-retained-budget.test.ts": "providers",
     "claude-intercept-on-demand.test.ts": "claude-integration",
     "main-account-hard-lock-thresholds.test.ts": "codex-integration",
diff --git a/src/adapters/openai-responses/passthrough.ts b/src/adapters/openai-responses/passthrough.ts
index bd7da14ab14..c02c6ee4916 100644
--- a/src/adapters/openai-responses/passthrough.ts
+++ b/src/adapters/openai-responses/passthrough.ts
@@ -8,7 +8,7 @@ import { Buffer } from "node:buffer";
 import type { IncomingMeta, ProviderAdapter } from "../base";
 import { namespacedToolName, type AdapterEvent, type OcxParsedRequest, type OcxProviderConfig, type OcxUsage, type TierDecision } from "../../types";
 import { applyCodexRoutingHint, CODEX_RESPONSES_LITE_HEADER, CODEX_ROUTING_HINT_HEADER } from "../../codex/forward-transport-headers";
-import { COMPACT_PROMPT, compactionItemToText, decodeCompactionSummary, isCompactionItemType } from "../../responses/compaction";
+import { COMPACT_PROMPT, compactionItemToText, decodeCompactionSummary, isCompactionItemType, transferCompactionCiphertextLease } from "../../responses/compaction";
 import { decodeServerSentEvents } from "../../lib/sse-decoder";
 import {
   CODEX_FORWARD_BASE_URL,
@@ -689,13 +689,15 @@ export function createResponsesPassthroughAdapter(provider: OcxProviderConfig):
         // completed snapshot so text is never double-counted.
         const text = snapshot || doneText || deltas;
         if (text) yield { type: "text_delta", text };
-        // The bridge owns the raw ciphertext lease once it receives done.
-        ciphertextTransferred = true;
-        yield {
+        const done: AdapterEvent = {
           type: "done",
           ...(usage ? { usage } : {}),
           ...(compactionEncryptedContent ? { compactionEncryptedContent } : {}),
         };
+        // Exact event ownership survives buffered collection without charging a second copy.
+        transferCompactionCiphertextLease(done, budget, compactionEncryptedContentBytes);
+        ciphertextTransferred = true;
+        yield done;
       } finally {
         budget.releaseRetained(
           deltasBytes + doneTextBytes + snapshotBytes + usageRawBytes
diff --git a/src/adapters/zed.ts b/src/adapters/zed.ts
index 4d43aeda350..106d73769d0 100644
--- a/src/adapters/zed.ts
+++ b/src/adapters/zed.ts
@@ -6,6 +6,7 @@ import { createOpenAIChatAdapter } from "./openai-chat";
 import { createResponsesPassthroughAdapter } from "./openai-responses";
 import type { AdapterEvent, OcxParsedRequest, OcxProviderConfig } from "../types";
 import { createTranslatorBudget, releaseTranslatedEvent, retainTranslatedEvent, type TranslatorBudget } from "../lib/translator-budget";
+import { releaseCompactionCiphertextLease } from "../responses/compaction";
 import { redactSecretString } from "../lib/redact";
 import {
   normalizeZedProvider,
@@ -379,17 +380,17 @@ export function createZedAdapter(provider: OcxProviderConfig): ProviderAdapter {
           try {
             retainTranslatedEvent(event, budget, events.at(-1));
           } catch (error) {
-            // A refused Responses done event never transfers its source ciphertext lease.
-            if (delegate?.provider === "open_ai" && event.type === "done" && event.compactionEncryptedContent) {
-              budget.releaseRetained(Buffer.byteLength(event.compactionEncryptedContent), { kind: "retained_collectors" });
-            }
+            releaseCompactionCiphertextLease(event, budget);
             throw error;
           }
           events.push(event);
         }
         return events;
       } catch (error) {
-        for (const event of events) releaseTranslatedEvent(event, budget);
+        for (const event of events) {
+          releaseTranslatedEvent(event, budget);
+          releaseCompactionCiphertextLease(event, budget);
+        }
         throw error;
       }
     },
diff --git a/src/bridge/response-json.ts b/src/bridge/response-json.ts
index 23f8dacc1af..f74ebccf77b 100644
--- a/src/bridge/response-json.ts
+++ b/src/bridge/response-json.ts
@@ -17,7 +17,7 @@ import {
   type OcxErrorPayload,
 } from "../lib/errors";
 import { mayBecomePatchEnvelope, repairFreeformToolInput } from "../responses/apply-patch-envelope";
-import { encodeCompactionSummary } from "../responses/compaction";
+import { encodeCompactionSummary, releaseCompactionCiphertextLease } from "../responses/compaction";
 import { compileCodeModeHelperInput, resolveCodeModeHelperName } from "../responses/code-mode-helper-compat";
 import { isTruncatedStopReason, truncationReasonFor } from "../responses/truncated-stop-reason";
 import { encodeReasoningEnvelope, type ReasoningEnvelope } from "../responses/reasoning-envelope";
@@ -59,21 +59,21 @@ export function buildResponseJSON(
 ): Record<string, unknown> {
   // Default-budget safety net: a caller that omits the budget gets a bounded
   // default (disposed with the call), never the unbounded append path.
-  if (options?.translatorBudget) {
-    const body = buildResponseJSONWithBudget(events, modelId, options);
-    // A buffered turn delivers its whole answer as one body, so nothing calls the per-frame
-    // recorder on the SSE bridge. Without this the attempt would persist adapter events with
-    // zero relayed ones, which is the loss signal -- raised on every non-streaming request.
-    if (options.recordBufferedDelivery !== false) {
-      attemptDeliveryRecorder(options.translatorBudget)?.noteBufferedDelivery(body);
+  const ownsBudget = !options?.translatorBudget;
+  const budget = options?.translatorBudget ?? createTranslatorBudget();
+  try {
+    const body = buildResponseJSONWithBudget(events, modelId, { ...options, translatorBudget: budget });
+    // Buffered delivery has no per-frame recorder; retain its existing one-shot attribution.
+    if (options?.translatorBudget && options.recordBufferedDelivery !== false) {
+      attemptDeliveryRecorder(budget)?.noteBufferedDelivery(body);
     }
     return body;
-  }
-  const budget = createTranslatorBudget();
-  try {
-    return buildResponseJSONWithBudget(events, modelId, { ...options, translatorBudget: budget });
   } finally {
-    budget.dispose();
+    for (const event of events) {
+      releaseTranslatedEvent(event, budget);
+      releaseCompactionCiphertextLease(event, budget);
+    }
+    if (ownsBudget) budget.dispose();
   }
 }
 
@@ -618,7 +618,7 @@ function buildResponseJSONWithBudget(
       type: "compaction", id: `cmp_${uuid()}`,
       encrypted_content: compactionEncryptedContent ?? encodeCompactionSummary(joinChunks(batchCompaction)),
     };
-    pushOutput(item, compactionEncryptedContent ? bytesOf(compactionEncryptedContent) : batchCompaction.bytes);
+    pushOutput(item, compactionEncryptedContent ? 0 : batchCompaction.bytes);
   }
 
   const failure = errorEvent ? adapterFailureFromEvent(errorEvent) : undefined;
diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts
index 7374d4be0b3..52b8038960f 100644
--- a/src/bridge/sse.ts
+++ b/src/bridge/sse.ts
@@ -22,7 +22,7 @@ import {
   repairFreeformToolInput,
 } from "../responses/apply-patch-envelope";
 import { progressiveFreeformInput } from "../responses/progressive-freeform-input";
-import { encodeCompactionSummary } from "../responses/compaction";
+import { encodeCompactionSummary, releaseCompactionCiphertextLease } from "../responses/compaction";
 import { compileCodeModeHelperInput, resolveCodeModeHelperName } from "../responses/code-mode-helper-compat";
 import { mayBecomeCodeModeShellInput } from "../responses/code-mode-shell-input";
 import { isTruncatedStopReason, truncationReasonFor } from "../responses/truncated-stop-reason";
@@ -714,6 +714,10 @@ export function bridgeToResponsesSSE(
         firstOutputReported = true;
         try { options?.onFirstOutput?.(); } catch { /* metrics must not break the stream */ }
       };
+      const releaseEvent = (event: AdapterEvent) => {
+        releaseTranslatedEvent(event, budget);
+        releaseCompactionCiphertextLease(event, budget);
+      };
       const it = events[Symbol.asyncIterator]();
       let iteratorStarted = false;
       let iteratorReturned = false;
@@ -734,7 +738,9 @@ export function bridgeToResponsesSSE(
         if (!iteratorStarted) {
           iteratorStarted = true;
           try {
-            void it.next().then(finishReturn, () => {}).catch(() => {});
+            void it.next().then(next => {
+              try { if (!next.done) releaseEvent(next.value); } finally { finishReturn(); }
+            }, () => {}).catch(() => {});
           } catch {
             /* synchronous iterator start failure is also best-effort */
           }
@@ -809,6 +815,7 @@ export function bridgeToResponsesSSE(
         while (!terminated && !closed && emittedFrames === emittedAtStart) {
           iteratorStarted = true;
           const next = await it.next();
+          try {
           // A cancel during this await disposes the owned budget; a late event
           // must never be processed or charged against it. Exit step() outright:
           // falling into EOF synthesis would let closeCurrentMessage() charge
@@ -1228,9 +1235,8 @@ export function bridgeToResponsesSSE(
                   encrypted_content: event.compactionEncryptedContent ?? encodeCompactionSummary(joinChunks(compaction)),
                 };
                 emit("response.output_item.done", { output_index: outputIndex, item });
-                retainFinishedItem(item as OutputItem, event.compactionEncryptedContent
-                  ? bytesOf(event.compactionEncryptedContent)
-                  : compaction.bytes);
+                retainFinishedItem(item as OutputItem, event.compactionEncryptedContent ? 0 : compaction.bytes);
+                releaseCompactionCiphertextLease(event, budget);
                 outputIndex++;
               }
               // Recognize every adapter's truncation vocabulary, not just the canonical pair.
@@ -1333,6 +1339,9 @@ export function bridgeToResponsesSSE(
             terminated = true;
             break;
           }
+          } finally {
+            if (!next.done) releaseEvent(next.value);
+          }
         }
       } catch (err) {
         if (isTranslatorBudgetExceededError(err)) {
diff --git a/src/responses/compaction.ts b/src/responses/compaction.ts
index b067e7c8587..516d92ba333 100644
--- a/src/responses/compaction.ts
+++ b/src/responses/compaction.ts
@@ -15,6 +15,25 @@
  * routed models get a short "history was compacted" note instead.
  */
 
+import type { TranslatorBudget } from "../lib/translator-budget";
+
+const ciphertextLeases = new WeakMap<object, { budget: TranslatorBudget; bytes: number }>();
+
+/** Attach an already-charged parser ciphertext lease to its exact terminal event. */
+export function transferCompactionCiphertextLease(event: object, budget: TranslatorBudget, bytes: number): void {
+  if (bytes === 0) return;
+  if (ciphertextLeases.has(event)) throw new Error("compaction ciphertext lease is already transferred");
+  ciphertextLeases.set(event, { budget, bytes });
+}
+
+/** Release only this event's transferred ciphertext lease, once and for its owning budget. */
+export function releaseCompactionCiphertextLease(event: object, budget: TranslatorBudget): void {
+  const lease = ciphertextLeases.get(event);
+  if (!lease || lease.budget !== budget) return;
+  ciphertextLeases.delete(event);
+  budget.releaseRetained(lease.bytes, { kind: "retained_collectors" });
+}
+
 export const OCX_COMPACTION_PREFIX = "ocx1:";
 
 export const OCX_NATIVE_REPLAY_RECOVERY_NOTE =
diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md
index b7ebb27522b..b1077a9a410 100644
--- a/structure/transports/byte-accounting.md
+++ b/structure/transports/byte-accounting.md
@@ -107,7 +107,13 @@ The buffered collector in `src/adapters/zed.ts` charges each delegated event bef
 and releases owned events when collection fails. Its translated stream is cancelled on early exit
 for every provider family. `src/adapters/openai-responses/passthrough.ts` releases partial text and
 usage collectors on every exit, and compaction ciphertext unless its lease transfers with the
-`done` event. Zed releases that source ciphertext lease if the collector cannot retain `done`.
+`done` event. The ciphertext lease is bound to the event object and its budget separately from
+serialized-event retention; an uncharged ciphertext field never releases another owner's bytes.
+Zed releases that source lease if collection fails. The SSE bridge releases both source leases
+after each consumed or discarded event, including cancellation before its first pull, late events,
+and processing failure. Buffered bridging keeps ciphertext charged through output admission and
+releases all source leases on success or failure. Output retention remains independently charged.
+`tests/responses/compaction-event-ownership.test.ts` covers those transitions and unrelated owners.
 
 Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate.
 Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged.
diff --git a/structure/transports/responses.md b/structure/transports/responses.md
index a8463f86cf1..c31ae1ba8c7 100644
--- a/structure/transports/responses.md
+++ b/structure/transports/responses.md
@@ -69,7 +69,7 @@ The Responses proxy does not treat transcript growth as repository progress. It
 boundaries, response items, tool names and payloads, adapter events, retained bytes, and elapsed
 silence. It cannot observe the client's workspace or prove whether a successful tool result changed
 repository state. Consequently, the active-turn and session-lane gates are concurrency admission
-limits, the translator budget is a live retained-byte limit, the response-state caps are cache
+limits, the translator budget follows [live retained-byte ownership](byte-accounting.md#stream-buffer-accounting), the response-state caps are cache
 retention limits, and the stall watchdog is a silence limit. None is a cumulative continuation or
 semantic no-progress budget.
 
diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json
index 564d82a3c54..c04bec3224f 100644
--- a/tests/fixtures/test-layout-expected.json
+++ b/tests/fixtures/test-layout-expected.json
@@ -1,4 +1,5 @@
 {
+  "compaction-event-ownership.test.ts": "responses",
   "zed-retained-budget.test.ts": "providers",
   "claude-intercept-on-demand.test.ts": "claude-integration",
   "main-account-hard-lock-thresholds.test.ts": "codex-integration",
diff --git a/tests/providers/zed-retained-budget.test.ts b/tests/providers/zed-retained-budget.test.ts
index 03fcaa46ed8..d91451b5414 100644
--- a/tests/providers/zed-retained-budget.test.ts
+++ b/tests/providers/zed-retained-budget.test.ts
@@ -2,6 +2,7 @@ import { afterEach, describe, expect, spyOn, test } from "bun:test";
 import { createZedAdapter, type ZedProvider } from "../../src/adapters/zed";
 import { createTranslatorBudget, releaseTranslatedEvent, TranslatorBudgetExceededError } from "../../src/lib/translator-budget";
 import { clearZedCaches } from "../../src/providers/zed";
+import { buildResponseJSON } from "../../src/bridge";
 import { parseRequest } from "../../src/responses/parser";
 
 const families: ZedProvider[] = ["anthropic", "google", "open_ai", "x_ai"];
@@ -90,6 +91,18 @@ describe("Zed Responses ciphertext failure ownership", () => {
   const completed = JSON.stringify({ event: { type: "response.completed", response: {
     output: [{ type: "compaction", encrypted_content: "fixture-ciphertext" }],
   } } }) + "\n";
+  test("successful ciphertext collection transfers both leases through buffered output", async () => {
+    const adapter = await configuredAdapter("open_ai");
+    const budget = createTranslatorBudget();
+    try {
+      const events = await adapter.parseResponse!(new Response(completed), budget);
+      expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(JSON.stringify(events)) + Buffer.byteLength("fixture-ciphertext"));
+      const result = buildResponseJSON(events, "fixture-model", { compaction: true, translatorBudget: budget });
+      const output = result.output as Record<string, unknown>[];
+      expect(output).toEqual([expect.objectContaining({ encrypted_content: "fixture-ciphertext" })]);
+      expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(JSON.stringify(output[0])));
+    } finally { budget.dispose(); }
+  });
   test("read failure after a completed snapshot releases ciphertext with no done owner", async () => {
     const adapter = await configuredAdapter("open_ai");
     const budget = createTranslatorBudget();
diff --git a/tests/responses/compaction-event-ownership.test.ts b/tests/responses/compaction-event-ownership.test.ts
new file mode 100644
index 00000000000..020e7f4399f
--- /dev/null
+++ b/tests/responses/compaction-event-ownership.test.ts
@@ -0,0 +1,211 @@
+import { describe, expect, spyOn, test } from "bun:test";
+import { createResponsesPassthroughAdapter } from "../../src/adapters/openai-responses";
+import { bridgeToResponsesSSE, buildResponseJSON } from "../../src/bridge";
+import { createTranslatorBudget, retainTranslatedEvent, retainTranslatedEventBatch, TranslatorBudgetExceededError, type TranslatorBudget } from "../../src/lib/translator-budget";
+import { releaseCompactionCiphertextLease } from "../../src/responses/compaction";
+import type { AdapterEvent } from "../../src/types";
+
+const ciphertext = "fixture-中文-ciphertext";
+const sentinelBytes = 137;
+function budgetWithSentinel() {
+  const budget = createTranslatorBudget({ maxTurnBytes: 4096 });
+  budget.chargeRetained(sentinelBytes, { kind: "retained_collectors" });
+  return budget;
+}
+function parser(budget: TranslatorBudget) {
+  const adapter = createResponsesPassthroughAdapter({ adapter: "openai-responses", baseUrl: "https://fixture.invalid/v1" });
+  const wire = `data: ${JSON.stringify({ type: "response.completed", response: {
+    output: [{ type: "compaction", encrypted_content: ciphertext }],
+  } })}\n\n`;
+  return adapter.parseStream(new Response(wire), budget);
+}
+function bridge(events: AsyncIterable<AdapterEvent>, budget: TranslatorBudget, compaction = true) {
+  return bridgeToResponsesSSE(events, "fixture-model", undefined, undefined, undefined, undefined, 0, {
+    translatorBudget: budget, compaction, stallTimeoutSec: 0,
+    timers: { setInterval: () => 0, clearInterval: () => {} },
+  });
+}
+function outputFromWire(wire: string): Record<string, unknown>[] {
+  const terminal = wire.split("\n\n").find(frame => frame.startsWith("event: response.completed\n"));
+  expect(terminal).toBeDefined();
+  return JSON.parse(terminal!.split("\ndata: ")[1]!).response.output;
+}
+async function retainedDone(budget: TranslatorBudget) {
+  const iterator = parser(budget);
+  const next = await iterator.next();
+  expect(next.done).toBe(false);
+  expect(next.value).toEqual({ type: "done", compactionEncryptedContent: ciphertext });
+  retainTranslatedEvent(next.value!, budget);
+  await iterator.return(undefined);
+  return next.value!;
+}
+async function* replay(event: AdapterEvent) { yield event; }
+
+// Real parser and bridge only: no provider send, fetch mock, or live credentials.
+describe("Responses bridge transferred event ownership", () => {
+  test("late done after cancellation releases both source leases, preserving another owner", async () => {
+    const budget = budgetWithSentinel();
+    const iterator = parser(budget);
+    const acquired = Promise.withResolvers<void>();
+    const release = Promise.withResolvers<void>();
+    const returned = Promise.withResolvers<void>();
+    const events: AsyncIterable<AdapterEvent> = { [Symbol.asyncIterator]() { return {
+      async next() {
+        const next = await iterator.next();
+        if (!next.done) retainTranslatedEvent(next.value, budget);
+        acquired.resolve();
+        await release.promise;
+        return next;
+      },
+      async return() { try { return await iterator.return(undefined); } finally { returned.resolve(); } },
+    }; } };
+    const reader = bridge(events, budget).getReader();
+    try {
+      await reader.read(); // response.created
+      await reader.read(); // response.in_progress
+      const pending = reader.read();
+      await acquired.promise;
+      expect(budget.snapshot().currentBytes).toBeGreaterThan(sentinelBytes);
+      await reader.cancel();
+      release.resolve();
+      await pending;
+      await returned.promise;
+      // The cancelled read resolves before the in-flight producer's continuation.
+      await new Promise(resolve => setTimeout(resolve, 0));
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally { release.resolve(); await reader.cancel(); reader.releaseLock(); budget.dispose(); }
+  });
+
+  test("cancel before first pull releases the bootstrap event without delivering it", async () => {
+    const budget = budgetWithSentinel();
+    const returned = Promise.withResolvers<void>();
+    let yielded = 0;
+    async function* events() {
+      try { yielded++; yield await retainedDone(budget); } finally { returned.resolve(); }
+    }
+    try {
+      await bridge(events(), budget).cancel();
+      await returned.promise;
+      expect(yielded).toBe(1);
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally { budget.dispose(); }
+  });
+
+  test("compaction continues release serialized heartbeat and text leases", async () => {
+    const budget = budgetWithSentinel();
+    const events: AdapterEvent[] = [{ type: "heartbeat" }, { type: "text_delta", text: "summary" }, { type: "done" }];
+    retainTranslatedEventBatch(events, budget);
+    try {
+      const output = outputFromWire(await new Response(bridge((async function* () { yield* events; })(), budget)).text());
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes + Buffer.byteLength(JSON.stringify(output[0])));
+    } finally { budget.dispose(); }
+  });
+
+  test("successful replacement releases raw ciphertext before terminal-frame admission", async () => {
+    const budget = createTranslatorBudget({ maxTurnBytes: 9400 });
+    const encrypted = "x".repeat(3000);
+    const adapter = createResponsesPassthroughAdapter({ adapter: "openai-responses", baseUrl: "https://fixture.invalid/v1" });
+    const wire = `data: ${JSON.stringify({ type: "response.completed", response: {
+      output: [{ type: "compaction", encrypted_content: encrypted }],
+    } })}\n\n`;
+    try {
+      const output = outputFromWire(await new Response(bridge(adapter.parseStream(new Response(wire), budget), budget)).text());
+      expect(output).toEqual([expect.objectContaining({ encrypted_content: encrypted })]);
+      expect(budget.snapshot().overflows).toBe(0);
+      expect(budget.snapshot().currentBytes).toBe(Buffer.byteLength(JSON.stringify(output[0])));
+    } finally { budget.dispose(); }
+  });
+
+  test("transferred ciphertext release is budget-bound and idempotent", async () => {
+    const budget = budgetWithSentinel();
+    const other = budgetWithSentinel();
+    const iterator = parser(budget);
+    try {
+      const next = await iterator.next();
+      await iterator.return(undefined);
+      const before = budget.snapshot().currentBytes;
+      releaseCompactionCiphertextLease(next.value!, other);
+      expect(budget.snapshot().currentBytes).toBe(before);
+      expect(other.snapshot().currentBytes).toBe(sentinelBytes);
+      releaseCompactionCiphertextLease(next.value!, budget);
+      releaseCompactionCiphertextLease(next.value!, budget);
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally { await iterator.return(undefined); other.dispose(); budget.dispose(); }
+  });
+
+  test.each(["stream", "buffered"] as const)("%s: failed finished-item reservation releases source leases", async mode => {
+    const budget = budgetWithSentinel();
+    const event = await retainedDone(budget);
+    const original = budget.reserveTransient.bind(budget);
+    const reserve = spyOn(budget, "reserveTransient").mockImplementation((bytes, scope) => {
+      if (scope.kind === "retained_collectors") throw new TranslatorBudgetExceededError(scope.kind, 1);
+      return original(bytes, scope);
+    });
+    try {
+      if (mode === "stream") {
+        const wire = await new Response(bridge(replay(event), budget)).text();
+        expect(wire).toContain("translation_buffer_limit");
+        expect(wire).not.toContain("event: response.completed");
+      } else {
+        expect(() => buildResponseJSON([event], "fixture-model", { compaction: true, translatorBudget: budget }))
+          .toThrow("buffer exceeded");
+      }
+      expect(reserve.mock.calls.some(([, scope]) => scope.kind === "retained_collectors")).toBe(true);
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally { reserve.mockRestore(); budget.dispose(); }
+  });
+
+  test.each(["stream", "buffered"] as const)("%s: successful handoff retains only output and unrelated bytes", async mode => {
+    const budget = budgetWithSentinel();
+    try {
+      const event = await retainedDone(budget);
+      const output = mode === "stream"
+        ? outputFromWire(await new Response(bridge(replay(event), budget)).text())
+        : buildResponseJSON([event], "fixture-model", { compaction: true, translatorBudget: budget }).output as Record<string, unknown>[];
+      expect(output).toEqual([expect.objectContaining({ type: "compaction", encrypted_content: ciphertext })]);
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes + Buffer.byteLength(JSON.stringify(output[0])));
+    } finally { budget.dispose(); }
+  });
+
+  test.each(["stream", "buffered"] as const)("%s: an uncharged ciphertext field cannot release another owner's bytes", async mode => {
+    const budget = budgetWithSentinel();
+    const event: AdapterEvent = { type: "done", compactionEncryptedContent: ciphertext };
+    try {
+      const output = mode === "stream"
+        ? outputFromWire(await new Response(bridge(replay(event), budget)).text())
+        : buildResponseJSON([event], "fixture-model", { compaction: true, translatorBudget: budget }).output as Record<string, unknown>[];
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes + Buffer.byteLength(JSON.stringify(output[0])));
+    } finally { budget.dispose(); }
+  });
+
+  test.each(["stream", "buffered"] as const)("%s: skipped compaction releases the unused ciphertext", async mode => {
+    const budget = budgetWithSentinel();
+    try {
+      const event = await retainedDone(budget);
+      const output = mode === "stream"
+        ? outputFromWire(await new Response(bridge(replay(event), budget, false)).text())
+        : buildResponseJSON([event], "fixture-model", { translatorBudget: budget }).output;
+      expect(output).toEqual([]);
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally { budget.dispose(); }
+  });
+  test.each(["stream", "buffered"] as const)("%s: truncated compaction releases ciphertext without installing history", async mode => {
+    const budget = budgetWithSentinel();
+    try {
+      const event = await retainedDone(budget);
+      if (event.type !== "done") throw new Error("expected done fixture");
+      event.stopReason = "max_tokens";
+      if (mode === "stream") {
+        const wire = await new Response(bridge(replay(event), budget)).text();
+        expect(wire).toContain("event: response.incomplete");
+        expect(wire).not.toContain('"type":"compaction"');
+      } else {
+        const result = buildResponseJSON([event], "fixture-model", { compaction: true, translatorBudget: budget });
+        expect(result.status).toBe("incomplete");
+        expect(result.output).toEqual([]);
+      }
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally { budget.dispose(); }
+  });
+
+});

From c6a1103b4e840f916e2419b533af793f29e95286 Mon Sep 17 00:00:00 2001
From: Epinephrine <luvs01@hanmail.net>
Date: Fri, 2 Oct 2026 07:58:54 -0700
Subject: [PATCH 4/4] fix(bridge): close iterator after failed cancellation
 bootstrap

---
 src/bridge/sse.ts                             |  3 +-
 .../compaction-event-ownership.test.ts        | 31 +++++++++++++++++++
 2 files changed, 33 insertions(+), 1 deletion(-)

diff --git a/src/bridge/sse.ts b/src/bridge/sse.ts
index 52b8038960f..e34c3bebdf6 100644
--- a/src/bridge/sse.ts
+++ b/src/bridge/sse.ts
@@ -740,8 +740,9 @@ export function bridgeToResponsesSSE(
           try {
             void it.next().then(next => {
               try { if (!next.done) releaseEvent(next.value); } finally { finishReturn(); }
-            }, () => {}).catch(() => {});
+            }, finishReturn).catch(() => {});
           } catch {
+            finishReturn();
             /* synchronous iterator start failure is also best-effort */
           }
           return;
diff --git a/tests/responses/compaction-event-ownership.test.ts b/tests/responses/compaction-event-ownership.test.ts
index 020e7f4399f..38963d800c3 100644
--- a/tests/responses/compaction-event-ownership.test.ts
+++ b/tests/responses/compaction-event-ownership.test.ts
@@ -91,6 +91,37 @@ describe("Responses bridge transferred event ownership", () => {
     } finally { budget.dispose(); }
   });
 
+  test.each(["throw", "reject"] as const)("bootstrap next() %s still closes its resource-owning iterator", async failure => {
+    const budget = budgetWithSentinel();
+    let cancelled = 0;
+    let returned = 0;
+    const source = new ReadableStream({ cancel() { cancelled++; } });
+    const reader = source.getReader();
+    const events: AsyncIterable<AdapterEvent> = { [Symbol.asyncIterator]() { return {
+      next() {
+        if (failure === "throw") throw new Error("fixture synchronous next failure");
+        return Promise.reject(new Error("fixture rejected next failure"));
+      },
+      async return() {
+        returned++;
+        await reader.cancel();
+        reader.releaseLock();
+        return { done: true, value: undefined };
+      },
+    }; } };
+    try {
+      await bridge(events, budget).cancel();
+      await new Promise(resolve => setTimeout(resolve, 0));
+      expect(returned).toBe(1);
+      expect(cancelled).toBe(1);
+      expect(source.locked).toBe(false);
+      expect(budget.snapshot().currentBytes).toBe(sentinelBytes);
+    } finally {
+      if (source.locked) { await reader.cancel(); reader.releaseLock(); }
+      budget.dispose();
+    }
+  });
+
   test("compaction continues release serialized heartbeat and text leases", async () => {
     const budget = budgetWithSentinel();
     const events: AdapterEvent[] = [{ type: "heartbeat" }, { type: "text_delta", text: "summary" }, { type: "done" }];