From c59b9afc71429cead486c65cd0cafc5b41af81f0 Mon Sep 17 00:00:00 2001 From: Jiayuan Date: Fri, 16 Jan 2026 01:51:08 +0800 Subject: [PATCH] Fix Codex ACP content ordering and session persistence issues Fixes two related issues when using Codex ACP agent: (1) ENOENT errors in session persistence due to concurrent writes using identical temp file names, and (2) ChatView content appearing out of order in real-time display due to async ACP message delivery. Uses unique temp file names with UUID suffix and adds sequence numbers to track update ordering across async boundaries. Co-Authored-By: Claude Haiku 4.5 --- src/main/conductor/AcpClientFactory.ts | 12 ++-- src/main/conductor/Conductor.ts | 2 +- src/main/index.ts | 5 +- src/main/session/SessionStore.ts | 83 +++++++++++++++++------- src/renderer/src/components/ChatView.tsx | 13 +++- src/renderer/src/hooks/useApp.ts | 2 + src/shared/electron-api.d.ts | 1 + src/shared/types/session.ts | 1 + 8 files changed, 85 insertions(+), 34 deletions(-) diff --git a/src/main/conductor/AcpClientFactory.ts b/src/main/conductor/AcpClientFactory.ts index 45cc8a56a..4dacf88d9 100644 --- a/src/main/conductor/AcpClientFactory.ts +++ b/src/main/conductor/AcpClientFactory.ts @@ -12,7 +12,7 @@ import type { import type { SessionStore } from '../session/SessionStore' export interface AcpClientCallbacks { - onSessionUpdate?: (update: SessionNotification) => void + onSessionUpdate?: (update: SessionNotification, sequenceNumber?: number) => void onPermissionRequest?: (params: RequestPermissionRequest) => Promise } @@ -60,18 +60,20 @@ export function createAcpClient(sessionId: string, options: AcpClientFactoryOpti console.log(`[ACP] raw update:`, params) } - // Store raw update to SessionStore (if available) + // Store raw update to SessionStore (if available) and get sequence number + let sequenceNumber: number | undefined if (sessionStore) { try { - await sessionStore.appendUpdate(sessionId, params) + const storedUpdate = await sessionStore.appendUpdate(sessionId, params) + sequenceNumber = storedUpdate.sequenceNumber } catch (err) { console.error('[Conductor] Failed to store session update:', err) } } - // Trigger UI callback + // Trigger UI callback with sequence number for ordering if (callbacks.onSessionUpdate) { - callbacks.onSessionUpdate(params) + callbacks.onSessionUpdate(params, sequenceNumber) } }, diff --git a/src/main/conductor/Conductor.ts b/src/main/conductor/Conductor.ts index b3b642af9..13da83688 100644 --- a/src/main/conductor/Conductor.ts +++ b/src/main/conductor/Conductor.ts @@ -29,7 +29,7 @@ import type { MessageContent, MessageContentItem } from '../../shared/types/mess import { formatHistoryForReplay, hasReplayableHistory } from './historyReplay' export interface SessionUpdateCallback { - (update: SessionNotification): void + (update: SessionNotification, sequenceNumber?: number): void } export interface ConductorEvents { diff --git a/src/main/index.ts b/src/main/index.ts index 0ed99b135..d86652205 100644 --- a/src/main/index.ts +++ b/src/main/index.ts @@ -64,11 +64,12 @@ app.whenReady().then(async () => { // Initialize conductor with event handlers conductor = new Conductor({ events: { - onSessionUpdate: (params) => { - // Forward ALL session updates to renderer + onSessionUpdate: (params, sequenceNumber) => { + // Forward ALL session updates to renderer with sequence number for ordering if (mainWindow && !mainWindow.isDestroyed()) { mainWindow.webContents.send(IPC_CHANNELS.AGENT_MESSAGE, { sessionId: params.sessionId, + sequenceNumber, update: params.update, done: false }) diff --git a/src/main/session/SessionStore.ts b/src/main/session/SessionStore.ts index 7d3c5d9eb..5b8755919 100644 --- a/src/main/session/SessionStore.ts +++ b/src/main/session/SessionStore.ts @@ -48,8 +48,11 @@ export class SessionStore { private sessionsIndex: Map = new Map() private loadedSessions: Map = new Map() - // Write locks to prevent concurrent writes - private writeLocks: Map> = new Map() + // Write queues to serialize concurrent writes (proper async mutex pattern) + private writeQueues: Map> = new Map() + + // Sequence counters per session for ordering updates (handles concurrent async updates) + private sequenceCounters: Map = new Map() constructor(basePath?: string) { this.basePath = basePath ?? getDefaultStoragePath() @@ -157,6 +160,8 @@ export class SessionStore { const content = await readFile(dataPath, 'utf-8') const sessionData = JSON.parse(content) as SessionData this.loadedSessions.set(sessionId, sessionData) + // Initialize sequence counter from loaded data + this.initSequenceCounter(sessionId, sessionData.updates) return sessionData } catch { console.error(`[SessionStore] Failed to load session: ${sessionId}`) @@ -167,16 +172,22 @@ export class SessionStore { /** * Append a session update */ - async appendUpdate(sessionId: string, update: SessionNotification): Promise { + async appendUpdate(sessionId: string, update: SessionNotification): Promise { // Ensure session is loaded const sessionData = await this.get(sessionId) if (!sessionData) { throw new Error(`Session not found: ${sessionId}`) } - // Append update + // Get next sequence number for this session (monotonically increasing) + const currentSeq = this.sequenceCounters.get(sessionId) ?? 0 + const nextSeq = currentSeq + 1 + this.sequenceCounters.set(sessionId, nextSeq) + + // Append update with sequence number const storedUpdate: StoredSessionUpdate = { timestamp: new Date().toISOString(), + sequenceNumber: nextSeq, update } sessionData.updates.push(storedUpdate) @@ -191,6 +202,22 @@ export class SessionStore { // Persist (debounce in production, immediate for now) await this.saveSessionData(sessionId) await this.saveIndex() + + return storedUpdate + } + + /** + * Initialize sequence counter when loading session from disk + */ + private initSequenceCounter(sessionId: string, updates: StoredSessionUpdate[]): void { + // Find the highest sequence number in existing updates + let maxSeq = 0 + for (const update of updates) { + if (update.sequenceNumber && update.sequenceNumber > maxSeq) { + maxSeq = update.sequenceNumber + } + } + this.sequenceCounters.set(sessionId, maxSeq) } /** @@ -301,22 +328,21 @@ export class SessionStore { } /** - * Atomic write with lock to prevent concurrent writes and corruption + * Atomic write with proper serialization to prevent concurrent writes and corruption. + * + * Uses a chained promise pattern to ensure writes to the same file are serialized. + * Each write waits for the previous write to complete before starting. */ private async atomicWrite( lockKey: string, filePath: string, getData: () => Promise ): Promise { - // Wait for any pending write to complete - const pendingWrite = this.writeLocks.get(lockKey) - if (pendingWrite) { - await pendingWrite - } - - // Create new write promise - const writePromise = (async () => { - const tempPath = `${filePath}.tmp.${Date.now()}` + // Create the actual write operation + const doWrite = async (): Promise => { + // Use unique temp file name: timestamp + random suffix to avoid collisions + // even when multiple writes happen within the same millisecond + const tempPath = `${filePath}.tmp.${Date.now()}.${randomUUID().slice(0, 8)}` try { const data = await getData() await writeFile(tempPath, data) @@ -332,18 +358,25 @@ export class SessionStore { } throw err } - })() - - this.writeLocks.set(lockKey, writePromise) - - try { - await writePromise - } finally { - // Only delete if this is still our promise - if (this.writeLocks.get(lockKey) === writePromise) { - this.writeLocks.delete(lockKey) - } } + + // Get the current write queue for this key (or resolved promise if none) + const previousWrite = this.writeQueues.get(lockKey) ?? Promise.resolve() + + // Chain this write after the previous one. + // Catch previous errors so they don't block the chain - each write should + // be independent and a failure in one shouldn't prevent subsequent writes. + const thisWrite = previousWrite.catch(() => {}).then(doWrite) + + // Immediately update the queue (synchronous) before any await. + // Store a non-rejecting version for chaining purposes. + this.writeQueues.set( + lockKey, + thisWrite.catch(() => {}) + ) + + // Return the actual promise which can reject if this write fails + return thisWrite } private countMessages(updates: StoredSessionUpdate[]): number { diff --git a/src/renderer/src/components/ChatView.tsx b/src/renderer/src/components/ChatView.tsx index 63d40216e..769e9949d 100644 --- a/src/renderer/src/components/ChatView.tsx +++ b/src/renderer/src/components/ChatView.tsx @@ -153,6 +153,17 @@ interface Message { } function groupUpdatesIntoMessages(updates: StoredSessionUpdate[]): Message[] { + // Sort updates by sequence number to ensure correct ordering despite async delivery + // Updates without sequence numbers (e.g., user messages, legacy data) keep their relative position + const sortedUpdates = [...updates].sort((a, b) => { + // If both have sequence numbers, sort by sequence + if (a.sequenceNumber !== undefined && b.sequenceNumber !== undefined) { + return a.sequenceNumber - b.sequenceNumber + } + // If only one has sequence number, keep relative order (stable sort) + return 0 + }) + const messages: Message[] = [] let currentBlocks: ContentBlock[] = [] // Track tool calls by ID to update them in place @@ -191,7 +202,7 @@ function groupUpdatesIntoMessages(updates: StoredSessionUpdate[]): Message[] { } } - for (const stored of updates) { + for (const stored of sortedUpdates) { const notification = stored.update const update = notification?.update if (!update || !('sessionUpdate' in update)) { diff --git a/src/renderer/src/hooks/useApp.ts b/src/renderer/src/hooks/useApp.ts index 716c5b3f5..805f28b58 100644 --- a/src/renderer/src/hooks/useApp.ts +++ b/src/renderer/src/hooks/useApp.ts @@ -138,9 +138,11 @@ export function useApp(): AppState & AppActions { // Pass through original update without any accumulation // ChatView is responsible for accumulating chunks into complete messages + // Include sequence number for proper ordering of concurrent updates setSessionUpdates((prev) => { const newUpdate = { timestamp: new Date().toISOString(), + sequenceNumber: message.sequenceNumber, update: { sessionId: message.sessionId, update: message.update diff --git a/src/shared/electron-api.d.ts b/src/shared/electron-api.d.ts index 599f8fc44..d206cfec5 100644 --- a/src/shared/electron-api.d.ts +++ b/src/shared/electron-api.d.ts @@ -12,6 +12,7 @@ import type { MessageContent } from './types/message' export interface AgentMessage { sessionId: string + sequenceNumber?: number // Monotonically increasing for ordering concurrent updates update: { sessionUpdate: string content?: { type: string; text: string } diff --git a/src/shared/types/session.ts b/src/shared/types/session.ts index 71bd7af25..fceec2a9e 100644 --- a/src/shared/types/session.ts +++ b/src/shared/types/session.ts @@ -55,6 +55,7 @@ export interface AskUserQuestionResponseUpdate { */ export interface StoredSessionUpdate { timestamp: string // Receive time + sequenceNumber?: number // Monotonically increasing sequence number for ordering (added for concurrent update handling) update: SessionNotification | { update: AskUserQuestionResponseUpdate } // Raw ACP data or custom update }