mirror of
https://github.com/stablyai/orca.git
synced 2026-10-17 08:03:14 +00:00
fix(native-chat): recover from 'No transcript found' race on a just-created session
Claude Code lazily flushes a new session's .jsonl, so the first transcript
read/tail can race the first write. When the file doesn't exist yet the reader
and cache returned a hard {error}, the renderer settled into phase 'error' with
no retry, and the live watcher no-op'd forever — so a session that is actually
alive stuck on 'No transcript found'.
Layered, additive fix (zero-regression):
- reader/cache: mark a missing file as a retryable notFound (null-resolve and
ENOENT-after-resolve); genuine non-ENOENT read/parse errors stay hard errors.
A notFound is never cached, so a later real read is not masked by a cached miss.
- renderer hook: on notFound, stay in 'loading' and retry readSession with a
bounded exponential backoff (1/2/4/8s, cap 10s); timer cleared on teardown.
- transcript-watch: install a backing-off resolve-poll (1s->10s cap) so a
not-yet-created file is picked up when it appears; blank-sessionId guard and
race-safe unsubscribe (no watcher/timer leak).
Fixes #8401
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -0,0 +1,111 @@
|
||||
import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
import { readNativeChatTranscript } from './transcript-reader'
|
||||
import {
|
||||
clearNativeChatTranscriptCache,
|
||||
readNativeChatTranscriptCached
|
||||
} from './transcript-read-cache'
|
||||
|
||||
// These cover the "No transcript found on a just-created session that is actually
|
||||
// alive" race (#8401): Claude Code lazily flushes the first turn, so the session
|
||||
// .jsonl briefly does not exist. A missing file must surface as a RETRYABLE
|
||||
// notFound, never a hard error, and a notFound must not be cached (else the later
|
||||
// real read is masked by the cached miss).
|
||||
|
||||
let tempRoots: string[] = []
|
||||
|
||||
beforeEach(() => {
|
||||
clearNativeChatTranscriptCache()
|
||||
tempRoots = []
|
||||
})
|
||||
|
||||
afterEach(async () => {
|
||||
await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true })))
|
||||
tempRoots = []
|
||||
})
|
||||
|
||||
function jsonLines(records: unknown[]): string {
|
||||
return records.map((record) => JSON.stringify(record)).join('\n')
|
||||
}
|
||||
|
||||
async function emptyProjectsDir(): Promise<string> {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-notfound-'))
|
||||
tempRoots.push(root)
|
||||
return join(root, 'projects')
|
||||
}
|
||||
|
||||
describe('readNativeChatTranscript — notFound vs hard error', () => {
|
||||
it('flags an unresolvable session as notFound (retryable), not a hard error', async () => {
|
||||
const result = await readNativeChatTranscript('claude', 'not-flushed-yet', {
|
||||
claudeProjectsDir: await emptyProjectsDir()
|
||||
})
|
||||
expect('error' in result).toBe(true)
|
||||
expect('notFound' in result && result.notFound).toBe(true)
|
||||
})
|
||||
|
||||
it('flags an ENOENT after resolve (file vanished/not-yet-there) as notFound', async () => {
|
||||
const result = await readNativeChatTranscript('claude', 'sess', {
|
||||
filePath: join(tmpdir(), 'orca-native-chat-8401-does-not-exist.jsonl')
|
||||
})
|
||||
expect('error' in result).toBe(true)
|
||||
expect('notFound' in result && result.notFound).toBe(true)
|
||||
})
|
||||
|
||||
it('keeps a genuine non-ENOENT read error a HARD error (no notFound)', async () => {
|
||||
// A directory path makes createReadStream fail with EISDIR — a real failure,
|
||||
// not a not-yet-flushed file, so it must stay a hard error the UI surfaces.
|
||||
const dir = await mkdtemp(join(tmpdir(), 'orca-native-chat-eisdir-'))
|
||||
tempRoots.push(dir)
|
||||
const result = await readNativeChatTranscript('claude', 'sess', { filePath: dir })
|
||||
expect('error' in result).toBe(true)
|
||||
expect('notFound' in result && result.notFound).toBeFalsy()
|
||||
})
|
||||
})
|
||||
|
||||
describe('readNativeChatTranscriptCached — notFound is retryable and never masks a real read', () => {
|
||||
it('returns a notFound (not a hard error) for a not-yet-created session file', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-notfound-cache-'))
|
||||
tempRoots.push(root)
|
||||
process.env.HOME = root
|
||||
const result = await readNativeChatTranscriptCached('claude', 'absent-session')
|
||||
expect('error' in result).toBe(true)
|
||||
expect('notFound' in result && result.notFound).toBe(true)
|
||||
})
|
||||
|
||||
it('does not cache the notFound miss: a later read after the file appears succeeds', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-notfound-appears-'))
|
||||
tempRoots.push(root)
|
||||
process.env.HOME = root
|
||||
const sessionId = 'lazy-flush'
|
||||
|
||||
// First read races the lazy first flush: the file does not exist yet.
|
||||
const first = await readNativeChatTranscriptCached('claude', sessionId)
|
||||
expect('notFound' in first && first.notFound).toBe(true)
|
||||
|
||||
// Claude Code flushes the transcript moments later.
|
||||
const projectDir = join(root, '.claude', 'projects', '-repo')
|
||||
await mkdir(projectDir, { recursive: true })
|
||||
await writeFile(
|
||||
join(projectDir, `${sessionId}.jsonl`),
|
||||
jsonLines([
|
||||
{
|
||||
type: 'user',
|
||||
uuid: 'u-0',
|
||||
timestamp: '2026-06-01T10:00:00.000Z',
|
||||
message: { role: 'user', content: 'hello there' }
|
||||
}
|
||||
])
|
||||
)
|
||||
|
||||
// The second read must re-resolve and read the now-present file, not serve a
|
||||
// cached miss (which would keep the pane stuck on "No transcript found").
|
||||
const second = await readNativeChatTranscriptCached('claude', sessionId)
|
||||
expect('messages' in second).toBe(true)
|
||||
if (!('messages' in second)) {
|
||||
return
|
||||
}
|
||||
expect(second.messages).toHaveLength(1)
|
||||
})
|
||||
})
|
||||
@@ -91,7 +91,9 @@ export async function readNativeChatTranscriptCached(
|
||||
): Promise<ReadTranscriptResult> {
|
||||
const filePath = await resolveSessionFilePath(agent, sessionId, { transcriptPath })
|
||||
if (!filePath) {
|
||||
return { error: `No transcript found for ${agent} session ${sessionId}` }
|
||||
// Retryable miss (session .jsonl not flushed yet, #8401). Return BEFORE the
|
||||
// cache write below so a later real read is never masked by a cached miss.
|
||||
return { error: `No transcript found for ${agent} session ${sessionId}`, notFound: true }
|
||||
}
|
||||
|
||||
const key = cacheKey(agent, filePath)
|
||||
@@ -104,7 +106,10 @@ export async function readNativeChatTranscriptCached(
|
||||
}
|
||||
|
||||
const result = await readNativeChatTranscript(agent, sessionId, { filePath })
|
||||
if (Number.isFinite(mtimeMs)) {
|
||||
// Never cache a notFound miss (an ENOENT after resolve): the file may appear
|
||||
// moments later, and a cached miss would keep the pane stuck on "No transcript
|
||||
// found" until the mtime changed. Genuine parses/errors cache as before.
|
||||
if (Number.isFinite(mtimeMs) && !('notFound' in result && result.notFound)) {
|
||||
setCached(key, { result, mtimeMs, bytes })
|
||||
}
|
||||
return result
|
||||
|
||||
@@ -9,7 +9,19 @@ import {
|
||||
} from './transcript-line-decoders'
|
||||
import { decodeTranscriptStream } from './transcript-stream-lines'
|
||||
|
||||
export type ReadTranscriptResult = { messages: NativeChatMessage[] } | { error: string }
|
||||
export type ReadTranscriptResult =
|
||||
| { messages: NativeChatMessage[] }
|
||||
// `notFound` marks a RETRYABLE miss (the session .jsonl isn't flushed yet — see
|
||||
// #8401), distinct from a hard read/parse error. Callers may retry a notFound
|
||||
// and must not cache it; a plain `error` is surfaced to the user immediately.
|
||||
| { error: string; notFound?: true }
|
||||
|
||||
// Why: a not-yet-created (or momentarily-vanished) transcript file surfaces as
|
||||
// ENOENT once resolve returns a path. That's the lazy-flush race, not a real
|
||||
// failure, so it becomes a retryable notFound; every other error stays hard.
|
||||
function isFileNotFoundError(err: unknown): boolean {
|
||||
return err instanceof Error && 'code' in err && (err as NodeJS.ErrnoException).code === 'ENOENT'
|
||||
}
|
||||
|
||||
export type ReadTranscriptOptions = ResolveSessionFileOptions & {
|
||||
/** Resolve directly to this file, skipping path discovery (used by tests). */
|
||||
@@ -30,7 +42,9 @@ export async function readNativeChatTranscript(
|
||||
): Promise<ReadTranscriptResult> {
|
||||
const filePath = options.filePath ?? (await resolveSessionFilePath(agent, sessionId, options))
|
||||
if (!filePath) {
|
||||
return { error: `No transcript found for ${agent} session ${sessionId}` }
|
||||
// The file hasn't been flushed yet (or was resolved away). Retryable, not a
|
||||
// hard error — the caller polls/backs off until the lazy first write lands.
|
||||
return { error: `No transcript found for ${agent} session ${sessionId}`, notFound: true }
|
||||
}
|
||||
try {
|
||||
if (agent === 'claude') {
|
||||
@@ -44,6 +58,9 @@ export async function readNativeChatTranscript(
|
||||
}
|
||||
return { error: `Unsupported agent for native chat transcript: ${agent}` }
|
||||
} catch (err) {
|
||||
if (isFileNotFoundError(err)) {
|
||||
return { error: errorMessage(err), notFound: true }
|
||||
}
|
||||
return { error: errorMessage(err) }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { appendFile, mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { appendFile, mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
|
||||
@@ -236,4 +236,66 @@ describe('subscribeNativeChatTranscript', () => {
|
||||
// Must not throw.
|
||||
sub.unsubscribe()
|
||||
})
|
||||
|
||||
it('installs the watcher once a not-yet-created session file appears (#8401 poll)', async () => {
|
||||
// The transcript file does not exist at subscribe time (Claude Code's lazy
|
||||
// first flush). The resolve-poll must pick it up once the agent writes it,
|
||||
// instead of leaving the pane's live tail permanently deaf.
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-poll-'))
|
||||
tempRoots.push(root)
|
||||
const projectsDir = join(root, 'projects')
|
||||
const projectDir = join(projectsDir, '-repo')
|
||||
await mkdir(projectDir, { recursive: true })
|
||||
const before = getActiveNativeChatWatcherCount()
|
||||
const seen: NativeChatMessage[] = []
|
||||
|
||||
const sub = await subscribeNativeChatTranscript({
|
||||
agent: 'claude',
|
||||
sessionId: 'lazy-flush',
|
||||
claudeProjectsDir: projectsDir,
|
||||
onAppend: (messages) => seen.push(...messages),
|
||||
debounceMs: 5,
|
||||
resolvePollMs: 15
|
||||
})
|
||||
|
||||
// No file yet → no watcher installed, but the poll is running.
|
||||
expect(getActiveNativeChatWatcherCount()).toBe(before)
|
||||
|
||||
const target = join(projectDir, 'lazy-flush.jsonl')
|
||||
await writeFile(target, claudeLine('u-1', 'user', 'first'))
|
||||
await waitFor(() => seen.some((m) => m.id === 'u-1'))
|
||||
expect(getActiveNativeChatWatcherCount()).toBe(before + 1)
|
||||
|
||||
// A later append on the now-watched file is tailed incrementally.
|
||||
await appendFile(target, claudeLine('a-1', 'assistant', 'reply'))
|
||||
await waitFor(() => seen.some((m) => m.id === 'a-1'))
|
||||
|
||||
sub.unsubscribe()
|
||||
expect(getActiveNativeChatWatcherCount()).toBe(before)
|
||||
})
|
||||
|
||||
it('stops the resolve-poll on unsubscribe before the file appears (no leak)', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'orca-native-chat-poll-cancel-'))
|
||||
tempRoots.push(root)
|
||||
const projectsDir = join(root, 'projects')
|
||||
await mkdir(projectsDir, { recursive: true })
|
||||
const before = getActiveNativeChatWatcherCount()
|
||||
|
||||
const sub = await subscribeNativeChatTranscript({
|
||||
agent: 'claude',
|
||||
sessionId: 'never-flushed',
|
||||
claudeProjectsDir: projectsDir,
|
||||
onAppend: () => {},
|
||||
resolvePollMs: 10
|
||||
})
|
||||
// Tear down while still polling (file never created).
|
||||
sub.unsubscribe()
|
||||
|
||||
// Even if the file appears later, the stopped poll installs no watcher.
|
||||
const projectDir = join(projectsDir, '-repo')
|
||||
await mkdir(projectDir, { recursive: true })
|
||||
await writeFile(join(projectDir, 'never-flushed.jsonl'), claudeLine('u-1', 'user', 'x'))
|
||||
await new Promise((resolve) => setTimeout(resolve, 40))
|
||||
expect(getActiveNativeChatWatcherCount()).toBe(before)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -29,6 +29,9 @@ export type SubscribeNativeChatTranscriptArgs = ResolveSessionFileOptions & {
|
||||
filePath?: string
|
||||
/** Coalesce window for rapid fs.watch events (ms). Defaults to 40ms. */
|
||||
debounceMs?: number
|
||||
/** Re-resolve interval while the session file doesn't exist yet (ms).
|
||||
* Defaults to RESOLVE_POLL_MS; overridable so tests run fast. */
|
||||
resolvePollMs?: number
|
||||
}
|
||||
|
||||
export type NativeChatTranscriptSubscription = {
|
||||
@@ -42,6 +45,17 @@ export type NativeChatTranscriptSubscription = {
|
||||
// mapping the full reader uses.
|
||||
const DEFAULT_DEBOUNCE_MS = 40
|
||||
|
||||
// Why: a just-created session's .jsonl is written lazily — an agent can take
|
||||
// from a few seconds to minutes to flush its first line (#8401) — so resolve can
|
||||
// return null right after subscribe. Re-resolve on a backing-off interval until
|
||||
// the file appears, then install the watcher; otherwise a live pane opened before
|
||||
// the first flush would never tail. The backoff keeps a long wait cheap (a fixed
|
||||
// fast poll would repeat the directory walk hundreds of times). Only runs while
|
||||
// the file is missing; a non-blank sessionId is required (blank has nothing to
|
||||
// resolve).
|
||||
const RESOLVE_POLL_INITIAL_MS = 1000
|
||||
const RESOLVE_POLL_MAX_MS = 10_000
|
||||
|
||||
// Why: process-wide count of live FSWatchers opened by this module. The U4 leak
|
||||
// test asserts this returns to zero after unsubscribe so a forgotten handle is
|
||||
// caught deterministically rather than relying on OS fd inspection.
|
||||
@@ -125,13 +139,13 @@ async function readAppendedMessages(
|
||||
export async function subscribeNativeChatTranscript(
|
||||
args: SubscribeNativeChatTranscriptArgs
|
||||
): Promise<NativeChatTranscriptSubscription> {
|
||||
const { agent, sessionId, onAppend, debounceMs } = args
|
||||
const { agent, sessionId, onAppend, debounceMs, resolvePollMs } = args
|
||||
const decode = lineDecoderForAgent(agent)
|
||||
const filePath = args.filePath ?? (await resolveSessionFilePath(agent, sessionId, args))
|
||||
const initialPath = args.filePath ?? (await resolveSessionFilePath(agent, sessionId, args))
|
||||
|
||||
if (!filePath || !decode) {
|
||||
// Nothing watchable — return a no-op teardown so callers can unconditionally
|
||||
// unsubscribe without null-checks.
|
||||
if (!decode) {
|
||||
// Unknown agent — nothing to decode. No-op teardown so callers can
|
||||
// unconditionally unsubscribe without null-checks.
|
||||
return { unsubscribe: () => {} }
|
||||
}
|
||||
|
||||
@@ -145,11 +159,19 @@ export async function subscribeNativeChatTranscript(
|
||||
let reading = false
|
||||
let pendingReadRequested = false
|
||||
let debounceTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let resolvePollTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let watcher: FSWatcher | null = null
|
||||
// The resolved path, set once the file exists (immediately, or via the poll).
|
||||
let currentFilePath: string | null = initialPath
|
||||
|
||||
async function drain(): Promise<void> {
|
||||
if (closed) {
|
||||
return
|
||||
}
|
||||
const path = currentFilePath
|
||||
if (path === null) {
|
||||
return
|
||||
}
|
||||
if (reading) {
|
||||
// A read is already in flight; mark that another pass is needed so rapid
|
||||
// successive appends coalesce without dropping the trailing one.
|
||||
@@ -161,12 +183,12 @@ export async function subscribeNativeChatTranscript(
|
||||
do {
|
||||
pendingReadRequested = false
|
||||
try {
|
||||
const currentSize = await fileSize(filePath!)
|
||||
const currentSize = await fileSize(path)
|
||||
if (currentSize < offset) {
|
||||
// Rotation/replacement/truncation: re-read from the top.
|
||||
offset = 0
|
||||
}
|
||||
const { messages, consumedTo } = await readAppendedMessages(filePath!, offset, decode!)
|
||||
const { messages, consumedTo } = await readAppendedMessages(path, offset, decode!)
|
||||
offset = consumedTo
|
||||
if (!closed && messages.length > 0) {
|
||||
onAppend(messages)
|
||||
@@ -196,19 +218,76 @@ export async function subscribeNativeChatTranscript(
|
||||
}, debounceMs ?? DEFAULT_DEBOUNCE_MS)
|
||||
}
|
||||
|
||||
let watcher: FSWatcher
|
||||
try {
|
||||
watcher = watch(filePath, scheduleDrain)
|
||||
} catch {
|
||||
// File vanished between resolve and watch — return a no-op teardown.
|
||||
return { unsubscribe: () => {} }
|
||||
function stopResolvePoll(): void {
|
||||
if (resolvePollTimer) {
|
||||
clearTimeout(resolvePollTimer)
|
||||
resolvePollTimer = null
|
||||
}
|
||||
}
|
||||
activeWatcherCount++
|
||||
|
||||
// Why: on some platforms fs.watch can miss the very first append that lands
|
||||
// between offset-seed and watcher install. Kick one debounced drain so a
|
||||
// turn written immediately after subscribe is still picked up.
|
||||
scheduleDrain()
|
||||
// Install the fs watcher on a now-present file. Returns false when watch()
|
||||
// throws (the file vanished again), so the poll caller keeps trying.
|
||||
function installWatcher(path: string): boolean {
|
||||
if (closed) {
|
||||
return false
|
||||
}
|
||||
let installed: FSWatcher
|
||||
try {
|
||||
installed = watch(path, scheduleDrain)
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
watcher = installed
|
||||
currentFilePath = path
|
||||
activeWatcherCount++
|
||||
// Why: on some platforms fs.watch can miss the very first append that lands
|
||||
// between offset-seed and watcher install. Kick one debounced drain so a
|
||||
// turn written immediately after install is still picked up.
|
||||
scheduleDrain()
|
||||
return true
|
||||
}
|
||||
|
||||
// Self-rescheduling backoff poll (setTimeout, not setInterval) so a slow resolve
|
||||
// can't overlap the next tick and the delay grows toward the cap on a long wait.
|
||||
function scheduleResolvePoll(delayMs: number): void {
|
||||
if (closed || watcher) {
|
||||
return
|
||||
}
|
||||
resolvePollTimer = setTimeout(() => {
|
||||
resolvePollTimer = null
|
||||
if (closed || watcher) {
|
||||
return
|
||||
}
|
||||
void resolveSessionFilePath(agent, sessionId, args)
|
||||
.then((resolved) => {
|
||||
// Re-check after the async resolve: unsubscribe may have run, or a
|
||||
// prior tick may have already installed the watcher (race-safe).
|
||||
if (closed || watcher) {
|
||||
return
|
||||
}
|
||||
if (resolved && installWatcher(resolved)) {
|
||||
return
|
||||
}
|
||||
scheduleResolvePoll(Math.min(delayMs * 2, RESOLVE_POLL_MAX_MS))
|
||||
})
|
||||
.catch(() => {
|
||||
// A transient resolve failure must not kill the poll; keep backing off.
|
||||
scheduleResolvePoll(Math.min(delayMs * 2, RESOLVE_POLL_MAX_MS))
|
||||
})
|
||||
}, delayMs)
|
||||
}
|
||||
|
||||
if (currentFilePath !== null) {
|
||||
if (!installWatcher(currentFilePath)) {
|
||||
// File vanished between resolve and watch — preserve the original no-op.
|
||||
return { unsubscribe: () => {} }
|
||||
}
|
||||
} else if (sessionId.trim()) {
|
||||
// #8401: the session .jsonl isn't flushed yet. Re-resolve on a backing-off
|
||||
// interval and install the watcher once it appears (then re-read from the
|
||||
// top). A blank sessionId has nothing to resolve, so it never polls.
|
||||
scheduleResolvePoll(resolvePollMs ?? RESOLVE_POLL_INITIAL_MS)
|
||||
}
|
||||
|
||||
return {
|
||||
unsubscribe: () => {
|
||||
@@ -216,12 +295,16 @@ export async function subscribeNativeChatTranscript(
|
||||
return
|
||||
}
|
||||
closed = true
|
||||
stopResolvePoll()
|
||||
if (debounceTimer) {
|
||||
clearTimeout(debounceTimer)
|
||||
debounceTimer = null
|
||||
}
|
||||
watcher.close()
|
||||
activeWatcherCount--
|
||||
if (watcher) {
|
||||
watcher.close()
|
||||
watcher = null
|
||||
activeWatcherCount--
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -813,7 +813,11 @@ export type AiVaultApi = {
|
||||
onWindowFocused: (callback: () => void) => () => void
|
||||
}
|
||||
|
||||
export type NativeChatReadSessionResult = { messages: NativeChatMessage[] } | { error: string }
|
||||
export type NativeChatReadSessionResult =
|
||||
| { messages: NativeChatMessage[] }
|
||||
// `notFound` marks a RETRYABLE miss (the session .jsonl isn't flushed yet,
|
||||
// #8401); the renderer stays in 'loading' and retries instead of hard-failing.
|
||||
| { error: string; notFound?: true }
|
||||
|
||||
/** Messages appended to a live-tailed transcript since the previous emit. */
|
||||
export type NativeChatAppendedMessages = NativeChatMessage[]
|
||||
|
||||
@@ -238,6 +238,41 @@ describe('useNativeChatLiveSession — transport routing', () => {
|
||||
expect(latest?.error).toBe('runtime too old')
|
||||
})
|
||||
|
||||
it('retries a notFound read on backoff instead of surfacing an error (#8401)', async () => {
|
||||
vi.useFakeTimers()
|
||||
try {
|
||||
const transport = getMockTransport('env-1')
|
||||
// First read races the lazy first flush → retryable notFound; the flush
|
||||
// lands before the backoff retry, which then reads the real transcript.
|
||||
transport.readSession
|
||||
.mockResolvedValueOnce({ error: 'No transcript found', notFound: true })
|
||||
.mockResolvedValueOnce({ messages: [assistant('a-1', 'flushed')] })
|
||||
|
||||
await render({
|
||||
paneKey: PANE,
|
||||
agent: AGENT,
|
||||
sessionId: SESSION,
|
||||
runtimeEnvironmentId: 'env-1'
|
||||
})
|
||||
|
||||
// notFound must NOT become an error: the pane stays loading and retries.
|
||||
expect(latest?.status).toBe('loading')
|
||||
expect(latest?.status).not.toBe('error')
|
||||
expect(transport.readSession).toHaveBeenCalledTimes(1)
|
||||
|
||||
// Advance past the first backoff step so the retry fires and resolves.
|
||||
await act(async () => {
|
||||
await vi.advanceTimersByTimeAsync(1000)
|
||||
})
|
||||
|
||||
expect(transport.readSession).toHaveBeenCalledTimes(2)
|
||||
expect(latest?.status).toBe('ready')
|
||||
expect(latest?.messages.map((m) => m.id)).toContain('a-1')
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('never calls the transport when there is no session id', async () => {
|
||||
await render({ paneKey: PANE, agent: AGENT, sessionId: null, runtimeEnvironmentId: 'env-1' })
|
||||
|
||||
|
||||
@@ -53,6 +53,14 @@ export type NativeChatLiveSession = NativeChatSession & {
|
||||
// Stable empty-base reference so a non-ready read doesn't churn the base axis.
|
||||
const EMPTY_MESSAGES: readonly NativeChatMessage[] = []
|
||||
|
||||
// A just-created session's transcript is written lazily by the agent, so the
|
||||
// first read can race the first flush and return notFound (#8401). Retry with a
|
||||
// bounded exponential backoff (kept in 'loading', never 'error') until the file
|
||||
// appears, so the pane recovers on its own instead of sticking on "No transcript
|
||||
// found". The cap keeps a genuinely-never-appearing session to a light poll.
|
||||
const NOT_FOUND_RETRY_INITIAL_MS = 1000
|
||||
const NOT_FOUND_RETRY_MAX_MS = 10_000
|
||||
|
||||
/** True when `whole`'s first `len` entries are referentially identical to
|
||||
* `prefix` — i.e. `whole` is `prefix` extended at the tail, so the incremental
|
||||
* assembler can splice just the suffix instead of resetting. */
|
||||
@@ -167,25 +175,39 @@ export function useNativeChatLiveSession(
|
||||
setAppended([])
|
||||
setHasMore(false)
|
||||
|
||||
void transport
|
||||
.readSession(agent, sessionId, limitRef.current, transcriptPath ?? undefined)
|
||||
.then((result) => {
|
||||
if (cancelled) {
|
||||
return
|
||||
}
|
||||
if (result && 'error' in result) {
|
||||
setRead({ phase: 'error', error: result.error })
|
||||
return
|
||||
}
|
||||
const messages = result?.messages ?? []
|
||||
setRead({ phase: 'ready', messages })
|
||||
setHasMore(hasMoreNativeChatHistory(messages.length, limitRef.current))
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
if (!cancelled) {
|
||||
setRead({ phase: 'error', error: err instanceof Error ? err.message : String(err) })
|
||||
}
|
||||
})
|
||||
// Backoff state for the notFound retry; both are cleared on teardown below.
|
||||
let retryTimer: ReturnType<typeof setTimeout> | null = null
|
||||
let retryDelayMs = NOT_FOUND_RETRY_INITIAL_MS
|
||||
|
||||
const attemptRead = (): void => {
|
||||
void transport
|
||||
.readSession(agent, sessionId, limitRef.current, transcriptPath ?? undefined)
|
||||
.then((result) => {
|
||||
if (cancelled) {
|
||||
return
|
||||
}
|
||||
if (result && 'error' in result) {
|
||||
if (result.notFound) {
|
||||
// The transcript isn't flushed yet — stay in 'loading' and retry
|
||||
// rather than hard-failing. The live watcher also polls for it.
|
||||
retryTimer = setTimeout(attemptRead, retryDelayMs)
|
||||
retryDelayMs = Math.min(retryDelayMs * 2, NOT_FOUND_RETRY_MAX_MS)
|
||||
return
|
||||
}
|
||||
setRead({ phase: 'error', error: result.error })
|
||||
return
|
||||
}
|
||||
const messages = result?.messages ?? []
|
||||
setRead({ phase: 'ready', messages })
|
||||
setHasMore(hasMoreNativeChatHistory(messages.length, limitRef.current))
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
if (!cancelled) {
|
||||
setRead({ phase: 'error', error: err instanceof Error ? err.message : String(err) })
|
||||
}
|
||||
})
|
||||
}
|
||||
attemptRead()
|
||||
|
||||
const subscriptionId = nextSubscriptionId()
|
||||
const unsubscribe = transport.subscribe(
|
||||
@@ -208,6 +230,12 @@ export function useNativeChatLiveSession(
|
||||
|
||||
return () => {
|
||||
cancelled = true
|
||||
if (retryTimer) {
|
||||
// Stop a pending notFound retry so a session swap/unmount can't fire a
|
||||
// stale read against the old session id.
|
||||
clearTimeout(retryTimer)
|
||||
retryTimer = null
|
||||
}
|
||||
// Desktop returns a sync unsubscribe fn; the web RPC bridge returns a
|
||||
// Promise instead (and can't deliver streaming callbacks). Calling a
|
||||
// Promise as a function crashed the whole chat view, so resolve it first
|
||||
|
||||
Reference in New Issue
Block a user