From cc336acec461fbf3b00f44b66b7680472db43ca4 Mon Sep 17 00:00:00 2001 From: Alexander Kirilin Date: Fri, 24 Jul 2026 20:16:44 -0400 Subject: [PATCH] fix(omp): fit native compaction after large tool output --- packages/agent/CHANGELOG.md | 4 + packages/agent/src/compaction/compaction.ts | 44 +++-- packages/agent/src/compaction/openai.ts | 134 ++++++++++++++- packages/agent/test/remote-compaction.test.ts | 162 ++++++++++++++++-- 4 files changed, 315 insertions(+), 29 deletions(-) diff --git a/packages/agent/CHANGELOG.md b/packages/agent/CHANGELOG.md index 2cb3f71dc..5de9a6d3c 100644 --- a/packages/agent/CHANGELOG.md +++ b/packages/agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Reduced oversized OpenAI native compaction requests by replacing only trailing tool-output bodies that exceed the model context window, while preserving calls, assistant history, and reasoning. + ## [17.1.2] - 2026-07-24 ### Added diff --git a/packages/agent/src/compaction/compaction.ts b/packages/agent/src/compaction/compaction.ts index b31015947..036d6868a 100644 --- a/packages/agent/src/compaction/compaction.ts +++ b/packages/agent/src/compaction/compaction.ts @@ -51,6 +51,7 @@ import { requestOpenAiRemoteCompaction, requestRemoteCompaction, shouldUseOpenAiRemoteCompaction, + trimRemoteCompactionInputToContextWindow, withOpenAiRemoteCompactionPreserveData, } from "./openai"; import autoHandoffThresholdFocusPrompt from "./prompts/auto-handoff-threshold-focus.md" with { type: "text" }; @@ -1290,7 +1291,7 @@ function buildOpenAiResponsesCompactionInput( messages: Message[], model: Model<"openai-responses" | "azure-openai-responses" | "openai-codex-responses">, previousReplacementHistory: Array> | undefined, -): unknown[] { +): Array> { const input = buildResponsesInput({ model, context: { messages }, @@ -1300,7 +1301,9 @@ function buildOpenAiResponsesCompactionInput( includeThinkingSignatures: true, repairOrphanOutputs: true, }); - return previousReplacementHistory ? [...previousReplacementHistory, ...input] : input; + return (previousReplacementHistory ? [...previousReplacementHistory, ...input] : input) as Array< + Record + >; } /** @@ -1415,20 +1418,33 @@ export async function compact( ); if (remoteHistory.length > 0) { try { - const request = buildCompactionV2Request( - model, + const instructions = summaryOptions.remoteInstructions ?? SUMMARIZATION_SYSTEM_PROMPT; + const tools = summaryOptions.tools + ? convertTools(summaryOptions.tools, model.compat.supportsStrictMode, model) + : undefined; + const trimmed = trimRemoteCompactionInputToContextWindow( remoteHistory, - summaryOptions.remoteInstructions ?? SUMMARIZATION_SYSTEM_PROMPT, - { - tools: summaryOptions.tools - ? convertTools(summaryOptions.tools, model.compat.supportsStrictMode, model) - : undefined, - reasoning: buildCompactionV2Reasoning(model, summaryOptions.thinkingLevel), - sessionId: summaryOptions.sessionId, - promptCacheKey: summaryOptions.promptCacheKey, - retainedMessageBudget: settings.v2RetainedMessageBudget, - }, + model.contextWindow, + instructions, + tools, ); + if (trimmed.rewrittenOutputs > 0) { + logger.info("Rewrote trailing tool outputs before OpenAI V2 remote compaction", { + model: model.id, + provider: model.provider, + rewrittenOutputs: trimmed.rewrittenOutputs, + estimatedTokensBefore: trimmed.estimatedTokensBefore, + estimatedTokensAfter: trimmed.estimatedTokensAfter, + contextWindow: model.contextWindow, + }); + } + const request = buildCompactionV2Request(model, trimmed.input, instructions, { + tools, + reasoning: buildCompactionV2Reasoning(model, summaryOptions.thinkingLevel), + sessionId: summaryOptions.sessionId, + promptCacheKey: summaryOptions.promptCacheKey, + retainedMessageBudget: settings.v2RetainedMessageBudget, + }); const remote = await withAuth( apiKey, key => diff --git a/packages/agent/src/compaction/openai.ts b/packages/agent/src/compaction/openai.ts index c745b54fd..a3ed49595 100644 --- a/packages/agent/src/compaction/openai.ts +++ b/packages/agent/src/compaction/openai.ts @@ -45,6 +45,7 @@ import { OPENAI_HEADERS, } from "@oh-my-pi/pi-catalog/wire/codex"; import { $env, logger, stringifyJson, structuredCloneJSON } from "@oh-my-pi/pi-utils"; +import { countTokens } from "../tokenizer"; export * from "./compaction-v2-streaming"; @@ -66,6 +67,120 @@ export const REMOTE_COMPACTION_TIMEOUT_MS = 180_000; const DEFAULT_AZURE_API_VERSION = "v1"; +export const CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE = "Output exceeded the available model context and was truncated"; + +const REMOTE_COMPACTION_REQUEST_OVERHEAD_TOKENS = 256; +const REMOTE_COMPACTION_IMAGE_TOKEN_ESTIMATE = 765; +const REMOTE_COMPACTION_ORIGINAL_IMAGE_TOKEN_ESTIMATE = 10_000; + +interface NormalizedEstimateValue { + value: unknown; + imageTokens: number; +} + +function normalizeRemoteCompactionEstimateValue(value: unknown): NormalizedEstimateValue { + if (Array.isArray(value)) { + const normalized: unknown[] = []; + let imageTokens = 0; + for (const item of value) { + const result = normalizeRemoteCompactionEstimateValue(item); + normalized.push(result.value); + imageTokens += result.imageTokens; + } + return { value: normalized, imageTokens }; + } + if (!value || typeof value !== "object") return { value, imageTokens: 0 }; + + const record = value as Record; + if (record.type === "input_image") { + return { + value: { ...record, image_url: "" }, + imageTokens: + record.detail === "original" + ? REMOTE_COMPACTION_ORIGINAL_IMAGE_TOKEN_ESTIMATE + : REMOTE_COMPACTION_IMAGE_TOKEN_ESTIMATE, + }; + } + + const normalized: Record = {}; + let imageTokens = 0; + for (const [key, item] of Object.entries(record)) { + const result = normalizeRemoteCompactionEstimateValue(item); + normalized[key] = result.value; + imageTokens += result.imageTokens; + } + return { value: normalized, imageTokens }; +} + +export interface TrimRemoteCompactionInputResult { + input: Array>; + rewrittenOutputs: number; + estimatedTokensBefore: number; + estimatedTokensAfter: number; +} + +function estimateRemoteCompactionInputTokens( + input: Array>, + instructions: string, + tools?: unknown[], +): number { + const normalized = normalizeRemoteCompactionEstimateValue({ instructions, input, ...(tools ? { tools } : {}) }); + const serialized = stringifyJson(normalized.value) ?? ""; + return countTokens(serialized) + normalized.imageTokens + REMOTE_COMPACTION_REQUEST_OVERHEAD_TOKENS; +} + +function rewriteToolOutputForContextWindow(item: Record): Record | undefined { + if (item.type === "function_call_output" || item.type === "custom_tool_call_output") { + return { ...item, output: CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE }; + } + if (item.type === "tool_search_output") { + return { ...item, tools: [] }; + } + return undefined; +} + +/** + * Preserve the full native transcript unless trailing tool outputs alone push a + * remote compaction request beyond the model window. Replacing only those + * outputs keeps call/result pairing and all earlier assistant/reasoning history, + * matching Codex's recovery path for oversized tool turns. + */ +export function trimRemoteCompactionInputToContextWindow( + input: Array>, + contextWindow: number | null | undefined, + instructions: string, + tools?: unknown[], +): TrimRemoteCompactionInputResult { + const estimatedTokensBefore = estimateRemoteCompactionInputTokens(input, instructions, tools); + if (!contextWindow || contextWindow <= 0 || estimatedTokensBefore <= contextWindow) { + return { + input, + rewrittenOutputs: 0, + estimatedTokensBefore, + estimatedTokensAfter: estimatedTokensBefore, + }; + } + + let rewrittenInput: Array> | undefined; + let estimatedTokensAfter = estimatedTokensBefore; + let rewrittenOutputs = 0; + for (let index = input.length - 1; index >= 0 && estimatedTokensAfter > contextWindow; index--) { + const rewritten = rewriteToolOutputForContextWindow(input[index]); + if (!rewritten) break; + rewrittenInput ??= input.slice(); + rewrittenInput[index] = rewritten; + rewrittenOutputs++; + estimatedTokensAfter = estimateRemoteCompactionInputTokens(rewrittenInput, instructions, tools); + } + + return { + input: rewrittenInput ?? input, + rewrittenOutputs, + estimatedTokensBefore, + estimatedTokensAfter, + }; +} + /** Race the caller's signal against the request timeout; `timeoutMs <= 0` disables the watchdog. */ function withRequestTimeout(signal: AbortSignal | undefined, timeoutMs: number): AbortSignal | undefined { if (timeoutMs <= 0) return signal; @@ -622,12 +737,23 @@ export async function requestOpenAiRemoteCompaction( ): Promise { const endpoint = resolveOpenAiCompactEndpoint(model); const requestModel = resolveOpenAiCompactModel(model); + const trimmed = trimRemoteCompactionInputToContextWindow(compactInput, model.contextWindow, instructions); + if (trimmed.rewrittenOutputs > 0) { + logger.info("Rewrote trailing tool outputs before OpenAI remote compaction", { + model: model.id, + provider: model.provider, + rewrittenOutputs: trimmed.rewrittenOutputs, + estimatedTokensBefore: trimmed.estimatedTokensBefore, + estimatedTokensAfter: trimmed.estimatedTokensAfter, + contextWindow: model.contextWindow, + }); + } const request: OpenAiRemoteCompactionRequest = { model: requestModel, - // Send full history to the endpoint - don't trim locally. - // The provider handles compression via the compaction endpoint. - // Trimming before sending loses assistant messages and thinking blocks. - input: compactInput, + // Preserve the native transcript. Only oversized trailing tool outputs are + // rewritten above, reducing the request without losing assistant turns, + // reasoning, or call/result pairing. + input: trimmed.input, instructions, }; const isAzureOpenAiResponses = (model.remoteCompaction?.api ?? model.api) === "azure-openai-responses"; diff --git a/packages/agent/test/remote-compaction.test.ts b/packages/agent/test/remote-compaction.test.ts index c561e63bd..ee68960f2 100644 --- a/packages/agent/test/remote-compaction.test.ts +++ b/packages/agent/test/remote-compaction.test.ts @@ -10,12 +10,14 @@ import { import { buildCompactionV2Request, buildOpenAiNativeHistory, + CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE, getCompactionV2PreserveData, requestCompactionV2Streaming, requestOpenAiRemoteCompaction, requestRemoteCompaction, shouldUseCompactionV2Streaming, shouldUseOpenAiRemoteCompaction, + trimRemoteCompactionInputToContextWindow, } from "@oh-my-pi/pi-agent-core/compaction/openai"; import * as ai from "@oh-my-pi/pi-ai"; import { getOpenAICodexTransportDetails } from "@oh-my-pi/pi-ai/providers/openai-codex-responses"; @@ -460,13 +462,19 @@ describe("buildOpenAiNativeHistory computer calls", () => { }); describe("remote compaction input forwarding", () => { - test("sends the full native history without local trimming", async () => { - // Contract: the compact endpoint owns compression. Trimming locally dropped - // assistant turns + encrypted reasoning before the provider ever saw them, - // so the client now forwards the full input untouched even on a tiny window. + test("rewrites an oversized trailing tool output without dropping native history", async () => { + // The compact endpoint still receives every call/result item. Only the + // trailing output body is replaced when the request cannot fit. + const nativeInput = [ + { type: "custom_tool_call", call_id: "call_apply_1", name: "apply_patch", input: "{}" }, + { type: "custom_tool_call_output", call_id: "call_apply_1", output: "patch applied".repeat(1_000) }, + ]; let requestInput: Array> | undefined; const fetchMock: FetchImpl = async (_input, init) => { - const body = JSON.parse(String(init?.body)) as { input: Array> }; + const body: unknown = JSON.parse(String(init?.body)); + if (!isRecord(body) || !Array.isArray(body.input) || !body.input.every(isRecord)) { + throw new Error("expected remote compaction input"); + } requestInput = body.input; return Response.json({ output: [{ type: "compaction_summary", summary: "compact" }], @@ -474,19 +482,87 @@ describe("remote compaction input forwarding", () => { }; await requestOpenAiRemoteCompaction( - makeOpenAiModel({ contextWindow: 1 }), + makeOpenAiModel({ contextWindow: 1_000 }), "test-key", - [ - { type: "custom_tool_call", call_id: "call_apply_1", name: "apply_patch", input: "x".repeat(10_000) }, - { type: "custom_tool_call_output", call_id: "call_apply_1", output: "patch applied".repeat(1_000) }, - ], + nativeInput, "compact", undefined, { fetch: fetchMock }, ); + const trimmed = trimRemoteCompactionInputToContextWindow(nativeInput, 1_000, "compact"); + expect(trimmed.estimatedTokensAfter).toBeLessThanOrEqual(1_000); expect(requestInput?.some(item => item.type === "custom_tool_call")).toBe(true); - expect(requestInput?.some(item => item.type === "custom_tool_call_output")).toBe(true); + expect(requestInput?.find(item => item.type === "custom_tool_call_output")?.output).toBe( + CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE, + ); + }); + + test("rewrites contiguous trailing outputs until the request fits", () => { + const input = [ + { type: "function_call", call_id: "call_1", name: "read", arguments: "{}" }, + { type: "function_call", call_id: "call_2", name: "read", arguments: "{}" }, + { type: "function_call_output", call_id: "call_1", output: "a".repeat(8_000) }, + { type: "function_call_output", call_id: "call_2", output: "b".repeat(8_000) }, + ]; + + const result = trimRemoteCompactionInputToContextWindow(input, 1_000, "compact"); + + expect(result.rewrittenOutputs).toBe(2); + expect(result.input.slice(0, 2)).toEqual(input.slice(0, 2)); + expect(result.input.slice(2).map(item => item.output)).toEqual([ + CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE, + CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE, + ]); + expect(result.estimatedTokensAfter).toBeLessThanOrEqual(1_000); + expect(input[2].output).toHaveLength(8_000); + }); + + test("charges inline images by vision tokens instead of serialized base64 size", () => { + const input = [ + { + type: "message", + role: "user", + content: [ + { type: "input_image", detail: "auto", image_url: `data:image/png;base64,${"a".repeat(80_000)}` }, + ], + }, + { type: "function_call_output", call_id: "call_1", output: "useful result" }, + ]; + + const result = trimRemoteCompactionInputToContextWindow(input, 2_000, "compact"); + + expect(result.rewrittenOutputs).toBe(0); + expect(result.input).toBe(input); + expect(result.estimatedTokensAfter).toBeLessThanOrEqual(2_000); + }); + + test("reserves the high-resolution patch budget for original-detail images", () => { + const input = [ + { + type: "message", + role: "user", + content: [ + { type: "input_image", detail: "original", image_url: `data:image/png;base64,${"a".repeat(80_000)}` }, + ], + }, + { type: "function_call_output", call_id: "call_1", output: "useful result".repeat(2_000) }, + ]; + + const result = trimRemoteCompactionInputToContextWindow(input, 12_000, "compact"); + + expect(result.estimatedTokensBefore).toBeGreaterThan(12_000); + expect(result.rewrittenOutputs).toBe(1); + expect(result.estimatedTokensAfter).toBeLessThanOrEqual(12_000); + }); + + test("returns the original input without allocation when it already fits", () => { + const input = [{ type: "function_call_output", call_id: "call_1", output: "small" }]; + + const result = trimRemoteCompactionInputToContextWindow(input, 1_000, "compact"); + + expect(result.rewrittenOutputs).toBe(0); + expect(result.input).toBe(input); }); }); @@ -1347,6 +1423,70 @@ describe("compact() remote compaction failure handling", () => { expect(completeSpy).not.toHaveBeenCalled(); }); + test("rewrites an oversized trailing tool output before V2 streaming compaction", async () => { + const preparation = makePreparation(); + preparation.settings = { ...preparation.settings, remoteStreamingV2Enabled: true }; + preparation.messagesToSummarize = [ + { role: "user", content: "inspect the large file", timestamp: 1 }, + { + role: "assistant", + content: [ + { type: "toolCall", id: "call_read_large|fc_read_large", name: "read", arguments: { path: "/tmp/x" } }, + ], + timestamp: 2, + provider: "openai", + model: "gpt-5", + api: "openai-responses", + usage: ZERO_USAGE, + stopReason: "toolUse", + }, + { + role: "toolResult", + toolCallId: "call_read_large|fc_read_large", + toolName: "read", + content: [{ type: "text", text: "large output".repeat(10_000) }], + isError: false, + timestamp: 3, + }, + ]; + preparation.recentMessages = []; + const model = makeOpenAiModel({ + contextWindow: 500, + remoteCompaction: { + enabled: true, + v2StreamingEnabled: true, + v2Endpoint: "https://compact.example/v1/responses", + }, + }); + let requestInput: Array> = []; + const fetchMock: FetchImpl = async (_input, init) => { + const body: unknown = JSON.parse(String(init?.body)); + if (!isRecord(body) || !Array.isArray(body.input) || !body.input.every(isRecord)) { + throw new Error("expected V2 compaction input"); + } + requestInput = body.input; + return sseResponse([ + { + type: "response.output_item.done", + output_index: 0, + item: { type: "compaction", encrypted_content: "enc_trimmed" }, + }, + { + type: "response.completed", + response: { usage: { input_tokens: 100, output_tokens: 2, total_tokens: 102 } }, + }, + ]); + }; + + await compact(preparation, model, "test-key", undefined, undefined, { fetch: fetchMock }); + + expect(requestInput.some(item => item.type === "function_call" && item.call_id === "call_read_large")).toBe(true); + expect(requestInput.find(item => item.type === "function_call_output")?.output).toBe( + CONTEXT_WINDOW_TRUNCATED_OUTPUT_MESSAGE, + ); + expect(requestInput.at(-1)).toEqual({ type: "compaction_trigger" }); + }); + test("re-expands a prior V2 compaction's originals when no candidate can reuse the replay", async () => { vi.spyOn(ai, "completeSimple").mockResolvedValue(localSummaryMessage("re-expanded local summary")); const compactionItem = { type: "compaction", encrypted_content: "enc_v2" };