diff --git a/docs/configuration.md b/docs/configuration.md index 3a18d9ca..88d8b8df 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -481,6 +481,14 @@ The cap is lifted for the cases that genuinely need the SDK to keep going — de > `/telemetry/logs` is what confirms the session was preserved. And a turn that > hits the cap having produced content but no forwardable tool call is reported > as `stop_reason: "max_tokens"` (truncated) instead of failing the request. +> +> A capped turn that produced *nothing* — no wire event, no captured tool call — +> never reached the tool boundary the cap exists to stop at, so the single turn +> bought nothing and cost the whole request. Meridian reissues that turn once +> with the cap lifted (`single_turn_cap_lifted` in `/telemetry/logs`), which is +> safe precisely because nothing had reached the client yet. An explicit +> `MERIDIAN_PASSTHROUGH_MAX_TURNS` suppresses the reissue too: the budget is +> then the operator's, and a second identical attempt would only spend a turn. ### Silent turns diff --git a/src/__tests__/passthrough-early-stop-integration.test.ts b/src/__tests__/passthrough-early-stop-integration.test.ts index 30f2a472..597feb4c 100644 --- a/src/__tests__/passthrough-early-stop-integration.test.ts +++ b/src/__tests__/passthrough-early-stop-integration.test.ts @@ -25,6 +25,13 @@ let yieldedCount = 0 let capturedQueryParams: any = null let capturedQueryParamsAll: any[] = [] let mockTerminalError: Error | undefined +/** + * Per-attempt SDK scripts, consumed one per `query()` call. Retry paths need + * the second attempt to behave differently from the first; without this every + * attempt replays the same fixture and a retry is invisible. Empty (the + * default) leaves `mockMessages` / `mockTerminalError` in charge. + */ +let mockAttemptScripts: Array<{ messages: any[]; terminalError?: Error }> = [] let forkSessionSequence = 0 let mockBaseSessionId = "test-session" let mockReturnedSessionIdOverride: string | undefined @@ -35,14 +42,15 @@ installSdkMock(() => ({ query: (params: any) => { capturedQueryParams = params capturedQueryParamsAll.push(params) - const terminalError = mockTerminalError + const script = mockAttemptScripts.length > 0 ? mockAttemptScripts.shift() : undefined + const terminalError = script ? script.terminalError : mockTerminalError const preHook = params?.options?.hooks?.PreToolUse?.[0]?.hooks?.[0] const returnedSessionId = mockReturnedSessionIdOverride ?? resolveMockSdkSessionId(params?.options, mockBaseSessionId) return (async function* () { let sawSyntheticDeny = false let sawResult = false - for (const msg of mockMessages) { + for (const msg of script ? script.messages : mockMessages) { yieldedCount++ if (msg?.type === "test_pre_tool_hook") { if (preHook) { @@ -111,7 +119,7 @@ installMcpToolsMock(() => ({ const { createProxyServer } = await import("../proxy/server") const { clearSessionCache } = await import("../proxy/session/cache") const { evictSharedSession, lookupSharedSession, setSessionStoreDir } = await import("../proxy/sessionStore") -const { telemetryStore } = await import("../telemetry") +const { diagnosticLog, telemetryStore } = await import("../telemetry") function userDenyMessage(toolUseId: string) { return { @@ -191,6 +199,7 @@ describe("Integration: passthrough early stop", () => { capturedQueryParams = null capturedQueryParamsAll = [] mockTerminalError = undefined + mockAttemptScripts = [] forkSessionSequence = 0 mockBaseSessionId = `test-session-${crypto.randomUUID()}` mockReturnedSessionIdOverride = undefined @@ -643,6 +652,35 @@ describe("Integration: passthrough early stop", () => { expect(resources.find((resource) => resource.locator.sessionId === wrongSessionId)?.state).toBe("retired") }) + it("stream: correlates a generic error with the fresh upstream session ID", async () => { + const requestId = `fresh-stream-error-request-${TEST_RUN_ID}` + telemetryStore.clear() + diagnosticLog.clear() + mockMessages = [messageStart("msg_fresh_stream_error")] + mockTerminalError = new Error("Claude Code process exited with code 1") + + const response = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + messages: [{ role: "user", content: "fresh stream error correlation" }], + }, "es-fresh-stream-error", { "x-request-id": requestId }) + expect(response.status).toBe(200) + expect(await response.text()).toContain("event: error") + const freshSessionId = capturedQueryParamsAll[0]?.options?.sessionId + expect(freshSessionId).toBeDefined() + + const log = diagnosticLog.getRecent({ limit: 200 }) + .find((entry) => entry.requestId === requestId && entry.level === "error") + expect(log).toBeDefined() + expect(log!.message).toContain(`session=${freshSessionId.slice(0, 8)}`) + + const row = telemetryStore.getRecent({ limit: 200 }) + .find((entry) => entry.requestId === requestId) + expect(row).toBeDefined() + expect(row!.sdkSessionId).toBe(freshSessionId) + }) + it("non-stream: rejects an SDK fork ID mismatch without advancing the shared mapping", async () => { const toolTurn = assistantMessage([ { type: "tool_use", id: "tu-id-mismatch", name: "read", input: { file_path: "a" } }, @@ -1544,20 +1582,42 @@ describe("Integration: passthrough early stop", () => { messageDelta("tool_use"), toolTurn, userDenyMessage("capped-stream-tool"), - { type: "result", subtype: "error_max_turns", is_error: true, session_id: "test-session" }, + { + type: "result", + subtype: "error_max_turns", + is_error: true, + session_id: "test-session", + usage: { output_tokens: 42 }, + }, ] mockTerminalError = new Error("Claude Code returned an error result: Reached maximum number of turns (1)") + const requestId = `capped-stream-recovery-${TEST_RUN_ID}` const first = await post(app, { model: "claude-sonnet-4-5", max_tokens: 400, stream: true, tools: [READ_TOOL], messages: [{ role: "user", content: "read x capped" }], - }, "es-capped-stream") + }, "es-capped-stream", { "x-request-id": requestId }) expect(first.status).toBe(200) - expect(await first.text()).toContain('"type":"tool_use"') + const firstBody = await first.text() + expect(firstBody).toContain('"type":"tool_use"') + expect(firstBody).toContain('"stop_reason":"tool_use"') + expect(firstBody).toContain('"output_tokens":42') expect(capturedQueryParamsAll[0].options.maxTurns).toBe(1) + const freshSessionId = capturedQueryParamsAll[0]?.options?.sessionId + expect(freshSessionId).toBeDefined() + // The hidden drain publishes this diagnostic after the client stream closes; + // wait for that real completion signal rather than sampling the log early. + let recoveryLog: any + for (let i = 0; i < 500 && !recoveryLog; i++) { + recoveryLog = diagnosticLog.getRecent({ limit: 200 }) + .find((entry) => entry.requestId === requestId && entry.message.includes("sdk_termination_recovered")) + if (!recoveryLog) await new Promise((resolve) => setTimeout(resolve, 10)) + } + expect(recoveryLog).toBeDefined() + expect(recoveryLog!.message).toContain(`session=${freshSessionId.slice(0, 8)}`) mockTerminalError = undefined mockMessages = [assistantMessage([{ type: "text", text: "the file says X" }])] @@ -1659,18 +1719,26 @@ describe("Integration: passthrough early stop", () => { ] mockTerminalError = new Error("Claude Code returned an error result: Reached maximum number of turns (1)") + const requestId = `capped-ns-recovery-${TEST_RUN_ID}` const first = await post(app, { model: "claude-sonnet-4-5", max_tokens: 400, stream: false, tools: [READ_TOOL], messages: [{ role: "user", content: "read y capped" }], - }, "es-capped-nonstream") + }, "es-capped-nonstream", { "x-request-id": requestId }) expect(first.status).toBe(200) const firstJson = await first.json() as any expect(firstJson.stop_reason).toBe("tool_use") expect(firstJson.content.some((b: any) => b.type === "tool_use")).toBe(true) + const freshSessionId = capturedQueryParamsAll[0]?.options?.sessionId + expect(freshSessionId).toBeDefined() + const recoveryLog = diagnosticLog.getRecent({ limit: 200 }) + .find((entry) => entry.requestId === requestId && entry.message.includes("sdk_termination_recovered")) + expect(recoveryLog).toBeDefined() + expect(recoveryLog!.message).toContain(`session=${freshSessionId.slice(0, 8)}`) + mockTerminalError = undefined mockMessages = [assistantMessage([{ type: "text", text: "the file says Y" }])] const second = await post(app, { @@ -1874,6 +1942,287 @@ describe("Integration: passthrough early stop", () => { expect(body).not.toContain('"stop_reason":"end_turn"') }) + // The failure the streaming capped-turn branch exists for: the turn streamed + // real content blocks before the cap refused it a next turn. 200 alone does + // not make that turn usable — a streaming response's status left with + // `message_start`, so an `event: error` in the body is what the client + // renders, and that is how "Reached maximum number of turns (1)" surfaced + // over text already on screen. Report truncation instead: no error frame, + // and the client can continue from what it has. + it("stream: a capped turn that streamed content reports truncation without an error frame", async () => { + mockMessages = [ + messageStart("msg_capped_text"), + textBlockStart(0), + textDelta(0, "half an answer"), + blockStop(0), + { + type: "result", + subtype: "error_max_turns", + is_error: true, + session_id: "test-session", + usage: { output_tokens: 42 }, + }, + ] + mockTerminalError = new Error("Claude Code returned an error result: Reached maximum number of turns (1)") + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "answer then stop" }], + }, "es-capped-text") + expect(res.status).toBe(200) + const body = await res.text() + expect(body).toContain("half an answer") + expect(body).toContain('"stop_reason":"max_tokens"') + expect(body).not.toContain("event: error") + expect(body).toContain('"output_tokens":42') + expect(body).toContain("event: message_stop") + }) + + // The lower boundary: honest degradation needs content to degrade. A capped + // turn that forwarded only the message envelope delivered nothing, so + // closing it cleanly would be the silent turn wearing `max_tokens` instead + // of `end_turn`. It keeps the error frame. + it("stream: a capped turn that forwarded no content still reports the failure", async () => { + mockMessages = [ + messageStart("msg_capped_empty"), + { type: "result", subtype: "error_max_turns", is_error: true, session_id: "test-session" }, + ] + mockTerminalError = new Error("Claude Code returned an error result: Reached maximum number of turns (1)") + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "nothing at all" }], + }, "es-capped-empty") + expect(res.status).toBe(200) + const body = await res.text() + expect(body).toContain("event: error") + }) + + // An empty text block is still an empty turn: content_block_start advances + // the client block index, but without a text delta the client received no + // actionable content. Keep the capped turn on the error path. + it("stream: a capped turn with an empty text block still reports the failure", async () => { + mockMessages = [ + messageStart("msg_capped_empty_block"), + textBlockStart(0), + blockStop(0), + { type: "result", subtype: "error_max_turns", is_error: true, session_id: "test-session" }, + ] + mockTerminalError = new Error("Claude Code returned an error result: Reached maximum number of turns (1)") + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "empty text block" }], + }, "es-capped-empty-block") + expect(res.status).toBe(200) + const body = await res.text() + expect(body).toContain("event: error") + }) + + + // The other boundary: a tool_use block reached the client while the hook + // captured nothing, which means those calls were refused rather than + // forwarded (forced-single overflow, duplicate abort, early-stop reversion). + // Ending that with `max_tokens` would leave a call the client is told + // neither to run nor to drop, so it stays on the error path. + it("stream: a capped turn with an uncaptured streamed tool call still reports the failure", async () => { + mockMessages = [ + messageStart("msg_capped_dangling"), + toolUseBlockStart(0, "read", "toolu_dangling"), + inputJsonDelta(0, '{"file_path":"/x"}'), + blockStop(0), + { type: "result", subtype: "error_max_turns", is_error: true, session_id: "test-session" }, + ] + mockTerminalError = new Error("Claude Code returned an error result: Reached maximum number of turns (1)") + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "call read" }], + }, "es-capped-dangling") + expect(res.status).toBe(200) + const body = await res.text() + expect(body).toContain("event: error") + }) + + // The production failure this branch missed: a resumed passthrough turn that + // ran for two minutes, yielded no wire event at all, and terminated + // `max_turns turns=1` — telemetry recorded 0 content blocks, 0 text events + // and a null TTFB, so nothing had reached the client. The cap is the + // proxy's own and that turn never reached the tool boundary the cap exists + // to stop at, so the single turn bought nothing and cost the whole request + // (the user's own identical retry then answered normally). With nothing + // yielded there is no envelope to corrupt: reissue the turn with the cap + // lifted instead of dressing an empty turn as truncation. + const CAPPED_TURN_ERROR = "Claude Code returned an error result: Reached maximum number of turns (1)" + const cappedEmptyAttempt = () => ({ + messages: [{ type: "result", subtype: "error_max_turns", is_error: true, session_id: "test-session" }], + terminalError: new Error(CAPPED_TURN_ERROR), + }) + + it("stream: reissues a capped turn that produced nothing, with the turn cap lifted", async () => { + mockAttemptScripts = [ + cappedEmptyAttempt(), + { + messages: [ + messageStart("msg_cap_lifted"), + textBlockStart(0), + textDelta(0, "answered after the lift"), + blockStop(0), + messageDelta("end_turn"), + { type: "result", subtype: "success", is_error: false, session_id: "test-session" }, + ], + }, + ] + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "answer me" }], + }, "es-capped-lift") + expect(res.status).toBe(200) + const body = await res.text() + expect(body).toContain("answered after the lift") + expect(body).not.toContain("event: error") + expect(capturedQueryParamsAll.length).toBe(2) + expect(capturedQueryParamsAll[0].options.maxTurns).toBe(1) + expect(capturedQueryParamsAll[1].options.maxTurns).toBe(3) + }) + + // Once. A turn that comes back empty with the budget already lifted is not a + // cap artifact, and reissuing it again would spend turns on a shape that has + // already refused to answer twice. + it("stream: lifts the turn cap once, then reports the failure", async () => { + mockAttemptScripts = [cappedEmptyAttempt(), cappedEmptyAttempt()] + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "answer me" }], + }, "es-capped-lift-once") + expect(res.status).toBe(200) + const body = await res.text() + expect(body).toContain("event: error") + expect(capturedQueryParamsAll.length).toBe(2) + expect(capturedQueryParamsAll[1].options.maxTurns).toBe(3) + }) + + it("non-stream: reissues a capped turn that produced nothing, with the turn cap lifted", async () => { + mockAttemptScripts = [ + cappedEmptyAttempt(), + { + messages: [ + assistantMessage([{ type: "text", text: "answered after the lift" }]), + { type: "result", subtype: "success", is_error: false, session_id: "test-session" }, + ], + }, + ] + + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: false, + tools: [READ_TOOL], + messages: [{ role: "user", content: "answer me" }], + }, "es-capped-lift-ns") + expect(res.status).toBe(200) + const json = await res.json() as any + expect(json.stop_reason).toBe("end_turn") + expect(json.content[0].text).toBe("answered after the lift") + expect(capturedQueryParamsAll[0].options.maxTurns).toBe(1) + expect(capturedQueryParamsAll[1].options.maxTurns).toBe(3) + }) + + // The production shape was `resume=true`. The reissue has to keep resuming + // the same SDK session — a lift that fell back to a cold fresh session would + // answer from an empty transcript — while still taking a fork target of its + // own, so it cannot publish into the transcript the refused attempt claimed. + it("stream: keeps the resume target when it lifts the cap on a resumed turn", async () => { + mockMessages = [assistantMessage([{ type: "text", text: "first answer" }])] + const first = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: false, + tools: [READ_TOOL], + messages: [{ role: "user", content: "just talk" }], + }, "es-capped-lift-resume") + expect(first.status).toBe(200) + + mockAttemptScripts = [ + cappedEmptyAttempt(), + { + messages: [ + messageStart("msg_cap_lifted_resume"), + textBlockStart(0), + textDelta(0, "answered after the lift"), + blockStop(0), + messageDelta("end_turn"), + { type: "result", subtype: "success", is_error: false, session_id: "test-session" }, + ], + }, + ] + const second = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [ + { role: "user", content: "just talk" }, + { role: "assistant", content: [{ type: "text", text: "first answer" }] }, + { role: "user", content: "and again" }, + ], + }, "es-capped-lift-resume") + expect(second.status).toBe(200) + expect(await second.text()).toContain("answered after the lift") + + expect(capturedQueryParamsAll.length).toBe(3) + const capped = capturedQueryParamsAll[1] + const lifted = capturedQueryParamsAll[2] + expect(capped.options.resume).toBe(initialManagedSessionId()) + expect(capped.options.maxTurns).toBe(1) + expect(lifted.options.resume).toBe(initialManagedSessionId()) + expect(lifted.options.maxTurns).toBe(3) + expect(lifted.options.sessionId).not.toBe(capped.options.sessionId) + }) + + // The gate is the budget the attempt asked for, not the number the SDK + // happens to print. With a pinned budget there is no proxy cap to lift, so + // the failure is reported as-is instead of buying a second identical turn. + it("stream: does not reissue when the turn budget was not the proxy's own cap", async () => { + process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS = "3" + try { + mockAttemptScripts = [cappedEmptyAttempt()] + const res = await post(app, { + model: "claude-sonnet-4-5", + max_tokens: 400, + stream: true, + tools: [READ_TOOL], + messages: [{ role: "user", content: "answer me" }], + }, "es-capped-lift-pinned") + expect(res.status).toBe(200) + expect(await res.text()).toContain("event: error") + expect(capturedQueryParamsAll.length).toBe(1) + expect(capturedQueryParamsAll[0].options.maxTurns).toBe(3) + } finally { + delete process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS + } + }) + it("non-stream: a capped turn that captured no tool call does not fail the request", async () => { mockMessages = [ assistantMessage([{ type: "thinking", thinking: "pondering", signature: "sig" }]), diff --git a/src/__tests__/query.test.ts b/src/__tests__/query.test.ts index 7695b71c..422e0bdf 100644 --- a/src/__tests__/query.test.ts +++ b/src/__tests__/query.test.ts @@ -2,7 +2,7 @@ * Tests for the SDK query options builder. */ import { describe, it, expect } from "bun:test" -import { buildQueryOptions, GIT_STATUS_PROVENANCE_NOTE, resolveQueryConfigDir, type QueryContext } from "../proxy/query" +import { buildQueryOptions, GIT_STATUS_PROVENANCE_NOTE, resolveQueryConfigDir, singleTurnCapLiftRaisesBudget, type QueryContext } from "../proxy/query" import { BLOCKED_BUILTIN_TOOLS, CLAUDE_CODE_ONLY_TOOLS, MCP_SERVER_NAME, ALLOWED_MCP_TOOLS } from "../proxy/tools" import { CHERRY_BLOCKED_BUILTIN_TOOLS, CHERRY_INCOMPATIBLE_TOOLS, CHERRY_WEB_TOOLS } from "../proxy/adapters/cherry" @@ -163,6 +163,28 @@ describe("buildQueryOptions", () => { } }) + it("lifts the single-turn cap when a reissue asks for it — the capped turn produced nothing to stop at", () => { + const result = buildQueryOptions(makeContext({ passthrough: true, liftSingleTurnCap: true })) + expect(result.options.maxTurns).toBe(3) + }) + + it("reports the lift as raising the budget only when the cap is the proxy's own", () => { + const prev = process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS + delete process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS + try { + expect(singleTurnCapLiftRaisesBudget(false)).toBe(true) + // An operator who pinned the budget owns it: the reissue would spend a + // second turn on an identical attempt, so the caller must not fire. + process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS = "1" + expect(singleTurnCapLiftRaisesBudget(false)).toBe(false) + process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS = "5" + expect(singleTurnCapLiftRaisesBudget(false)).toBe(false) + } finally { + if (prev === undefined) delete process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS + else process.env.MERIDIAN_PASSTHROUGH_MAX_TURNS = prev + } + }) + it("keeps maxTurns at 4 with deferred tools — ToolSearch discovery is a real round-trip, so the cap must not apply (#547)", () => { const result = buildQueryOptions(makeContext({ passthrough: true, hasDeferredTools: true })) expect(result.options.maxTurns).toBe(4) diff --git a/src/__tests__/silent-turn-recovery.test.ts b/src/__tests__/silent-turn-recovery.test.ts index c110f78f..398d1d44 100644 --- a/src/__tests__/silent-turn-recovery.test.ts +++ b/src/__tests__/silent-turn-recovery.test.ts @@ -84,6 +84,7 @@ installMcpToolsMock(() => ({ })) const { createProxyServer } = await import("../proxy/server") +const { diagnosticLog, telemetryStore } = await import("../telemetry") const ev = (event: any) => ({ type: "stream_event", event, parent_tool_use_id: null, @@ -155,7 +156,7 @@ const msgEnd = () => [ ev({ type: "message_stop" }), ] -async function post(app: any, body: any, session = "silent-session") { +async function post(app: any, body: any, session = "silent-session", extraHeaders: Record = {}) { return app.fetch(new Request("http://localhost/v1/messages", { method: "POST", headers: { @@ -163,6 +164,7 @@ async function post(app: any, body: any, session = "silent-session") { "x-api-key": "dummy", "x-opencode-session": session, "user-agent": "opencode/1.0.0", + ...extraHeaders, }, body: JSON.stringify(body), })) @@ -386,6 +388,28 @@ describe("silent-turn recovery", () => { expect(queryCalls[2].options.resume).toBe(queryCalls[1].options.sessionId) }) + it("classifies a message envelope without content blocks as no_blocks", async () => { + process.env.MERIDIAN_SILENT_TURN_RECOVERY = "0" + scripted = [[msgStart()]] + const requestId = `silent-no-blocks-${crypto.randomUUID()}` + + const response = await post(app, REQUEST, "silent-no-blocks", { "x-request-id": requestId }) + const body = await read(response) + expect(response.status).toBe(200) + expect(body).toContain("message_start") + expect(body).toContain("message_stop") + + const log = diagnosticLog.getRecent({ limit: 200 }) + .find((entry: any) => entry.requestId === requestId && entry.message.includes("silent_turn")) + expect(log).toBeDefined() + expect(log!.message).toContain("reason=no_blocks") + + const row = telemetryStore.getRecent({ limit: 200 }) + .find((entry: any) => entry.requestId === requestId) + expect(row).toBeDefined() + expect(row!.contentBlocks).toBe(0) + }) + it("kill switch keeps detection but skips the extra turn", async () => { process.env.MERIDIAN_SILENT_TURN_RECOVERY = "0" scripted = [ diff --git a/src/proxy/query.ts b/src/proxy/query.ts index 017287c3..b36e4400 100644 --- a/src/proxy/query.ts +++ b/src/proxy/query.ts @@ -116,6 +116,14 @@ export interface QueryContext { * digest turn. */ earlyStop?: boolean + /** + * Reissue escape hatch for the single-turn cap. A capped turn that produced + * nothing at all — no wire event, no captured tool call — spent the budget + * without ever reaching the tool boundary the cap exists to stop at, so the + * caller reissues it once with the cap off. Never set on a first attempt; + * see the retry site in server.ts. + */ + liftSingleTurnCap?: boolean /** SDK session ID for resume (if continuing a session) */ resumeSessionId?: string /** Whether this is an undo operation */ @@ -240,6 +248,7 @@ function computePassthroughMaxTurns( hasDeferredTools: boolean, advisorModel: string | undefined, singleTurnHandoff: boolean, + liftSingleTurnCap: boolean, ): number { const deferredBump = hasDeferredTools ? 1 : 0 const defaultBase = 3 + deferredBump @@ -256,11 +265,33 @@ function computePassthroughMaxTurns( // silently override a value someone set to work around a client quirk. const operatorPinned = env("PASSTHROUGH_MAX_TURNS") !== undefined && configured > 0 const advisorBump = advisorModel ? 3 : 0 - if (singleTurnHandoff && !operatorPinned) return 1 + if (singleTurnHandoff && !liftSingleTurnCap && !operatorPinned) return 1 const base = configured > 0 ? configured : defaultBase return base + advisorBump } +/** + * Whether reissuing a capped passthrough turn with `liftSingleTurnCap` would + * actually raise the budget. + * + * Asked only about an attempt whose requested `maxTurns` was 1 — the caller + * reads that off the options it built — so `singleTurnHandoff` is a settled + * fact here, not an assumption: no other combination produces a budget of 1 + * except an operator pin. Which is the one case this answers false for: the + * cap is then theirs, not the proxy's, and the reissue would spend a second + * turn on an identical attempt. Answered by comparing the real computation + * against itself rather than by a copy of its conditions, so the two cannot + * drift. + */ +export function singleTurnCapLiftRaisesBudget( + hasDeferredTools: boolean, + advisorModel?: string, +): boolean { + const capped = computePassthroughMaxTurns(hasDeferredTools, advisorModel, true, false) + const lifted = computePassthroughMaxTurns(hasDeferredTools, advisorModel, true, true) + return lifted > capped +} + /** * Build an addendum that tells the model which path belongs to the real user. * Applied when the SDK subprocess runs in one directory on the proxy host but @@ -385,6 +416,7 @@ export function buildQueryOptions(ctx: QueryContext, abortController?: AbortCont // Every condition here is one that needs the SDK to keep going // past the tool boundary; see computePassthroughMaxTurns. ctx.earlyStop !== false && !hasDeferredTools && !ctx.advisorModel && !outputFormat, + ctx.liftSingleTurnCap === true, ) : 200, cwd: workingDirectory, diff --git a/src/proxy/server.ts b/src/proxy/server.ts index 1d8177f2..6aa9d900 100644 --- a/src/proxy/server.ts +++ b/src/proxy/server.ts @@ -77,7 +77,7 @@ import { translateResponsesToAnthropic, translateAnthropicToResponses, createRes import { extractAdvisorModel, extractSystemText, getLastUserMessage, stripAdvisorTools, stripNonStandardStreamFields, consolidateMultimodalOntoLastUser, MULTIMODAL_TYPES, buildToolUseIndex, describeToolCall, frameReplayTurns } from "./messages" import { requireAuth, authEnabled } from "./auth" import { detectAdapter } from "./adapters/detect" -import { buildQueryOptions, resolveQueryConfigDir, type QueryContext } from "./query" +import { buildQueryOptions, resolveQueryConfigDir, singleTurnCapLiftRaisesBudget, type QueryContext } from "./query" import { normalizeEffort } from "./effort" import { parseOutputFormat, structuredOutputText } from "./structuredOutput" import { runTransformHook, buildPipeline, createRequestContext } from "./transform" @@ -3196,6 +3196,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe let busySessionFork = false let sawUnresumableRefusal = false let managedCreationAttemptStarted = false + let singleTurnCapLifted = false while (true) { if (managedForkTarget) { if (managedCreationAttemptStarted) { @@ -3216,11 +3217,18 @@ export function createProxyServer(config: Partial = {}): ProxyServe // message arrives (release sites: assistant arrival in the // consumer loop, attempt error, loop exit). turnGenerating = true + // The turn budget THIS attempt asked for. The cap-lift branch + // below reissues only a turn the proxy itself capped at 1, + // read from the options it built rather than inferred from the + // SDK's "Reached maximum number of turns (N)" wording — which + // is an optional parse, and would make an uncapped budget that + // happens to report 1 look like the proxy's own cap. + let attemptMaxTurns: number | undefined try { if (resumeSessionId) resumedMappingMayBeAdvanced = true - for await (const event of runSdkQueryAttempt(buildQueryOptions({ + const attemptQuery = buildQueryOptions({ prompt: makePrompt(), model, workingDirectory, clientWorkingDirectory, systemContext, claudeExecutable, - passthrough, stream: false, sdkAgents, passthroughMcp, cleanEnv: profileEnv, envOverrides, hasDeferredTools, earlyStop: earlyStopEnabled, + passthrough, stream: false, sdkAgents, passthroughMcp, cleanEnv: profileEnv, envOverrides, hasDeferredTools, earlyStop: earlyStopEnabled, liftSingleTurnCap: singleTurnCapLifted, resumeSessionId, isUndo: sdkUndo, resumeSessionAtUuid: undoRollbackUuid ?? passthroughToolCallAssistantUuid, forkSession: busySessionFork || undefined, forkSessionId: managedForkTarget?.sessionId, sdkHooks, blockedTools: pipelineCtx.blockedTools, incompatibleTools: pipelineCtx.incompatibleTools, mcpServerName: adapter.getMcpServerName(), allowedMcpTools: pipelineCtx.allowedMcpTools, onStderr, effort, thinking, taskBudget, outputFormat, betas, settingSources, codeSystemPrompt: sdkFeatures.codeSystemPrompt, clientSystemPrompt: sdkFeatures.clientSystemPrompt === false ? false : undefined, @@ -3233,7 +3241,9 @@ export function createProxyServer(config: Partial = {}): ProxyServe ? sdkFeatures.additionalDirectories.split(",").map(d => d.trim()).filter(Boolean) : undefined, advisorModel, - }, requestAbort.controller), requestAbort.controller.signal, requestMeta, "non_stream", managedSdkAttemptLocators())) { + }, requestAbort.controller) + attemptMaxTurns = attemptQuery.options.maxTurns + for await (const event of runSdkQueryAttempt(attemptQuery, requestAbort.controller.signal, requestMeta, "non_stream", managedSdkAttemptLocators())) { // Capture Claude Max subscription quota updates emitted by // the SDK as rate_limit_event. We snapshot them in this // profile's slot of the (per-profile-scoped) rate limit @@ -3455,6 +3465,34 @@ export function createProxyServer(config: Partial = {}): ProxyServe } } + // Same lower boundary as the streaming path: a capped turn + // that produced nothing spent the single-turn budget without + // reaching the tool boundary the cap exists to stop at. + // Nothing was yielded (guarded above), so reissue it once + // with the cap lifted rather than answering an empty turn. + // + // Only for a turn this proxy capped at 1 (`attemptMaxTurns`): + // a turn that burned a real multi-turn budget would be + // reissued into an identical attempt. + if ( + passthrough && + !singleTurnCapLifted && + attemptMaxTurns === 1 && + capturedToolUses.length === 0 && + extractSdkTermination(errMsg).reason === "max_turns" && + singleTurnCapLiftRaisesBudget(hasDeferredTools, advisorModel) + ) { + singleTurnCapLifted = true + claudeLog("passthrough.single_turn_cap_lifted", { mode: "non_stream", model }) + diagnosticLog.session( + `${requestMeta.requestId} single_turn_cap_lifted mode=non_stream model=${model} ` + + `resume=${Boolean(resumeSessionId)}`, + requestMeta.requestId, + ) + plog(`[PROXY] ${requestMeta.requestId} capped turn produced nothing — retrying with the turn cap lifted`) + continue + } + throw error } } @@ -3741,7 +3779,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe if (canRecoverAsToolUse) { diagnosticLog.session( `${requestMeta.requestId} sdk_termination_recovered ${formatSdkTermination(sdkTerm, { - model, requestSource, isResume, hasDeferredTools, sdkSessionId: resumeSessionId, + model, requestSource, isResume, hasDeferredTools, sdkSessionId: currentSessionId || resumeSessionId, })} captured=${capturedToolUses.length}`, requestMeta.requestId, ) @@ -4088,6 +4126,9 @@ export function createProxyServer(config: Partial = {}): ProxyServe let heartbeatCount = 0 let streamEventsSeen = 0 let eventsForwarded = 0 + // Unlike eventsForwarded, this counts only content_block_start + // events that reached the client. + let contentBlocksForwarded = 0 let textEventsForwarded = 0 // Characters of forwarded text — the announce classification is a // length test (see turnOutcome.ts). @@ -4249,6 +4290,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe let busySessionFork = false let sawUnresumableRefusal = false let managedCreationAttemptStarted = false + let singleTurnCapLifted = false while (true) { if (managedForkTarget) { @@ -4266,11 +4308,15 @@ export function createProxyServer(config: Partial = {}): ProxyServe // stderr emitted by THIS attempt's subprocess only — retries // must not re-match a previous attempt's refusal text. const attemptStderrStart = stderrLines.length + // The turn budget THIS attempt asked for — see the non-stream + // twin above for why the cap-lift branch reads it here rather + // than from the SDK's termination wording. + let attemptMaxTurns: number | undefined try { if (resumeSessionId) resumedMappingMayBeAdvanced = true - for await (const event of runSdkQueryAttempt(buildQueryOptions({ + const attemptQuery = buildQueryOptions({ prompt: makePrompt(), model, workingDirectory, clientWorkingDirectory, systemContext, claudeExecutable, - passthrough, stream: true, sdkAgents, passthroughMcp, cleanEnv: profileEnv, envOverrides, hasDeferredTools, earlyStop: earlyStopEnabled, + passthrough, stream: true, sdkAgents, passthroughMcp, cleanEnv: profileEnv, envOverrides, hasDeferredTools, earlyStop: earlyStopEnabled, liftSingleTurnCap: singleTurnCapLifted, resumeSessionId, isUndo: sdkUndo, resumeSessionAtUuid: undoRollbackUuid ?? passthroughToolCallAssistantUuid, forkSession: busySessionFork || undefined, forkSessionId: managedForkTarget?.sessionId, sdkHooks, blockedTools: pipelineCtx.blockedTools, incompatibleTools: pipelineCtx.incompatibleTools, mcpServerName: adapter.getMcpServerName(), allowedMcpTools: pipelineCtx.allowedMcpTools, onStderr, effort, thinking, taskBudget, outputFormat, betas, settingSources, codeSystemPrompt: sdkFeatures.codeSystemPrompt, clientSystemPrompt: sdkFeatures.clientSystemPrompt === false ? false : undefined, @@ -4283,7 +4329,9 @@ export function createProxyServer(config: Partial = {}): ProxyServe ? sdkFeatures.additionalDirectories.split(",").map(d => d.trim()).filter(Boolean) : undefined, advisorModel, - }, requestAbort.controller), requestAbort.controller.signal, requestMeta, "stream", managedSdkAttemptLocators())) { + }, requestAbort.controller) + attemptMaxTurns = attemptQuery.options.maxTurns + for await (const event of runSdkQueryAttempt(attemptQuery, requestAbort.controller.signal, requestMeta, "stream", managedSdkAttemptLocators())) { // Same SDK rate-limit capture as the non-stream path. if ((event as any).type === "rate_limit_event") { rateLimitStore.record(profile.id, (event as any).rate_limit_info) @@ -4480,6 +4528,45 @@ export function createProxyServer(config: Partial = {}): ProxyServe } } + // A capped turn that produced NOTHING never reached the + // tool boundary the cap exists to stop at, so the single + // turn bought nothing and cost the whole turn. Observed in + // production as a resumed opus[1m] passthrough turn that + // ran 108s, yielded no wire event at all, and terminated + // `max_turns turns=1`; the client's own identical retry + // then answered normally. Reissue it once with the cap + // lifted — the refused turn is the one that would have + // answered. + // + // Safe by the guard at the top of this catch: nothing was + // yielded downstream, so no SSE frame, no message_start + // and no committed priority exposure can be duplicated by + // a second attempt. `capturedToolUses` is checked too + // because a capped turn WITH captured calls is already + // recoverable as a tool_use envelope downstream, and that + // is the cheaper answer. `attemptMaxTurns === 1` keeps it + // to turns this proxy capped itself: an uncapped budget + // that ran out is a different failure, and reissuing it + // would spend a turn on an identical attempt. + if ( + passthrough && + !singleTurnCapLifted && + attemptMaxTurns === 1 && + capturedToolUses.length === 0 && + extractSdkTermination(errMsg).reason === "max_turns" && + singleTurnCapLiftRaisesBudget(hasDeferredTools, advisorModel) + ) { + singleTurnCapLifted = true + claudeLog("passthrough.single_turn_cap_lifted", { mode: "stream", model }) + diagnosticLog.session( + `${requestMeta.requestId} single_turn_cap_lifted mode=stream model=${model} ` + + `resume=${Boolean(resumeSessionId)}`, + requestMeta.requestId, + ) + plog(`[PROXY] ${requestMeta.requestId} capped turn produced nothing — retrying with the turn cap lifted`) + continue + } + throw error } } @@ -4863,6 +4950,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe break } eventsForwarded += 1 + if (eventType === "content_block_start") contentBlocksForwarded += 1 } // Track envelope integrity: which forwarded blocks are open. @@ -5105,10 +5193,14 @@ export function createProxyServer(config: Partial = {}): ProxyServe // In particular, message_delta is client permission to finalize. if (pendingStructuredFrames.length > 0) { clientAssistantContentExposed = true + let structuredFramesForwarded = 0 for (const frame of pendingStructuredFrames) { - safeEnqueue(frame.payload, frame.source) + if (safeEnqueue(frame.payload, frame.source)) { + structuredFramesForwarded += 1 + if (frame.source === "structured_block_start") contentBlocksForwarded += 1 + } } - eventsForwarded += pendingStructuredFrames.length + eventsForwarded += structuredFramesForwarded pendingStructuredFrames = [] messageStartEmitted = true textEventsForwarded += 1 @@ -5132,7 +5224,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe const classifyNow = () => classifyTurnOutcome({ textEvents: textEventsForwarded, toolUses: streamedToolUseIds.size, - blocksForwarded: eventsForwarded, + blocksForwarded: contentBlocksForwarded, }) const preRecoveryOutcome = classifyNow() // @@ -5432,9 +5524,10 @@ export function createProxyServer(config: Partial = {}): ProxyServe capturedToolUses.splice(capturedBeforeRecovery) } else if (silentTurnRecovered) { for (const lifted of recoveryLiftedFrames) { - safeEnqueue(encoder.encode( + const delivered = safeEnqueue(encoder.encode( `event: ${lifted.frame.type}\ndata: ${JSON.stringify(lifted.frame)}\n\n`, ), `silent_recovery_${lifted.kind}`) + if (delivered && lifted.kind === "block_start") contentBlocksForwarded += 1 if (lifted.kind === "block_start") { eventsForwarded += 1 } else if (lifted.kind === "text_delta") { @@ -5457,7 +5550,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe if (silentTurnRecovered && preRecoveryOutcome.kind === "silent") { diagnosticLog.session( `${requestMeta.requestId} silent_turn reason=${preRecoveryOutcome.reason} ` + - `blocks=${eventsForwarded} out=${lastUsage?.output_tokens ?? 0} ` + + `blocks=${contentBlocksForwarded} out=${lastUsage?.output_tokens ?? 0} ` + `recovery=succeeded`, requestMeta.requestId, ) @@ -5511,14 +5604,15 @@ export function createProxyServer(config: Partial = {}): ProxyServe streamedToolUseIds.add(tu.id) // content_block_start - safeEnqueue(encoder.encode( + if (safeEnqueue(encoder.encode( `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: blockIndex, content_block: { type: "tool_use", id: tu.id, name: tu.name, input: {} } })}\n\n` - ), "passthrough_tool_block_start") - + ), "passthrough_tool_block_start")) { + contentBlocksForwarded += 1 + } // input_json_delta with the full input safeEnqueue(encoder.encode( `event: content_block_delta\ndata: ${JSON.stringify({ @@ -5556,13 +5650,15 @@ export function createProxyServer(config: Partial = {}): ProxyServe const streamFileChangeSummary = formatFileChangeSummary(fileChanges) if (streamFileChangeSummary && messageStartEmitted) { const fcBlockIndex = nextClientBlockIndex++ - safeEnqueue(encoder.encode( + if (safeEnqueue(encoder.encode( `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: fcBlockIndex, content_block: { type: "text", text: "" }, })}\n\n` - ), "file_changes_block_start") + ), "file_changes_block_start")) { + contentBlocksForwarded += 1 + } safeEnqueue(encoder.encode( `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", @@ -5654,7 +5750,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe ttfbMs: requestMeta.ttfbMs ?? null, upstreamDurationMs: requestMeta.sdkActiveDurationMs, totalDurationMs: streamTotalDurationMs, - contentBlocks: eventsForwarded, + contentBlocks: contentBlocksForwarded, textEvents: textEventsForwarded, error: null, inputTokens: lastUsage?.input_tokens, @@ -5687,7 +5783,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe // means "the loop just lost a turn". diagnosticLog.session( `${requestMeta.requestId} silent_turn reason=${turnOutcome.reason} ` + - `blocks=${eventsForwarded} out=${lastUsage?.output_tokens ?? 0} ` + + `blocks=${contentBlocksForwarded} out=${lastUsage?.output_tokens ?? 0} ` + `recovery=${silentTurnRecoveryAttempted ? (silentTurnRecovered ? "succeeded" : "failed") : "off"}`, requestMeta.requestId, ) @@ -5884,7 +5980,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe requestSource, isResume, hasDeferredTools, - sdkSessionId: resumeSessionId, + sdkSessionId: currentSessionId || resumeSessionId, })} captured=${capturedToolUses.length}`, requestMeta.requestId, ) @@ -5904,13 +6000,15 @@ export function createProxyServer(config: Partial = {}): ProxyServe const tu = unseenToolUses[i]! const blockIndex = nextClientBlockIndex++ streamedToolUseIds.add(tu.id) - safeEnqueue(encoder.encode( + if (safeEnqueue(encoder.encode( `event: content_block_start\ndata: ${JSON.stringify({ type: "content_block_start", index: blockIndex, content_block: { type: "tool_use", id: tu.id, name: tu.name, input: {} } })}\n\n` - ), "recover_tool_block_start") + ), "recover_tool_block_start")) { + contentBlocksForwarded += 1 + } safeEnqueue(encoder.encode( `event: content_block_delta\ndata: ${JSON.stringify({ type: "content_block_delta", @@ -6001,7 +6099,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe `event: message_delta\ndata: ${JSON.stringify({ type: "message_delta", delta: { stop_reason: "tool_use", stop_sequence: null }, - usage: { output_tokens: 0 } + usage: { output_tokens: lastUsage?.output_tokens ?? 0 } })}\n\n` ), "recover_message_delta") safeEnqueue(encoder.encode( @@ -6029,7 +6127,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe toolCount, lineageType, messageCount: allMessages.length, - sdkSessionId: resumeSessionId, + sdkSessionId: currentSessionId || resumeSessionId, status: 200, queueWaitMs: recoverQueueWaitMs, sessionQueueWaitMs: requestMeta.sessionQueueWaitMs, @@ -6038,7 +6136,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe ttfbMs: requestMeta.ttfbMs ?? null, upstreamDurationMs: requestMeta.sdkActiveDurationMs, totalDurationMs: recoverTotalMs, - contentBlocks: eventsForwarded + unseenToolUses.length, + contentBlocks: contentBlocksForwarded, textEvents: textEventsForwarded, error: null, // The capped tool handoff makes this the ordinary path for a @@ -6065,14 +6163,136 @@ export function createProxyServer(config: Partial = {}): ProxyServe return } + // The streaming counterpart of the non-streaming capped-turn + // branch above. The turn spent its single-turn budget without + // producing a forwardable tool call, but content already reached + // the client. Falling through would answer a half-delivered turn + // with an error frame, which a client can only surface as a hard + // failure over an answer that is already on screen — the + // "Reached maximum number of turns (1)" stream error reported + // against a turn whose text had rendered. + // + // Close it the way every other cut-off turn is closed instead: + // `max_tokens`, the wire's word for truncation, and no error + // event. `end_turn` would be the silent-turn lie #768 exists to + // prevent, and the error frame is a dead end where the client + // could otherwise continue. The non-streaming path has answered + // this shape honestly since the turn budget dropped to one; its + // comment already claims streaming does the same, and this is + // what makes that true. + // + // Nothing durable changes on this path: with no captured tool + // calls there is no checkpoint to publish and no mapping to + // advance, exactly as on the throwing path it replaces. + // + // Two shapes are deliberately left on the error path. + // + // A turn that forwarded only `message_start`, or an empty text + // block, delivered no actionable content. `eventsForwarded` + // includes envelope events and `nextClientBlockIndex` includes + // non-text blocks, so neither is a content oracle. Use the same + // text-delta count as classifyTurnOutcome instead. + // + // A tool_use block already on the wire with nothing captured + // means the hook never let those calls stand (forced-single + // overflow, duplicate abort, early-stop reversion). Ending that + // with `max_tokens` leaves a call the client is told neither to + // run nor to discard, so it keeps the error it gets today. + if ( + passthrough && + sdkTerm.reason === "max_turns" && + capturedToolUses.length === 0 && + streamedToolUseIds.size === 0 && + messageStartEmitted && + textEventsForwarded > 0 + ) { + flushOpenClientBlocks("capped_turn") + diagnosticLog.session( + `${requestMeta.requestId} sdk_termination_truncated ${formatSdkTermination(sdkTerm, { + model, + requestSource, + isResume, + hasDeferredTools, + sdkSessionId: currentSessionId || resumeSessionId, + })} blocks=${nextClientBlockIndex}`, + requestMeta.requestId, + ) + claudeLog("passthrough.capped_turn_truncated", { + mode: "stream", + blocks: nextClientBlockIndex, + }) + plog(`[PROXY] ${requestMeta.requestId} capped turn produced no forwardable tool call — reporting as truncated`) + safeEnqueue(encoder.encode( + `event: message_delta\ndata: ${JSON.stringify({ + type: "message_delta", + delta: { stop_reason: "max_tokens", stop_sequence: null }, + usage: { output_tokens: lastUsage?.output_tokens ?? 0 } + })}\n\n` + ), "capped_turn_message_delta") + safeEnqueue(encoder.encode( + `event: message_stop\ndata: {"type":"message_stop"}\n\n` + ), "capped_turn_message_stop") + + if (lastUsage) logUsage(requestMeta.requestId, lastUsage) + const cappedTotalMs = Date.now() - requestStartAt + const cappedQueueWaitMs = totalQueueWaitMs(requestMeta) + telemetryStore.record({ + requestId: requestMeta.requestId, + timestamp: Date.now(), + adapter: adapter.name, + profileId: profile.id, + requestSource, + model, + requestModel: body.model || undefined, + mode: "stream", + isResume, + isPassthrough: passthrough, + hasDeferredTools, + deferredToolCount: hasDeferredTools ? deferredToolCount : undefined, + toolCount, + lineageType, + messageCount: allMessages.length, + sdkSessionId: currentSessionId || resumeSessionId, + status: 200, + queueWaitMs: cappedQueueWaitMs, + sessionQueueWaitMs: requestMeta.sessionQueueWaitMs, + sdkQueueWaitMs: requestMeta.sdkQueueWaitMs, + proxyOverheadMs: Math.max(0, cappedTotalMs - cappedQueueWaitMs - requestMeta.sdkActiveDurationMs), + ttfbMs: requestMeta.ttfbMs ?? null, + upstreamDurationMs: requestMeta.sdkActiveDurationMs, + totalDurationMs: cappedTotalMs, + contentBlocks: contentBlocksForwarded, + textEvents: textEventsForwarded, + error: null, + inputTokens: lastUsage?.input_tokens, + outputTokens: lastUsage?.output_tokens, + cacheReadInputTokens: lastUsage?.cache_read_input_tokens, + cacheCreationInputTokens: lastUsage?.cache_creation_input_tokens, + cacheHitRate: computeCacheHitRate(lastUsage), + ...(envelopeViolations.length > 0 ? { envelopeViolations: [...envelopeViolations] } : {}), + }) + + if (!streamClosed) { + try { controller.close() } catch {} + streamClosed = true + } + return + } + + // What the client actually got is part of the failure, not a + // detail: the recovery and truncation branches above key off it, + // so a bare termination line leaves the operator unable to tell + // an error over rendered text from one that delivered nothing. + // Reconstructing it from telemetry's null TTFB is archaeology. diagnosticLog.error( `${requestMeta.requestId} ${formatSdkTermination(sdkTerm, { model, requestSource, isResume, hasDeferredTools, - sdkSessionId: resumeSessionId, - })}`, + sdkSessionId: currentSessionId || resumeSessionId, + })} envelope=${messageStartEmitted ? "open" : "unopened"} blocks=${contentBlocksForwarded} ` + + `text=${textEventsForwarded} tools=${capturedToolUses.length}/${streamedToolUseIds.size}`, requestMeta.requestId, ) @@ -6097,7 +6317,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe toolCount, lineageType, messageCount: allMessages.length, - sdkSessionId: resumeSessionId, + sdkSessionId: currentSessionId || resumeSessionId, status: streamErr.status, queueWaitMs: streamErrQueueWaitMs, sessionQueueWaitMs: requestMeta.sessionQueueWaitMs, @@ -6106,7 +6326,7 @@ export function createProxyServer(config: Partial = {}): ProxyServe ttfbMs: requestMeta.ttfbMs ?? null, upstreamDurationMs: requestMeta.sdkActiveDurationMs, totalDurationMs: streamErrTotalMs, - contentBlocks: eventsForwarded, + contentBlocks: contentBlocksForwarded, textEvents: textEventsForwarded, error: streamErr.type, })