mirror of
https://github.com/stablyai/orca.git
synced 2026-10-21 08:03:33 +00:00
fix(terminal): restore preserved remote PTYs after host relaunch (#12990)
* fix(terminal): foreground preserved daemon PTYs * fix(terminal): keep snapshot sequence domains distinct * test(terminal): use active reconnect control * test(terminal): await reconnect control activation * test(terminal): validate reconnect with fresh control * test(terminal): tighten host restart evidence * fix(terminal): retry preserved PTY attach after inventory * fix(terminal): retry attach after overlapping inventory --------- Co-authored-by: Jinwoo-H <Jinwoo-H@users.noreply.github.com>
This commit is contained in:
@@ -637,6 +637,7 @@ describe('registerPtyHandlers', () => {
|
||||
const write = vi.fn()
|
||||
const pauseProducer = vi.fn()
|
||||
const resumeProducer = vi.fn()
|
||||
const setPtyBackgrounded = vi.fn()
|
||||
const shutdown = vi.fn()
|
||||
let dataHandler: ((payload: { id: string; data: string }) => void) | null = null
|
||||
let exitHandler: ((payload: { id: string; code: number }) => void) | null = null
|
||||
@@ -650,6 +651,7 @@ describe('registerPtyHandlers', () => {
|
||||
resize: vi.fn(),
|
||||
pauseProducer,
|
||||
resumeProducer,
|
||||
setPtyBackgrounded,
|
||||
kill: vi.fn(),
|
||||
shutdown,
|
||||
sendSignal: vi.fn(),
|
||||
@@ -688,6 +690,7 @@ describe('registerPtyHandlers', () => {
|
||||
write,
|
||||
pauseProducer,
|
||||
resumeProducer,
|
||||
setPtyBackgrounded,
|
||||
shutdown,
|
||||
getBufferSnapshot,
|
||||
emitData: (id: string, data: string) => dataHandler?.({ id, data }),
|
||||
@@ -15608,6 +15611,50 @@ describe('registerPtyHandlers', () => {
|
||||
})
|
||||
|
||||
describe('hidden renderer delivery gate', () => {
|
||||
it('foregrounds a preserved daemon PTY after handler recreation loses sync memory', async () => {
|
||||
const daemon = installObservableDaemonTestProvider()
|
||||
const firstRuntime = {
|
||||
setPtyController: vi.fn(),
|
||||
hasRawTerminalViewSubscriber: vi.fn(() => false),
|
||||
createPreAllocatedTerminalHandle: vi.fn(() => null),
|
||||
registerPty: vi.fn(),
|
||||
noteTerminalSpawnCommand: vi.fn(),
|
||||
onPtySpawned: vi.fn(),
|
||||
onPtyExit: vi.fn(),
|
||||
onPtyData: vi.fn()
|
||||
}
|
||||
registerPtyHandlers(mainWindow as never, firstRuntime as never)
|
||||
const result = (await handlers.get('pty:spawn')!(null, {
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
sessionId: 'daemon-session'
|
||||
})) as { id: string }
|
||||
|
||||
getPtySetRendererPtyVisibleListener()(null, { id: result.id, visible: false })
|
||||
expect(daemon.setPtyBackgrounded).toHaveBeenLastCalledWith(result.id, true)
|
||||
|
||||
daemon.setPtyBackgrounded.mockClear()
|
||||
handlers.clear()
|
||||
let rawSubscriberPresent = false
|
||||
const nextRuntime = {
|
||||
setPtyController: vi.fn(),
|
||||
hasRawTerminalViewSubscriber: vi.fn(() => rawSubscriberPresent),
|
||||
onRemoteTerminalViewPresenceChanged: null as ((id: string) => void) | null,
|
||||
registerRawTerminalViewSubscriber(id: string): void {
|
||||
rawSubscriberPresent = true
|
||||
this.onRemoteTerminalViewPresenceChanged?.(id)
|
||||
}
|
||||
}
|
||||
registerPtyHandlers(mainWindow as never, nextRuntime as never)
|
||||
|
||||
nextRuntime.registerRawTerminalViewSubscriber(result.id)
|
||||
// Repeated presence signals must remain deduplicated.
|
||||
nextRuntime.onRemoteTerminalViewPresenceChanged?.(result.id)
|
||||
|
||||
expect(daemon.setPtyBackgrounded).toHaveBeenCalledOnce()
|
||||
expect(daemon.setPtyBackgrounded).toHaveBeenCalledWith(result.id, false)
|
||||
})
|
||||
|
||||
it('drops hidden PTY data after model ingestion and emits one out-of-band restore marker', async () => {
|
||||
vi.useFakeTimers()
|
||||
const runtime = {
|
||||
|
||||
+1
-1
@@ -2516,7 +2516,7 @@ export function registerPtyHandlers(
|
||||
function syncPtyBackgroundedDelivery(id: string, caller: string): void {
|
||||
const background =
|
||||
rendererPtyIsKnownHidden(id) && !(runtime?.hasRawTerminalViewSubscriber?.(id) ?? false)
|
||||
if ((backgroundedDeliverySyncByPty.get(id) ?? false) === background) {
|
||||
if (backgroundedDeliverySyncByPty.get(id) === background) {
|
||||
return
|
||||
}
|
||||
const provider = tryGetProviderForPty(id)
|
||||
|
||||
@@ -3111,6 +3111,7 @@ export class OrcaRuntimeService {
|
||||
// concurrent first-subscribes and keeps later subscribes no-ops; it is
|
||||
// cleared per lifecycle generation (exit/respawn) and on failed attempts.
|
||||
private subscriberDrivenProviderAttachesByPtyId = new Map<string, Promise<boolean>>()
|
||||
private subscriberDrivenProviderAttachInventoryWaiters = new Set<string>()
|
||||
// Why: a spawn through this app already attaches its provider stream, so
|
||||
// subscriber-driven attach and the never-attached read fallback must target
|
||||
// only inventory-discovered sessions no local spawn published this
|
||||
@@ -10597,6 +10598,7 @@ export class OrcaRuntimeService {
|
||||
this.legacyWorkerRecoveredPtys.delete(ptyId)
|
||||
// Why: a respawn under the same session id needs its own subscriber-driven attach.
|
||||
this.subscriberDrivenProviderAttachesByPtyId.delete(ptyId)
|
||||
this.subscriberDrivenProviderAttachInventoryWaiters.delete(ptyId)
|
||||
this.spawnPublishedPtys.delete(ptyId)
|
||||
// Why: a provider response belongs to the process generation that issued
|
||||
// it; a respawn must neither reuse its frame nor join its in-flight call.
|
||||
@@ -10806,10 +10808,7 @@ export class OrcaRuntimeService {
|
||||
// A spawn published (or admission pending) this generation already
|
||||
// attaches the provider stream; a replacement under a reused id must not
|
||||
// read as the discovered never-attached session it replaced.
|
||||
if (
|
||||
this.spawnPublishedPtys.has(ptyId) ||
|
||||
this.pendingPtyRegistrationIncarnations.has(ptyId)
|
||||
) {
|
||||
if (this.spawnPublishedPtys.has(ptyId) || this.pendingPtyRegistrationIncarnations.has(ptyId)) {
|
||||
return false
|
||||
}
|
||||
// SSH panes have their own lease/reattach machinery.
|
||||
@@ -10846,6 +10845,31 @@ export class OrcaRuntimeService {
|
||||
})
|
||||
}
|
||||
|
||||
private reconcileSubscriberDrivenProviderAttach(ptyId: string): void {
|
||||
if (!this.hasRemoteTerminalViewSubscriber(ptyId)) {
|
||||
return
|
||||
}
|
||||
const pending = this.subscriberDrivenProviderAttachesByPtyId.get(ptyId)
|
||||
if (!pending) {
|
||||
this.ensureSubscriberDrivenProviderAttach(ptyId)
|
||||
return
|
||||
}
|
||||
if (this.subscriberDrivenProviderAttachInventoryWaiters.has(ptyId)) {
|
||||
return
|
||||
}
|
||||
this.subscriberDrivenProviderAttachInventoryWaiters.add(ptyId)
|
||||
void pending.then((attached) => {
|
||||
this.subscriberDrivenProviderAttachInventoryWaiters.delete(ptyId)
|
||||
if (attached || !this.hasRemoteTerminalViewSubscriber(ptyId)) {
|
||||
return
|
||||
}
|
||||
if (this.subscriberDrivenProviderAttachesByPtyId.get(ptyId) === pending) {
|
||||
this.subscriberDrivenProviderAttachesByPtyId.delete(ptyId)
|
||||
}
|
||||
this.ensureSubscriberDrivenProviderAttach(ptyId)
|
||||
})
|
||||
}
|
||||
|
||||
/** Mark a raw-output viewer without transferring terminal query authority. */
|
||||
registerRawTerminalViewSubscriber(ptyId: string): () => void {
|
||||
this.rawTerminalViewSubscriberCounts.set(
|
||||
@@ -29268,6 +29292,7 @@ export class OrcaRuntimeService {
|
||||
this.restoredOrchestrationAuthorityByPtyId.delete(session.id)
|
||||
}
|
||||
pty.controllerTitle = session.title?.trim() || null
|
||||
this.reconcileSubscriberDrivenProviderAttach(session.id)
|
||||
}
|
||||
// Why: fire-and-forget so this listing hot path doesn't serialize a relay round-trip per session and a throw can't abort the sweep below.
|
||||
this.refreshPtyForegroundAgent(session.id)
|
||||
|
||||
@@ -2623,7 +2623,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
const size = runtime.getTerminalSize(ptyId)
|
||||
const displayMode = runtime.getMobileDisplayMode(ptyId)
|
||||
const layoutSeq = runtime.getLayout(ptyId)?.seq
|
||||
const snapshotFrameSeq = serialized?.seq ?? layoutSeq
|
||||
// Why: layout versions and output offsets are different sequence domains.
|
||||
const snapshotOutputSeq = serialized?.seq
|
||||
emit({
|
||||
type: 'subscribed',
|
||||
@@ -2664,7 +2664,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
cols: serialized?.cols ?? size?.cols ?? 80,
|
||||
rows: serialized?.rows ?? size?.rows ?? 24,
|
||||
displayMode,
|
||||
seq: snapshotFrameSeq,
|
||||
seq: snapshotOutputSeq,
|
||||
cwd: serialized?.cwd,
|
||||
truncated: initialOutputOverflowed,
|
||||
truncatedByByteBudget: serialized?.truncatedByByteBudget,
|
||||
@@ -3424,8 +3424,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
const displayMode = runtime.getMobileDisplayMode(ptyId)
|
||||
// Why: layout seq is the mobile stale-event filter's high-water mark (undefined pre-transition is fail-open). See docs/mobile-terminal-layout-state-machine.md.
|
||||
const layoutSeq = runtime.getLayout(ptyId)?.seq
|
||||
const snapshotFrameSeq = serialized?.seq ?? layoutSeq
|
||||
// Why: track the seq that actually covered the buffered chunks (recovery snapshots advance it) or an absorbed query gets zero replies.
|
||||
// Why: only an output offset can cover buffered chunks; layout versions are a separate sequence domain.
|
||||
let snapshotOutputSeq = serialized?.seq
|
||||
emit({
|
||||
type: 'subscribed',
|
||||
@@ -3445,7 +3444,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
||||
cols: serialized?.cols ?? size?.cols ?? params.viewport?.cols ?? 80,
|
||||
rows: serialized?.rows ?? size?.rows ?? params.viewport?.rows ?? 24,
|
||||
displayMode,
|
||||
seq: snapshotFrameSeq,
|
||||
seq: snapshotOutputSeq,
|
||||
cwd: serialized?.cwd,
|
||||
truncated: initialOutputOverflowed,
|
||||
truncatedByByteBudget: serialized?.truncatedByByteBudget,
|
||||
|
||||
@@ -444,6 +444,58 @@ function startSourceRangeOverflowHarness(options: {
|
||||
}
|
||||
|
||||
describe('terminal multiplex RPC', () => {
|
||||
it('keeps layout versions out of the output sequence domain', async () => {
|
||||
let dataListener: ((data: string, meta?: RuntimeTerminalDataMeta) => void) | undefined
|
||||
const harness = startDesktopMultiplexSubscribe({
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 675 }),
|
||||
serializeTerminalBuffer: vi.fn().mockResolvedValue({
|
||||
data: 'restored terminal',
|
||||
cols: 120,
|
||||
rows: 40,
|
||||
source: 'renderer'
|
||||
}),
|
||||
subscribeToTerminalData: vi.fn((_ptyId, listener) => {
|
||||
dataListener = listener
|
||||
return vi.fn()
|
||||
})
|
||||
})
|
||||
await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
|
||||
sendDesktopMultiplexSubscribe(harness.handlers)
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
harness.messages.some((message) => JSON.parse(message).result?.type === 'subscribed')
|
||||
).toBe(true)
|
||||
)
|
||||
|
||||
const subscribed = harness.messages
|
||||
.map((message) => JSON.parse(message).result)
|
||||
.find((event) => event?.type === 'subscribed')
|
||||
const snapshotStart = harness.binaryFrames
|
||||
.map(decodeTerminalStreamFrame)
|
||||
.find((frame) => frame?.opcode === TerminalStreamOpcode.SnapshotStart)
|
||||
expect(subscribed.seq).toBe(675)
|
||||
if (!snapshotStart) {
|
||||
throw new Error('Missing multiplex snapshot start frame')
|
||||
}
|
||||
const snapshotPayload = decodeTerminalStreamJson(snapshotStart.payload)
|
||||
expect(snapshotPayload).toMatchObject({ kind: 'scrollback' })
|
||||
expect(snapshotPayload).not.toHaveProperty('seq')
|
||||
|
||||
harness.binaryFrames.splice(0)
|
||||
dataListener?.('live', { seq: 4, rawLength: 4 })
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
harness.binaryFrames.some((bytes) => {
|
||||
const frame = decodeTerminalStreamFrame(bytes)
|
||||
return frame?.opcode === TerminalStreamOpcode.Output && frame.seq === 4
|
||||
})
|
||||
).toBe(true)
|
||||
)
|
||||
|
||||
harness.cleanups.get('terminal-multiplex:conn-desktop-first-paint')?.()
|
||||
await harness.dispatchPromise
|
||||
})
|
||||
|
||||
it.each(['headless', 'renderer'] as const)(
|
||||
'commits a source-range replacement only after the %s snapshot publishes',
|
||||
async (source) => {
|
||||
|
||||
@@ -833,15 +833,13 @@ describe('terminal subscribe buffering', () => {
|
||||
const decodedFrames = binaryFrames
|
||||
.map((frame) => decodeTerminalStreamFrame(frame))
|
||||
.filter((frame): frame is NonNullable<typeof frame> => frame !== null)
|
||||
const snapshotStarts = decodedFrames.filter(
|
||||
const snapshotStart = decodedFrames.find(
|
||||
(frame) => frame.opcode === TerminalStreamOpcode.SnapshotStart
|
||||
)
|
||||
// Why seq 1 (layout seq): a no-output-seq snapshot falls back to the
|
||||
// layout seq on the wire; the recovered data still ships as the first
|
||||
// and only scrollback snapshot.
|
||||
expect(snapshotStarts.map((frame) => decodeTerminalStreamJson(frame.payload))).toEqual([
|
||||
expect.objectContaining({ kind: 'scrollback', seq: 1 })
|
||||
])
|
||||
// Why: layout versions and output offsets are different sequence domains.
|
||||
const snapshotInfo = decodeTerminalStreamJson(snapshotStart!.payload)
|
||||
expect(snapshotInfo).toMatchObject({ kind: 'scrollback' })
|
||||
expect(snapshotInfo).not.toHaveProperty('seq')
|
||||
const snapshotText = decodedFrames
|
||||
.filter((frame) => frame.opcode === TerminalStreamOpcode.SnapshotChunk)
|
||||
.map((frame) => decodeTerminalStreamText(frame.payload))
|
||||
|
||||
@@ -37,6 +37,12 @@ type RuntimeInternals = {
|
||||
) => unknown
|
||||
issuePtyHandle: (pty: unknown) => string
|
||||
headlessTerminals: Map<string, unknown>
|
||||
subscriberDrivenProviderAttachesByPtyId: Map<string, Promise<boolean>>
|
||||
subscriberDrivenProviderAttachInventoryWaiters: Set<string>
|
||||
refreshPtyWorktreeRecordsWithControllerInventory: (
|
||||
worktrees: [],
|
||||
targetWorktreeId: string | null
|
||||
) => Promise<unknown>
|
||||
}
|
||||
|
||||
function internals(runtime: OrcaRuntimeService): RuntimeInternals {
|
||||
@@ -56,11 +62,21 @@ function createDaemonProviderModel(opts: { snapshotCapable: boolean }) {
|
||||
const sessions = new Map<string, DaemonSessionModel>()
|
||||
const attachCalls: string[] = []
|
||||
const resizeCalls: [string, number, number][] = []
|
||||
let nextAttachBarrier: { promise: Promise<void>; result: boolean } | null = null
|
||||
let runtime: OrcaRuntimeService | null = null
|
||||
const controller = {
|
||||
write: () => true,
|
||||
kill: () => true,
|
||||
getForegroundProcess: async () => null,
|
||||
listProcesses: async () =>
|
||||
[...sessions.entries()].map(([id, session]) => ({
|
||||
id,
|
||||
cwd: '/tmp/wt',
|
||||
worktreeId: WORKTREE_ID,
|
||||
title: '',
|
||||
cols: session.cols,
|
||||
rows: session.rows
|
||||
})),
|
||||
hasRendererSerializer: () => false,
|
||||
getSize: (ptyId: string) => {
|
||||
const session = sessions.get(ptyId)
|
||||
@@ -72,6 +88,14 @@ function createDaemonProviderModel(opts: { snapshotCapable: boolean }) {
|
||||
},
|
||||
attach: async (ptyId: string) => {
|
||||
attachCalls.push(ptyId)
|
||||
const barrier = nextAttachBarrier
|
||||
nextAttachBarrier = null
|
||||
if (barrier) {
|
||||
await barrier.promise
|
||||
if (!barrier.result) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
const session = sessions.get(ptyId)
|
||||
// Attach-only: an absent session is refused, never created.
|
||||
if (!session) {
|
||||
@@ -100,6 +124,14 @@ function createDaemonProviderModel(opts: { snapshotCapable: boolean }) {
|
||||
sessions,
|
||||
attachCalls,
|
||||
resizeCalls,
|
||||
deferNextAttach(result: boolean): () => void {
|
||||
let release!: () => void
|
||||
const promise = new Promise<void>((resolve) => {
|
||||
release = resolve
|
||||
})
|
||||
nextAttachBarrier = { promise, result }
|
||||
return release
|
||||
},
|
||||
bind(target: OrcaRuntimeService) {
|
||||
runtime = target
|
||||
},
|
||||
@@ -376,6 +408,39 @@ describe('subscriber-driven daemon attach (never-activated tab)', () => {
|
||||
await vi.waitFor(() => expect(outputText(harness)).toContain('late daemon hello'))
|
||||
})
|
||||
|
||||
it('retries an existing subscriber when provider inventory becomes ready', async () => {
|
||||
const { runtime, model, handle } = setupNeverAttachedDaemonSession({
|
||||
snapshotCapable: false
|
||||
})
|
||||
model.sessions.delete(PTY_ID)
|
||||
const refuseFirstAttach = model.deferNextAttach(false)
|
||||
const harness = startMultiplex(runtime)
|
||||
await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
|
||||
|
||||
sendSubscribe(harness, 1, handle)
|
||||
await waitForSubscribed(harness, 1)
|
||||
await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID]))
|
||||
expect(internals(runtime).subscriberDrivenProviderAttachesByPtyId.has(PTY_ID)).toBe(true)
|
||||
expect(runtime.hasRemoteTerminalViewSubscriber(PTY_ID)).toBe(true)
|
||||
expect(model.emitData(PTY_ID, 'before inventory\r\n')).toBe(false)
|
||||
|
||||
model.sessions.set(PTY_ID, { cols: 100, rows: 30, attached: false, screen: '' })
|
||||
await internals(runtime).refreshPtyWorktreeRecordsWithControllerInventory([], null)
|
||||
await internals(runtime).refreshPtyWorktreeRecordsWithControllerInventory([], null)
|
||||
expect(model.attachCalls).toEqual([PTY_ID])
|
||||
expect([...internals(runtime).subscriberDrivenProviderAttachInventoryWaiters]).toEqual([PTY_ID])
|
||||
refuseFirstAttach()
|
||||
|
||||
await vi.waitFor(() => expect(model.attachCalls).toEqual([PTY_ID, PTY_ID]))
|
||||
await vi.waitFor(() =>
|
||||
expect(internals(runtime).subscriberDrivenProviderAttachInventoryWaiters.has(PTY_ID)).toBe(
|
||||
false
|
||||
)
|
||||
)
|
||||
expect(model.emitData(PTY_ID, 'after inventory\r\n')).toBe(true)
|
||||
await vi.waitFor(() => expect(outputText(harness)).toContain('after inventory'))
|
||||
})
|
||||
|
||||
it('does not subscriber-attach or provider-read a session this app already spawned', async () => {
|
||||
const { runtime, model, handle } = setupNeverAttachedDaemonSession({
|
||||
snapshotCapable: true,
|
||||
|
||||
Reference in New Issue
Block a user