mirror of
https://github.com/stablyai/orca.git
synced 2026-10-28 16:02:55 +00:00
Harden restored PTY snapshot sequencing
Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
@@ -506,6 +506,7 @@ describe('DaemonPtyAdapter (IPtyProvider)', () => {
|
||||
expect(second.launchAgent).toBe('droid')
|
||||
expect(second.snapshot).toBeDefined()
|
||||
expect(second.snapshot).toContain('hello from shell')
|
||||
expect(second.snapshotOutputSequence).toBe('hello from shell\r\n'.length)
|
||||
})
|
||||
|
||||
it('includes rehydrateSequences in snapshot when terminal modes are active', async () => {
|
||||
|
||||
@@ -410,6 +410,9 @@ export class DaemonPtyAdapter implements IPtyProvider {
|
||||
snapshot: snapshotPayload,
|
||||
snapshotCols: result.snapshot.cols,
|
||||
snapshotRows: result.snapshot.rows,
|
||||
...(typeof result.snapshot.outputSequence === 'number'
|
||||
? { snapshotOutputSequence: result.snapshot.outputSequence }
|
||||
: {}),
|
||||
...(typeof kittyKeyboardFlags === 'number' && kittyKeyboardFlags > 0
|
||||
? { snapshotKittyKeyboardFlags: kittyKeyboardFlags }
|
||||
: {}),
|
||||
|
||||
@@ -10437,6 +10437,42 @@ describe('registerPtyHandlers', () => {
|
||||
expect(mockProc.proc.write).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('anchors runtime output sequencing from a provider reattach snapshot', async () => {
|
||||
setLocalPtyProvider({
|
||||
spawn: vi.fn(async () => ({
|
||||
id: 'pty-restored',
|
||||
isReattach: true,
|
||||
snapshotOutputSequence: 900
|
||||
})),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
kill: vi.fn(),
|
||||
shutdown: vi.fn(),
|
||||
onData: vi.fn(() => vi.fn()),
|
||||
onExit: vi.fn(() => vi.fn()),
|
||||
listProcesses: vi.fn(async () => []),
|
||||
getForegroundProcess: vi.fn(async () => null)
|
||||
} as never)
|
||||
const runtime = {
|
||||
setPtyController: vi.fn(),
|
||||
noteTerminalSpawnCommand: vi.fn(),
|
||||
anchorPtyOutputSequenceFromProviderSnapshot: vi.fn(),
|
||||
onPtySpawned: vi.fn(),
|
||||
onPtyData: vi.fn(),
|
||||
onPtyExit: vi.fn(),
|
||||
createPreAllocatedTerminalHandle: vi.fn(() => null),
|
||||
preAllocateHandleForPty: vi.fn()
|
||||
}
|
||||
registerPtyHandlers(mainWindow as never, runtime as never)
|
||||
|
||||
await handlers.get('pty:spawn')!(null, { cols: 80, rows: 24 })
|
||||
|
||||
expect(runtime.anchorPtyOutputSequenceFromProviderSnapshot).toHaveBeenCalledWith(
|
||||
'pty-restored',
|
||||
900
|
||||
)
|
||||
})
|
||||
|
||||
it('seeds headless terminal state with cold-restore cwd metadata', async () => {
|
||||
const oscLinks = [{ row: 0, startCol: 0, endCol: 8, uri: 'https://example.com/restored' }]
|
||||
const coldRestore = {
|
||||
|
||||
@@ -3126,6 +3126,12 @@ export function registerPtyHandlers(
|
||||
trustedTerminalHandleEnv.add(args.preAllocatedHandle)
|
||||
}
|
||||
result = await provider.spawn(spawnOptions)
|
||||
if (typeof result.snapshotOutputSequence === 'number') {
|
||||
runtime?.anchorPtyOutputSequenceFromProviderSnapshot?.(
|
||||
result.id,
|
||||
result.snapshotOutputSequence
|
||||
)
|
||||
}
|
||||
} catch (err) {
|
||||
const rawMessage = err instanceof Error ? err.message : String(err)
|
||||
const spawnError = normalizeNodePtySpawnError(err)
|
||||
@@ -4044,6 +4050,12 @@ export function registerPtyHandlers(
|
||||
}
|
||||
spawnTiming.mark('options')
|
||||
result = await provider.spawn(spawnOptions)
|
||||
if (typeof result.snapshotOutputSequence === 'number') {
|
||||
runtime?.anchorPtyOutputSequenceFromProviderSnapshot?.(
|
||||
result.id,
|
||||
result.snapshotOutputSequence
|
||||
)
|
||||
}
|
||||
spawnTiming.mark('provider_spawn')
|
||||
} catch (err) {
|
||||
// Why: a failed spawn must not leave a stale hidden mark on a session
|
||||
|
||||
@@ -117,6 +117,9 @@ export type PtySpawnResult = {
|
||||
* writing the snapshot so ANSI cursor positions land correctly. */
|
||||
snapshotCols?: number
|
||||
snapshotRows?: number
|
||||
/** Provider output sequence at snapshot capture. Main uses this to keep
|
||||
* post-reattach live bytes in the same deduplication domain. */
|
||||
snapshotOutputSequence?: number
|
||||
/** Kitty keyboard flags persisted in the daemon snapshot, threaded so the
|
||||
* re-seeded runtime emulator answers hidden `CSI ? u` with the real flags
|
||||
* (terminal-query-authority.md §kitty). Never replayed into a renderer
|
||||
|
||||
@@ -7117,6 +7117,24 @@ describe('OrcaRuntimeService', () => {
|
||||
expect(runtime.getPtyOutputSequence('pty-1')).toBe(0)
|
||||
})
|
||||
|
||||
it('anchors post-reattach live output to the provider snapshot sequence', async () => {
|
||||
const { runtime } = createSideEffectRuntime()
|
||||
|
||||
// Simulate a stream-socket byte winning the race with the spawn response.
|
||||
runtime.onPtyData('pty-restored', 'queued', 100)
|
||||
expect(runtime.anchorPtyOutputSequenceFromProviderSnapshot('pty-restored', 900)).toBe(906)
|
||||
runtime.onPtyData('pty-restored', 'fresh', 101)
|
||||
|
||||
expect(runtime.getPtyOutputSequence('pty-restored')).toBe(911)
|
||||
await expect(runtime.serializeMainTerminalBuffer('pty-restored')).resolves.toMatchObject({
|
||||
data: expect.stringContaining('queuedfresh'),
|
||||
seq: 911,
|
||||
source: 'headless'
|
||||
})
|
||||
// Renderer remounts can attach the same provider PTY again in one main lifetime.
|
||||
expect(runtime.anchorPtyOutputSequenceFromProviderSnapshot('pty-restored', 911)).toBe(911)
|
||||
})
|
||||
|
||||
it('carries the synthetic permission BEL as a bell fact', () => {
|
||||
const { runtime, batches } = createSideEffectRuntime()
|
||||
syncSinglePty(runtime)
|
||||
|
||||
@@ -2249,6 +2249,7 @@ export class OrcaRuntimeService {
|
||||
private titleObservationSequence = 0
|
||||
private headlessTerminals = new Map<string, RuntimeHeadlessTerminal>()
|
||||
private ptyOutputSequenceById = new Map<string, number>()
|
||||
private providerSequenceAnchoredPtys = new Set<string>()
|
||||
private recentPtyPathCandidatesById = new Map<string, string[]>()
|
||||
// Why: OSC 9999 status can span PTY chunks. Keeping parser state in the
|
||||
// runtime lets hidden/model-owned terminals observe agent state without a
|
||||
@@ -6599,6 +6600,29 @@ export class OrcaRuntimeService {
|
||||
return this.ptyOutputSequenceById.get(ptyId) ?? 0
|
||||
}
|
||||
|
||||
anchorPtyOutputSequenceFromProviderSnapshot(ptyId: string, providerSequence: number): number {
|
||||
if (
|
||||
this.providerSequenceAnchoredPtys.has(ptyId) ||
|
||||
!Number.isFinite(providerSequence) ||
|
||||
providerSequence < 0
|
||||
) {
|
||||
return this.getPtyOutputSequence(ptyId)
|
||||
}
|
||||
const baseline = Math.floor(providerSequence)
|
||||
const anchoredSequence = baseline + this.getPtyOutputSequence(ptyId)
|
||||
this.ptyOutputSequenceById.set(ptyId, anchoredSequence)
|
||||
this.providerSequenceAnchoredPtys.add(ptyId)
|
||||
const headless = this.headlessTerminals.get(ptyId)
|
||||
if (headless) {
|
||||
// Why: daemon bytes can reach main just before spawn resolves. Queue the
|
||||
// baseline behind those writes so their emulator sequence is rebased too.
|
||||
headless.writeChain = headless.writeChain.then(() => {
|
||||
headless.outputSequence += baseline
|
||||
})
|
||||
}
|
||||
return anchoredSequence
|
||||
}
|
||||
|
||||
subscribeToTerminalData(
|
||||
ptyId: string,
|
||||
listener: (data: string, meta?: { seq?: number; rawLength?: number; cwd?: string }) => void
|
||||
@@ -8269,6 +8293,7 @@ export class OrcaRuntimeService {
|
||||
this.clearWaitBlockedCheckState(ptyId)
|
||||
this.recentPtyPathCandidatesById.delete(ptyId)
|
||||
this.ptyOutputSequenceById.delete(ptyId)
|
||||
this.providerSequenceAnchoredPtys.delete(ptyId)
|
||||
this.agentStatusOscProcessorsByPtyId.delete(ptyId)
|
||||
this.terminalSpawnCommandsByPtyId.delete(ptyId)
|
||||
this.disposePtyTitleTracker(ptyId)
|
||||
@@ -20486,6 +20511,7 @@ export class OrcaRuntimeService {
|
||||
this.clearWaitBlockedCheckState(ptyId)
|
||||
this.recentPtyPathCandidatesById.delete(ptyId)
|
||||
this.ptyOutputSequenceById.delete(ptyId)
|
||||
this.providerSequenceAnchoredPtys.delete(ptyId)
|
||||
this.agentStatusOscProcessorsByPtyId.delete(ptyId)
|
||||
this.terminalSpawnCommandsByPtyId.delete(ptyId)
|
||||
this.disposePtyTitleTracker(ptyId)
|
||||
|
||||
@@ -0,0 +1,94 @@
|
||||
import { expect, it, vi } from 'vitest'
|
||||
import { RpcDispatcher } from './dispatcher'
|
||||
import type { RpcRequest } from './core'
|
||||
import { TERMINAL_METHODS } from './methods/terminal'
|
||||
import type { OrcaRuntimeService } from '../orca-runtime'
|
||||
import type { RuntimeTerminalWait } from '../../../shared/runtime-types'
|
||||
import {
|
||||
TerminalStreamOpcode,
|
||||
decodeTerminalStreamFrame,
|
||||
decodeTerminalStreamText
|
||||
} from '../../../shared/terminal-stream-protocol'
|
||||
|
||||
const request: RpcRequest = {
|
||||
id: 'req-provider-sequence',
|
||||
authToken: 'tok',
|
||||
method: 'terminal.subscribe',
|
||||
params: {
|
||||
terminal: 'terminal-restored',
|
||||
client: { id: 'phone-1', type: 'mobile' },
|
||||
capabilities: { terminalBinaryStream: 1 }
|
||||
}
|
||||
}
|
||||
|
||||
it('replays post-capture output in the provider snapshot sequence domain', async () => {
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
const cleanups = new Map<string, () => void>()
|
||||
let dataListener:
|
||||
| ((data: string, meta?: { seq?: number; rawLength?: number }) => void)
|
||||
| undefined
|
||||
let resolveSnapshot:
|
||||
| ((snapshot: { data: string; cols: number; rows: number; seq: number }) => void)
|
||||
| undefined
|
||||
const runtime = {
|
||||
getRuntimeId: () => 'test-runtime',
|
||||
registerRemoteTerminalViewSubscriber: () => () => {},
|
||||
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-restored' }),
|
||||
handleMobileSubscribe: vi.fn().mockResolvedValue(true),
|
||||
handleMobileUnsubscribe: vi.fn(),
|
||||
subscribeToTerminalData: vi.fn((_ptyId, listener) => {
|
||||
dataListener = listener
|
||||
return vi.fn()
|
||||
}),
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
||||
serializeTerminalBuffer: vi.fn(
|
||||
() =>
|
||||
new Promise<{ data: string; cols: number; rows: number; seq: number }>((resolve) => {
|
||||
resolveSnapshot = resolve
|
||||
})
|
||||
),
|
||||
getTerminalSize: vi.fn().mockReturnValue({ cols: 80, rows: 24 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
isTerminalAlternateScreen: vi.fn().mockReturnValue(false),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
||||
cleanups.set(id, cleanup)
|
||||
}),
|
||||
cleanupSubscription: vi.fn((id: string) => {
|
||||
cleanups.get(id)?.()
|
||||
cleanups.delete(id)
|
||||
}),
|
||||
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {}))
|
||||
} as unknown as OrcaRuntimeService
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(request, vi.fn(), {
|
||||
connectionId: 'conn-restored',
|
||||
sendBinary: (bytes) => {
|
||||
binaryFrames.push(bytes)
|
||||
}
|
||||
})
|
||||
|
||||
await vi.waitFor(() => expect(resolveSnapshot).toBeDefined())
|
||||
resolveSnapshot?.({ data: 'restored snapshot', cols: 80, rows: 24, seq: 900 })
|
||||
// Why: provider attach anchoring makes this first new chunk continue at
|
||||
// 905 instead of restarting at 5, so snapshot reconciliation keeps it.
|
||||
dataListener?.('fresh', { seq: 905, rawLength: 5 })
|
||||
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
binaryFrames.some((bytes) => {
|
||||
const frame = decodeTerminalStreamFrame(bytes)
|
||||
return (
|
||||
frame?.opcode === TerminalStreamOpcode.Output &&
|
||||
decodeTerminalStreamText(frame.payload) === 'fresh'
|
||||
)
|
||||
})
|
||||
).toBe(true)
|
||||
)
|
||||
|
||||
runtime.cleanupSubscription('terminal-restored:phone-1')
|
||||
await dispatchPromise
|
||||
})
|
||||
Reference in New Issue
Block a user