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

import { appendFile, mkdtemp, rm, writeFile } from 'fs/promises'
import { tmpdir } from 'os'
import { join } from 'path'
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
import type { NativeChatMessage } from '../../shared/native-chat-types'
import { getActiveNativeChatWatcherCount, subscribeNativeChatTranscript } from './transcript-watch'

let tempRoots: string[] = []

beforeEach(() => {
  tempRoots = []
})

afterEach(async () => {
  await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true })))
  tempRoots = []
})

async function tempFile(initial: string): Promise<string> {
  const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-watch-'))
  tempRoots.push(root)
  const filePath = join(root, 'rollout.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 = 2000): 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('subscribeNativeChatTranscript', () => {
  it('re-emits from the top on first drain so appended turns are never dropped', async () => {
    const filePath = await tempFile(claudeLine('u-1', 'user', 'first'))
    const batches: NativeChatMessage[][] = []

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: (messages) => batches.push(messages),
      debounceMs: 5
    })

    await appendFile(filePath, claudeLine('a-1', 'assistant', 'reply'))
    await waitFor(() => batches.flat().some((m) => m.id === 'a-1'))

    sub.unsubscribe()

    // Seed-at-0 means the first drain re-reads the whole file; the assembler
    // dedups by id. The appended turn must appear; the pre-existing line may
    // appear too (collapsed downstream by id).
    const ids = batches.flat().map((m) => m.id)
    expect(ids).toContain('a-1')
  })

  it('appends a turn in the gap between initial read and first watcher drain exactly once', async () => {
    // Simulate the read/subscribe race: a turn lands after the caller's
    // readSession EOF but before the watcher's first drain. Seeding at 0 means
    // the first drain reads it; the assembler later dedups by deterministic id.
    const filePath = await tempFile(claudeLine('u-1', 'user', 'first'))
    const seen: NativeChatMessage[] = []

    // The gap turn is written BEFORE subscribe completes its first drain.
    await appendFile(filePath, claudeLine('a-gap', 'assistant', 'raced reply'))

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: (messages) => seen.push(...messages),
      debounceMs: 5
    })

    await waitFor(() => seen.some((m) => m.id === 'a-gap'))
    sub.unsubscribe()

    // The raced turn is present, and not duplicated within a single drain pass.
    expect(seen.filter((m) => m.id === 'a-gap')).toHaveLength(1)
  })

  it('recovers cleanly when a read throws (subscription not left deaf)', async () => {
    const filePath = await tempFile(claudeLine('u-1', 'user', 'hi'))
    const seen: NativeChatMessage[] = []

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: (messages) => seen.push(...messages),
      debounceMs: 5
    })

    // Make the file unreadable mid-flight (EACCES on the read path). The drain's
    // try/catch must break and reset `reading` in finally so a later append
    // still tails once permissions are restored.
    await waitFor(() => seen.some((m) => m.id === 'u-1'))
    const { chmod } = await import('fs/promises')
    await chmod(filePath, 0o000)
    await appendFile(filePath, claudeLine('a-1', 'assistant', 'reply')).catch(() => {})
    // Give the watcher a chance to attempt (and fail) a drain.
    await new Promise((resolve) => setTimeout(resolve, 40))
    await chmod(filePath, 0o644)
    await appendFile(filePath, claudeLine('a-2', 'assistant', 'recovered'))

    await waitFor(() => seen.some((m) => m.id === 'a-2'))
    sub.unsubscribe()
    expect(seen.some((m) => m.id === 'a-2')).toBe(true)
  })

  it('releases the watcher on unsubscribe (no leak)', async () => {
    const filePath = await tempFile(claudeLine('u-1', 'user', 'hi'))
    const before = getActiveNativeChatWatcherCount()

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: () => {},
      debounceMs: 5
    })
    expect(getActiveNativeChatWatcherCount()).toBe(before + 1)

    sub.unsubscribe()
    expect(getActiveNativeChatWatcherCount()).toBe(before)

    // Idempotent: a second unsubscribe must not under-count.
    sub.unsubscribe()
    expect(getActiveNativeChatWatcherCount()).toBe(before)
  })

  it('coalesces rapid successive appends without dropping messages', async () => {
    const filePath = await tempFile(claudeLine('u-1', 'user', 'hi'))
    const seen: NativeChatMessage[] = []

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: (messages) => seen.push(...messages),
      debounceMs: 10
    })

    // Fire several appends back-to-back within the debounce window.
    await appendFile(filePath, claudeLine('a-1', 'assistant', 'one'))
    await appendFile(filePath, claudeLine('a-2', 'assistant', 'two'))
    await appendFile(filePath, claudeLine('a-3', 'assistant', 'three'))

    await waitFor(() => ['a-1', 'a-2', 'a-3'].every((id) => seen.some((m) => m.id === id)))
    sub.unsubscribe()

    // Order is preserved for the appended turns (the seed re-read may also carry
    // the pre-existing u-1, which the assembler dedups downstream).
    const appendedIds = seen.map((m) => m.id).filter((id) => id !== 'u-1')
    expect(appendedIds).toEqual(['a-1', 'a-2', 'a-3'])
  })

  it('waits for an incomplete trailing JSONL line before advancing the offset', async () => {
    const filePath = await tempFile(claudeLine('u-1', 'user', 'hi'))
    const seen: NativeChatMessage[] = []

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: (messages) => seen.push(...messages),
      debounceMs: 5
    })

    await waitFor(() => seen.some((m) => m.id === 'u-1'))

    const line = claudeLine('a-partial', 'assistant', 'split reply')
    const splitAt = Math.floor(line.length / 2)
    await appendFile(filePath, line.slice(0, splitAt))
    await new Promise((resolve) => setTimeout(resolve, 40))
    expect(seen.some((m) => m.id === 'a-partial')).toBe(false)

    await appendFile(filePath, line.slice(splitAt))
    await waitFor(() => seen.some((m) => m.id === 'a-partial'))

    sub.unsubscribe()
    expect(seen.filter((m) => m.id === 'a-partial')).toHaveLength(1)
  })

  it('survives file replacement / rotation (offset reset on shrink)', async () => {
    const filePath = await tempFile(
      claudeLine('u-1', 'user', 'old') + claudeLine('a-1', 'assistant', 'old-reply')
    )
    const seen: NativeChatMessage[] = []

    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: 'ignored',
      filePath,
      onAppend: (messages) => seen.push(...messages),
      debounceMs: 5
    })

    // Replace the file with shorter content (simulates rotation to a new,
    // smaller session file at the same resolved path).
    await writeFile(filePath, claudeLine('u-2', 'user', 'fresh'))
    await waitFor(() => seen.some((m) => m.id === 'u-2'))

    // A subsequent append on the rotated file is still tailed.
    await appendFile(filePath, claudeLine('a-2', 'assistant', 'fresh-reply'))
    await waitFor(() => seen.some((m) => m.id === 'a-2'))

    sub.unsubscribe()
    const ids = seen.map((m) => m.id)
    expect(ids).toContain('u-2')
    expect(ids).toContain('a-2')
  })

  it('returns a no-op unsubscribe when the file cannot be resolved', async () => {
    const before = getActiveNativeChatWatcherCount()
    const sub = await subscribeNativeChatTranscript({
      agent: 'claude',
      sessionId: '',
      onAppend: () => {}
    })
    expect(getActiveNativeChatWatcherCount()).toBe(before)
    // Must not throw.
    sub.unsubscribe()
  })
})