mirror of
https://github.com/stablyai/orca.git
synced 2026-10-28 00:02:54 +00:00
fix(native-chat): harden remote transcript recovery
This commit is contained in:
@@ -62,23 +62,24 @@ function memoryProvider(files: Map<string, Buffer>) {
|
||||
return { bytes, bytesRead: bytes.length }
|
||||
}
|
||||
)
|
||||
const stat = vi.fn(async (filePath: string): Promise<FileStat> => {
|
||||
const bytes = files.get(filePath)
|
||||
if (!bytes) {
|
||||
throw Object.assign(new Error(`ENOENT: no such file or directory, stat '${filePath}'`), {
|
||||
code: 'ENOENT'
|
||||
})
|
||||
}
|
||||
return { size: bytes.length, type: 'file', mtime: 1, mtimeMs: 1, dev: 7, ino: 11 }
|
||||
})
|
||||
const provider = {
|
||||
async stat(filePath: string): Promise<FileStat> {
|
||||
const bytes = files.get(filePath)
|
||||
if (!bytes) {
|
||||
throw Object.assign(new Error(`ENOENT: no such file or directory, stat '${filePath}'`), {
|
||||
code: 'ENOENT'
|
||||
})
|
||||
}
|
||||
return { size: bytes.length, type: 'file', mtime: 1, mtimeMs: 1, dev: 7, ino: 11 }
|
||||
},
|
||||
stat,
|
||||
readFile,
|
||||
readFileRange,
|
||||
async supportsFileRangeRead() {
|
||||
return true
|
||||
}
|
||||
} as unknown as IFilesystemProvider
|
||||
return { provider, readFile, readFileRange }
|
||||
return { provider, readFile, readFileRange, stat }
|
||||
}
|
||||
|
||||
function routedArgs(transcriptPath: string): {
|
||||
@@ -128,19 +129,24 @@ describe('native chat SSH transcript host routing (#13663)', () => {
|
||||
const transcriptPath = join(root, 'session.jsonl')
|
||||
await writeFile(transcriptPath, claudeLine('desktop-poison', 'wrong host'))
|
||||
const files = new Map([[transcriptPath, Buffer.from(claudeLine('remote-real', 'right host'))]])
|
||||
const { provider, readFile, readFileRange } = memoryProvider(files)
|
||||
const { provider, readFile, readFileRange, stat } = memoryProvider(files)
|
||||
mocks.getProvider.mockReturnValue(provider)
|
||||
const controller = new AbortController()
|
||||
|
||||
const result = await readNativeChatTranscriptTail({
|
||||
...routedArgs(transcriptPath),
|
||||
limit: 40
|
||||
} as Parameters<typeof readNativeChatTranscriptTail>[0])
|
||||
const result = await readNativeChatTranscriptTail(
|
||||
{
|
||||
...routedArgs(transcriptPath),
|
||||
limit: 40
|
||||
} as Parameters<typeof readNativeChatTranscriptTail>[0],
|
||||
controller.signal
|
||||
)
|
||||
|
||||
expect(result).toMatchObject({ messages: [{ id: 'remote-real' }], hasMore: false })
|
||||
expect(result).not.toMatchObject({ messages: [{ id: 'desktop-poison' }] })
|
||||
expect(mocks.getProvider).toHaveBeenCalledWith('ssh-owner')
|
||||
expect(readFile).not.toHaveBeenCalled()
|
||||
expect(readFileRange).toHaveBeenCalled()
|
||||
expect(stat).toHaveBeenCalledWith(transcriptPath, { signal: controller.signal })
|
||||
})
|
||||
|
||||
it('retries consecutive ranged snapshots when the remote file is replaced mid-read', async () => {
|
||||
@@ -509,6 +515,19 @@ describe('native chat SSH transcript host routing (#13663)', () => {
|
||||
expect(getActiveNativeChatWatcherCount()).toBe(before + 1)
|
||||
expect(snapshots.some((snapshot) => snapshot.ids.includes('before-loss'))).toBe(true)
|
||||
expect(snapshots.at(-1)?.ids).toEqual(['after-restore'])
|
||||
|
||||
const firstOutageFrames = snapshots.filter(
|
||||
(snapshot) => snapshot.error === 'Transcript unverifiable on the remote host'
|
||||
).length
|
||||
changeProvider(undefined)
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
snapshots.filter(
|
||||
(snapshot) => snapshot.error === 'Transcript unverifiable on the remote host'
|
||||
)
|
||||
).toHaveLength(firstOutageFrames + 1)
|
||||
)
|
||||
expect(getActiveNativeChatWatcherCount()).toBe(before)
|
||||
} finally {
|
||||
subscription.unsubscribe()
|
||||
}
|
||||
@@ -544,10 +563,31 @@ describe('native chat SSH transcript host routing (#13663)', () => {
|
||||
}
|
||||
})
|
||||
|
||||
it('emits one unresolved advisory while an exact remote path is still missing', async () => {
|
||||
it('emits one advisory while a registered provider remains unreachable', async () => {
|
||||
const transcriptPath = '/tmp/unreachable-provider-session.jsonl'
|
||||
const { provider, stat } = memoryProvider(new Map())
|
||||
stat.mockRejectedValue(Object.assign(new Error('connection lost'), { code: 'CONNECTION_LOST' }))
|
||||
mocks.getProvider.mockReturnValue(provider)
|
||||
const errors: (string | undefined)[] = []
|
||||
const subscription = await subscribeNativeChatTranscript({
|
||||
...routedArgs(transcriptPath),
|
||||
onInitialSnapshot: (_messages, _hasMore, _beforeOffset, error) => errors.push(error),
|
||||
onAppend: () => {},
|
||||
resolvePollIntervalMs: 5
|
||||
})
|
||||
|
||||
try {
|
||||
await vi.waitFor(() => expect(stat.mock.calls.length).toBeGreaterThanOrEqual(3))
|
||||
expect(errors).toEqual(['Transcript unverifiable on the remote host'])
|
||||
} finally {
|
||||
subscription.unsubscribe()
|
||||
}
|
||||
})
|
||||
|
||||
it('replaces an unverifiable advisory after the remote host becomes reachable', async () => {
|
||||
const transcriptPath = '/tmp/not-yet-written.jsonl'
|
||||
const { provider } = memoryProvider(new Map())
|
||||
mocks.getProvider.mockReturnValue(provider)
|
||||
mocks.getProvider.mockReturnValue(undefined)
|
||||
const errors: (string | undefined)[] = []
|
||||
const subscription = await subscribeNativeChatTranscript({
|
||||
...routedArgs(transcriptPath),
|
||||
@@ -558,9 +598,12 @@ describe('native chat SSH transcript host routing (#13663)', () => {
|
||||
})
|
||||
|
||||
try {
|
||||
await vi.waitFor(() => expect(errors).toEqual(['Transcript unverifiable on the remote host']))
|
||||
changeProvider(provider)
|
||||
await vi.waitFor(() =>
|
||||
expect(errors).toEqual([
|
||||
'No transcript found for this session on this machine. If the agent runs on a remote host its transcript lives there; otherwise it may not have been written yet.'
|
||||
'Transcript unverifiable on the remote host',
|
||||
'No transcript found for this session on its remote host. It may not have been written yet.'
|
||||
])
|
||||
)
|
||||
} finally {
|
||||
|
||||
@@ -44,7 +44,7 @@ export async function createSshTranscriptRangeFs(
|
||||
async function stat(filePath: string, statSignal?: AbortSignal): Promise<TranscriptFileStamp> {
|
||||
statSignal?.throwIfAborted()
|
||||
requireSameProvider(connectionId, providerSnapshot)
|
||||
const stamp = await provider.stat(filePath)
|
||||
const stamp = await provider.stat(filePath, { signal: statSignal })
|
||||
requireSameProvider(connectionId, providerSnapshot)
|
||||
const mtimeMs = stamp.mtimeMs ?? stamp.mtime
|
||||
return {
|
||||
|
||||
@@ -4,8 +4,8 @@ import { createSshTranscriptRangeFs } from './ssh-transcript-range-fs'
|
||||
import {
|
||||
DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS,
|
||||
isTranscriptHostUnverifiableError,
|
||||
TRANSCRIPT_UNVERIFIABLE_MESSAGE,
|
||||
UNRESOLVED_TRANSCRIPT_MESSAGE
|
||||
REMOTE_UNRESOLVED_TRANSCRIPT_MESSAGE,
|
||||
TRANSCRIPT_UNVERIFIABLE_MESSAGE
|
||||
} from './transcript-host-verdict'
|
||||
import { nativeChatLineDecoderForAgent } from './transcript-tail-reader'
|
||||
import { installTranscriptWatcher } from './transcript-watch-engine'
|
||||
@@ -87,7 +87,7 @@ export function subscribeSshNativeChatTranscript(
|
||||
}
|
||||
unresolvedNoticeEmitted = true
|
||||
try {
|
||||
args.onInitialSnapshot([], false, 0, UNRESOLVED_TRANSCRIPT_MESSAGE)
|
||||
args.onInitialSnapshot([], false, 0, REMOTE_UNRESOLVED_TRANSCRIPT_MESSAGE)
|
||||
} catch {
|
||||
// The advisory cannot own retry liveness when its subscriber is closing.
|
||||
}
|
||||
@@ -147,6 +147,7 @@ export function subscribeSshNativeChatTranscript(
|
||||
result?.unsubscribe()
|
||||
return
|
||||
}
|
||||
unverifiableEmitted = false
|
||||
if (result) {
|
||||
installed = result
|
||||
retryMs = args.resolvePollIntervalMs ?? INITIAL_RETRY_MS
|
||||
|
||||
@@ -6,6 +6,8 @@ export const TRANSCRIPT_UNVERIFIABLE_MESSAGE = 'Transcript unverifiable on the r
|
||||
export const DEFAULT_TRANSCRIPT_UNRESOLVED_NOTICE_MS = 60_000
|
||||
export const UNRESOLVED_TRANSCRIPT_MESSAGE =
|
||||
'No transcript found for this session on this machine. If the agent runs on a remote host its transcript lives there; otherwise it may not have been written yet.'
|
||||
export const REMOTE_UNRESOLVED_TRANSCRIPT_MESSAGE =
|
||||
'No transcript found for this session on its remote host. It may not have been written yet.'
|
||||
|
||||
export class TranscriptHostUnverifiableError extends Error {
|
||||
readonly verdict = 'unverifiable' as const
|
||||
|
||||
@@ -81,7 +81,7 @@ export type IFilesystemProvider = {
|
||||
): Promise<FileStat>
|
||||
writeFileBase64(filePath: string, contentBase64: string): Promise<void>
|
||||
writeFileBase64Chunk(filePath: string, contentBase64: string, append: boolean): Promise<void>
|
||||
stat(filePath: string): Promise<FileStat>
|
||||
stat(filePath: string, options?: { signal?: AbortSignal }): Promise<FileStat>
|
||||
lstat?(filePath: string): Promise<FileStat>
|
||||
deletePath(targetPath: string, recursive?: boolean): Promise<void>
|
||||
createFile(filePath: string): Promise<void>
|
||||
|
||||
@@ -15,6 +15,17 @@ function providerWith(request: ReturnType<typeof vi.fn>): SshFilesystemProvider
|
||||
return new SshFilesystemProvider('conn-1', mux)
|
||||
}
|
||||
|
||||
describe('SshFilesystemProvider.stat', () => {
|
||||
it('forwards an abort signal to the multiplexer', async () => {
|
||||
const controller = new AbortController()
|
||||
const request = vi.fn().mockResolvedValue({ size: 0, type: 'file', mtime: 1 })
|
||||
await providerWith(request).stat('/x.jsonl', { signal: controller.signal })
|
||||
expect(request).toHaveBeenCalledWith('fs.stat', expect.anything(), {
|
||||
signal: controller.signal
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
describe('SshFilesystemProvider.readFileRange', () => {
|
||||
it('decodes the base64 window the relay returned', async () => {
|
||||
const payload = Buffer.from('hello')
|
||||
|
||||
@@ -209,8 +209,11 @@ export class SshFilesystemProvider implements IFilesystemProvider {
|
||||
}
|
||||
}
|
||||
|
||||
async stat(filePath: string): Promise<FileStat> {
|
||||
return (await this.mux.request('fs.stat', { filePath })) as FileStat
|
||||
async stat(filePath: string, options?: { signal?: AbortSignal }): Promise<FileStat> {
|
||||
const request = options?.signal
|
||||
? this.mux.request('fs.stat', { filePath }, options)
|
||||
: this.mux.request('fs.stat', { filePath })
|
||||
return (await request) as FileStat
|
||||
}
|
||||
|
||||
async lstat(filePath: string): Promise<FileStat> {
|
||||
|
||||
Reference in New Issue
Block a user