fix(native-chat): add reconciliation to catch missed transcript updates

Periodic reconciliation catches file changes that fs.watch may miss on
remote filesystems. Event-driven drains use a bounded max-wait timer to
ensure responsiveness. Extracts timer logic into a scheduler module for
testability and maintainability.

src/main/native-chat/transcript-file-version.ts

@@ -0,0 +1,30 @@
+import { stat } from 'node:fs/promises'
+
+export type TranscriptFileVersion = {
+  identity: string
+  size: number
+  mtimeMs: number
+  ctimeMs: number
+}
+
+export async function readTranscriptFileVersion(filePath: string): Promise<TranscriptFileVersion> {
+  const value = await stat(filePath)
+  return {
+    identity: `${value.dev}:${value.ino}`,
+    size: value.size,
+    mtimeMs: value.mtimeMs,
+    ctimeMs: value.ctimeMs
+  }
+}
+
+export function transcriptFileVersionChanged(
+  current: TranscriptFileVersion,
+  previous: TranscriptFileVersion
+): boolean {
+  return (
+    current.identity !== previous.identity ||
+    current.size !== previous.size ||
+    current.mtimeMs !== previous.mtimeMs ||
+    current.ctimeMs !== previous.ctimeMs
+  )
+}

src/main/native-chat/transcript-watch-contract.ts

@@ -28,6 +28,8 @@ export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & {
   debounceMs?: number
   /** Test-only override for the production resolve-poll backoff. */
   resolvePollIntervalMs?: number
+  /** Test-only override for the host-side watcher reconciliation interval. */
+  reconciliationIntervalMs?: number
 }
 
 export type NativeChatTranscriptSubscription = {

src/main/native-chat/transcript-watch-engine.ts

@@ -2,6 +2,11 @@ import { watch, type FSWatcher } from 'node:fs'
 import { open, stat } from 'node:fs/promises'
 import { basename, dirname } from 'node:path'
 import type { NativeChatMessage, NativeChatTurnLifecycle } from '../../shared/native-chat-types'
+import {
+  readTranscriptFileVersion,
+  transcriptFileVersionChanged,
+  type TranscriptFileVersion
+} from './transcript-file-version'
 import {
   readIncrementalTranscriptMessages,
   resetIncrementalTranscriptState,
@@ -13,8 +18,8 @@ import type {
   NativeChatTranscriptSubscription,
   SubscribeNativeChatTranscriptArgs
 } from './transcript-watch-contract'
+import { createTranscriptWatchScheduler } from './transcript-watch-scheduler'
 
-const DEFAULT_DEBOUNCE_MS = 40
 const ROTATION_RETRY_MS = 25
 const MAX_ROTATION_RETRY_MS = 2_000
 let activeWatcherCount = 0
@@ -23,11 +28,6 @@ export function getActiveNativeChatWatcherCount(): number {
   return activeWatcherCount
 }
 
-async function fileVersion(filePath: string): Promise<{ identity: string; size: number }> {
-  const value = await stat(filePath)
-  return { identity: `${value.dev}:${value.ino}`, size: value.size }
-}
-
 async function boundaryFingerprint(filePath: string, offset: number): Promise<string> {
   if (offset <= 0) {
     return ''
@@ -62,7 +62,7 @@ export async function installTranscriptWatcher(
   } catch {
     return null
   }
-  const { onAppend, onInitialSnapshot, onReplace, initialLimit, debounceMs } = args
+  const { onAppend, onInitialSnapshot, onReplace, initialLimit } = args
   const decodeLifecycle = nativeChatTurnLifecycleDecoderForAgent(args.agent)
 
   const state: IncrementalTranscriptState = {
@@ -72,7 +72,7 @@ export async function installTranscriptWatcher(
     pendingBytes: 0,
     droppingOversizedRecord: false
   }
-  let watchedIdentity: string | null = null
+  let watchedVersion: TranscriptFileVersion | null = null
   let watchedBoundary = ''
   let initialDrain = true
   // Guards the one-time error snapshot emitted when the initial drain throws, so
@@ -82,36 +82,20 @@ export async function installTranscriptWatcher(
   let reading = false
   let pendingReadRequested = false
   let rotationRetryCount = 0
-  let debounceTimer: ReturnType<typeof setTimeout> | null = null
   let watcher: FSWatcher | null = null
   let watcherNeedsRebind = false
 
-  function scheduleDrain(): void {
-    if (closed) {
-      return
-    }
-    if (debounceTimer) {
-      clearTimeout(debounceTimer)
-    }
-    debounceTimer = setTimeout(() => {
-      debounceTimer = null
-      void drain()
-    }, debounceMs ?? DEFAULT_DEBOUNCE_MS)
-  }
-
   function scheduleRotationRetry(): void {
-    if (closed || debounceTimer) {
+    if (closed) {
       return
     }
     const retryDelay = Math.min(
       ROTATION_RETRY_MS * 2 ** Math.min(rotationRetryCount, 7),
       MAX_ROTATION_RETRY_MS
     )
-    rotationRetryCount += 1
-    debounceTimer = setTimeout(() => {
-      debounceTimer = null
-      void drain()
-    }, retryDelay)
+    if (scheduler.scheduleRetry(retryDelay)) {
+      rotationRetryCount += 1
+    }
   }
 
   async function readAndEmitAppends(): Promise<void> {
@@ -136,7 +120,7 @@ export async function installTranscriptWatcher(
   }
 
   async function drainOnce(): Promise<void> {
-    const current = await fileVersion(filePath)
+    const current = await readTranscriptFileVersion(filePath)
     const currentBoundary = await boundaryFingerprint(filePath, state.offset)
     if (closed) {
       return
@@ -144,7 +128,7 @@ export async function installTranscriptWatcher(
     if (watcherNeedsRebind) {
       bindWatcher()
     }
-    const identityChanged = watchedIdentity !== null && current.identity !== watchedIdentity
+    const identityChanged = watchedVersion !== null && current.identity !== watchedVersion.identity
     const contentReplaced =
       identityChanged ||
       current.size < state.offset ||
@@ -152,7 +136,7 @@ export async function installTranscriptWatcher(
     if (contentReplaced) {
       resetIncrementalTranscriptState(state)
     }
-    watchedIdentity = current.identity
+    watchedVersion = current
 
     const replacementSnapshot =
       // Why: 0 is a valid window — an explicit undefined check keeps an empty
@@ -267,15 +251,45 @@ export async function installTranscriptWatcher(
     }
   }
 
+  async function reconcile(): Promise<void> {
+    if (closed) {
+      return
+    }
+    try {
+      const current = await readTranscriptFileVersion(filePath)
+      if (closed) {
+        return
+      }
+      const versionChanged =
+        watchedVersion === null || transcriptFileVersionChanged(current, watchedVersion)
+      if (versionChanged || current.size !== state.offset || watcherNeedsRebind) {
+        await drain()
+      }
+    } catch {
+      // Why: a missing/replaced path needs the existing capped rotation retry,
+      // even when fs.watch stayed silent about the transition.
+      await drain()
+    }
+  }
+
+  const scheduler = createTranscriptWatchScheduler({
+    debounceMs: args.debounceMs,
+    reconciliationIntervalMs: args.reconciliationIntervalMs,
+    drain: () => void drain(),
+    reconcile
+  })
+
   function bindWatcher(): void {
     const watchedName = basename(filePath)
     // Why: file watchers stay bound to an unlinked inode on macOS. Watching
     // the parent keeps observing a successor even after a long recreate gap.
     const nextWatcher = watch(dirname(filePath), (_event, changedName) => {
       if (changedName === null || changedName.toString() === watchedName) {
-        scheduleDrain()
+        scheduler.scheduleEventDrain()
       }
     })
+    // Why: an active tail should not keep a headless runtime alive during shutdown.
+    nextWatcher.unref?.()
     nextWatcher.on('error', () => {
       // Why: Windows can emit EPERM when a watched directory disappears. An
       // error listener prevents a process crash and retries against its successor.
@@ -298,7 +312,8 @@ export async function installTranscriptWatcher(
     return null
   }
   activeWatcherCount++
-  scheduleDrain()
+  scheduler.startReconciliation()
+  scheduler.scheduleEventDrain()
 
   return {
     watching: true,
@@ -307,10 +322,7 @@ export async function installTranscriptWatcher(
         return
       }
       closed = true
-      if (debounceTimer) {
-        clearTimeout(debounceTimer)
-        debounceTimer = null
-      }
+      scheduler.dispose()
       watcher?.close()
       watcher = null
       activeWatcherCount--

src/main/native-chat/transcript-watch-liveness.test.ts

@@ -0,0 +1,148 @@
+import { EventEmitter } from 'node:events'
+import type { FSWatcher } from 'node:fs'
+import type * as NodeFs from 'node:fs'
+import { appendFile, mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+const { watchers, watchCallbacks, watchMock } = vi.hoisted(() => ({
+  watchers: [] as (EventEmitter & {
+    close: ReturnType<typeof vi.fn>
+    unref: ReturnType<typeof vi.fn>
+  })[],
+  watchCallbacks: [] as ((event: string, filename: string | Buffer | null) => void)[],
+  watchMock: vi.fn()
+}))
+
+vi.mock('node:fs', async () => {
+  const actual = await vi.importActual<typeof NodeFs>('node:fs')
+  watchMock.mockImplementation((_path, callback) => {
+    const watcher = Object.assign(new EventEmitter(), { close: vi.fn(), unref: vi.fn() })
+    watchers.push(watcher)
+    watchCallbacks.push(callback)
+    return watcher as unknown as FSWatcher
+  })
+  return { ...actual, watch: watchMock }
+})
+
+import { subscribeNativeChatTranscript } from './transcript-watch'
+
+const roots: string[] = []
+
+afterEach(async () => {
+  vi.useRealTimers()
+  watchers.length = 0
+  watchCallbacks.length = 0
+  watchMock.mockClear()
+  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })))
+})
+
+async function tempFile(initial: string): Promise<string> {
+  const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-watch-liveness-'))
+  roots.push(root)
+  const filePath = join(root, 'transcript.jsonl')
+  await writeFile(filePath, initial)
+  return filePath
+}
+
+function claudeLine(uuid: string, role: 'user' | 'assistant', text: string): string {
+  return `${JSON.stringify({
+    type: role,
+    uuid,
+    timestamp: '2026-06-01T10:00:00.000Z',
+    message: { role, content: role === 'user' ? text : [{ type: 'text', text }] }
+  })}\n`
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise<void> {
+  const start = Date.now()
+  while (!predicate()) {
+    if (Date.now() - start > timeoutMs) {
+      throw new Error('timed out waiting for condition')
+    }
+    await new Promise((resolve) => setTimeout(resolve, 10))
+  }
+}
+
+describe('native chat transcript watcher liveness', () => {
+  it('reconciles an append when fs.watch omits the callback', async () => {
+    const filePath = await tempFile(claudeLine('seed', 'user', 'hello'))
+    const snapshots = vi.fn()
+    const appends = vi.fn()
+    const subscription = await subscribeNativeChatTranscript({
+      agent: 'claude',
+      sessionId: 'session',
+      filePath,
+      onInitialSnapshot: snapshots,
+      onAppend: appends,
+      debounceMs: 0,
+      reconciliationIntervalMs: 20
+    })
+    await waitFor(() => snapshots.mock.calls.length === 1)
+
+    await appendFile(filePath, claudeLine('missed-event', 'assistant', 'recovered'))
+    // The mocked watcher never invokes watchCallbacks[0], reproducing a silent
+    // missed/coalesced host event while the file itself remains valid.
+    await waitFor(() => appends.mock.calls.flat(2).some((message) => message.id === 'missed-event'))
+
+    subscription.unsubscribe()
+    expect(watchCallbacks).toHaveLength(1)
+  })
+
+  it('drains within the max wait while matching events remain sustained', async () => {
+    const filePath = await tempFile(claudeLine('seed', 'user', 'hello'))
+    const snapshots = vi.fn()
+    const appends = vi.fn()
+    const subscription = await subscribeNativeChatTranscript({
+      agent: 'claude',
+      sessionId: 'session',
+      filePath,
+      onInitialSnapshot: snapshots,
+      onAppend: appends,
+      debounceMs: 40,
+      reconciliationIntervalMs: 10_000
+    })
+    await waitFor(() => snapshots.mock.calls.length === 1)
+    await appendFile(filePath, claudeLine('sustained-events', 'assistant', 'bounded'))
+
+    for (let index = 0; index < 20; index += 1) {
+      watchCallbacks[0]!('change', 'transcript.jsonl')
+      await new Promise((resolve) => setTimeout(resolve, 20))
+    }
+
+    expect(appends.mock.calls.flat(2).some((message) => message.id === 'sustained-events')).toBe(
+      true
+    )
+    subscription.unsubscribe()
+  })
+
+  it('clears every timer and cannot rebind after unsubscribe', async () => {
+    vi.useFakeTimers()
+    const filePath = await tempFile('')
+    const snapshots = vi.fn()
+    const appends = vi.fn()
+    const subscription = await subscribeNativeChatTranscript({
+      agent: 'claude',
+      sessionId: 'session',
+      filePath,
+      onInitialSnapshot: snapshots,
+      onAppend: appends,
+      reconciliationIntervalMs: 20
+    })
+
+    subscription.unsubscribe()
+    expect(watchers[0]!.unref).toHaveBeenCalledOnce()
+    expect(watchers[0]!.close).toHaveBeenCalledOnce()
+    expect(vi.getTimerCount()).toBe(0)
+
+    watchCallbacks[0]!('change', 'transcript.jsonl')
+    watchers[0]!.emit('error', new Error('late watcher error'))
+    await vi.advanceTimersByTimeAsync(10_000)
+
+    expect(watchMock).toHaveBeenCalledOnce()
+    expect(snapshots).not.toHaveBeenCalled()
+    expect(appends).not.toHaveBeenCalled()
+    expect(vi.getTimerCount()).toBe(0)
+  })
+})

src/main/native-chat/transcript-watch-scheduler.ts

@@ -0,0 +1,101 @@
+const DEFAULT_DEBOUNCE_MS = 40
+const EVENT_DRAIN_MAX_WAIT_MS = 250
+const DEFAULT_RECONCILIATION_MS = 1_000
+
+type TranscriptWatchSchedulerOptions = {
+  debounceMs?: number
+  reconciliationIntervalMs?: number
+  drain: () => void
+  reconcile: () => Promise<void>
+}
+
+export type TranscriptWatchScheduler = {
+  scheduleEventDrain: () => void
+  scheduleRetry: (delayMs: number) => boolean
+  startReconciliation: () => void
+  dispose: () => void
+}
+
+function unrefTimer(timer: ReturnType<typeof setTimeout>): void {
+  timer.unref?.()
+}
+
+export function createTranscriptWatchScheduler(
+  options: TranscriptWatchSchedulerOptions
+): TranscriptWatchScheduler {
+  const debounceMs = options.debounceMs ?? DEFAULT_DEBOUNCE_MS
+  const maxEventWaitMs = Math.max(debounceMs, EVENT_DRAIN_MAX_WAIT_MS)
+  const reconciliationIntervalMs = options.reconciliationIntervalMs ?? DEFAULT_RECONCILIATION_MS
+  let disposed = false
+  let drainTimer: ReturnType<typeof setTimeout> | null = null
+  let firstEventAt: number | null = null
+  let reconciliationTimer: ReturnType<typeof setTimeout> | null = null
+
+  function clearDrainTimer(): void {
+    if (!drainTimer) {
+      return
+    }
+    clearTimeout(drainTimer)
+    drainTimer = null
+  }
+
+  function fireDrain(): void {
+    clearDrainTimer()
+    firstEventAt = null
+    options.drain()
+  }
+
+  function armDrain(delayMs: number): void {
+    drainTimer = setTimeout(fireDrain, delayMs)
+    unrefTimer(drainTimer)
+  }
+
+  function armReconciliation(): void {
+    if (disposed || reconciliationTimer) {
+      return
+    }
+    reconciliationTimer = setTimeout(() => {
+      reconciliationTimer = null
+      // Why: wait for the host-side stat/drain check before rearming so a slow
+      // remote filesystem cannot accumulate overlapping reconciliation work.
+      void options.reconcile().then(armReconciliation, armReconciliation)
+    }, reconciliationIntervalMs)
+    unrefTimer(reconciliationTimer)
+  }
+
+  return {
+    scheduleEventDrain(): void {
+      if (disposed) {
+        return
+      }
+      const now = Date.now()
+      firstEventAt ??= now
+      const remainingMaxWait = Math.max(0, maxEventWaitMs - (now - firstEventAt))
+      clearDrainTimer()
+      armDrain(Math.min(debounceMs, remainingMaxWait))
+    },
+    scheduleRetry(delayMs: number): boolean {
+      if (disposed || drainTimer) {
+        return false
+      }
+      firstEventAt = null
+      armDrain(delayMs)
+      return true
+    },
+    startReconciliation(): void {
+      armReconciliation()
+    },
+    dispose(): void {
+      if (disposed) {
+        return
+      }
+      disposed = true
+      clearDrainTimer()
+      if (reconciliationTimer) {
+        clearTimeout(reconciliationTimer)
+        reconciliationTimer = null
+      }
+      firstEventAt = null
+    }
+  }
+}