diff --git a/cli/src/codex/shared/queue.ts b/cli/src/codex/shared/queue.ts index 0ddb01bc..e715ac52 100644 --- a/cli/src/codex/shared/queue.ts +++ b/cli/src/codex/shared/queue.ts @@ -80,8 +80,8 @@ export class SharedCodexQueue { } while (cursor); return items; } - reconcile(): Promise { return this.serial(() => this.reconcileNow()); } - private async reconcileNow(): Promise { + reconcile(): Promise { return this.serial(() => this.reconcileNow()); } + private async reconcileNow(): Promise { const present = new Set(); for (const item of await this.list()) { const id = item.clientUserMessageId; present.add(id); @@ -94,13 +94,15 @@ export class SharedCodexQueue { } // Absence could mean consumed, removed, or lost during engine restart. // Only an item event/history or a successful delete can decide which. + const dequeued: string[] = []; for (const [id, entry] of Object.entries(this.entries)) { - if (entry.state === 'queued' && !present.has(id)) entry.state = 'unknown'; + if (entry.state === 'queued' && !present.has(id)) { entry.state = 'unknown'; dequeued.push(id); } } await this.save(); const unknown = Object.entries(this.entries).filter(([, entry]) => entry.state === 'unknown').map(([id]) => id); if (unknown.length) this.uncertain(unknown); await this.publishReleased(); + return dequeued; } private async publishReleased(): Promise { diff --git a/cli/src/codex/shared/root.test.ts b/cli/src/codex/shared/root.test.ts index ae027642..f400eba4 100644 --- a/cli/src/codex/shared/root.test.ts +++ b/cli/src/codex/shared/root.test.ts @@ -61,6 +61,8 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] }) const updateState = vi.fn((fn: (value: AgentState) => AgentState) => { state = fn(state); }); const rpc = new Map Promise>(); const send = vi.fn(); + const consumed = vi.fn(); + const uncertain = vi.fn(); const hubArchivedListeners: Array<() => void> = []; const session = { sessionId: 'sid', getMetadata: () => metadata, @@ -74,7 +76,7 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] }) }, rpcHandlerManager: { registerHandler: (name: string, handler: (raw: unknown) => Promise) => rpc.set(name, handler) }, sendSessionEvent() {}, sendAgentMessage: send, emitSessionReady() {}, - sendUserMessage() {}, emitMessagesConsumed() {}, emitSteerIndeterminate() {}, syncNativeQueuedMessage() {}, + sendUserMessage() {}, emitMessagesConsumed: consumed, emitSteerIndeterminate: uncertain, syncNativeQueuedMessage() {}, sendSessionDeath() {}, async flush() {}, close() {} } as unknown as ApiSessionClient; const end = opts?.end ?? (async () => { throw new Error('Unexpected root archive'); }); @@ -94,7 +96,8 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] }) abandoned(): void; }; return { - root, native, rpc, send, metadata: () => metadata, state: () => state, updateState, + root, native, rpc, send, consumed, uncertain, metadata: () => metadata, state: () => state, updateState, + drain: () => (root as unknown as { notifications: Promise }).notifications, reconnect: () => reconnect?.(), emitHubArchived: () => { for (const listener of hubArchivedListeners) listener(); }, hubArchivedListenerCount: () => hubArchivedListeners.length, @@ -102,6 +105,89 @@ async function fixture(opts?: { hubArchived?: boolean; end?: RootHost['end'] }) }; } +async function startCompaction(f: Awaited>, id = 'input', turnId = 'compacting-turn') { + f.native.queue = [{ id: `native-${id}`, clientUserMessageId: id, input: [{ type: 'text', text: id }] }]; + await f.root.refresh(); + f.native.queue = []; + // Native removes input before announcing the turn and before recording userMessage. + f.native.notify('thread/queue/changed', { threadId: 'thread' }); + await vi.waitFor(() => expect(f.uncertain).toHaveBeenCalledWith(expect.arrayContaining([id]))); + const turn = { id: turnId, status: 'inProgress', items: [] }; + f.native.thread.turns.push(turn); + f.native.notify('turn/started', { threadId: 'thread', turn }); + f.native.notify('item/started', { threadId: 'thread', turnId, item: { id: 'compact', type: 'contextCompaction' } }); + return turn; +} + +describe('compaction delivery confirmation', () => { + it('does not turn uncertainty discovered by a cold snapshot into live compaction evidence', async () => { + const f = await fixture(); + f.native.queue = [{ id: 'old-native', clientUserMessageId: 'old', input: [] }]; + await f.root.refresh(); + f.native.queue = []; + await f.root.refresh(); + expect(f.uncertain).toHaveBeenCalledWith(['old']); + f.native.thread.turns.push({ id: 'later', status: 'inProgress', items: [] }); + f.native.notify('turn/started', { threadId: 'thread', turn: { id: 'later' } }); + f.native.notify('item/started', { threadId: 'thread', turnId: 'later', item: { id: 'c', type: 'contextCompaction' } }); + await f.drain(); + expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }); + }); + + it('reports only freshly dequeued input as waiting, without acknowledging or replaying it', async () => { + const f = await fixture(); + const request = vi.spyOn(f.root.client, 'request'); + const turn = await startCompaction(f); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] })); + expect(f.consumed).not.toHaveBeenCalled(); + expect(request.mock.calls.some(([method]) => method === 'thread/queue/add')).toBe(false); + f.native.notify('item/started', { threadId: 'thread', turnId: turn.id, + item: { id: 'user', type: 'userMessage', clientId: 'input', content: [] } }); + await vi.waitFor(() => expect(f.consumed).toHaveBeenCalledWith(['input'], { steered: undefined })); + expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }); + }); + + it.each(['completed', 'interrupted', 'failed'])('restores uncertainty when the turn is %s without consumption', async status => { + const f = await fixture(); + const turn = await startCompaction(f); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] })); + turn.status = status; + f.native.notify('turn/completed', { threadId: 'thread', turn }); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] })); + expect(f.consumed).not.toHaveBeenCalled(); + await startCompaction(f, 'next', 'next-turn'); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['next'] })); + }); + + it('withdraws waiting at compaction completion and ignores child and stale turn events', async () => { + const f = await fixture(); + const turn = await startCompaction(f); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] })); + for (const [threadId, turnId] of [['child', turn.id], ['thread', 'stale']]) { + f.native.notify('item/completed', { threadId, turnId, item: { id: 'other', type: 'contextCompaction' } }); + } + await f.drain(); + expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] }); + f.native.notify('item/completed', { threadId: 'thread', turnId: turn.id, item: { id: 'compact', type: 'contextCompaction' } }); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] })); + expect(f.consumed).not.toHaveBeenCalled(); + }); + + it('does not keep waiting after native disconnect or stopping', async () => { + const f = await fixture(); + await startCompaction(f); + await vi.waitFor(() => expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: ['input'] })); + f.native.initialized = false; + f.native.abandoned(); + expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }); + await vi.waitFor(() => expect(f.native.initialized).toBe(true)); + await f.root.refresh(); + expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }); + f.root.stopAccepting(); + expect(f.state()).toMatchObject({ codexCompactionPendingMessageIds: [] }); + }); +}); + async function completePlan(f: Awaited>, status = 'completed') { await f.root.applySettings({ collaborationMode: 'plan' }); const turn = { id: 'plan-turn', status: 'inProgress', items: [{ id: 'plan-item', type: 'plan', text: '# Implement me' }] }; diff --git a/cli/src/codex/shared/root.ts b/cli/src/codex/shared/root.ts index 187c0153..93811c5f 100644 --- a/cli/src/codex/shared/root.ts +++ b/cli/src/codex/shared/root.ts @@ -54,6 +54,11 @@ export class SharedCodexRoot { private work: Promise = Promise.resolve(); private notifications = Promise.resolve(); private currentTurn: string | undefined; + // Ephemeral observations only: dequeue/compaction never prove acceptance. + private readonly dequeuedInputs = new Set(); + private deliveryTurn?: string; + private compactingTurn?: string; + private publishedCompactionIds?: string; private latestTurn?: { id: string; status: string; planId?: string }; private publishedPlanId: string | null | undefined; private submittingPlan = false; @@ -137,6 +142,7 @@ export class SharedCodexRoot { this.projection?.reset(); this.steeringActive = undefined; this.publishedPlanId = undefined; + this.publishedCompactionIds = undefined; void this.refresh().then(() => this.refreshChildren(false)).then(() => this.queue.replay()) .catch(error => logger.debug('[Codex shared] hub resync', error)); }); @@ -186,7 +192,11 @@ export class SharedCodexRoot { if (this.threadId && this.threadId !== threadId) throw new Error('Cannot retarget a shared HAPI session'); this.threadId = threadId; this.queue = new SharedCodexQueue(this.client, threadId, join(this.host.directory, `${this.session.sessionId}.queue.json`), - (ids, steered) => this.session.emitMessagesConsumed(ids, { steered }), ids => this.session.emitSteerIndeterminate(ids), + (ids, steered) => { + this.session.emitMessagesConsumed(ids, { steered }); + for (const id of ids) this.dequeuedInputs.delete(id); + this.publishCompaction(); + }, ids => this.session.emitSteerIndeterminate(ids), (id, input) => this.session.syncNativeQueuedMessage(id, input === null ? null : inputText(input)), ids => this.session.setSteerDeliveryState(ids, 'queued')); await this.queue.load(); @@ -197,7 +207,7 @@ export class SharedCodexRoot { // advertise a destructive operation we cannot make atomic yet. conversationHistory: { forkCurrent: true, forkAtMessage: true, rewindToMessage: false } } })); - this.session.updateAgentState(state => ({ ...state, controlledByUser: false, startingMode: undefined, codexPlanProposalId: null, requests: {}, + this.session.updateAgentState(state => ({ ...state, controlledByUser: false, startingMode: undefined, codexPlanProposalId: null, codexCompactionPendingMessageIds: [], requests: {}, completedRequests: { ...state.completedRequests, ...Object.fromEntries(Object.entries(state.requests ?? {}).map(([id, request]) => [id, { ...request, completedAt: Date.now(), status: 'canceled' as const }])) } })); @@ -213,11 +223,21 @@ export class SharedCodexRoot { this.heartbeat = setInterval(() => this.alive(), 2_000); this.alive(); this.session.emitSessionReady(); } private publishSteering(): void { + this.publishCompaction(); const active = Boolean(this.currentTurn) && !this.stopping && !this.closed && !this.reconnecting && this.client.isInitialized(); if (this.steeringActive === active) return; this.steeringActive = active; this.session.updateAgentState(state => ({ ...state, steeringActive: active })); } + private publishCompaction(): void { + const active = this.compactingTurn && this.compactingTurn === this.currentTurn + && !this.closed && !this.stopping && !this.reconnecting && this.client.isInitialized(); + const ids = active ? [...this.dequeuedInputs].filter(id => this.queue?.state(id) === 'unknown').sort() : []; + const key = JSON.stringify(ids); + if (key === this.publishedCompactionIds) return; + this.publishedCompactionIds = key; + this.session.updateAgentState(state => ({ ...state, codexCompactionPendingMessageIds: ids })); + } private alive(): void { this.publishSteering(); this.publishPlan(); @@ -273,8 +293,25 @@ export class SharedCodexRoot { const name = p.threadName ?? undefined; this.session.updateMetadata(metadata => ({ ...metadata, name })); } if (method === 'thread/archived') { await this.host.end(this, false); return; } - if (method === 'thread/queue/changed') await this.queue.reconcile(); - await this.projection.notification(method, params, modelAtReceipt); this.alive(); + // Process these in notification order: queue removal can precede turn/started. + // Older unknown input must not acquire a later turn's compaction status. + if (method === 'turn/started') { + const id = string(record(p.turn).id); + if (this.deliveryTurn && this.deliveryTurn !== id) this.dequeuedInputs.clear(); + this.deliveryTurn = id; + this.compactingTurn = undefined; + } + if (method === 'turn/completed' && record(p.turn).id === this.deliveryTurn) { + this.dequeuedInputs.clear(); this.deliveryTurn = undefined; this.compactingTurn = undefined; + } + if (p.turnId === this.currentTurn && record(p.item).type === 'contextCompaction') { + if (method === 'item/started') this.compactingTurn = string(p.turnId); + if (method === 'item/completed') { this.compactingTurn = undefined; this.dequeuedInputs.clear(); } + } + if (method === 'thread/queue/changed') { + for (const id of await this.queue.reconcile()) this.dequeuedInputs.add(id); + } + await this.projection.notification(method, params, modelAtReceipt); this.publishCompaction(); this.alive(); } async readThread(threadId = this.threadId): Promise> { let thread = record(record(await this.client.request('thread/read', { threadId, includeTurns: false })).thread); @@ -342,7 +379,9 @@ export class SharedCodexRoot { } private async reconnect(): Promise { if (this.closed || this.stopping || this.reconnecting || !this.threadId) return; - this.reconnecting = true; this.publishSteering(); this.publishPlan(); this.permissions.close(); + this.reconnecting = true; + this.dequeuedInputs.clear(); this.deliveryTurn = undefined; this.compactingTurn = undefined; + this.publishSteering(); this.publishPlan(); this.permissions.close(); try { while (!this.closed && !this.stopping) { try { diff --git a/docs/guide/codex-shared-sessions.md b/docs/guide/codex-shared-sessions.md index 8505fe94..e954da2c 100644 --- a/docs/guide/codex-shared-sessions.md +++ b/docs/guide/codex-shared-sessions.md @@ -70,6 +70,13 @@ Native pending queue text is mirrored to HAPI. Native edits and acknowledged removals update it. Losing a queue entry during transport recovery is not proof that it was canceled or executed. +During live context compaction, Web may show **Compacting context — waiting for +message acceptance** for input just removed from the native queue. This is not +an acceptance acknowledgement: only the native user-message event or history +confirms consumption. Compaction completion, turn completion/interruption or a +native disconnect withdraws this temporary status. Unconfirmed input retains +the existing unknown-delivery recovery controls and is never automatically replayed. + Codex arbitrates terminal/Web answers. A submitted Web answer is only a candidate. Native `serverRequest/resolved` closes the prompt with neutral **Resolved in Codex** status; HAPI does not invent the winner or their answer. diff --git a/shared/src/schemas.ts b/shared/src/schemas.ts index 40da2b47..3d716917 100644 --- a/shared/src/schemas.ts +++ b/shared/src/schemas.ts @@ -215,6 +215,8 @@ export const AgentStateSchema = z.object({ controlledByUser: z.boolean().nullish(), // Current actionable shared Codex proposal; content remains in the transcript. codexPlanProposalId: z.string().nullish(), + // Observed dequeue during live context compaction; not a consumption ACK. + codexCompactionPendingMessageIds: z.array(z.string()).nullish(), // True while the CLI is delivering a queued message into the active turn // (Steer). Surfaced so the web can reflect the inject in progress. steeringActive: z.boolean().nullish(), diff --git a/web/src/components/AssistantChat/QueuedMessagesBar.test.tsx b/web/src/components/AssistantChat/QueuedMessagesBar.test.tsx index 794a4851..176cb7b7 100644 --- a/web/src/components/AssistantChat/QueuedMessagesBar.test.tsx +++ b/web/src/components/AssistantChat/QueuedMessagesBar.test.tsx @@ -198,6 +198,41 @@ afterEach(() => { vi.unstubAllGlobals() }) +describe('QueuedMessagesBar compaction confirmation', () => { + it('shows waiting only for the observed message, then restores recovery controls when compaction ends', () => { + mocks.messageWindowState = { messages: [ + makeQueuedMessage(null, 'compacting', { deliveryState: 'indeterminate' }), + makeQueuedMessage(null, 'older', { deliveryState: 'indeterminate' }), + ] } + const client = new QueryClient() + const bar = (ids?: string[]) => ( + + + + ) + const view = render(bar(['local-compacting'])) + expect(screen.getAllByText('queuedMessages.compactionPending')).toHaveLength(1) + expect(screen.getAllByText('queuedMessages.steerOutcomeUnknown')).toHaveLength(1) + expect(screen.getAllByRole('button', { name: 'queuedMessages.retryOutcome' })).toHaveLength(1) + view.rerender(bar()) + expect(screen.queryByText('queuedMessages.compactionPending')).toBeNull() + expect(screen.getAllByText('queuedMessages.steerOutcomeUnknown')).toHaveLength(2) + expect(screen.getAllByRole('button', { name: 'queuedMessages.retryOutcome' })).toHaveLength(2) + }) + + it('does not change a queued message into an uncertain or consumed message', () => { + mocks.messageWindowState = { messages: [makeQueuedMessage(null, 'queued')] } + render( + + ) + expect(screen.queryByText('queuedMessages.compactionPending')).toBeNull() + expect(screen.queryByText('queuedMessages.steerOutcomeUnknown')).toBeNull() + expect(mocks.markMessagesConsumed).not.toHaveBeenCalled() + }) +}) + describe('QueuedMessagesBar layout', () => { it('keeps the queue footer flush with the composer area', () => { renderQueuedMessage() diff --git a/web/src/components/AssistantChat/QueuedMessagesBar.tsx b/web/src/components/AssistantChat/QueuedMessagesBar.tsx index 17478eb3..16490798 100644 --- a/web/src/components/AssistantChat/QueuedMessagesBar.tsx +++ b/web/src/components/AssistantChat/QueuedMessagesBar.tsx @@ -199,9 +199,11 @@ export function QueuedMessagesBar({ pendingScheduleRevision, onEdit, canSteer, + compactionPendingMessageIds, }: { sessionId: string api: ApiClient | null + compactionPendingMessageIds?: readonly string[] /** Current composer schedule, used only to guard an asynchronous edit restore. */ pendingSchedule: PendingSchedule | null /** Monotonic per-session revision; schedule selections win over an async edit restore. */ @@ -351,6 +353,8 @@ export function QueuedMessagesBar({ const editText = getQueuedMessageEditText(preview) const hasAttachments = attachmentNames.length > 0 const localId = msg.localId ?? msg.id + const waitingForCompaction = msg.deliveryState === 'indeterminate' + && compactionPendingMessageIds?.includes(localId) === true const isPending = cancelMutation.isPending || queuedOperationPending const canCancel = computeCanCancel({ id: msg.id, localId: msg.localId, isPending }) @@ -519,8 +523,8 @@ export function QueuedMessagesBar({ ) : null} {msg.deliveryState === 'indeterminate' ? ( -
- {t('queuedMessages.steerOutcomeUnknown')} +
+ {t(waitingForCompaction ? 'queuedMessages.compactionPending' : 'queuedMessages.steerOutcomeUnknown')}
) : null} {hasAttachments ? ( @@ -547,7 +551,7 @@ export function QueuedMessagesBar({ )}
- {msg.deliveryState === 'indeterminate' ? ( + {msg.deliveryState === 'indeterminate' && !waitingForCompaction ? (