diff --git a/src/main/native-chat/transcript-file-version.ts b/src/main/native-chat/transcript-file-version.ts new file mode 100644 index 00000000000..69634c0a957 --- /dev/null +++ b/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 { + 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 + ) +} diff --git a/src/main/native-chat/transcript-watch-contract.ts b/src/main/native-chat/transcript-watch-contract.ts index 2942594b370..f2e46f3581b 100644 --- a/src/main/native-chat/transcript-watch-contract.ts +++ b/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 = { diff --git a/src/main/native-chat/transcript-watch-engine.ts b/src/main/native-chat/transcript-watch-engine.ts index 6ce52de4787..1bcc0eccfc2 100644 --- a/src/main/native-chat/transcript-watch-engine.ts +++ b/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 { 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 | 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 { @@ -136,7 +120,7 @@ export async function installTranscriptWatcher( } async function drainOnce(): Promise { - 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 { + 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-- diff --git a/src/main/native-chat/transcript-watch-liveness.test.ts b/src/main/native-chat/transcript-watch-liveness.test.ts new file mode 100644 index 00000000000..3c06b0534cc --- /dev/null +++ b/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 + unref: ReturnType + })[], + watchCallbacks: [] as ((event: string, filename: string | Buffer | null) => void)[], + watchMock: vi.fn() +})) + +vi.mock('node:fs', async () => { + const actual = await vi.importActual('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 { + 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 { + 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) + }) +}) diff --git a/src/main/native-chat/transcript-watch-scheduler.ts b/src/main/native-chat/transcript-watch-scheduler.ts new file mode 100644 index 00000000000..983b1802166 --- /dev/null +++ b/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 +} + +export type TranscriptWatchScheduler = { + scheduleEventDrain: () => void + scheduleRetry: (delayMs: number) => boolean + startReconciliation: () => void + dispose: () => void +} + +function unrefTimer(timer: ReturnType): 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 | null = null + let firstEventAt: number | null = null + let reconciliationTimer: ReturnType | 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 + } + } +}