mirror of
https://github.com/stablyai/orca.git
synced 2026-10-23 08:03:39 +00:00
fix(runtime): wait for queued writes in park snapshots
When serializing recovery snapshots for hidden output, repeatedly drain the headless terminal write chain to capture all queued data (including SSH suffixes). Fall back to the provider buffer if writes never settle, preventing incomplete snapshots.
This commit is contained in:
@@ -10643,6 +10643,104 @@ describe('OrcaRuntimeService', () => {
|
||||
expect(snapshot?.lastTitle).toBe('Codex ready')
|
||||
})
|
||||
|
||||
it('includes a queued SSH suffix admitted across multiple recovery drains', async () => {
|
||||
const { runtime } = createSideEffectRuntime()
|
||||
runtime.onPtyData('pty-1', 'seed\r\n', 100)
|
||||
const state = (
|
||||
runtime as unknown as {
|
||||
headlessTerminals: Map<
|
||||
string,
|
||||
{
|
||||
emulator: { write: (data: string) => Promise<void> }
|
||||
writeChain: Promise<void>
|
||||
}
|
||||
>
|
||||
}
|
||||
).headlessTerminals.get('pty-1')!
|
||||
await state.writeChain
|
||||
|
||||
const originalWrite = state.emulator.write.bind(state.emulator)
|
||||
let releaseBlockedWrite: () => void = () => {}
|
||||
let markBlockedWriteStarted: () => void = () => {}
|
||||
const blockedWriteStarted = new Promise<void>((resolve) => {
|
||||
markBlockedWriteStarted = resolve
|
||||
})
|
||||
const blockedWrite = new Promise<void>((resolve) => {
|
||||
releaseBlockedWrite = resolve
|
||||
})
|
||||
vi.spyOn(state.emulator, 'write').mockImplementation(async (data) => {
|
||||
if (data === 'blocked\r\n') {
|
||||
markBlockedWriteStarted()
|
||||
await blockedWrite
|
||||
}
|
||||
await originalWrite(data)
|
||||
if (data === 'blocked\r\n') {
|
||||
runtime.onPtyData('pty-1', 'newer tail\r\n', 102)
|
||||
} else if (data === 'newer tail\r\n') {
|
||||
runtime.onPtyData('pty-1', 'newest tail\r\n', 103)
|
||||
}
|
||||
})
|
||||
|
||||
runtime.onPtyData('pty-1', 'blocked\r\n', 101)
|
||||
const snapshotPromise = runtime.serializeHiddenOutputRecoveryBuffer('pty-1')
|
||||
await blockedWriteStarted
|
||||
releaseBlockedWrite()
|
||||
|
||||
await expect(snapshotPromise).resolves.toMatchObject({
|
||||
data: expect.stringContaining('newest tail'),
|
||||
seq: 'seed\r\nblocked\r\nnewer tail\r\nnewest tail\r\n'.length
|
||||
})
|
||||
})
|
||||
|
||||
it('falls back instead of returning a recovery snapshot that never stabilizes', async () => {
|
||||
const { runtime } = createSideEffectRuntime()
|
||||
const serializeBuffer = vi.fn().mockResolvedValue({
|
||||
data: 'stale renderer content\r\n',
|
||||
cols: 80,
|
||||
rows: 24
|
||||
})
|
||||
const serializeProviderBuffer = vi.fn().mockResolvedValue(null)
|
||||
runtime.onPtyData('pty-1', 'seed\r\n', 100)
|
||||
const state = (
|
||||
runtime as unknown as {
|
||||
headlessTerminals: Map<
|
||||
string,
|
||||
{
|
||||
emulator: { write: (data: string) => Promise<void> }
|
||||
writeChain: Promise<void>
|
||||
}
|
||||
>
|
||||
}
|
||||
).headlessTerminals.get('pty-1')!
|
||||
await state.writeChain
|
||||
runtime.setPtyController({
|
||||
write: () => true,
|
||||
kill: () => true,
|
||||
getForegroundProcess: async () => null,
|
||||
serializeBuffer,
|
||||
serializeProviderBuffer,
|
||||
hasRendererSerializer: () => true
|
||||
})
|
||||
|
||||
const originalWrite = state.emulator.write.bind(state.emulator)
|
||||
let queuedWrites = 0
|
||||
vi.spyOn(state.emulator, 'write').mockImplementation(async (data) => {
|
||||
await originalWrite(data)
|
||||
if (queuedWrites < 66) {
|
||||
queuedWrites += 1
|
||||
runtime.onPtyData('pty-1', `queued-${queuedWrites}\r\n`, 100 + queuedWrites)
|
||||
}
|
||||
})
|
||||
|
||||
runtime.onPtyData('pty-1', 'queued-0\r\n', 101)
|
||||
|
||||
await expect(runtime.serializeHiddenOutputRecoveryBuffer('pty-1')).resolves.toBeNull()
|
||||
expect(serializeProviderBuffer).toHaveBeenCalledOnce()
|
||||
expect(serializeBuffer).not.toHaveBeenCalled()
|
||||
await state.writeChain
|
||||
expect(queuedWrites).toBe(66)
|
||||
})
|
||||
|
||||
it('returns a title-only replay snapshot and never historical attention', () => {
|
||||
const { runtime } = createSideEffectRuntime()
|
||||
syncSinglePty(runtime)
|
||||
|
||||
@@ -12704,13 +12704,18 @@ export class OrcaRuntimeService {
|
||||
scrollbackAnsi?: string
|
||||
pendingEscapeTailAnsi?: string
|
||||
} | null> {
|
||||
const hadHeadlessState = this.headlessTerminals.has(ptyId)
|
||||
const headlessSnapshot = await this.serializeHeadlessTerminalBuffer(ptyId, {
|
||||
...opts,
|
||||
includeEmpty: true
|
||||
includeEmpty: true,
|
||||
requireSettled: true
|
||||
})
|
||||
if (headlessSnapshot) {
|
||||
return headlessSnapshot
|
||||
}
|
||||
if (hadHeadlessState || this.headlessTerminals.has(ptyId)) {
|
||||
return this.serializeProviderTerminalBuffer(ptyId, opts)
|
||||
}
|
||||
// Why: hidden-output recovery is initiated by the desktop renderer. If the
|
||||
// runtime has not built headless state yet, the mounted xterm is still the
|
||||
// best available state and avoids a false "snapshot unavailable" result.
|
||||
@@ -13621,7 +13626,7 @@ export class OrcaRuntimeService {
|
||||
|
||||
private async serializeHeadlessTerminalBuffer(
|
||||
ptyId: string,
|
||||
opts: { scrollbackRows?: number; includeEmpty?: boolean } = {}
|
||||
opts: { scrollbackRows?: number; includeEmpty?: boolean; requireSettled?: boolean } = {}
|
||||
): Promise<{
|
||||
data: string
|
||||
cols: number
|
||||
@@ -13643,7 +13648,26 @@ export class OrcaRuntimeService {
|
||||
if (!state) {
|
||||
return null
|
||||
}
|
||||
await state.writeChain
|
||||
let drainedWriteChain = state.writeChain
|
||||
await drainedWriteChain
|
||||
if (opts.requireSettled) {
|
||||
let followupDrains = 0
|
||||
while (
|
||||
state.writeChain !== drainedWriteChain &&
|
||||
followupDrains < HEADLESS_SNAPSHOT_FOLLOWUP_DRAIN_MAX
|
||||
) {
|
||||
drainedWriteChain = state.writeChain
|
||||
await drainedWriteChain
|
||||
followupDrains += 1
|
||||
}
|
||||
// Why: a partial park snapshot outranks relay replay; fall back instead of dropping a still-queued SSH suffix.
|
||||
if (state.writeChain !== drainedWriteChain) {
|
||||
return null
|
||||
}
|
||||
}
|
||||
if (this.headlessTerminals.get(ptyId) !== state) {
|
||||
return null
|
||||
}
|
||||
// Why: normal history is separated from an active alternate frame, so the
|
||||
// caller's scrollback policy can be honored without painting it into alt.
|
||||
const isAlternateScreen = state.emulator.isAlternateScreen
|
||||
@@ -38641,6 +38665,7 @@ const MAX_TERMINAL_PREVIEW_CHARS = 32 * 1024
|
||||
export const AUTHORITATIVE_TERMINAL_SNAPSHOT_TIMEOUT_MS = 8_000
|
||||
const VISIBLE_TERMINAL_SNAPSHOT_TIMEOUT_MS = 750
|
||||
const VISIBLE_TERMINAL_SNAPSHOT_RETRY_MS = 1_000
|
||||
const HEADLESS_SNAPSHOT_FOLLOWUP_DRAIN_MAX = 64
|
||||
const MAX_PREVIEW_LINES = 6
|
||||
const MAX_PREVIEW_CHARS = 300
|
||||
const WORKTREE_STATUS_PRIORITY: Record<RuntimeWorktreeStatus, number> = {
|
||||
|
||||
Reference in New Issue
Block a user