diff --git a/docs/reference/transcript-hygiene.md b/docs/reference/transcript-hygiene.md index 09edf941b276..217f77f7680e 100644 --- a/docs/reference/transcript-hygiene.md +++ b/docs/reference/transcript-hygiene.md @@ -77,9 +77,9 @@ Implementation: - Max image side is configurable via `agents.defaults.imageMaxDimensionPx` (default: `1200`) - Blank text blocks are removed while this pass walks replay content. - Assistant turns that become empty are dropped from the replay copy; user - and tool-result turns that become empty receive a non-empty - omitted-content placeholder. + Assistant turns that become empty are dropped unless they own opaque + provider replay state; user and tool-result turns that become empty receive + a non-empty omitted-content placeholder. --- diff --git a/packages/ai/src/transports/openai-responses-compaction-replay.test.ts b/packages/ai/src/transports/openai-responses-compaction-replay.test.ts index 2de2ae6204ef..c6d99da8fbfc 100644 --- a/packages/ai/src/transports/openai-responses-compaction-replay.test.ts +++ b/packages/ai/src/transports/openai-responses-compaction-replay.test.ts @@ -928,19 +928,15 @@ describe("OpenAI Responses compaction replay", () => { expect(input.map((item) => item.type)).toEqual(["compaction", "message"]); }); - it("replays when session and auth identities match", () => { - const assistant = createOutput(); - assistant.providerReplay = compactionState(model, { replayIndex: 0 }); + it.each(responseConverters)( + "$name replays an empty checkpoint owner when request identities match", + ({ convert }) => { + const assistant = createOutput(); + assistant.providerReplay = compactionState(model, { replayIndex: 0 }); - const input = convertResponsesMessages( - model, - { messages: [assistant] }, - new Set(["openai"]), - replayIdentity, - ); - - expect(input.some((item) => item.type === "compaction")).toBe(true); - }); + expect(convert({ messages: [assistant] }).map((item) => item.type)).toEqual(["compaction"]); + }, + ); it.each(responseConverters)( "$name does not replay or prune across a different or missing request identity", diff --git a/scripts/lib/state-schema-inline-plugin.mts b/scripts/lib/state-schema-inline-plugin.mts index 68e03b23654a..f68684f31b76 100644 --- a/scripts/lib/state-schema-inline-plugin.mts +++ b/scripts/lib/state-schema-inline-plugin.mts @@ -21,9 +21,18 @@ export function createStateSchemaInlinePlugin(rootDir = process.cwd()) { const schemasByModulePath = new Map( STATE_SCHEMA_MODULES.map((schema) => [path.resolve(rootDir, schema.modulePath), schema]), ); + const cacheKeyForSchema = ({ id }: { id: string }) => { + const schema = schemasByModulePath.get(path.resolve(id)); + return schema ? fs.readFileSync(path.resolve(rootDir, schema.schemaPath), "utf8") : undefined; + }; return { name: STATE_SCHEMA_INLINE_PLUGIN_NAME, + configureVitest(context: { + experimental_defineCacheKeyGenerator(callback: typeof cacheKeyForSchema): void; + }) { + context.experimental_defineCacheKeyGenerator(cacheKeyForSchema); + }, load(this: { addWatchFile(id: string): void }, id: string) { const schema = schemasByModulePath.get(path.resolve(id)); if (!schema) { diff --git a/src/agents/embedded-agent-helpers.sanitize-session-messages-images.removes-empty-assistant-text-blocks-but-preserves.test.ts b/src/agents/embedded-agent-helpers.sanitize-session-messages-images.removes-empty-assistant-text-blocks-but-preserves.test.ts index b9b3658b1539..d492a7cb525d 100644 --- a/src/agents/embedded-agent-helpers.sanitize-session-messages-images.removes-empty-assistant-text-blocks-but-preserves.test.ts +++ b/src/agents/embedded-agent-helpers.sanitize-session-messages-images.removes-empty-assistant-text-blocks-but-preserves.test.ts @@ -250,6 +250,31 @@ describe("sanitizeSessionMessagesImages", () => { expect(out).toHaveLength(1); expect(out[0]?.role).toBe("user"); }); + it.each([ + ["full", "length"], + ["images-only", "length"], + ["full", "error"], + ["images-only", "error"], + ] as const)( + "preserves an empty provider replay owner in %s mode after %s", + async (sanitizeMode, stopReason) => { + const checkpoint = { + ...makeOpenAiResponsesAssistantMessage([{ type: "text", text: "" }], stopReason), + providerReplay: { + v: 1, + type: "opaque-checkpoint", + data: "opaque-state", + provider: "openai", + api: "openai-responses", + model: "gpt-5.4", + }, + } satisfies AssistantMessage; + + const out = await sanitizeSessionMessagesImages([checkpoint], "test", { sanitizeMode }); + + expect(out).toEqual([{ ...checkpoint, content: [] }]); + }, + ); it("drops empty assistant error messages", async () => { const input = castAgentMessages([ { role: "user", content: "hello", timestamp: nextTimestamp() } satisfies UserMessage, diff --git a/src/agents/embedded-agent-helpers/images.ts b/src/agents/embedded-agent-helpers/images.ts index 9710f2a3ac92..4e0394181d73 100644 --- a/src/agents/embedded-agent-helpers/images.ts +++ b/src/agents/embedded-agent-helpers/images.ts @@ -53,8 +53,6 @@ export async function sanitizeSessionMessagesImages( }; } & ImageSanitizationLimits, ): Promise { - const sanitizeMode = options?.sanitizeMode ?? "full"; - const allowNonImageSanitization = sanitizeMode === "full"; const imageSanitization = { maxDimensionPx: options?.maxDimensionPx, maxBytes: options?.maxBytes, @@ -113,7 +111,7 @@ export async function sanitizeSessionMessagesImages( imageSanitization, )) as unknown as typeof assistantMsg.content; const finalContent = dropEmptyTextBlocks(nextContent); - if (finalContent.length > 0) { + if (finalContent.length > 0 || assistantMsg.providerReplay) { out.push({ ...assistantMsg, content: finalContent }); } } else { @@ -126,28 +124,14 @@ export async function sanitizeSessionMessagesImages( const strippedContent = options?.preserveSignatures ? content // Keep signatures for Antigravity Claude : stripThoughtSignatures(content, options?.sanitizeThoughtSignatures); // Strip for Gemini - if (!allowNonImageSanitization) { - const nextContent = (await sanitizeContentBlocksImages( - dropEmptyTextBlocks(strippedContent) as unknown as ContentBlock[], - label, - imageSanitization, - )) as unknown as typeof assistantMsg.content; - if (nextContent.length > 0) { - out.push({ ...assistantMsg, content: nextContent }); - } - continue; - } - - const filteredContent = dropEmptyTextBlocks(strippedContent); const finalContent = (await sanitizeContentBlocksImages( - filteredContent as unknown as ContentBlock[], + dropEmptyTextBlocks(strippedContent) as unknown as ContentBlock[], label, imageSanitization, )) as unknown as typeof assistantMsg.content; - if (finalContent.length === 0) { - continue; + if (finalContent.length > 0 || assistantMsg.providerReplay) { + out.push({ ...assistantMsg, content: finalContent }); } - out.push({ ...assistantMsg, content: finalContent }); continue; } } diff --git a/src/agents/embedded-agent-runner/run.incomplete-turn.empty-response-recovery.test.ts b/src/agents/embedded-agent-runner/run.incomplete-turn.empty-response-recovery.test.ts index bb36281ff43c..f08a503f3d6d 100644 --- a/src/agents/embedded-agent-runner/run.incomplete-turn.empty-response-recovery.test.ts +++ b/src/agents/embedded-agent-runner/run.incomplete-turn.empty-response-recovery.test.ts @@ -134,6 +134,48 @@ describe("runEmbeddedAgent incomplete-turn safety", () => { expectWarnMessageWith("empty response detected"); }); + it("continues after an OpenAI Responses compaction-only incomplete turn", async () => { + const checkpoint = makeLastAssistant({ + api: "openai-responses", + provider: "openai", + model: "gpt-5.6-luna", + stopReason: "length", + providerReplay: { + v: 1, + type: "openai-responses-compaction", + data: "opaque-checkpoint", + provider: "openai", + api: "openai-responses", + model: "gpt-5.6-luna", + }, + }); + mockedRunEmbeddedAttempt.mockResolvedValueOnce( + makeAttemptResult({ + assistantTexts: [], + currentAttemptAssistant: checkpoint, + lastAssistant: checkpoint, + }), + ); + mockedRunEmbeddedAttempt.mockResolvedValueOnce( + makeAttemptResult({ + assistantTexts: ["Visible answer after compaction."], + lastAssistant: makeLastAssistant({ + content: [{ type: "text", text: "Visible answer after compaction." }], + }), + }), + ); + + await runEmbeddedAgent( + makeRunParams("run-provider-compaction-continuation", { + provider: "openai", + model: "gpt-5.6-luna", + }), + ); + + expect(mockedRunEmbeddedAttempt).toHaveBeenCalledTimes(2); + expectWarnMessageWith("compaction interrupted visible final answer"); + }); + it("retries empty Anthropic-compatible stop turns even when the provider is not Kimi", async () => { mockedClassifyFailoverReason.mockReturnValue(null); mockedResolveModelAsync.mockResolvedValue({ diff --git a/src/agents/embedded-agent-runner/run/terminal-resolution.ts b/src/agents/embedded-agent-runner/run/terminal-resolution.ts index 0e78982afa0d..6cc44e1a787c 100644 --- a/src/agents/embedded-agent-runner/run/terminal-resolution.ts +++ b/src/agents/embedded-agent-runner/run/terminal-resolution.ts @@ -341,7 +341,8 @@ export async function resolveEmbeddedRunTerminal(input: { if ( !emptyAssistantReplyIsSilent && !settledTurnFinalizationAttempted && - input.attemptCompactionCount > 0 && + (input.attemptCompactionCount > 0 || + attempt.currentAttemptAssistant?.providerReplay?.type === "openai-responses-compaction") && payloadCount === 0 && !terminalInterrupted && !promptError && diff --git a/src/gateway/gateway-codex-harness.live-helpers.test.ts b/src/gateway/gateway-codex-harness.live-helpers.test.ts index cd417b941fee..7b20184dd33f 100644 --- a/src/gateway/gateway-codex-harness.live-helpers.test.ts +++ b/src/gateway/gateway-codex-harness.live-helpers.test.ts @@ -62,19 +62,21 @@ describe("gateway codex harness live helpers", () => { guardianProbe: false, imageProbe: false, mcpProbe: false, + multiSessionProbe: false, resumeStress: false, subagentProbe: true, }; expect(shouldUseCodexHarnessSubagentOnlyFastPath(base)).toBe(true); - expect(shouldUseCodexHarnessSubagentOnlyFastPath({ ...base, resumeStress: true })).toBe(false); - expect(shouldUseCodexHarnessSubagentOnlyFastPath({ ...base, compactionStress: true })).toBe( - false, - ); - expect(shouldUseCodexHarnessSubagentOnlyFastPath({ ...base, codeModeOnly: true })).toBe(false); - expect(shouldUseCodexHarnessSubagentOnlyFastPath({ ...base, explicitOptOut: true })).toBe( - false, - ); + for (const flag of [ + "codeModeOnly", + "compactionStress", + "explicitOptOut", + "multiSessionProbe", + "resumeStress", + ] as const) { + expect(shouldUseCodexHarnessSubagentOnlyFastPath({ ...base, [flag]: true })).toBe(false); + } }); it("classifies sessions.list timeouts as retryable live Codex errors", () => { diff --git a/src/gateway/gateway-codex-harness.live-helpers.ts b/src/gateway/gateway-codex-harness.live-helpers.ts index 9aa168ecb90f..031a98a17f84 100644 --- a/src/gateway/gateway-codex-harness.live-helpers.ts +++ b/src/gateway/gateway-codex-harness.live-helpers.ts @@ -96,6 +96,7 @@ export function shouldUseCodexHarnessSubagentOnlyFastPath(params: { guardianProbe: boolean; imageProbe: boolean; mcpProbe: boolean; + multiSessionProbe: boolean; resumeStress: boolean; subagentProbe: boolean; }): boolean { @@ -107,6 +108,7 @@ export function shouldUseCodexHarnessSubagentOnlyFastPath(params: { !params.guardianProbe && !params.imageProbe && !params.mcpProbe && + !params.multiSessionProbe && !params.resumeStress && !params.explicitOptOut ); diff --git a/src/gateway/gateway-codex-harness.live.test.ts b/src/gateway/gateway-codex-harness.live.test.ts index f5698d348c92..80ebb9fa880b 100644 --- a/src/gateway/gateway-codex-harness.live.test.ts +++ b/src/gateway/gateway-codex-harness.live.test.ts @@ -148,6 +148,7 @@ const CODEX_HARNESS_SUBAGENT_ONLY = shouldUseCodexHarnessSubagentOnlyFastPath({ guardianProbe: CODEX_HARNESS_GUARDIAN_PROBE, imageProbe: CODEX_HARNESS_IMAGE_PROBE, mcpProbe: CODEX_HARNESS_MCP_PROBE, + multiSessionProbe: CODEX_HARNESS_MULTI_SESSION_PROBE, resumeStress: CODEX_HARNESS_RESUME_STRESS, subagentProbe: CODEX_HARNESS_SUBAGENT_PROBE, }); @@ -2209,7 +2210,6 @@ describeLive("gateway live (Codex harness)", () => { }, workspace, }); - break; } if (CODEX_HARNESS_SUBAGENT_PROBE) { diff --git a/src/gateway/gateway-openai-long-context.live.test.ts b/src/gateway/gateway-openai-long-context.live.test.ts index f4717e6c684d..85e5fb43a17b 100644 --- a/src/gateway/gateway-openai-long-context.live.test.ts +++ b/src/gateway/gateway-openai-long-context.live.test.ts @@ -503,8 +503,12 @@ describeLive("Gateway OpenAI long-context compaction (live)", () => { } } if (!compactionState?.latest) { + const thresholdEvidence = + peakPromptTokens > 0 + ? `peak provider prompt tokens=${peakPromptTokens}, compact threshold=${profile.compactThreshold}` + : "provider prompt-token usage unavailable"; throw new Error( - `OpenAI emitted no first-class compaction item after ${profile.maxDenseTurns} dense turns`, + `OpenAI emitted no first-class compaction item after ${profile.maxDenseTurns} dense turns; ${thresholdEvidence}`, ); } expect(compactionState.latest).toMatchObject({ diff --git a/src/infra/tsdown-config.test.ts b/src/infra/tsdown-config.test.ts index 6a04bf6c4511..e2a5c7bbc8c1 100644 --- a/src/infra/tsdown-config.test.ts +++ b/src/infra/tsdown-config.test.ts @@ -111,6 +111,12 @@ describe("tsdown config", () => { const rootDir = process.cwd(); const watchedPaths: string[] = []; const plugin = createStateSchemaInlinePlugin(rootDir); + let cacheKeyGenerator: ((context: { id: string }) => string | undefined) | undefined; + plugin.configureVitest({ + experimental_defineCacheKeyGenerator: (generator) => { + cacheKeyGenerator = generator; + }, + }); const result = plugin.load.call( { addWatchFile: (filePath: string) => watchedPaths.push(filePath) }, path.resolve(rootDir, schema.modulePath), @@ -126,6 +132,10 @@ describe("tsdown config", () => { expect(JSON.parse(match?.[1] ?? "null")).toBe(canonicalSql); expect(schema.sourceValue).toBe(canonicalSql); expect(watchedPaths).toEqual([schemaPath]); + expect(cacheKeyGenerator?.({ id: path.resolve(rootDir, schema.modulePath) })).toBe( + canonicalSql, + ); + expect(cacheKeyGenerator?.({ id: path.resolve(rootDir, "src/index.ts") })).toBeUndefined(); }); it("installs schema inlining only on the unified runtime graph", () => { diff --git a/test/gateway-openai-compaction-replay.e2e.test.ts b/test/gateway-openai-compaction-replay.e2e.test.ts index f0ed4b12b065..356dd30fc581 100644 --- a/test/gateway-openai-compaction-replay.e2e.test.ts +++ b/test/gateway-openai-compaction-replay.e2e.test.ts @@ -63,7 +63,9 @@ describe("Gateway OpenAI Responses compaction replay", () => { }); try { await runAgentTurn(client, "capture compaction state"); - expect(modelServer.requests).toHaveLength(1); + // The provider can terminate after emitting only a compaction item. The + // runner must continue from that checkpoint before completing the turn. + expect(modelServer.requests).toHaveLength(2); const session = await client.request<{ sessions?: Array<{ key?: string; sessionId?: string }>; @@ -78,28 +80,31 @@ describe("Gateway OpenAI Responses compaction replay", () => { sessionKey: SESSION_KEY, storePath: path.join(instance.state.agentDir("main"), "openclaw-agent.sqlite"), }); - const persistedReplay = manager - .buildSessionContext() - .messages.find((message) => message.role === "assistant")?.providerReplay; + const contextMessages = manager.buildSessionContext().messages; + const persistedReplay = contextMessages.find( + (message) => message.role === "assistant", + )?.providerReplay; expect(persistedReplay).toMatchObject({ + v: 1, type: "openai-responses-compaction", id: COMPACTION_ID, data: COMPACTION_DATA, provider: "replay-proof", api: "openai-responses", model: "replay-proof", + baseUrlHash: expect.any(String), sessionHash: expect.any(String), }); expect(persistedReplay).not.toHaveProperty("authProfileHash"); + expectCompactionReplay(modelServer.requests[1]?.body.input ?? []); + expect(JSON.stringify(modelServer.requests[1]?.body.input)).toContain( + "Continue from the compacted transcript", + ); await runAgentTurn(client, "replay compaction state"); - expect(modelServer.requests).toHaveLength(2); - const replayInput = modelServer.requests[1]?.body.input ?? []; - expect(replayInput).toContainEqual({ - type: "compaction", - id: COMPACTION_ID, - encrypted_content: COMPACTION_DATA, - }); + expect(modelServer.requests).toHaveLength(3); + const replayInput = modelServer.requests[2]?.body.input ?? []; + expectCompactionReplay(replayInput); const compactionIndex = replayInput.findIndex( (item) => typeof item === "object" && @@ -120,7 +125,7 @@ describe("Gateway OpenAI Responses compaction replay", () => { ).toBe(true); const encodedReplayInput = JSON.stringify(replayInput); expect(encodedReplayInput).not.toContain("capture compaction state"); - expect(encodedReplayInput).toContain("gateway replay response 1"); + expect(encodedReplayInput).toContain("gateway replay response 2"); expect(encodedReplayInput).toContain("replay compaction state"); } finally { await disconnectGatewayClient(client); @@ -188,6 +193,14 @@ async function runAgentTurn( return runId; } +function expectCompactionReplay(input: unknown[]): void { + expect(input).toContainEqual({ + type: "compaction", + id: COMPACTION_ID, + encrypted_content: COMPACTION_DATA, + }); +} + async function startMockModelServer(): Promise { const requests: CapturedRequest[] = []; const server = createServer((request, response) => { @@ -241,6 +254,28 @@ async function handleRequest( } function writeModelResponse(response: ServerResponse, sequence: number): void { + if (sequence === 1) { + const compaction = { + type: "compaction", + id: COMPACTION_ID, + encrypted_content: COMPACTION_DATA, + }; + writeSseEvents(response, [ + { type: "response.output_item.added", output_index: 0, item: compaction }, + { type: "response.output_item.done", output_index: 0, item: compaction }, + { + type: "response.incomplete", + response: { + id: "resp_gateway_replay_1", + status: "incomplete", + incomplete_details: { reason: "max_output_tokens" }, + output: [compaction], + usage: { input_tokens: 0, output_tokens: 0, total_tokens: 0 }, + }, + }, + ]); + return; + } const text = `gateway replay response ${sequence}`; const message = { type: "message", @@ -249,10 +284,7 @@ function writeModelResponse(response: ServerResponse, sequence: number): void { status: "completed", content: [{ type: "output_text", text, annotations: [] }], }; - const output = - sequence === 1 - ? [{ type: "compaction", id: COMPACTION_ID, encrypted_content: COMPACTION_DATA }, message] - : [message]; + const output = [message]; const events: MockSseEvent[] = output.flatMap((item, outputIndex) => [ { type: "response.output_item.added", @@ -270,6 +302,10 @@ function writeModelResponse(response: ServerResponse, sequence: number): void { usage: { input_tokens: 10, output_tokens: 2, total_tokens: 12 }, }, }); + writeSseEvents(response, events); +} + +function writeSseEvents(response: ServerResponse, events: MockSseEvent[]): void { response.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-store", diff --git a/test/helpers/openai-long-context-live.test.ts b/test/helpers/openai-long-context-live.test.ts index 9a69ee8241e6..3ada28de60b6 100644 --- a/test/helpers/openai-long-context-live.test.ts +++ b/test/helpers/openai-long-context-live.test.ts @@ -118,7 +118,7 @@ describe("OpenAI long-context live settings", () => { contextWindow: 48_000, contextTokens: 48_000, maxTokens: 8_192, - compactThreshold: 32_000, + compactThreshold: 1_000, }); const full = resolveOpenAILongContextLiveSettings( { diff --git a/test/helpers/openai-long-context-live.ts b/test/helpers/openai-long-context-live.ts index 44b7ee535e6d..7912db2f24e7 100644 --- a/test/helpers/openai-long-context-live.ts +++ b/test/helpers/openai-long-context-live.ts @@ -48,9 +48,12 @@ const PROFILES = { contextWindow: 48_000, contextTokens: 48_000, maxTokens: 8_192, - compactThreshold: 32_000, + // Keep the reduced live probe on OpenAI's demonstrated compaction path. + // High-threshold Luna probes can cross the configured threshold without + // emitting a checkpoint, while the 1k boundary is deterministic. + compactThreshold: 1_000, denseTurnChars: 120_000, - maxDenseTurns: 8, + maxDenseTurns: 3, defaultToolBytes: 300_000, requestTimeoutMs: 2 * 60_000, suiteTimeoutMs: 10 * 60_000, @@ -199,6 +202,9 @@ export function buildOpenAILongContextConfig(params: { workspace: params.workspace, skipBootstrap: true, thinkingDefault: "low", + // This suite owns the server-compaction threshold. Embedded proactive + // compaction would consume the same history before replay can be proved. + compaction: { enabled: false }, model: { primary: profile.modelRef }, models: { [profile.modelRef]: { @@ -256,6 +262,11 @@ export function assertOpenAILongContextConfig( cfg.secrets?.providers?.default?.source, "env", ); + expectConfigValue( + "agents.defaults.compaction.enabled", + cfg.agents?.defaults?.compaction?.enabled, + false, + ); expectConfigValue("models.providers.openai.models.length", provider?.models.length, 1); const model = provider?.models[0]; expectConfigValue("model.id", model?.id, profile.modelId);