diff --git a/cli/src/codex/codexRemoteLauncher.test.ts b/cli/src/codex/codexRemoteLauncher.test.ts index 9629d3ca..29b80e3b 100644 --- a/cli/src/codex/codexRemoteLauncher.test.ts +++ b/cli/src/codex/codexRemoteLauncher.test.ts @@ -3439,11 +3439,11 @@ describe('codexRemoteLauncher', () => { expect(harness.compactThreadIds).toEqual(['thread-1']); expect(sessionEvents).toContainEqual({ type: 'message', - message: 'Compaction started' + message: '📦 Compaction started' }); expect(sessionEvents).toContainEqual({ type: 'message', - message: 'Compaction completed' + message: '📦 Compaction completed' }); }); @@ -3463,7 +3463,7 @@ describe('codexRemoteLauncher', () => { expect(harness.startTurnMessages).toEqual(['first message']); expect(sessionEvents).not.toContainEqual({ type: 'message', - message: 'Compaction completed' + message: '📦 Compaction completed' }); harness.dispatchNotification?.('item/completed', { @@ -3482,7 +3482,7 @@ describe('codexRemoteLauncher', () => { expect(harness.startTurnMessages).toEqual(['first message', 'after compact']); expect(sessionEvents).toContainEqual({ type: 'message', - message: 'Compaction completed' + message: '📦 Compaction completed' }); }); @@ -3499,7 +3499,7 @@ describe('codexRemoteLauncher', () => { expect(harness.compactThreadIds).toEqual(['thread-1']); expect(sessionEvents).toContainEqual({ type: 'message', - message: 'Compaction completed' + message: '📦 Compaction completed' }); expect(session.thinking).toBe(false); }); @@ -3533,4 +3533,220 @@ describe('codexRemoteLauncher', () => { message: '/compact does not accept arguments' }); }); + + it('announces automatic compaction when it starts, before it completes', async () => { + harness.suppressTurnCompletion = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(harness.startTurnMessages).toEqual(['first message']); + }); + + harness.dispatchNotification?.('item/started', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }); + + await vi.waitFor(() => { + expect(sessionEvents).toContainEqual({ + type: 'message', + message: '📦 Compaction started' + }); + }); + + harness.dispatchNotification?.('item/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }); + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turn: { id: 'turn-1', status: 'completed' } + }); + + await expect(running).resolves.toBe('exit'); + const startCount = sessionEvents.filter( + (event) => event.type === 'message' && event.message === '📦 Compaction started' + ).length; + expect(startCount).toBe(1); + }); + + it('dedups duplicate automatic compaction starts and keeps foreign-thread starts out of the parent stream', async () => { + harness.suppressTurnCompletion = true; + const { session, sessionEvents, codexMessages } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(harness.startTurnMessages).toEqual(['first message']); + }); + + const startParams = { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }; + harness.dispatchNotification?.('item/started', startParams); + harness.dispatchNotification?.('item/started', startParams); + + await vi.waitFor(() => { + expect(sessionEvents).toContainEqual({ + type: 'message', + message: '📦 Compaction started' + }); + }); + + harness.dispatchNotification?.('item/started', { + threadId: 'foreign-thread', + turnId: 'foreign-turn', + item: { id: 'compact-item-foreign', type: 'contextCompaction' } + }); + + harness.dispatchNotification?.('item/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }); + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turn: { id: 'turn-1', status: 'completed' } + }); + + await expect(running).resolves.toBe('exit'); + const startCount = sessionEvents.filter( + (event) => event.type === 'message' && event.message === '📦 Compaction started' + ).length; + expect(startCount).toBe(1); + expect(codexMessages).toContainEqual(expect.objectContaining({ + type: 'agent-run-trace' + })); + }); + + it('announces successive compactions in one turn without suppression', async () => { + harness.suppressTurnCompletion = true; + const { session, sessionEvents } = createSessionStub(['first message']); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(harness.startTurnMessages).toEqual(['first message']); + }); + + harness.dispatchNotification?.('item/started', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }); + harness.dispatchNotification?.('item/started', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-2', type: 'contextCompaction' } + }); + + await vi.waitFor(() => { + const startCount = sessionEvents.filter( + (event) => event.type === 'message' && event.message === '📦 Compaction started' + ).length; + expect(startCount).toBe(2); + }); + + harness.dispatchNotification?.('item/completed', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-2', type: 'contextCompaction' } + }); + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turn: { id: 'turn-1', status: 'completed' } + }); + + await expect(running).resolves.toBe('exit'); + }); + + it('suppresses automatic compaction start notification while manual /compact is in flight', async () => { + harness.suppressTurnCompletion = true; + harness.deferCompactCompletion = true; + const { session, sessionEvents, rpcHandlers } = createSessionStub( + ['first message', '/compact'], + createMode(), + false, + false + ); + + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { + expect(harness.compactThreadIds).toEqual(['thread-1']); + }); + await vi.waitFor(() => { + expect(sessionEvents).toContainEqual({ + type: 'message', + message: '📦 Compaction started' + }); + }); + const startCountBefore = sessionEvents.filter( + (event) => event.type === 'message' && event.message === '📦 Compaction started' + ).length; + + harness.dispatchNotification?.('item/started', { + threadId: 'thread-1', + turnId: 'compact-turn', + item: { id: 'compact-item-manual', type: 'contextCompaction' } + }); + + harness.dispatchNotification?.('item/completed', { + threadId: 'thread-1', + turnId: 'compact-turn', + item: { id: 'compact-item-manual', type: 'contextCompaction' } + }); + harness.dispatchNotification?.('turn/completed', { + threadId: 'thread-1', + turn: { id: 'compact-turn', status: 'completed' } + }); + + await rpcHandlers.get('switch')?.({}); + await expect(running).resolves.toBe('switch'); + const startCount = sessionEvents.filter( + (event) => event.type === 'message' && event.message === '📦 Compaction started' + ).length; + expect(startCount).toBe(startCountBefore); + }); + + it('binds deferred manual compact to its actual turn, ignoring a stale compaction start', async () => { + harness.suppressTurnCompletion = true; + harness.deferCompactCompletion = true; + const { session, sessionEvents } = createSessionStub(['first message', '/compact'], createMode(), true); + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { expect(harness.compactThreadIds).toEqual(['thread-1']); }); + harness.dispatchNotification?.('item/started', { threadId: 'thread-1', turnId: 'stale-turn', item: { id: 'stale-item', type: 'contextCompaction' } }); + harness.dispatchNotification?.('turn/started', { threadId: 'thread-1', turn: { id: 'compact-actual' } }); + harness.dispatchNotification?.('item/completed', { threadId: 'thread-1', turnId: 'compact-actual', item: { id: 'compact-item-1', type: 'contextCompaction' } }); + harness.dispatchNotification?.('turn/completed', { threadId: 'thread-1', turn: { id: 'compact-actual', status: 'completed' } }); + const exitReason = await running; + expect(exitReason).toBe('exit'); + expect(sessionEvents).toContainEqual({ type: 'message', message: '📦 Compaction completed' }); + }); + + it('drops unscoped compaction starts while child agents are active', async () => { + harness.suppressTurnCompletion = true; + const { session, sessionEvents } = createSessionStub(['first message']); + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { expect(harness.startTurnMessages).toEqual(['first message']); }); + harness.dispatchNotification?.('item/started', { item: { id: 'call-spawn', type: 'collabAgentToolCall', tool: 'spawnAgent', prompt: 'x', receiverThreadIds: [] } }); + harness.dispatchNotification?.('item/started', { item: { id: 'compact-item-1', type: 'contextCompaction' } }); + harness.dispatchNotification?.('turn/completed', { threadId: 'thread-1', turn: { id: 'turn-1', status: 'completed' } }); + await expect(running).resolves.toBe('exit'); + expect(sessionEvents.filter((e) => e.type === 'message' && e.message === '📦 Compaction started')).toHaveLength(0); + }); + + it('emits child compaction trace with exact agent id, type and statusText', async () => { + harness.suppressTurnCompletion = true; + const { session, codexMessages } = createSessionStub(['first message']); + const running = codexRemoteLauncher(session as never); + await vi.waitFor(() => { expect(harness.startTurnMessages).toEqual(['first message']); }); + harness.dispatchNotification?.('item/started', { threadId: 'child-1', turnId: 'cturn-1', item: { id: 'compact-child-1', type: 'contextCompaction' } }); + harness.dispatchNotification?.('turn/completed', { threadId: 'thread-1', turn: { id: 'turn-1', status: 'completed' } }); + await expect(running).resolves.toBe('exit'); + const traces = codexMessages.filter((m) => (m as Record).type === 'agent-run-trace'); + expect(traces).toContainEqual(expect.objectContaining({ agentId: 'child-1', message: expect.objectContaining({ type: 'compaction_started', statusText: '📦 Compaction started' }) })); + }); }); diff --git a/cli/src/codex/codexRemoteLauncher.ts b/cli/src/codex/codexRemoteLauncher.ts index 299f8686..b34899ba 100644 --- a/cli/src/codex/codexRemoteLauncher.ts +++ b/cli/src/codex/codexRemoteLauncher.ts @@ -750,6 +750,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { const isScopeSensitiveCodexEvent = (type: string): boolean => { return type === 'token_count' || type === 'context_compacted' + || type === 'compaction_started' || type === 'thread_goal_updated' || type === 'thread_goal_cleared'; }; @@ -1519,6 +1520,16 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return; } + if (msgType === 'compaction_started') { + emitAgentRunTraceMessage(agentId, { + type: 'compaction_started', + statusText: '📦 Compaction started', + id: randomUUID() + }); + updateActivity('Compacting context', 'compact'); + return; + } + if (msgType === 'task_started') { runtime.reasoningPreview = ''; runtime.finalMessage = null; @@ -3255,6 +3266,17 @@ class CodexRemoteLauncher extends RemoteLauncherBase { id: randomUUID() }); } + if (msgType === 'compaction_started') { + const threadId = eventThreadId ?? this.currentThreadId; + const manualInFlight = manualCompact + && (!threadId || manualCompact.threadId === threadId) + && (!manualCompact.turnId || !eventTurnId || manualCompact.turnId === eventTurnId); + if (!manualInFlight) { + const message = '📦 Compaction started'; + messageBuffer.addMessage(message, 'status'); + session.sendSessionEvent({ type: 'message', message }); + } + } if (msgType === 'plan_update') { session.sendAgentMessage({ type: 'tool-call', @@ -4022,7 +4044,7 @@ class CodexRemoteLauncher extends RemoteLauncherBase { return true; } - sendVisibleStatus('Compaction started'); + sendVisibleStatus('📦 Compaction started'); const compactCompletion = beginManualCompact(threadId); void compactCompletion.catch(() => {}); try { @@ -4030,10 +4052,10 @@ class CodexRemoteLauncher extends RemoteLauncherBase { signal: this.abortController.signal }); await compactCompletion; - sendVisibleStatus('Compaction completed'); + sendVisibleStatus('📦 Compaction completed'); } catch (error) { const detail = error instanceof Error ? error.message : String(error); - sendVisibleStatus(`Compaction failed: ${detail}`); + sendVisibleStatus(`📦 Compaction failed: ${detail}`); } finally { if (manualCompact?.threadId === threadId) { const compact = manualCompact; diff --git a/cli/src/codex/shared/projection.test.ts b/cli/src/codex/shared/projection.test.ts index 3d421dc9..6695db40 100644 --- a/cli/src/codex/shared/projection.test.ts +++ b/cli/src/codex/shared/projection.test.ts @@ -106,4 +106,45 @@ describe('shared history projection', () => { await projection.notification('item/completed', { threadId: 'child', turnId: 'turn', item: { id: 'prompt', type: 'userMessage', content: [{ type: 'text', text: 'child prompt' }], clientId: 'cid' } }); expect(user).not.toHaveBeenCalled(); expect(committed).not.toHaveBeenCalled(); }); + + it('announces parent compaction start via session event, deduping repeats', async () => { + const send = vi.fn(); const sendEvent = vi.fn(); + const session = { getMetadata: () => ({}), sendAgentMessage: send, sendSessionEvent: sendEvent } as unknown as ApiSessionClient; + const projection = new SharedCodexProjection(session, 'thread', async () => {}); + const params = { threadId: 'thread', turnId: 'turn', item: { id: 'c1', type: 'contextCompaction' } }; + await projection.notification('item/started', params); + await projection.notification('item/started', params); + await projection.notification('item/started', { threadId: 'thread', turnId: 'turn', item: { id: 'c2', type: 'contextCompaction' } }); + expect(send).not.toHaveBeenCalled(); + expect(sendEvent.mock.calls).toEqual([ + [{ type: 'message', message: '📦 Compaction started' }], + [{ type: 'message', message: '📦 Compaction started' }] + ]); + }); + it('keeps child compaction start as trace with statusText, never a parent status', async () => { + const send = vi.fn(); const sendEvent = vi.fn(); + const session = { getMetadata: () => ({}), sendAgentMessage: send, sendSessionEvent: sendEvent } as unknown as ApiSessionClient; + const projection = new SharedCodexProjection(session, 'child', async () => {}, 'root'); + await projection.notification('item/started', { threadId: 'child', turnId: 'turn', item: { id: 'c1', type: 'contextCompaction' } }); + expect(sendEvent).not.toHaveBeenCalled(); + expect(send).toHaveBeenCalledTimes(1); + const [body] = send.mock.calls[0]; + expect(body.type).toBe('agent-run-trace'); + expect(body.message.type).toBe('compaction_started'); + expect(body.message.statusText).toBe('📦 Compaction started'); + }); + it('emits live compaction start but not for completed history replay', async () => { + const send = vi.fn(); const sendEvent = vi.fn(); + const session = { getMetadata: () => ({}), sendAgentMessage: send, sendSessionEvent: sendEvent } as unknown as ApiSessionClient; + const projection = new SharedCodexProjection(session, 'thread', async () => {}); + await projection.history({ turns: [{ id: 'turn', status: 'completed', + items: [{ id: 'c1', type: 'contextCompaction' }] }] }); + expect(sendEvent).not.toHaveBeenCalled(); + expect(send.mock.calls.map(([b]) => b).some((b) => b.message === '📦 Compaction started')).toBe(false); + expect(send.mock.calls.map(([b]) => b)).toContainEqual(expect.objectContaining({ type: 'context_compacted' })); + send.mockClear(); + await projection.notification('item/started', { threadId: 'thread', turnId: 'live', item: { id: 'c9', type: 'contextCompaction' } }); + expect(send).not.toHaveBeenCalled(); + expect(sendEvent).toHaveBeenCalledWith({ type: 'message', message: '📦 Compaction started' }); + }); }); diff --git a/cli/src/codex/shared/projection.ts b/cli/src/codex/shared/projection.ts index 6c8c03eb..9a4e2471 100644 --- a/cli/src/codex/shared/projection.ts +++ b/cli/src/codex/shared/projection.ts @@ -90,7 +90,7 @@ export class SharedCodexProjection { await this.project(method, params, modelAtReceipt); } - private async project(method: string, params: unknown, modelAtReceipt?: string): Promise { + private async project(method: string, params: unknown, modelAtReceipt?: string, fromHistory = false): Promise { if (method.startsWith('codex/event/')) return; const p = record(params); const item = record(p.item); @@ -146,6 +146,20 @@ export class SharedCodexProjection { this.send({ type: 'tool-call', name: event.name, callId, input: event.input ?? event.arguments }, key); } else if (event.type === 'codex_tool_call_end' && callId) { this.send({ type: 'tool-call-result', callId, output: event.output, is_error: event.is_error }, key); + } else if (event.type === 'compaction_started') { + // History replay re-projects every completed item's start; + // a finished compaction must not announce a live status. + if (fromHistory) continue; + if (this.parentThreadId) { + this.send({ type: 'compaction_started', statusText: '📦 Compaction started' }, key); + } else { + // Status UI listens on session events; an agent message + // would render as a normal assistant bubble instead. + // Dedup is in-memory only (this.emitted set). + if (this.emitted.has(key)) continue; + this.emitted.add(key); + this.session.sendSessionEvent({ type: 'message', message: '📦 Compaction started' }); + } } else if (event.type === 'token_count' || event.type === 'context_compacted' || event.type.startsWith('thread_goal_')) { const model = event.type === 'token_count' && turnId ? this.turnModels.get(turnId) : undefined; this.send({ ...event, ...(model ? { model } : {}), flavor: 'codex', scope: { role: 'parent', threadId: this.threadId }, scope_role: 'parent', thread_id: this.threadId }, key); @@ -191,7 +205,7 @@ export class SharedCodexProjection { if (!this.parentThreadId && string(record(item).id) && pendingTitle && !this.completedTitles.has(titleKey)) { this.pendingTitles.set(titleKey, pendingTitle); } - await this.project('item/started', params); + await this.project('item/started', params, undefined, true); // Active snapshots can contain partial assistant text. Do not // settle it under the final stable id and suppress completion. if (turn.status !== 'inProgress' || record(item).status === 'completed' || record(item).type === 'userMessage') { @@ -203,7 +217,7 @@ export class SharedCodexProjection { } this.pendingTitles.delete(titleKey); } - await this.project('item/completed', params); + await this.project('item/completed', params, undefined, true); } } } diff --git a/cli/src/codex/utils/appServerEventConverter.test.ts b/cli/src/codex/utils/appServerEventConverter.test.ts index 00e18eb7..6898e58a 100644 --- a/cli/src/codex/utils/appServerEventConverter.test.ts +++ b/cli/src/codex/utils/appServerEventConverter.test.ts @@ -1042,6 +1042,75 @@ describe('AppServerEventConverter', () => { ]); }); + it('converts compaction start notifications into scoped compaction_started events', () => { + const converter = new AppServerEventConverter(); + const events = converter.handleNotification('item/started', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }); + + expect(events).toEqual([{ + type: 'compaction_started', + thread_id: 'thread-1', + turn_id: 'turn-1', + compaction_id: 'compact-item-1' + }]); + }); + + it('dedups duplicate compaction starts per compaction id but allows successive compactions', () => { + const converter = new AppServerEventConverter(); + const params = { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + }; + + expect(converter.handleNotification('item/started', params)).toHaveLength(1); + expect(converter.handleNotification('item/started', params)).toEqual([]); + expect(converter.handleNotification('codex/event/item_started', { + msg: { + type: 'item_started', + thread_id: 'thread-1', + turn_id: 'turn-1', + item_id: 'compact-item-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + } + })).toEqual([]); + expect(converter.handleNotification('item/started', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-2', type: 'contextCompaction' } + })).toHaveLength(1); + }); + + it('drops malformed compaction starts without an item id', () => { + const converter = new AppServerEventConverter(); + expect(converter.handleNotification('item/started', { + threadId: 'thread-1', + item: { type: 'contextCompaction' } + })).toEqual([]); + }); + + it('scopes compaction start dedup by thread so a reused item id still announces', () => { + const converter = new AppServerEventConverter(); + expect(converter.handleNotification('item/started', { + threadId: 'thread-1', + turnId: 'turn-1', + item: { id: 'compact-item-1', type: 'contextCompaction' } + })).toHaveLength(1); + expect(converter.handleNotification('item/started', { + threadId: 'thread-2', + turnId: 'turn-9', + item: { id: 'compact-item-1', type: 'contextCompaction' } + })).toEqual([{ + type: 'compaction_started', + thread_id: 'thread-2', + turn_id: 'turn-9', + compaction_id: 'compact-item-1' + }]); + }); + it('ignores compacted notifications without thread ids', () => { const converter = new AppServerEventConverter(); @@ -1118,4 +1187,18 @@ describe('AppServerEventConverter', () => { debug.mockRestore(); }); + + it('clears compaction start dedup on reset', () => { + const converter = new AppServerEventConverter(); + const params = { threadId: 'thread-1', turnId: 'turn-1', item: { id: 'compact-item-1', type: 'contextCompaction' } }; + expect(converter.handleNotification('item/started', params)).toHaveLength(1); + converter.reset(); + expect(converter.handleNotification('item/started', params)).toHaveLength(1); + }); + + it('announces legacy compaction start without scope', () => { + const converter = new AppServerEventConverter(); + const events = converter.handleNotification('item/started', { item: { id: 'compact-legacy', type: 'contextCompaction' } }); + expect(events).toEqual([{ type: 'compaction_started', compaction_id: 'compact-legacy' }]); + }); }); diff --git a/cli/src/codex/utils/appServerEventConverter.ts b/cli/src/codex/utils/appServerEventConverter.ts index 1421b826..4ac506c1 100644 --- a/cli/src/codex/utils/appServerEventConverter.ts +++ b/cli/src/codex/utils/appServerEventConverter.ts @@ -502,6 +502,7 @@ export class AppServerEventConverter { private readonly lastCommandOutputDeltaByItemId = new Map(); private readonly rawAgentToolCallIds = new Set(); private readonly rawAgentToolNames = new Map(); + private readonly announcedCompactionStarts = new Set(); private handleWrappedCodexEvent(paramsRecord: Record): ConvertedEvent[] | null { const msg = asRecord(paramsRecord.msg); @@ -970,6 +971,15 @@ export class AppServerEventConverter { } if (itemType === 'contextcompaction') { + if (method === 'item/started') { + const compactionKey = `${asString(eventScope.thread_id) ?? ''}\0${itemId}`; + if (this.announcedCompactionStarts.has(compactionKey)) { + return events; + } + this.announcedCompactionStarts.add(compactionKey); + events.push(scoped({ type: 'compaction_started', compaction_id: itemId })); + return events; + } if (method === 'item/completed') { const threadId = asString(eventScope.thread_id); const turnId = asString(eventScope.turn_id); @@ -1218,5 +1228,6 @@ export class AppServerEventConverter { this.lastCommandOutputDeltaByItemId.clear(); this.rawAgentToolCallIds.clear(); this.rawAgentToolNames.clear(); + this.announcedCompactionStarts.clear(); } }