Skip to content

[WRONG BRANCH] claude: avoid retaining streamed reasoning fallback to prevent aggregate OOM #450

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
luvs01 wants to merge 1 commit into from
Closed

[WRONG BRANCH] claude: avoid retaining streamed reasoning fallback to prevent aggregate OOM #450

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 4 additions & 29 deletions src/claude/outbound.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -215,8 +215,6 @@ interface OpenBlock {
callId?: string;
/** Last fixed-size reasoning identity (item + summary/content index) seen by this block. */
reasoningPartKey?: string;
thinkingBuf?: string;
thinkingBufBytes?: number;
reasoningSig?: string;
}

Expand Down Expand Up @@ -260,12 +258,6 @@ export function responsesSseToAnthropicSse(
const bytes = queuedLiveFrameBytes.shift();
if (bytes !== undefined) translatorBudget.releaseRetained(bytes, { kind: "live_transient" });
};
const releaseThinkingBuffer = (block: OpenBlock | null | undefined) => {
if (block?.kind !== "thinking") return;
translatorBudget.releaseRetained(block.thinkingBufBytes ?? 0, { kind: "reasoning" });
block.thinkingBufBytes = 0;
};

return new ReadableStream<Uint8Array>({
start(controller) {
const emit = (name: string, data: Rec) => {
Expand Down Expand Up @@ -308,14 +300,15 @@ export function responsesSseToAnthropicSse(
open.webSearchArgsEmitted = true;
}
if (open.kind === "thinking") {
const signature = open.reasoningSig ?? encodeReasoningEnvelope({ txt: open.thinkingBuf ?? "" });
// Streamed thinking has already been delivered to the client. Keep the fallback
// signature bounded instead of retaining and re-encoding the entire untrusted block.
const signature = open.reasoningSig ?? encodeReasoningEnvelope({ txt: "" });

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 7, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve streamed thinking when replaying the bounded fallback

When a streamed reasoning block has no genuine upstream signature and Claude Code later replays that assistant block with a tool result, src/claude/inbound.ts preserves the visible thinking in summary alongside this envelope, but src/responses/parser.ts:284 evaluates envelope?.txt ?? text; because the envelope explicitly contains txt: "", it selects the empty value and then drops the reasoning item at line 300. This removes the reasoning_content preceding the replayed function call, so strict providers such as DeepSeek can reject the continuation instead of completing the tool loop. Keep the fallback bounded without setting an overriding empty txt, or make the parser fall back to the visible summary when this marker is empty.

AGENTS.md reference: src/AGENTS.md:L19-L19

Useful? React with 👍 / 👎.

emit("content_block_delta", {
type: "content_block_delta", index: open.index,
delta: { type: "signature_delta", signature },
});
}
emit("content_block_stop", { type: "content_block_stop", index: open.index });
releaseThinkingBuffer(open);
if (open.callId) translatorBudget.closeCall(open.callId);
open = null;
};
Expand All @@ -328,7 +321,7 @@ export function responsesSseToAnthropicSse(
? { type: "text", text: "" }
: { type: "thinking", thinking: "", signature: "" };
emit("content_block_start", { type: "content_block_start", index, content_block: contentBlock });
open = { kind, index, thinkingBuf: "", thinkingBufBytes: 0 };
open = { kind, index };
};
const finish = (stopReason: string, usage: unknown) => {
if (terminated) return;
Expand All @@ -355,7 +348,6 @@ export function responsesSseToAnthropicSse(
if (terminated && (code !== "translation_buffer_limit" || terminalDelivered)) return;
terminated = true;
if (code === "translation_buffer_limit") {
releaseThinkingBuffer(open);
if (open?.callId) translatorBudget.closeCall(open.callId);
open = null;
terminalDelivered = true;
Expand Down Expand Up @@ -421,21 +413,6 @@ export function responsesSseToAnthropicSse(
const partKey = `${boundedReasoningIdentity(data.item_id)}:${slot}`;
const needsPartSeparator = active.reasoningPartKey !== undefined
&& active.reasoningPartKey !== partKey;
const appended = `${needsPartSeparator ? "\n\n" : ""}${data.delta}`;
const previous = active.thinkingBuf ?? "";
const previousBytes = active.thinkingBufBytes ?? 0;
const nextBytes = appendedUtf8Bytes(previous, previousBytes, appended);
const scope = { kind: "reasoning" } as const;
const reservation = translatorBudget.reserveTransient(nextBytes, scope);
try {
active.thinkingBuf = previous + appended;
active.thinkingBufBytes = nextBytes;
reservation.commitRetained();
translatorBudget.releaseRetained(previousBytes, scope);
} catch (error) {
reservation.release();
throw error;
}
if (needsPartSeparator) {
emit("content_block_delta", {
type: "content_block_delta", index: active.index,
Expand Down Expand Up @@ -762,7 +739,6 @@ export function responsesSseToAnthropicSse(
fail(413, "upstream translation buffer exceeded the safe limit", false, "translation_buffer_limit");
} else fail(500, err instanceof Error ? err.message : String(err));
} finally {
releaseThinkingBuffer(open);
translatorBudget.releaseRetained(bufferBytes, { kind: "live_transient" });
if (pingTimer !== undefined) clearInterval(pingTimer);
reader.releaseLock();
Expand All @@ -776,7 +752,6 @@ export function responsesSseToAnthropicSse(
cancel(reason) {
cancelled = true;
while (queuedLiveFrameBytes.length > 0) releaseDeliveredFrame();
releaseThinkingBuffer(open);
if (open?.callId) translatorBudget.closeCall(open.callId);
if (pingTimer !== undefined) clearInterval(pingTimer);
return reader?.cancel(reason);
Expand Down
120 changes: 11 additions & 109 deletions tests/claude-integration/claude-outbound.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,8 +275,7 @@ describe("claude outbound SSE", () => {
"**A**\n\nOne.\n\n**B**\n\nTwo.",
"Three.",
]);
expect(decodeReasoningEnvelope(thinkingBlocks[0].signature)?.txt)
.toBe("**A**\n\nOne.\n\n**B**\n\nTwo.");
expect(decodeReasoningEnvelope(thinkingBlocks[0].signature)).toEqual({ txt: "" });

// Parity: the non-streaming translator joins the same summary parts identically.
const json = responsesJsonToAnthropicMessage({
Expand All @@ -289,10 +288,9 @@ describe("claude outbound SSE", () => {
expect(jsonThinking.thinking).toBe("**A**\n\nOne.\n\n**B**\n\nTwo.");
});

test("reasoning fallback buffering is bounded and releases its retained budget", async () => {
test("streamed reasoning does not accumulate retained fallback text", async () => {
const budget = createTestTranslatorBudget({ maxTurnBytes: 8 * 1024 });
let reasoningCommitted = 0;
let reasoningReleased = 0;
const trackedBudget: TranslatorBudget = {
openCall: id => budget.openCall(id),
closeCall: id => budget.closeCall(id),
Expand All @@ -310,10 +308,7 @@ describe("claude outbound SSE", () => {
budget.chargeRetained(bytes, scope);
if (scope.kind === "reasoning") reasoningCommitted += bytes;
},
releaseRetained(bytes, scope) {
budget.releaseRetained(bytes, scope);
if (scope.kind === "reasoning") reasoningReleased += bytes;
},
releaseRetained: (bytes, scope) => budget.releaseRetained(bytes, scope),
observeAcceptedRequestCopy: bytes => budget.observeAcceptedRequestCopy(bytes),
observeExternallyCapped: (kind, bytes) => budget.observeExternallyCapped(kind, bytes),
snapshot: () => budget.snapshot(),
Expand All @@ -326,115 +321,22 @@ describe("claude outbound SSE", () => {
content_index: 0,
delta: `${index}:` + "x".repeat(512),
})),
sse("response.completed", { response: { status: "completed", usage: {} } }),
];
const events = await collectEvents(responsesSseToAnthropicSse(
streamFromChunks(frames),
"m",
{ translatorBudget: trackedBudget },
{ translatorBudget: trackedBudget, pingIntervalMs: 0 },
));

expect(events.at(-1)).toMatchObject({
name: "error",
data: { error: { type: "request_too_large", code: "translation_buffer_limit" } },
});
expect(budget.snapshot().overflows).toBe(1);
expect(reasoningCommitted).toBeGreaterThan(0);
expect(reasoningReleased).toBe(reasoningCommitted);
expect(events.at(-1)?.name).toBe("message_stop");
expect(budget.snapshot().overflows).toBe(0);
expect(reasoningCommitted).toBe(0);
const signature = events.find(event => event.data.delta?.type === "signature_delta")?.data.delta.signature;
expect(signature.length).toBeLessThan(256);
expect(decodeReasoningEnvelope(signature)).toEqual({ txt: "" });
});

for (const terminal of ["eof", "failed", "completed", "incomplete"] as const) {
for (const buffered of [false, true]) {
test(`closure-only reasoning overflow: ${terminal}, ${buffered ? "collector" : "stream"}`, async () => {
// All small deltas fit, including replacement reservations. Closing needs
// the retained 32 KiB text PLUS its base64 signature frame. Capture the
// generated stream before collection: concurrent collector retention can
// exceed a shared budget during ingestion instead of exercising closure.
// Collection below reuses this SAME budget, without resetting it.
const budget = createTestTranslatorBudget({ maxTurnBytes: 70 * 1024 });
let reasoningBytes = 0;
let maxReasoningBytes = 0;
let reasoningBytesAtOverflow = -1;
const trackedBudget: TranslatorBudget = {
openCall: id => budget.openCall(id),
closeCall: id => budget.closeCall(id),
reserveTransient(bytes, scope) {
let reservation: ReturnType<TranslatorBudget["reserveTransient"]>;
try { reservation = budget.reserveTransient(bytes, scope); }
catch (error) { reasoningBytesAtOverflow = reasoningBytes; throw error; }
return {
commitRetained() {
reservation.commitRetained();
if (scope.kind === "reasoning") {
reasoningBytes += bytes;
maxReasoningBytes = Math.max(maxReasoningBytes, reasoningBytes);
}
},
release: () => reservation.release(),
};
},
chargeRetained: (bytes, scope) => budget.chargeRetained(bytes, scope),
releaseRetained(bytes, scope) {
if (scope.kind === "reasoning") reasoningBytes -= bytes;
budget.releaseRetained(bytes, scope);
},
observeAcceptedRequestCopy: bytes => budget.observeAcceptedRequestCopy(bytes),
observeExternallyCapped: (kind, bytes) => budget.observeExternallyCapped(kind, bytes),
snapshot: () => budget.snapshot(),
dispose: () => budget.dispose(),
};
const text = "x".repeat(32 * 1024);
const frames = Array.from({ length: 128 }, () => sse("response.reasoning_text.delta", {
item_id: "rs_closure", content_index: 0, delta: text.slice(0, 256),
}));
if (terminal !== "eof") {
frames.push(sse(`response.${terminal}`, { response: terminal === "failed"
? { error: { message: "upstream failure", status: 502 } }
: terminal === "incomplete"
? { status: "incomplete", incomplete_details: { reason: "max_output_tokens" }, usage: {} }
: { status: "completed", usage: {} } }));
// Neither a repeated completion nor a later failure may add a terminal.
frames.push(sse("response.completed", { response: { status: "completed", usage: {} } }));
frames.push(sse("response.failed", { response: { error: { message: "late failure" } } }));
}
const stream = responsesSseToAnthropicSse(streamFromChunks(frames), "m", {
translatorBudget: trackedBudget, pingIntervalMs: 0,
});
const captured = buffered ? await new Response(stream).text() : undefined;
const capturedFrames = captured?.split("\n\n").filter(Boolean).map(frame => `${frame}\n\n`);
const events = await collectEvents(capturedFrames ? streamFromChunks(capturedFrames) : stream);
const deltas = events.filter(event => event.data.delta?.type === "thinking_delta");
expect(deltas.map(event => event.data.delta.thinking).join("")).toBe(text);
expect(events.filter(event => event.name === "error")).toHaveLength(1);
expect(events.at(-1)).toMatchObject({ name: "error", data: { type: "error", error: {
type: "request_too_large", code: "translation_buffer_limit",
} } });
expect(JSON.stringify(events.at(-1)).length).toBeLessThan(1024);
expect(events.some(event => event.name === "message_stop" || event.name === "message_delta" || event.name === "content_block_stop")).toBe(false);
expect(events.some(event => event.data.delta?.type === "signature_delta")).toBe(false);
if (capturedFrames) {
expect(capturedFrames.join("")).toBe(captured);
expect(reasoningBytesAtOverflow).toBe(text.length);
expect(reasoningBytes).toBe(0);
expect(budget.snapshot().overflows).toBe(1);
// Feed the actual generated frames, without inventing an error event or
// collecting one huge chunk that introduces a different buffer limit.
const message = await collectAnthropicMessage(streamFromChunks(capturedFrames), "m", trackedBudget);
expect(message).toMatchObject({ type: "error", error: {
type: "request_too_large", code: "translation_buffer_limit",
} });
expect(message).not.toHaveProperty("content");
expect(message).not.toHaveProperty("stop_reason");
}
// These prove failure happened after all text was retained, not while
// ingesting a delta, and the error path released the thinking reservation.
expect(reasoningBytesAtOverflow).toBe(text.length);
expect(maxReasoningBytes).toBeGreaterThanOrEqual(text.length);
expect(reasoningBytes).toBe(0);
expect(budget.snapshot().overflows).toBe(1);
});
}
}

test("same-part deltas and index-free reasoning frames never get a separator", async () => {
const samePart = [
sse("response.created", { response: { id: "resp_1", status: "in_progress" } }),
Expand Down
Loading