fix(omp): fit native compaction after large tool output
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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<Record<string, unknown>> | undefined,
|
||||
): unknown[] {
|
||||
): Array<Record<string, unknown>> {
|
||||
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<string, unknown>
|
||||
>;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -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 =>
|
||||
|
||||
@@ -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<string, unknown>;
|
||||
if (record.type === "input_image") {
|
||||
return {
|
||||
value: { ...record, image_url: "<image>" },
|
||||
imageTokens:
|
||||
record.detail === "original"
|
||||
? REMOTE_COMPACTION_ORIGINAL_IMAGE_TOKEN_ESTIMATE
|
||||
: REMOTE_COMPACTION_IMAGE_TOKEN_ESTIMATE,
|
||||
};
|
||||
}
|
||||
|
||||
const normalized: Record<string, unknown> = {};
|
||||
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<Record<string, unknown>>;
|
||||
rewrittenOutputs: number;
|
||||
estimatedTokensBefore: number;
|
||||
estimatedTokensAfter: number;
|
||||
}
|
||||
|
||||
function estimateRemoteCompactionInputTokens(
|
||||
input: Array<Record<string, unknown>>,
|
||||
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<string, unknown>): Record<string, unknown> | 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<Record<string, unknown>>,
|
||||
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<Record<string, unknown>> | 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<OpenAiRemoteCompactionResponse> {
|
||||
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";
|
||||
|
||||
@@ -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<Record<string, unknown>> | undefined;
|
||||
const fetchMock: FetchImpl = async (_input, init) => {
|
||||
const body = JSON.parse(String(init?.body)) as { input: Array<Record<string, unknown>> };
|
||||
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<Record<string, unknown>> = [];
|
||||
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" };
|
||||
|
||||
Reference in New Issue
Block a user