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 <noreply@anthropic.com>

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<RequestPermissionResponse>
 }
 
@@ -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)
       }
     },
 

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 {

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
           })

src/main/session/SessionStore.ts

@@ -48,8 +48,11 @@ export class SessionStore {
   private sessionsIndex: Map<string, MulticaSession> = new Map()
   private loadedSessions: Map<string, SessionData> = new Map()
 
-  // Write locks to prevent concurrent writes
-  private writeLocks: Map<string, Promise<void>> = new Map()
+  // Write queues to serialize concurrent writes (proper async mutex pattern)
+  private writeQueues: Map<string, Promise<void>> = new Map()
+
+  // Sequence counters per session for ordering updates (handles concurrent async updates)
+  private sequenceCounters: Map<string, number> = 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<void> {
+  async appendUpdate(sessionId: string, update: SessionNotification): Promise<StoredSessionUpdate> {
     // 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<string>
   ): Promise<void> {
-    // 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<void> => {
+      // 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 {

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)) {

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

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 }

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
 }