From c3012ea5208cb0940df83021d96b3fa91e9dfbb0 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 13 May 2026 21:52:21 +0800 Subject: [PATCH] Add Codex context compaction support (#990) --- .../app/src/components/agent-stream-view.tsx | 8 +- .../message-compaction-label.test.ts | 19 ++ .../components/message-compaction-label.ts | 17 ++ packages/app/src/components/message.tsx | 8 +- .../providers/codex-app-server-agent.test.ts | 193 ++++++++++++++++++ .../agent/providers/codex-app-server-agent.ts | 176 +++++++++++++++- 6 files changed, 413 insertions(+), 8 deletions(-) create mode 100644 packages/app/src/components/message-compaction-label.test.ts create mode 100644 packages/app/src/components/message-compaction-label.ts diff --git a/packages/app/src/components/agent-stream-view.tsx b/packages/app/src/components/agent-stream-view.tsx index 8f7dfa4a5..7bd100e84 100644 --- a/packages/app/src/components/agent-stream-view.tsx +++ b/packages/app/src/components/agent-stream-view.tsx @@ -540,7 +540,13 @@ const AgentStreamViewComponent = forwardRef; case "compaction": - return ; + return ( + + ); default: return null; diff --git a/packages/app/src/components/message-compaction-label.test.ts b/packages/app/src/components/message-compaction-label.test.ts new file mode 100644 index 000000000..670ab8541 --- /dev/null +++ b/packages/app/src/components/message-compaction-label.test.ts @@ -0,0 +1,19 @@ +import { describe, expect, it } from "vitest"; + +import { getCompactionMarkerLabel } from "./message-compaction-label"; + +describe("getCompactionMarkerLabel", () => { + it("renders loading, automatic, manual, tokenized, and fallback labels", () => { + expect(getCompactionMarkerLabel({ status: "loading" })).toBe("Compacting..."); + expect(getCompactionMarkerLabel({ status: "completed", trigger: "auto" })).toBe( + "Context automatically compacted", + ); + expect(getCompactionMarkerLabel({ status: "completed", trigger: "manual" })).toBe( + "Context manually compacted", + ); + expect(getCompactionMarkerLabel({ status: "completed", preTokens: 12_345 })).toBe( + "Context compacted (12K tokens)", + ); + expect(getCompactionMarkerLabel({ status: "completed" })).toBe("Context compacted"); + }); +}); diff --git a/packages/app/src/components/message-compaction-label.ts b/packages/app/src/components/message-compaction-label.ts new file mode 100644 index 000000000..052d1d690 --- /dev/null +++ b/packages/app/src/components/message-compaction-label.ts @@ -0,0 +1,17 @@ +export interface CompactionMarkerLabelInput { + status: "loading" | "completed"; + trigger?: "auto" | "manual"; + preTokens?: number; +} + +export function getCompactionMarkerLabel({ + status, + trigger, + preTokens, +}: CompactionMarkerLabelInput): string { + if (status === "loading") return "Compacting..."; + if (trigger === "auto") return "Context automatically compacted"; + if (trigger === "manual") return "Context manually compacted"; + if (preTokens) return `Context compacted (${Math.round(preTokens / 1000)}K tokens)`; + return "Context compacted"; +} diff --git a/packages/app/src/components/message.tsx b/packages/app/src/components/message.tsx index b53cb2f01..8864a778a 100644 --- a/packages/app/src/components/message.tsx +++ b/packages/app/src/components/message.tsx @@ -92,6 +92,7 @@ import { import { PlanCard } from "./plan-card"; import { useToolCallSheet } from "./tool-call-sheet"; import { ToolCallDetailsContent } from "./tool-call-details"; +import { getCompactionMarkerLabel } from "./message-compaction-label"; import { useAttachmentPreviewUrl } from "@/attachments/use-attachment-preview-url"; import { persistAttachmentFromBytes, persistAttachmentFromDataUrl } from "@/attachments/service"; import type { DaemonClient } from "@server/client/daemon-client"; @@ -1978,6 +1979,7 @@ export const ActivityLog = memo(function ActivityLog({ interface CompactionMarkerProps { status: "loading" | "completed"; + trigger?: "auto" | "manual"; preTokens?: number; } @@ -2008,12 +2010,10 @@ const compactionStylesheet = StyleSheet.create((theme) => ({ export const CompactionMarker = memo(function CompactionMarker({ status, + trigger, preTokens, }: CompactionMarkerProps) { - let label: string; - if (status === "loading") label = "Compacting..."; - else if (preTokens) label = `Context compacted (${Math.round(preTokens / 1000)}K tokens)`; - else label = "Context compacted"; + const label = getCompactionMarkerLabel({ status, trigger, preTokens }); return ( diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts index 7729df953..3308607db 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.test.ts @@ -1290,6 +1290,10 @@ describe("Codex app-server provider", () => { id: "message-history", text: "History loaded.", }, + { + type: "contextCompaction", + id: "compact-history", + }, ], }, ], @@ -1318,6 +1322,14 @@ describe("Codex app-server provider", () => { messageId: "message-history", }, }, + { + type: "timeline", + provider: "codex", + item: { + type: "compaction", + status: "completed", + }, + }, ]); }); @@ -1380,6 +1392,116 @@ describe("Codex app-server provider", () => { ]); }); + test("emits Codex context compaction markers from live thread items", () => { + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("item/started", { + threadId: "test-thread", + item: { + type: "contextCompaction", + id: "compact-live", + }, + }); + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "contextCompaction", + id: "compact-live", + }, + }); + + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "compaction", + status: "loading", + }, + }, + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "compaction", + status: "completed", + }, + }, + ]); + }); + + test("emits and dedupes Codex thread/compacted notifications", () => { + const session = createSession(); + session.activeForegroundTurnId = null; + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("thread/compacted", { + threadId: "test-thread", + turnId: "legacy-compact-turn", + }); + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "contextCompaction", + id: "legacy-compact-item", + }, + }); + + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + turnId: "legacy-compact-turn", + item: { + type: "compaction", + status: "completed", + }, + }, + ]); + }); + + test("emits consecutive Codex thread/compacted notifications", () => { + const session = createSession(); + session.activeForegroundTurnId = null; + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + + asInternals(session).handleNotification("thread/compacted", { + threadId: "test-thread", + turnId: "legacy-compact-turn-1", + }); + asInternals(session).handleNotification("thread/compacted", { + threadId: "test-thread", + turnId: "legacy-compact-turn-2", + }); + + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + turnId: "legacy-compact-turn-1", + item: { + type: "compaction", + status: "completed", + }, + }, + { + type: "timeline", + provider: "codex", + turnId: "legacy-compact-turn-2", + item: { + type: "compaction", + status: "completed", + }, + }, + ]); + }); + test("does not replace a persisted Codex thread when app-server resume fails", async () => { const session = createSession({ thinkingOptionId: "medium" }); session.currentThreadId = "archived-thread-id"; @@ -1450,6 +1572,77 @@ describe("Codex app-server provider", () => { ]); }); + test("lists /compact and sends Codex compaction out of band", async () => { + const requests: Array<{ method: string; params: unknown }> = []; + const session = createSession(); + session.client = { + request: vi.fn(async (method: string, params: unknown) => { + requests.push({ method, params }); + if (method === "thread/loaded/list") { + return { data: ["test-thread"] }; + } + if (method === "skills/list") { + return { data: [] }; + } + return {}; + }), + }; + + await expect(session.listCommands?.()).resolves.toContainEqual({ + name: "compact", + description: "Summarize conversation to prevent hitting the context limit", + argumentHint: "", + }); + + const handler = session.tryHandleOutOfBand?.("/compact"); + expect(handler).not.toBeNull(); + + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + await handler?.run({ emit: (event) => events.push(event) }); + asInternals(session).handleNotification("item/started", { + threadId: "test-thread", + item: { + type: "contextCompaction", + id: "manual-compact", + }, + }); + asInternals(session).handleNotification("item/completed", { + threadId: "test-thread", + item: { + type: "contextCompaction", + id: "manual-compact", + }, + }); + + expect(requests).toContainEqual({ + method: "thread/compact/start", + params: { threadId: "test-thread" }, + }); + expect(events).toEqual([ + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "compaction", + status: "loading", + trigger: "manual", + }, + }, + { + type: "timeline", + provider: "codex", + turnId: "test-turn", + item: { + type: "compaction", + status: "completed", + trigger: "manual", + }, + }, + ]); + }); + test("maps question responses from headers back to question ids and completes the tool call", async () => { const session = createSession(); const events: AgentStreamEvent[] = []; diff --git a/packages/server/src/server/agent/providers/codex-app-server-agent.ts b/packages/server/src/server/agent/providers/codex-app-server-agent.ts index 2fe97fa86..e5637c7c0 100644 --- a/packages/server/src/server/agent/providers/codex-app-server-agent.ts +++ b/packages/server/src/server/agent/providers/codex-app-server-agent.ts @@ -98,6 +98,7 @@ const CODEX_TOOL_THREAD_ITEM_TYPES = new Set([ "webSearch", "collabAgentToolCall", ]); +const CODEX_CONTEXT_COMPACTION_TYPE = "contextCompaction"; const CODEX_PLAN_IMPLEMENTATION_PROMPT_PREFIX = "The user approved the plan. Implement it now. Do not restate or revise the plan unless blocked."; @@ -1546,6 +1547,11 @@ function threadItemToTimeline( return mapCodexThreadPlanItem(normalizedItem); case "reasoning": return mapCodexThreadReasoningItem(normalizedItem); + case CODEX_CONTEXT_COMPACTION_TYPE: + return { + type: "compaction", + status: "completed", + }; default: return null; } @@ -1729,6 +1735,13 @@ const ItemLifecycleNotificationSchema = z }) .passthrough(); +const ContextCompactedNotificationSchema = z + .object({ + threadId: z.string(), + turnId: z.string().optional(), + }) + .passthrough(); + const CodexEventTurnAbortedNotificationSchema = z .object({ msg: z @@ -1958,6 +1971,7 @@ type ParsedCodexNotification = itemId: string; delta: string | null; } + | { kind: "context_compacted"; threadId: string; turnId: string | null } | { kind: "invalid_payload"; method: string; params: unknown } | { kind: "unknown_method"; method: string; params: unknown }; @@ -2056,6 +2070,22 @@ const CodexNotificationSchema = z.union([ params, }), ), + z + .object({ method: z.literal("thread/compacted"), params: ContextCompactedNotificationSchema }) + .transform( + ({ params }): ParsedCodexNotification => ({ + kind: "context_compacted", + threadId: params.threadId, + turnId: params.turnId ?? null, + }), + ), + z.object({ method: z.literal("thread/compacted"), params: z.unknown() }).transform( + ({ method, params }): ParsedCodexNotification => ({ + kind: "invalid_payload", + method, + params, + }), + ), z .object({ method: z.literal("item/agentMessage/delta"), @@ -2679,6 +2709,12 @@ class CodexAppServerAgentSession implements AgentSession { private warnedIncompleteEditToolCallIds = new Set(); private latestUsage: AgentUsage | undefined; private latestPlanResult: { callId: string; text: string; turnId: string | null } | null = null; + private pendingManualCompactionStarts = 0; + private compactionTriggerByItemId = new Map(); + // Codex can report one completed compaction through both channels: + // `thread/compacted` and a completed `contextCompaction` item. + private unpairedCompactionNotificationCompletions = 0; + private unpairedCompactionItemCompletions = 0; private connected = false; private collaborationModes: Array<{ name: string; @@ -3610,7 +3646,13 @@ class CodexAppServerAgentSession implements AgentSession { appServerSkills.length === 0 ? await listCodexSkills(this.config.cwd, this.deps.workspaceGitService) : []; - const builtin: AgentSlashCommand[] = []; + const builtin: AgentSlashCommand[] = [ + { + name: "compact", + description: "Summarize conversation to prevent hitting the context limit", + argumentHint: "", + }, + ]; if (this.goalsEnabled) { builtin.push({ name: "goal", @@ -3626,10 +3668,26 @@ class CodexAppServerAgentSession implements AgentSession { tryHandleOutOfBand( prompt: AgentPromptInput, ): { run(ctx: { emit: (event: AgentStreamEvent) => void }): Promise } | null { - if (!this.goalsEnabled) return null; if (typeof prompt !== "string") return null; const parsed = this.parseSlashCommandInput(prompt); - if (!parsed || parsed.commandName !== "goal") return null; + if (!parsed) return null; + + if (parsed.commandName === "compact") { + return { + run: async ({ emit }) => { + const error = await this.executeCompactCommand(); + if (error) { + emit({ + type: "timeline", + provider: CODEX_PROVIDER, + item: { type: "assistant_message", text: formatOutOfBandStatusMessage(error) }, + }); + } + }, + }; + } + + if (!this.goalsEnabled || parsed.commandName !== "goal") return null; const subcommand = parseGoalSubcommand(parsed.args); return { @@ -3644,6 +3702,33 @@ class CodexAppServerAgentSession implements AgentSession { }; } + private async executeCompactCommand(): Promise { + try { + await this.connect(); + if (this.currentThreadId) { + await this.ensureThreadLoaded(); + } else { + await this.ensureThread(); + } + if (!this.client || !this.currentThreadId) { + throw new Error("Codex thread is not available"); + } + this.pendingManualCompactionStarts += 1; + try { + await this.client.request("thread/compact/start", { + threadId: this.currentThreadId, + }); + } catch (error) { + this.pendingManualCompactionStarts = Math.max(0, this.pendingManualCompactionStarts - 1); + throw error; + } + return null; + } catch (error) { + const message = error instanceof Error ? error.message : "unknown error"; + return `Failed to compact context: ${message}`; + } + } + private async executeGoalSubcommand(subcommand: GoalSubcommand): Promise { if (subcommand.kind === "usage") { return "Usage: /goal |pause|resume|clear"; @@ -3875,6 +3960,9 @@ class CodexAppServerAgentSession implements AgentSession { case "token_usage_updated": this.handleTokenUsageUpdatedNotification(parsed); return; + case "context_compacted": + this.handleContextCompactedNotification(parsed); + return; case "agent_message_delta": case "reasoning_delta": case "exec_command_output_delta": @@ -4201,6 +4289,8 @@ class CodexAppServerAgentSession implements AgentSession { this.pendingFileChangeOutputDeltas.clear(); this.pendingAssistantMessageBoundary = false; this.warnedIncompleteEditToolCallIds.clear(); + this.unpairedCompactionNotificationCompletions = 0; + this.unpairedCompactionItemCompletions = 0; } private handlePlanUpdatedNotification( @@ -4240,6 +4330,65 @@ class CodexAppServerAgentSession implements AgentSession { } } + private resolveContextCompactionTrigger(itemId?: string): "auto" | "manual" | undefined { + if (itemId) { + const known = this.compactionTriggerByItemId.get(itemId); + if (known) { + return known; + } + } + if (this.pendingManualCompactionStarts > 0) { + this.pendingManualCompactionStarts -= 1; + return "manual"; + } + return undefined; + } + + private createContextCompactionTimelineItem( + status: "loading" | "completed", + itemId?: string, + ): Extract { + const trigger = this.resolveContextCompactionTrigger(itemId); + if (itemId && trigger) { + if (status === "loading") { + this.compactionTriggerByItemId.set(itemId, trigger); + } else { + this.compactionTriggerByItemId.delete(itemId); + } + } + return { + type: "compaction", + status, + ...(trigger ? { trigger } : {}), + }; + } + + private isContextCompactionItem(item: { type?: string; [key: string]: unknown }): boolean { + return ( + normalizeCodexThreadItemType(typeof item.type === "string" ? item.type : undefined) === + CODEX_CONTEXT_COMPACTION_TYPE + ); + } + + private handleContextCompactedNotification( + parsed: Extract, + ): void { + if (parsed.threadId !== this.currentThreadId) { + return; + } + if (this.unpairedCompactionItemCompletions > 0) { + this.unpairedCompactionItemCompletions -= 1; + return; + } + this.unpairedCompactionNotificationCompletions += 1; + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: this.createContextCompactionTimelineItem("completed"), + ...(parsed.turnId ? { turnId: parsed.turnId } : {}), + }); + } + private handleExecCommandStartedNotification( parsed: Extract, ): void { @@ -4357,6 +4506,19 @@ class CodexAppServerAgentSession implements AgentSession { if (parsed.source === "codex_event") { return; } + if (this.isContextCompactionItem(parsed.item)) { + if (this.unpairedCompactionNotificationCompletions > 0) { + this.unpairedCompactionNotificationCompletions -= 1; + return; + } + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: this.createContextCompactionTimelineItem("completed", parsed.item.id), + }); + this.unpairedCompactionItemCompletions += 1; + return; + } const timelineItem = threadItemToTimeline(parsed.item, { includeUserMessage: false, cwd: this.config.cwd ?? null, @@ -4487,6 +4649,14 @@ class CodexAppServerAgentSession implements AgentSession { if (parsed.source === "codex_event") { return; } + if (this.isContextCompactionItem(parsed.item)) { + this.emitEvent({ + type: "timeline", + provider: CODEX_PROVIDER, + item: this.createContextCompactionTimelineItem("loading", parsed.item.id), + }); + return; + } const timelineItem = threadItemToTimeline(parsed.item, { includeUserMessage: false, cwd: this.config.cwd ?? null,