fix(terminal): recover rejected paired-runtime input (STA-2830) (#12675)
With a desktop client paired to a remote Orca runtime, terminal panes could report connected, writable, and `terminal.send` returning accepted — yet keystrokes never reached the agent. No error, no banner, no recovery; input silently vanished. The ticket was really two bugs. The attach half was already fixed by #12589 (subscriber-driven daemon attach), confirmed by reproducing against current main. This fixes the remaining half: a write the host refuses had no way to tell anyone. A capability-negotiated `WriteUnavailable` opcode carries that refusal back to the client, where it feeds the pane's pre-existing recovery hook. Capability gating matters because decoders reject unknown opcodes on desktop — and, worse, silently drop them on mobile — so the signal is negotiated in the subscribe handshake. Verified per direction: an old host strips the unknown Subscribe key, an old client omits it so the host never emits, and capability cannot be inherited across resubscribe. Independent review then found the signal was being delivered and discarded: recovery demanded an authoritative liveness answer, and `pty:hasPty` had no `remote:` guard, so a paired pane's id fell through to the LOCAL provider, which returned false, and recovery bailed before remounting. Every test stopped at the transport boundary, so all of them passed while the pane stayed just as stuck. `pty:kill` already had exactly that guard. The fix makes main answer LESS rather than claim more: `pty:hasPty` now returns unknown for a `remote:` id instead of a fabricated false, because main cannot speak for another host's PTY. The remount is then authorized by positive evidence — the process that owns the PTY stating it refused this specific write over a live negotiated connection — not by inference from silence. Local and app-SSH ids keep the probe, where a false genuinely means the shell died. Nothing is destroyed on this path; the remount rebuilds the renderer over the session it already had. An end-to-end test now carries a rejected write from the host through to an actual remount, which no prior test did. A surviving mutant was also killed: the legacy-binary capability gate could previously be deleted with nothing turning red. The reliability gate stays experimental — live paired journeys and mixed installed-release evidence remain uncollected. Fixes STA-2830.
This commit is contained in:
Jinwoo Hong
GitHub
parent
06780260c0
commit
d15939c5fd
|
|
@ -6066,6 +6066,120 @@
|
|||
],
|
||||
"demotionRule": "Cannot promote if provider failure can close panes or if the oracle is screenshot-only."
|
||||
},
|
||||
{
|
||||
"id": "terminal-input.remote-write-rejection-recovery",
|
||||
"title": "Rejected paired-runtime terminal input remounts the pane",
|
||||
"maturity": "experimental",
|
||||
"protection": "partial",
|
||||
"owner": "terminal-runtime",
|
||||
"layer": "paired-runtime-stream-contract",
|
||||
"surfaces": [
|
||||
"terminal multiplex input",
|
||||
"legacy binary terminal input",
|
||||
"one-shot terminal.send fallback",
|
||||
"pane recovery",
|
||||
"pty:hasPty liveness routing"
|
||||
],
|
||||
"platforms": ["macos", "linux", "windows"],
|
||||
"providers": ["paired-runtime"],
|
||||
"coveredPlatforms": ["macos"],
|
||||
"coveredProviders": ["paired-runtime"],
|
||||
"coverageNotes": "One end-to-end contract runs the real dispatcher, renderer multiplexer, remote transport, and pty-connection together and requires the tab remount, so the signal is proven past the transport callback it used to die behind. Focused contracts cover capability negotiation, legacy binary subscriptions, stream-id reuse, pane lifecycle reuse, the one-shot JSON fallback, and main refusing to answer liveness for a `remote:` id. Live headed/headless paired-runtime and mixed installed releases remain uncollected.",
|
||||
"motivatingLinks": [
|
||||
"https://linear.app/stably/issue/STA-2830",
|
||||
"https://github.com/stablyai/orca/issues/11124"
|
||||
],
|
||||
"invariant": "When a paired-runtime client accepts terminal input locally but the authoritative host rejects the PTY write, a capability-compatible stream must notify only that current pane generation and that notification must end in an actual tab remount — no local liveness probe may veto it, because main owns no registry entry for a `remote:` id and must answer unknown for one. A host must never send the new opcode to a legacy or un-negotiated client, and a late rejection must never recover a replacement stream or pane lifecycle.",
|
||||
"oracle": "Wire the real dispatcher to the real renderer multiplexer, remote transport, and pty-connection over one bridged subscription; type into the pane, reject the authoritative runtime send before any process write, and require both the WriteUnavailable frame and a remountTerminalTabForRecovery call — repeated for every answer main can produce for a `remote:` id (fabricated dead, unknown, thrown). Separately: require one frame only for a capability-declaring client by driving an un-negotiated legacy binary subscriber first and a capable one second on the same runtime, so the capable frame proves the rejection had already been processed for both. Reuse the stream id before releasing a held rejection and require no signal; replace the stream or detach and reattach the same handle before releasing held failures and require no stale recovery. Require pty:hasPty to answer null for a `remote:` id without consulting the local provider.",
|
||||
"commands": [
|
||||
"pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/renderer/src/runtime/runtime-terminal-stream.test.ts src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.test.ts tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts src/renderer/src/components/terminal-pane/terminal-pane-recovery.test.ts src/main/ipc/pty.test.ts --reporter=dot"
|
||||
],
|
||||
"testFiles": [
|
||||
"src/main/runtime/rpc/terminal-multiplex.test.ts",
|
||||
"src/renderer/src/runtime/runtime-terminal-stream.test.ts",
|
||||
"src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts",
|
||||
"src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.test.ts",
|
||||
"tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts",
|
||||
"src/renderer/src/components/terminal-pane/terminal-pane-recovery.test.ts",
|
||||
"src/main/ipc/pty.test.ts"
|
||||
],
|
||||
"assertionRefs": [
|
||||
{
|
||||
"file": "src/main/runtime/rpc/terminal-multiplex.test.ts",
|
||||
"assertions": [
|
||||
"rejected authoritative input emits WriteUnavailable only for capable multiplex and legacy binary clients",
|
||||
"an un-negotiated legacy binary subscriber never receives the rejection opcode",
|
||||
"a late rejection cannot target a replacement stream with the same id"
|
||||
]
|
||||
},
|
||||
{
|
||||
"file": "src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.test.ts",
|
||||
"assertions": [
|
||||
"WriteUnavailable reaches the current pane recovery callback without a fatal error",
|
||||
"superseded streams and same-handle pane lifecycles ignore delayed rejections",
|
||||
"a rejected one-shot runtime fallback invokes pane recovery"
|
||||
]
|
||||
},
|
||||
{
|
||||
"file": "tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts",
|
||||
"assertions": [
|
||||
"a host-rejected write travels dispatcher to multiplexer to transport to pty-connection and remounts the tab",
|
||||
"no answer the local liveness probe can give for a `remote:` id blocks that remount"
|
||||
]
|
||||
},
|
||||
{
|
||||
"file": "src/renderer/src/components/terminal-pane/terminal-pane-recovery.test.ts",
|
||||
"assertions": [
|
||||
"input-rejected-by-host recovery consults no liveness probe",
|
||||
"input-rejected-by-host still coalesces under the shared recovery cooldown"
|
||||
]
|
||||
},
|
||||
{
|
||||
"file": "src/main/ipc/pty.test.ts",
|
||||
"assertions": [
|
||||
"pty:hasPty answers unknown for a paired-runtime handle instead of the local provider's fabricated dead"
|
||||
]
|
||||
}
|
||||
],
|
||||
"evidenceRuns": [
|
||||
{
|
||||
"date": "2026-08-05",
|
||||
"runner": "local",
|
||||
"platform": "macos",
|
||||
"command": "pnpm exec vitest run --config config/vitest.config.ts src/main/runtime/rpc/terminal-multiplex.test.ts src/renderer/src/runtime/runtime-terminal-stream.test.ts src/renderer/src/runtime/remote-runtime-terminal-parse-backpressure.test.ts src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.test.ts tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts src/renderer/src/components/terminal-pane/terminal-pane-recovery.test.ts src/main/ipc/pty.test.ts --reporter=dot",
|
||||
"result": "passed",
|
||||
"durationSeconds": 15,
|
||||
"summary": "678 passed / 0 failed with the end-to-end remount contract, the un-negotiated legacy binary gate, and the pty:hasPty liveness-routing contract added."
|
||||
}
|
||||
],
|
||||
"runtimeBudget": {
|
||||
"p95Seconds": 45,
|
||||
"scope": "focused paired-runtime host and renderer contracts plus the end-to-end remount chain"
|
||||
},
|
||||
"flakeHistory": {
|
||||
"status": "unknown",
|
||||
"evidence": "Deterministic controlled-promise tests pass locally; CI and soak history have not started."
|
||||
},
|
||||
"redGreenEvidence": {
|
||||
"status": "complete",
|
||||
"evidence": "On origin/main e5f49e0e1d, the test-only dispatcher oracle subscribed, accepted client input, invoked the authoritative host send once, recorded no process write, and failed only because no WriteUnavailable frame returned. The end-to-end contract was then red at the last hop on the delivery-only implementation — the host frame arrived and no remount followed — for all three liveness answers (3 failed / 0 passed), and green after the recovery routing fix (3 passed). Deleting the legacy-binary capability gate makes terminal-multiplex red (1 failed / 62 passed); deleting the pty:hasPty `remote:` guard makes the main contract red with the fabricated `false`; reverting either half of the renderer routing makes the end-to-end contract red (3 failed). Every mutation was restored by re-applying the edit and re-verified green."
|
||||
},
|
||||
"performanceBudget": {
|
||||
"required": true,
|
||||
"evidence": "The change adds one optional capability field, one constant-time outcome classification per existing write, and one rejection-only frame/callback. The recovery routing adds one string comparison per rejection and removes an IPC round-trip on that path; the pty:hasPty guard is a prefix test that short-circuits a provider lookup. It adds no polling, timer, provider listing, subprocess, retained payload, or cross-pane fanout."
|
||||
},
|
||||
"promotionCriteria": [
|
||||
"Collect headed and headless paired-runtime journeys with a rejected host write.",
|
||||
"Collect mixed installed-release evidence in both client/server directions.",
|
||||
"Collect CI soak history with no unexplained flakes."
|
||||
],
|
||||
"knownGaps": [
|
||||
"Live headed and headless paired-runtime journeys are not collected.",
|
||||
"Linux, Windows, mobile, and mixed installed-release runs are not collected.",
|
||||
"Dead-record connected-state correction remains tracked separately in STA-2896."
|
||||
],
|
||||
"demotionRule": "Keep experimental or demote if rejected input can remain silent, the signal stops short of a remount, a legacy or un-negotiated client receives an unknown opcode, a liveness probe fabricates an answer for a `remote:` id, a stale failure recovers a replacement pane, or the deterministic contract flakes without an identified product or harness bug."
|
||||
},
|
||||
{
|
||||
"id": "terminal-provider.wsl-restore-contract",
|
||||
"title": "WSL terminals preserve launch identity, liveness, and restore boundaries",
|
||||
|
|
|
|||
|
|
@ -7071,6 +7071,41 @@ describe('registerPtyHandlers', () => {
|
|||
await expect(handlers.get('pty:hasPty')!(null, { id: 'maybe-pty' })).resolves.toBe(null)
|
||||
})
|
||||
|
||||
it('never answers liveness for a paired-runtime handle from the local registry', async () => {
|
||||
const hasPty = vi.fn(() => false)
|
||||
setLocalPtyProvider({
|
||||
spawn: vi.fn(),
|
||||
write: vi.fn(),
|
||||
resize: vi.fn(),
|
||||
shutdown: vi.fn(),
|
||||
sendSignal: vi.fn(),
|
||||
getCwd: vi.fn(),
|
||||
getInitialCwd: vi.fn(),
|
||||
clearBuffer: vi.fn(),
|
||||
acknowledgeDataEvent: vi.fn(),
|
||||
hasChildProcesses: vi.fn(),
|
||||
getForegroundProcess: vi.fn(),
|
||||
serialize: vi.fn(),
|
||||
revive: vi.fn(),
|
||||
onData: vi.fn(() => () => {}),
|
||||
onReplay: vi.fn(() => () => {}),
|
||||
onExit: vi.fn(() => () => {}),
|
||||
listProcesses: vi.fn(async () => []),
|
||||
attach: vi.fn(),
|
||||
hasPty,
|
||||
getDefaultShell: vi.fn(),
|
||||
getProfiles: vi.fn()
|
||||
} as never)
|
||||
registerPtyHandlers(mainWindow as never)
|
||||
|
||||
// The local provider would happily report `false` here — it just doesn't
|
||||
// hold the id. Callers read that as "the shell died" (STA-2830).
|
||||
await expect(
|
||||
handlers.get('pty:hasPty')!(null, { id: 'remote:env-1@@terminal-1' })
|
||||
).resolves.toBe(null)
|
||||
expect(hasPty).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('lists duplicate SSH relay session ids as distinct app sessions', async () => {
|
||||
registerPtyHandlers(mainWindow as never)
|
||||
const shutdownA = vi.fn(async () => undefined)
|
||||
|
|
|
|||
|
|
@ -7159,6 +7159,13 @@ export function registerPtyHandlers(
|
|||
)
|
||||
|
||||
ipcMain.handle('pty:hasPty', async (_event, args: { id: string }): Promise<boolean | null> => {
|
||||
if (typeof args?.id !== 'string' || args.id.startsWith('remote:')) {
|
||||
// Why: same routing hazard pty:kill guards against — ptyOwnership never holds
|
||||
// a runtime terminal handle and parseAppSshPtyId ignores it, so the lookup
|
||||
// falls through to the local provider and its "not in my table" reads as an
|
||||
// authoritative dead. That is a fabricated answer about another host's PTY.
|
||||
return null
|
||||
}
|
||||
const ownedConnectionId = ptyOwnership.get(args.id)
|
||||
const parsedSshId = ownedConnectionId === undefined ? parseAppSshPtyId(args.id) : null
|
||||
const provider = parsedSshId
|
||||
|
|
|
|||
|
|
@ -135,6 +135,7 @@ type TerminalMultiplexStream = {
|
|||
ackInFlightBytes: number
|
||||
ackWindowBytes: number
|
||||
supportsOutputPause: boolean
|
||||
supportsWriteUnavailable: boolean
|
||||
outputPaused: boolean
|
||||
supportsDesktopViewportClaims: boolean
|
||||
desktopClaimTail: Promise<boolean>
|
||||
|
|
@ -315,6 +316,13 @@ function resolveMobileFloorClientId(
|
|||
return null
|
||||
}
|
||||
|
||||
type TerminalStreamInputOutcome = 'delivered' | 'rejected' | 'failed'
|
||||
|
||||
function isTerminalStreamInputRejection(error: unknown): boolean {
|
||||
const message = error instanceof Error ? error.message : String(error)
|
||||
return message.includes('terminal_not_writable') || message.includes('terminal_handle_stale')
|
||||
}
|
||||
|
||||
async function sendTerminalStreamInput(
|
||||
runtime: OrcaRuntimeService,
|
||||
args: {
|
||||
|
|
@ -323,14 +331,14 @@ async function sendTerminalStreamInput(
|
|||
client: TerminalViewportClient | undefined
|
||||
isMobile: boolean
|
||||
}
|
||||
): Promise<void> {
|
||||
): Promise<TerminalStreamInputOutcome> {
|
||||
const action = { text: args.text, enter: false, interrupt: false }
|
||||
const clientId = args.isMobile ? args.client?.id : undefined
|
||||
const floorClaim: MobileInputFloorClaimHolder = { current: null }
|
||||
try {
|
||||
if (!clientId) {
|
||||
await runtime.sendTerminal(args.terminal, action)
|
||||
return
|
||||
const result = await runtime.sendTerminal(args.terminal, action)
|
||||
return result.accepted ? 'delivered' : 'rejected'
|
||||
}
|
||||
const result = await runtime.sendTerminal(args.terminal, action, {
|
||||
reserveWrite: (writePtyId) => {
|
||||
|
|
@ -344,9 +352,12 @@ async function sendTerminalStreamInput(
|
|||
})
|
||||
if (!result.accepted) {
|
||||
floorClaim.current?.rollback()
|
||||
return 'rejected'
|
||||
}
|
||||
} catch {
|
||||
return 'delivered'
|
||||
} catch (error) {
|
||||
floorClaim.current?.rollback()
|
||||
return isTerminalStreamInputRejection(error) ? 'rejected' : 'failed'
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1017,7 +1028,8 @@ const TerminalSubscribe = TerminalHandle.extend({
|
|||
.object({
|
||||
terminalBinaryStream: z.literal(1).optional(),
|
||||
desktopViewportClaims: z.literal(1).optional(),
|
||||
mobileInputLeaseOnly: z.literal(1).optional()
|
||||
mobileInputLeaseOnly: z.literal(1).optional(),
|
||||
writeUnavailable: z.literal(1).optional()
|
||||
})
|
||||
.optional()
|
||||
})
|
||||
|
|
@ -1038,7 +1050,8 @@ const TerminalMultiplexSubscribeFrame = TerminalHandle.extend({
|
|||
ackOutput: z.literal(1).optional(),
|
||||
ackOutputSourceRanges: z.literal(1).optional(),
|
||||
desktopViewportClaims: z.literal(1).optional(),
|
||||
outputPause: z.literal(1).optional()
|
||||
outputPause: z.literal(1).optional(),
|
||||
writeUnavailable: z.literal(1).optional()
|
||||
})
|
||||
.optional()
|
||||
})
|
||||
|
|
@ -1675,6 +1688,20 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
sendFrame(streamId, TerminalStreamOpcode.Error, encodeTerminalStreamText(message))
|
||||
emit({ type: 'error', streamId, message })
|
||||
}
|
||||
const notifyStreamWriteUnavailable = (
|
||||
stream: TerminalMultiplexStream,
|
||||
outcome: TerminalStreamInputOutcome
|
||||
): void => {
|
||||
if (
|
||||
closed ||
|
||||
streams.get(stream.streamId) !== stream ||
|
||||
outcome !== 'rejected' ||
|
||||
!stream.supportsWriteUnavailable
|
||||
) {
|
||||
return
|
||||
}
|
||||
sendFrame(stream.streamId, TerminalStreamOpcode.WriteUnavailable)
|
||||
}
|
||||
const sendResizedFrame = (
|
||||
stream: TerminalMultiplexStream,
|
||||
event: { cols: number; rows: number; displayMode: string; reason: string; seq?: number }
|
||||
|
|
@ -2127,16 +2154,17 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
}
|
||||
// Mobile already has the higher-priority floor, so a rejected desktop claim must not suppress later phone input.
|
||||
const inputClaimTail = stream.isMobile ? Promise.resolve(true) : stream.desktopClaimTail
|
||||
void inputClaimTail.then((claimed) => {
|
||||
void inputClaimTail.then(async (claimed) => {
|
||||
if (!claimed || isTerminalInputLockedForClient(runtime, stream.ptyId, stream.client)) {
|
||||
return
|
||||
}
|
||||
return sendTerminalStreamInput(runtime, {
|
||||
const outcome = await sendTerminalStreamInput(runtime, {
|
||||
terminal: stream.terminal,
|
||||
text,
|
||||
client: stream.client,
|
||||
isMobile: stream.isMobile
|
||||
})
|
||||
notifyStreamWriteUnavailable(stream, outcome)
|
||||
})
|
||||
return
|
||||
}
|
||||
|
|
@ -2476,6 +2504,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
ackInFlightBytes: 0,
|
||||
ackWindowBytes: TERMINAL_MULTIPLEX_ACK_STREAM_INITIAL_WINDOW_BYTES,
|
||||
supportsOutputPause: request.capabilities?.outputPause === 1,
|
||||
supportsWriteUnavailable: request.capabilities?.writeUnavailable === 1,
|
||||
outputPaused: false,
|
||||
supportsDesktopViewportClaims: request.capabilities?.desktopViewportClaims === 1,
|
||||
desktopClaimTail: Promise.resolve(true),
|
||||
|
|
@ -2892,6 +2921,7 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
: runtime.getRendererTerminalSerializerGeneration(ptyId)
|
||||
: 0
|
||||
const supportsDesktopViewportClaims = params.capabilities?.desktopViewportClaims === 1
|
||||
const supportsWriteUnavailable = params.capabilities?.writeUnavailable === 1
|
||||
if (mobileInputLeaseOnly && clientId) {
|
||||
let closed = false
|
||||
let resolveStream = (): void => {}
|
||||
|
|
@ -3139,12 +3169,15 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
|
|||
if (!claimed || isTerminalInputLockedForClient(runtime, ptyId, params.client)) {
|
||||
return
|
||||
}
|
||||
await sendTerminalStreamInput(runtime, {
|
||||
const outcome = await sendTerminalStreamInput(runtime, {
|
||||
terminal: params.terminal,
|
||||
text,
|
||||
client: params.client,
|
||||
isMobile
|
||||
})
|
||||
if (!closed && outcome === 'rejected' && supportsWriteUnavailable) {
|
||||
sendFrame(TerminalStreamOpcode.WriteUnavailable)
|
||||
}
|
||||
})
|
||||
return
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ import {
|
|||
import { SshPtyOutputIntake, type SshPtyOutputDataEvent } from '../../ipc/ssh-pty-output-intake'
|
||||
|
||||
const SET_OUTPUT_PAUSED_OPCODE = 16 as TerminalStreamOpcode
|
||||
const WRITE_UNAVAILABLE_OPCODE = 17 as TerminalStreamOpcode
|
||||
|
||||
function stubRuntime(overrides: Partial<OrcaRuntimeService> = {}): OrcaRuntimeService {
|
||||
const serializeAuthoritativeTerminalBuffer =
|
||||
|
|
@ -131,7 +132,8 @@ function startDesktopMultiplexSubscribe(
|
|||
}
|
||||
|
||||
function sendDesktopMultiplexSubscribe(
|
||||
handlers: Map<number, (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void>
|
||||
handlers: Map<number, (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void>,
|
||||
capabilities: Record<string, 1> = { ackOutput: 1, desktopViewportClaims: 1 }
|
||||
) {
|
||||
handlers.get(0)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
|
|
@ -143,7 +145,7 @@ function sendDesktopMultiplexSubscribe(
|
|||
streamId: 7,
|
||||
terminal: 'terminal-1',
|
||||
client: { id: 'desktop-1', type: 'desktop' },
|
||||
capabilities: { ackOutput: 1, desktopViewportClaims: 1 },
|
||||
capabilities,
|
||||
viewport: { cols: 120, rows: 40 }
|
||||
})
|
||||
})
|
||||
|
|
@ -151,6 +153,135 @@ function sendDesktopMultiplexSubscribe(
|
|||
)
|
||||
}
|
||||
|
||||
describe('terminal multiplex rejected input signalling', () => {
|
||||
it('reports when locally accepted input never reaches the process', async () => {
|
||||
const processWrites: string[] = []
|
||||
const sendTerminal = vi.fn().mockRejectedValue(new Error('terminal_not_writable'))
|
||||
const harness = startDesktopMultiplexSubscribe({
|
||||
sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal']
|
||||
})
|
||||
await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
|
||||
sendDesktopMultiplexSubscribe(harness.handlers, {
|
||||
ackOutput: 1,
|
||||
desktopViewportClaims: 1,
|
||||
writeUnavailable: 1
|
||||
})
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
harness.messages.some((message) => JSON.parse(message).result?.type === 'subscribed')
|
||||
).toBe(true)
|
||||
)
|
||||
harness.binaryFrames.splice(0)
|
||||
|
||||
const clientInputHandler = harness.handlers.get(7)
|
||||
expect(clientInputHandler).toBeDefined()
|
||||
clientInputHandler?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
streamId: 7,
|
||||
seq: 2,
|
||||
payload: encodeTerminalStreamText('x')
|
||||
})
|
||||
)!
|
||||
)
|
||||
const clientReportedAccepted = true
|
||||
|
||||
expect(clientReportedAccepted).toBe(true)
|
||||
await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalledOnce())
|
||||
expect(processWrites).toEqual([])
|
||||
await vi.waitFor(() =>
|
||||
expect(harness.binaryFrames.some((frame) => frame[2] === WRITE_UNAVAILABLE_OPCODE)).toBe(true)
|
||||
)
|
||||
})
|
||||
|
||||
it('does not send an unknown opcode to a legacy client', async () => {
|
||||
const sendTerminal = vi.fn().mockRejectedValue(new Error('terminal_not_writable'))
|
||||
const harness = startDesktopMultiplexSubscribe({
|
||||
sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal']
|
||||
})
|
||||
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)
|
||||
)
|
||||
harness.binaryFrames.splice(0)
|
||||
|
||||
harness.handlers.get(7)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
streamId: 7,
|
||||
seq: 2,
|
||||
payload: encodeTerminalStreamText('x')
|
||||
})
|
||||
)!
|
||||
)
|
||||
|
||||
await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalledOnce())
|
||||
expect(harness.binaryFrames.some((frame) => frame[2] === WRITE_UNAVAILABLE_OPCODE)).toBe(false)
|
||||
})
|
||||
|
||||
it('does not report a late rejection to a replacement stream with the same id', async () => {
|
||||
let settleWrite: (result: {
|
||||
handle: string
|
||||
accepted: boolean
|
||||
bytesWritten: number
|
||||
}) => void = () => {}
|
||||
const hostWrite = new Promise<{ handle: string; accepted: boolean; bytesWritten: number }>(
|
||||
(resolve) => {
|
||||
settleWrite = resolve
|
||||
}
|
||||
)
|
||||
const sendTerminal = vi.fn(() => hostWrite)
|
||||
const harness = startDesktopMultiplexSubscribe({
|
||||
sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal']
|
||||
})
|
||||
await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true))
|
||||
const capabilities = { ackOutput: 1 as const, writeUnavailable: 1 as const }
|
||||
sendDesktopMultiplexSubscribe(harness.handlers, capabilities)
|
||||
await vi.waitFor(() => expect(harness.handlers.has(7)).toBe(true))
|
||||
harness.binaryFrames.splice(0)
|
||||
|
||||
harness.handlers.get(7)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
streamId: 7,
|
||||
seq: 2,
|
||||
payload: encodeTerminalStreamText('old')
|
||||
})
|
||||
)!
|
||||
)
|
||||
await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalledOnce())
|
||||
harness.handlers.get(7)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Unsubscribe,
|
||||
streamId: 7,
|
||||
seq: 3,
|
||||
payload: new Uint8Array()
|
||||
})
|
||||
)!
|
||||
)
|
||||
sendDesktopMultiplexSubscribe(harness.handlers, capabilities)
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
harness.messages.filter((message) => JSON.parse(message).result?.type === 'subscribed')
|
||||
).toHaveLength(2)
|
||||
)
|
||||
harness.binaryFrames.splice(0)
|
||||
|
||||
settleWrite({ handle: 'terminal-1', accepted: false, bytesWritten: 0 })
|
||||
await hostWrite
|
||||
await Promise.resolve()
|
||||
|
||||
expect(harness.binaryFrames.some((frame) => frame[2] === WRITE_UNAVAILABLE_OPCODE)).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
function sendDesktopSourceRangeSubscribe(
|
||||
handlers: Map<number, (frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void>
|
||||
) {
|
||||
|
|
@ -3625,6 +3756,187 @@ describe('terminal multiplex RPC', () => {
|
|||
await dispatchPromise
|
||||
})
|
||||
|
||||
it('reports rejected input on a capable legacy binary stream', async () => {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
const handlers = new Map<
|
||||
number,
|
||||
(frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
|
||||
>()
|
||||
const cleanups = new Map<string, () => void>()
|
||||
const sendTerminal = vi.fn().mockRejectedValue(new Error('terminal_not_writable'))
|
||||
const runtime = stubRuntime({
|
||||
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
||||
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
|
||||
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
||||
cleanups.set(id, cleanup)
|
||||
}),
|
||||
cleanupSubscription: vi.fn((id: string) => cleanups.get(id)?.()),
|
||||
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
||||
sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal'],
|
||||
updateDesktopViewport: vi.fn().mockResolvedValue(true)
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.subscribe', {
|
||||
terminal: 'terminal-1',
|
||||
client: { id: 'desktop-1', type: 'desktop' },
|
||||
capabilities: { terminalBinaryStream: 1, writeUnavailable: 1 }
|
||||
}),
|
||||
(message) => messages.push(message),
|
||||
{
|
||||
connectionId: 'conn-subscribe-rejected-input',
|
||||
sendBinary: (bytes) => {
|
||||
binaryFrames.push(bytes)
|
||||
},
|
||||
registerBinaryStreamHandler: (streamId, handler) => {
|
||||
handlers.set(streamId, handler)
|
||||
return () => handlers.delete(streamId)
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((message) => JSON.parse(message).result?.type === 'subscribed')).toBe(
|
||||
true
|
||||
)
|
||||
)
|
||||
const streamId = JSON.parse(
|
||||
messages.find((message) => JSON.parse(message).result?.type === 'subscribed')!
|
||||
).result.streamId as number
|
||||
binaryFrames.splice(0)
|
||||
handlers.get(streamId)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
streamId,
|
||||
seq: 1,
|
||||
payload: encodeTerminalStreamText('x')
|
||||
})
|
||||
)!
|
||||
)
|
||||
|
||||
await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalledOnce())
|
||||
await vi.waitFor(() =>
|
||||
expect(binaryFrames.some((frame) => frame[2] === WRITE_UNAVAILABLE_OPCODE)).toBe(true)
|
||||
)
|
||||
runtime.cleanupSubscription('terminal-1:desktop-1')
|
||||
await dispatchPromise
|
||||
})
|
||||
|
||||
it('never sends the rejection opcode to an un-negotiated legacy binary stream', async () => {
|
||||
// The mobile client is exactly this subscriber: it declares
|
||||
// terminalBinaryStream and nothing else, and its vendored opcode enum knows
|
||||
// nothing past 12, so an unsolicited 17 is an unknown opcode on that wire.
|
||||
// A capable desktop subscriber shares the runtime and is driven second, so
|
||||
// its frame proves the rejection had already been processed for both.
|
||||
const cleanups = new Map<string, () => void>()
|
||||
const sendTerminal = vi.fn().mockRejectedValue(new Error('terminal_not_writable'))
|
||||
const runtime = stubRuntime({
|
||||
resolveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
||||
serializeTerminalBuffer: vi.fn().mockResolvedValue(null),
|
||||
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
registerSubscriptionCleanup: vi.fn((id: string, cleanup: () => void) => {
|
||||
cleanups.set(id, cleanup)
|
||||
}),
|
||||
cleanupSubscription: vi.fn((id: string) => cleanups.get(id)?.()),
|
||||
waitForTerminal: vi.fn(() => new Promise<RuntimeTerminalWait>(() => {})),
|
||||
sendTerminal: sendTerminal as unknown as OrcaRuntimeService['sendTerminal'],
|
||||
updateDesktopViewport: vi.fn().mockResolvedValue(true),
|
||||
handleMobileSubscribe: vi.fn(),
|
||||
handleMobileUnsubscribe: vi.fn(),
|
||||
updateMobileViewport: vi.fn().mockResolvedValue(true)
|
||||
})
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
|
||||
async function subscribeLegacyBinary(
|
||||
client: { id: string; type: 'mobile' | 'desktop' },
|
||||
capabilities: Record<string, 1>
|
||||
) {
|
||||
const messages: string[] = []
|
||||
const binaryFrames: Uint8Array<ArrayBufferLike>[] = []
|
||||
const handlers = new Map<
|
||||
number,
|
||||
(frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
|
||||
>()
|
||||
const dispatchPromise = dispatcher.dispatchStreaming(
|
||||
makeRequest('terminal.subscribe', { terminal: 'terminal-1', client, capabilities }),
|
||||
(message) => messages.push(message),
|
||||
{
|
||||
connectionId: `conn-legacy-${client.id}`,
|
||||
sendBinary: (bytes) => {
|
||||
binaryFrames.push(bytes)
|
||||
},
|
||||
registerBinaryStreamHandler: (streamId, handler) => {
|
||||
handlers.set(streamId, handler)
|
||||
return () => handlers.delete(streamId)
|
||||
}
|
||||
}
|
||||
)
|
||||
await vi.waitFor(() =>
|
||||
expect(messages.some((message) => JSON.parse(message).result?.type === 'subscribed')).toBe(
|
||||
true
|
||||
)
|
||||
)
|
||||
const streamId = JSON.parse(
|
||||
messages.find((message) => JSON.parse(message).result?.type === 'subscribed')!
|
||||
).result.streamId as number
|
||||
binaryFrames.splice(0)
|
||||
return {
|
||||
binaryFrames,
|
||||
dispatchPromise,
|
||||
sendInput: () =>
|
||||
handlers.get(streamId)?.(
|
||||
decodeTerminalStreamFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
streamId,
|
||||
seq: 1,
|
||||
payload: encodeTerminalStreamText('x')
|
||||
})
|
||||
)!
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
const legacy = await subscribeLegacyBinary(
|
||||
{ id: 'mobile-1', type: 'mobile' },
|
||||
{ terminalBinaryStream: 1 }
|
||||
)
|
||||
const capable = await subscribeLegacyBinary(
|
||||
{ id: 'desktop-1', type: 'desktop' },
|
||||
{ terminalBinaryStream: 1, writeUnavailable: 1 }
|
||||
)
|
||||
|
||||
legacy.sendInput()
|
||||
capable.sendInput()
|
||||
|
||||
await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalledTimes(2))
|
||||
await vi.waitFor(() =>
|
||||
expect(capable.binaryFrames.some((frame) => frame[2] === WRITE_UNAVAILABLE_OPCODE)).toBe(true)
|
||||
)
|
||||
expect(legacy.binaryFrames.some((frame) => frame[2] === WRITE_UNAVAILABLE_OPCODE)).toBe(false)
|
||||
|
||||
runtime.cleanupSubscription('terminal-1:mobile-1')
|
||||
runtime.cleanupSubscription('terminal-1:desktop-1')
|
||||
await Promise.all([legacy.dispatchPromise, capable.dispatchPromise])
|
||||
})
|
||||
|
||||
it('owns and releases a viewport floor for legacy JSON desktop streams', async () => {
|
||||
const messages: string[] = []
|
||||
const cleanups = new Map<string, () => void>()
|
||||
|
|
|
|||
|
|
@ -4007,10 +4007,17 @@ export function connectPanePty(
|
|||
}
|
||||
const storePtyId = useAppStore.getState().ptyIdsByTabId?.[deps.tabId]?.[0] ?? null
|
||||
const undeliverablePtyId = transport.getPtyId() ?? storePtyId
|
||||
// Why the split: for a local (daemon/app-SSH) id main's registry can answer,
|
||||
// and a `false` there means the shell really died — the dead-session
|
||||
// reconcile owns that teardown and a remount would race it. For a `remote:`
|
||||
// id main owns no registry entry, so `pty:hasPty` routes to the local
|
||||
// provider and fabricates "dead"; that answer blocked every recovery this
|
||||
// signal exists to trigger (STA-2830). The host's own rejection replaces it.
|
||||
const hostRejectedRemoteInput = providerRejected && isRemoteRuntimePtyId(undeliverablePtyId)
|
||||
void requestTerminalPaneRecovery({
|
||||
tabId: deps.tabId,
|
||||
ptyId: undeliverablePtyId,
|
||||
reason: 'input-undeliverable',
|
||||
reason: hostRejectedRemoteInput ? 'input-rejected-by-host' : 'input-undeliverable',
|
||||
terminalRecoveryGeneration,
|
||||
terminalRecoveryInstanceId: terminalRecoveryInstance.id,
|
||||
// Why: pty:hasPty answers null for ids the local registry doesn't own,
|
||||
|
|
|
|||
|
|
@ -41,7 +41,13 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
terminal: string
|
||||
client: { id: string; type: string }
|
||||
viewport?: { cols: number; rows: number }
|
||||
capabilities?: { desktopViewportClaims?: 1 }
|
||||
capabilities?: {
|
||||
ackOutput?: 1
|
||||
ackOutputSourceRanges?: 1
|
||||
desktopViewportClaims?: 1
|
||||
outputPause?: 1
|
||||
writeUnavailable?: 1
|
||||
}
|
||||
} {
|
||||
const frames = subscriptionSendBinary.mock.calls
|
||||
.map((call) => decodeTerminalStreamFrame(call[0]))
|
||||
|
|
@ -55,7 +61,13 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
terminal: string
|
||||
client: { id: string; type: string }
|
||||
viewport?: { cols: number; rows: number }
|
||||
capabilities?: { desktopViewportClaims?: 1 }
|
||||
capabilities?: {
|
||||
ackOutput?: 1
|
||||
ackOutputSourceRanges?: 1
|
||||
desktopViewportClaims?: 1
|
||||
outputPause?: 1
|
||||
writeUnavailable?: 1
|
||||
}
|
||||
}>(frame.payload)
|
||||
if (!payload) {
|
||||
throw new Error('invalid terminal subscribe payload')
|
||||
|
|
@ -265,7 +277,8 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
ackOutput: 1,
|
||||
ackOutputSourceRanges: 1,
|
||||
desktopViewportClaims: 1,
|
||||
outputPause: 1
|
||||
outputPause: 1,
|
||||
writeUnavailable: 1
|
||||
})
|
||||
)
|
||||
expect(runtimeSubscribe).toHaveBeenCalledWith(
|
||||
|
|
@ -284,6 +297,84 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
})
|
||||
})
|
||||
|
||||
it('reports a rejected multiplex write through the pane recovery callback', async () => {
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const onWriteUnavailable = vi.fn()
|
||||
const onError = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:terminal-1',
|
||||
cols: 120,
|
||||
rows: 40,
|
||||
callbacks: { onError, onWriteUnavailable }
|
||||
})
|
||||
await vi.waitFor(() => expect(subscriptionSendBinary).toHaveBeenCalled())
|
||||
const { streamId } = latestSubscribePayload()
|
||||
|
||||
subscriptionCallbacks?.onBinary?.(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.WriteUnavailable,
|
||||
streamId,
|
||||
seq: 1,
|
||||
payload: new Uint8Array()
|
||||
})
|
||||
)
|
||||
|
||||
expect(onWriteUnavailable).toHaveBeenCalledOnce()
|
||||
expect(onError).not.toHaveBeenCalled()
|
||||
transport.destroy?.()
|
||||
})
|
||||
|
||||
it('does not report a rejected write from a superseded multiplex stream', async () => {
|
||||
const callbacksByAttempt: NonNullable<typeof subscriptionCallbacks>[] = []
|
||||
runtimeSubscribe.mockImplementation(
|
||||
async (_args: unknown, callbacks: NonNullable<typeof subscriptionCallbacks>) => {
|
||||
callbacksByAttempt.push(callbacks)
|
||||
subscriptionCallbacks = callbacks
|
||||
queueMicrotask(() => callbacks.onResponse({ ok: true, result: { type: 'ready' } }))
|
||||
return { unsubscribe: vi.fn(), sendBinary: subscriptionSendBinary }
|
||||
}
|
||||
)
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const onWriteUnavailable = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
|
||||
resolvedPaneHandle = 'terminal-old'
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:env-1@@terminal-old',
|
||||
callbacks: { onWriteUnavailable }
|
||||
})
|
||||
await vi.waitFor(() => expect(callbacksByAttempt).toHaveLength(1))
|
||||
const oldStreamId = latestSubscribePayload().streamId
|
||||
resolvedPaneHandle = 'terminal-new'
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:env-1@@terminal-new',
|
||||
callbacks: { onWriteUnavailable }
|
||||
})
|
||||
await vi.waitFor(() => expect(latestSubscribePayload().terminal).toBe('terminal-new'))
|
||||
|
||||
callbacksByAttempt[0]?.onBinary?.(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.WriteUnavailable,
|
||||
streamId: oldStreamId,
|
||||
seq: 1,
|
||||
payload: new Uint8Array()
|
||||
})
|
||||
)
|
||||
|
||||
expect(onWriteUnavailable).not.toHaveBeenCalled()
|
||||
transport.destroy?.()
|
||||
})
|
||||
|
||||
it('does not report attachment health until the authoritative PTY snapshot arrives', async () => {
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', { worktreeId: 'wt-1' })
|
||||
|
|
@ -3431,6 +3522,159 @@ describe('createRemoteRuntimePtyTransport', () => {
|
|||
expect(transport.isConnected()).toBe(false)
|
||||
})
|
||||
|
||||
it('reports rejected input from the one-shot runtime fallback', async () => {
|
||||
vi.useFakeTimers()
|
||||
runtimeSubscribe.mockImplementation(
|
||||
async (_args: unknown, callbacks: typeof subscriptionCallbacks) => {
|
||||
subscriptionCallbacks = callbacks
|
||||
return { unsubscribe: vi.fn(), sendBinary: subscriptionSendBinary }
|
||||
}
|
||||
)
|
||||
try {
|
||||
const defaultRuntimeCall = runtimeCall.getMockImplementation()
|
||||
runtimeCall.mockImplementation((args: { method: string }) => {
|
||||
if (args.method === 'terminal.send') {
|
||||
return Promise.resolve({
|
||||
ok: true,
|
||||
result: { send: { handle: 'terminal-1', accepted: false, bytesWritten: 0 } }
|
||||
})
|
||||
}
|
||||
return defaultRuntimeCall?.(args)
|
||||
})
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const onWriteUnavailable = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:env-1@@terminal-1',
|
||||
callbacks: { onWriteUnavailable }
|
||||
})
|
||||
await vi.waitFor(() => expect(transport.getPtyId()).toBe('remote:env-1@@terminal-1'))
|
||||
expect(transport.sendInput('x')).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(8)
|
||||
|
||||
await vi.waitFor(() => expect(onWriteUnavailable).toHaveBeenCalledOnce())
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('reports terminal_not_writable from the one-shot runtime fallback', async () => {
|
||||
vi.useFakeTimers()
|
||||
runtimeSubscribe.mockImplementation(
|
||||
async (_args: unknown, callbacks: typeof subscriptionCallbacks) => {
|
||||
subscriptionCallbacks = callbacks
|
||||
return { unsubscribe: vi.fn(), sendBinary: subscriptionSendBinary }
|
||||
}
|
||||
)
|
||||
try {
|
||||
const defaultRuntimeCall = runtimeCall.getMockImplementation()
|
||||
runtimeCall.mockImplementation((args: { method: string }) => {
|
||||
if (args.method === 'terminal.send') {
|
||||
return Promise.resolve({
|
||||
ok: false,
|
||||
error: { code: 'internal_error', message: 'terminal_not_writable' }
|
||||
})
|
||||
}
|
||||
return defaultRuntimeCall?.(args)
|
||||
})
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const onWriteUnavailable = vi.fn()
|
||||
const onError = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:env-1@@terminal-1',
|
||||
callbacks: { onWriteUnavailable, onError }
|
||||
})
|
||||
await vi.waitFor(() => expect(transport.getPtyId()).toBe('remote:env-1@@terminal-1'))
|
||||
expect(transport.sendInput('x')).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(8)
|
||||
|
||||
await vi.waitFor(() => expect(onWriteUnavailable).toHaveBeenCalledOnce())
|
||||
expect(onError).not.toHaveBeenCalled()
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('does not report a delayed fallback rejection after same-handle reattach', async () => {
|
||||
vi.useFakeTimers()
|
||||
runtimeSubscribe.mockImplementation(
|
||||
async (_args: unknown, callbacks: typeof subscriptionCallbacks) => {
|
||||
subscriptionCallbacks = callbacks
|
||||
return { unsubscribe: vi.fn(), sendBinary: subscriptionSendBinary }
|
||||
}
|
||||
)
|
||||
try {
|
||||
let settleSend: (response: unknown) => void = () => {}
|
||||
const sendResponse = new Promise((resolve) => {
|
||||
settleSend = resolve
|
||||
})
|
||||
const defaultRuntimeCall = runtimeCall.getMockImplementation()
|
||||
runtimeCall.mockImplementation((args: { method: string }) => {
|
||||
if (args.method === 'terminal.send') {
|
||||
return sendResponse
|
||||
}
|
||||
return defaultRuntimeCall?.(args)
|
||||
})
|
||||
const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport')
|
||||
const oldWriteUnavailable = vi.fn()
|
||||
const replacementWriteUnavailable = vi.fn()
|
||||
const transport = createRemoteRuntimePtyTransport('env-1', {
|
||||
worktreeId: 'wt-1',
|
||||
tabId: 'tab-1',
|
||||
leafId: 'pane:1'
|
||||
})
|
||||
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:env-1@@terminal-1',
|
||||
callbacks: { onWriteUnavailable: oldWriteUnavailable }
|
||||
})
|
||||
await vi.waitFor(() => expect(transport.getPtyId()).toBe('remote:env-1@@terminal-1'))
|
||||
expect(transport.sendInput('old')).toBe(true)
|
||||
await vi.advanceTimersByTimeAsync(8)
|
||||
await vi.waitFor(() =>
|
||||
expect(runtimeCall).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ method: 'terminal.send' })
|
||||
)
|
||||
)
|
||||
|
||||
transport.detach?.()
|
||||
transport.attach({
|
||||
existingPtyId: 'remote:env-1@@terminal-1',
|
||||
callbacks: { onWriteUnavailable: replacementWriteUnavailable }
|
||||
})
|
||||
await vi.waitFor(() =>
|
||||
expect(
|
||||
runtimeCall.mock.calls.filter((call) => call[0].method === 'terminal.resolvePane')
|
||||
).toHaveLength(2)
|
||||
)
|
||||
settleSend({
|
||||
ok: true,
|
||||
result: { send: { handle: 'terminal-1', accepted: false, bytesWritten: 0 } }
|
||||
})
|
||||
await sendResponse
|
||||
await Promise.resolve()
|
||||
|
||||
expect(oldWriteUnavailable).not.toHaveBeenCalled()
|
||||
expect(replacementWriteUnavailable).not.toHaveBeenCalled()
|
||||
transport.destroy?.()
|
||||
} finally {
|
||||
vi.useRealTimers()
|
||||
}
|
||||
})
|
||||
|
||||
it('drops pending input when attaching a different remote terminal handle', async () => {
|
||||
vi.useFakeTimers()
|
||||
runtimeSubscribe.mockImplementation(
|
||||
|
|
|
|||
|
|
@ -1251,8 +1251,15 @@ export function createRemoteRuntimePtyTransport(
|
|||
}
|
||||
}
|
||||
|
||||
function notifyWriteUnavailable(): void {
|
||||
if (!destroyed) {
|
||||
storedCallbacks.onWriteUnavailable?.()
|
||||
}
|
||||
}
|
||||
|
||||
const inputBatcher = createRemoteRuntimePtyTextBatcher(REMOTE_TERMINAL_INPUT_FLUSH_MS, (text) => {
|
||||
const targetHandle = handle
|
||||
const targetLifecycleEpoch = lifecycleEpoch
|
||||
if (!connected || !targetHandle || recoveryBlocksIo()) {
|
||||
return
|
||||
}
|
||||
|
|
@ -1265,16 +1272,32 @@ export function createRemoteRuntimePtyTransport(
|
|||
pendingClaimInput += text
|
||||
return
|
||||
}
|
||||
void callRuntime('terminal.send', {
|
||||
void callRuntime<{ send: RuntimeTerminalSend }>('terminal.send', {
|
||||
terminal: targetHandle,
|
||||
text,
|
||||
client: { id: clientId, type: 'desktop' },
|
||||
...(desiredViewport ? { viewport: desiredViewport, claimViewport: true as const } : {})
|
||||
}).catch((error) => {
|
||||
if (handle === targetHandle) {
|
||||
handleRemoteTerminalError(error)
|
||||
}
|
||||
})
|
||||
.then((result) => {
|
||||
if (
|
||||
connected &&
|
||||
lifecycleEpoch === targetLifecycleEpoch &&
|
||||
handle === targetHandle &&
|
||||
result.send.accepted !== true
|
||||
) {
|
||||
notifyWriteUnavailable()
|
||||
}
|
||||
})
|
||||
.catch((error) => {
|
||||
if (lifecycleEpoch !== targetLifecycleEpoch || handle !== targetHandle) {
|
||||
return
|
||||
}
|
||||
if (runtimeTerminalErrorMessage(error).includes('terminal_not_writable')) {
|
||||
notifyWriteUnavailable()
|
||||
} else {
|
||||
handleRemoteTerminalError(error)
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
function sendViewportUpdate(cols: number, rows: number, claim = false): void {
|
||||
|
|
@ -1825,6 +1848,11 @@ export function createRemoteRuntimePtyTransport(
|
|||
setDriverForPty(subscribedPtyId, driver)
|
||||
}
|
||||
},
|
||||
onWriteUnavailable: () => {
|
||||
if (isCurrentSubscription()) {
|
||||
notifyWriteUnavailable()
|
||||
}
|
||||
},
|
||||
onTransportClose: ({ recoverable, retryWithBackoff }) => {
|
||||
transportClosed = true
|
||||
if (generation !== subscriptionGeneration) {
|
||||
|
|
|
|||
|
|
@ -443,6 +443,56 @@ describe('requestTerminalPaneRecovery', () => {
|
|||
expect(mocks.remountTerminalTabForRecovery).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
// A `remote:` id has no entry in main's registry, so pty:hasPty routes it to
|
||||
// the local provider. Every answer that path can produce blocked the remount
|
||||
// this signal exists to trigger (STA-2830); none of them is evidence.
|
||||
describe('host-rejected input', () => {
|
||||
for (const [label, liveness] of [
|
||||
['a fabricated dead answer', async () => false],
|
||||
['an explicit unknown', async () => null],
|
||||
[
|
||||
'a failed probe',
|
||||
async () => {
|
||||
throw new Error('ipc down')
|
||||
}
|
||||
]
|
||||
] as [string, () => Promise<boolean | null>][]) {
|
||||
it(`recovers even though the local probe would give ${label}`, async () => {
|
||||
mocks.hasPty.mockImplementation(liveness)
|
||||
|
||||
const result = await requestTerminalPaneRecovery({
|
||||
tabId: 'tab-1',
|
||||
ptyId: 'remote:env-1@@terminal-1',
|
||||
reason: 'input-rejected-by-host',
|
||||
requireAuthoritativeLiveness: true,
|
||||
endpointReplaced: true
|
||||
})
|
||||
|
||||
expect(result).toBe(true)
|
||||
expect(mocks.hasPty).not.toHaveBeenCalled()
|
||||
expect(mocks.remountTerminalTabForRecovery).toHaveBeenCalledWith('tab-1')
|
||||
})
|
||||
}
|
||||
|
||||
it('still coalesces under the shared cooldown', async () => {
|
||||
expect(
|
||||
await requestTerminalPaneRecovery({
|
||||
tabId: 'tab-1',
|
||||
ptyId: 'remote:env-1@@terminal-1',
|
||||
reason: 'input-rejected-by-host'
|
||||
})
|
||||
).toBe(true)
|
||||
expect(
|
||||
await requestTerminalPaneRecovery({
|
||||
tabId: 'tab-1',
|
||||
ptyId: 'remote:env-1@@terminal-1',
|
||||
reason: 'input-rejected-by-host'
|
||||
})
|
||||
).toBe(false)
|
||||
expect(mocks.remountTerminalTabForRecovery).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
|
||||
it('never throws when the store surface is partial (timer/callback contexts)', async () => {
|
||||
// Regression: recovery fires from stall-watch timers and write callbacks;
|
||||
// an environment with a partial store (mocked suites, teardown races) must
|
||||
|
|
|
|||
|
|
@ -20,6 +20,13 @@ export type TerminalPaneRecoveryReason =
|
|||
| 'write-stalled'
|
||||
| 'replay-wedged'
|
||||
| 'input-undeliverable'
|
||||
// The paired runtime that owns the PTY refused this write and said so on the
|
||||
// wire. Distinct from 'input-undeliverable' because it skips the liveness
|
||||
// probe: main's registry holds no entry for a `remote:` id, so `pty:hasPty`
|
||||
// routes it to the local provider and answers a fabricated "dead". The
|
||||
// rejection frame is the evidence instead — it came from the process that
|
||||
// owns the PTY, over a connection that is by construction still up.
|
||||
| 'input-rejected-by-host'
|
||||
// A restore was requested for a certified-dead pipeline (reveal path).
|
||||
| 'restore-blocked'
|
||||
|
||||
|
|
@ -36,7 +43,9 @@ type RecoveryRequest = {
|
|||
/** Remote panes (runtime mirrors, app-SSH) must prove the PTY alive before
|
||||
* an input-undeliverable remount: pty:hasPty answers null for ids the local
|
||||
* registry doesn't own, and treating null as "proceed" would let a
|
||||
* disconnected remote pane churn reconnects on every cooldown window. */
|
||||
* disconnected remote pane churn reconnects on every cooldown window. That
|
||||
* churn needs a *disconnected* pane, which is why 'input-rejected-by-host'
|
||||
* is exempt: its evidence arrives over a live connection. */
|
||||
requireAuthoritativeLiveness?: boolean
|
||||
/** The provider rejected the write because its endpoint stopped accepting
|
||||
* writes, so re-attach MAY land on a *fresh* shell (a respawn; a transient
|
||||
|
|
@ -183,7 +192,9 @@ function cancelPendingRecoveryRetry(tabId: string): void {
|
|||
*
|
||||
* For 'input-undeliverable' the PTY is liveness-checked first: a dead PTY is
|
||||
* the dead-session reconcile's job (it tears down and reports "Process
|
||||
* exited"), and remounting there would race it.
|
||||
* exited"), and remounting there would race it. 'input-rejected-by-host' skips
|
||||
* that probe — see the reason's declaration. Nothing here destroys a session
|
||||
* either way: a remount rebuilds the renderer over the PTY it already had.
|
||||
*/
|
||||
export async function requestTerminalPaneRecovery(request: RecoveryRequest): Promise<boolean> {
|
||||
if (!isCurrentTerminalRecoveryRequest(request)) {
|
||||
|
|
@ -199,6 +210,8 @@ export async function requestTerminalPaneRecovery(request: RecoveryRequest): Pro
|
|||
}
|
||||
return false
|
||||
}
|
||||
// 'input-rejected-by-host' is deliberately absent: no local probe can speak
|
||||
// for the id it carries, and its evidence already came from the PTY's owner.
|
||||
if (request.reason === 'input-undeliverable') {
|
||||
if (!request.ptyId) {
|
||||
return false
|
||||
|
|
|
|||
|
|
@ -73,6 +73,7 @@ export type RemoteRuntimeMultiplexedTerminalCallbacks = {
|
|||
onDriverChanged?: (
|
||||
driver: { kind: 'idle' } | { kind: 'desktop' } | { kind: 'mobile'; clientId: string }
|
||||
) => void
|
||||
onWriteUnavailable?: () => void
|
||||
onTransportClose?: (event: { recoverable: boolean; retryWithBackoff?: boolean }) => void
|
||||
}
|
||||
|
||||
|
|
@ -506,6 +507,7 @@ class RemoteRuntimeTerminalMultiplexer {
|
|||
ackOutput: 1,
|
||||
ackOutputSourceRanges: 1,
|
||||
outputPause: 1,
|
||||
writeUnavailable: 1,
|
||||
...(args.client.type === 'desktop' ? { desktopViewportClaims: 1 } : {})
|
||||
}
|
||||
})
|
||||
|
|
@ -735,6 +737,10 @@ class RemoteRuntimeTerminalMultiplexer {
|
|||
return
|
||||
}
|
||||
stream.watchdog.recordInbound()
|
||||
if (frame.opcode === TerminalStreamOpcode.WriteUnavailable) {
|
||||
stream.callbacks.onWriteUnavailable?.()
|
||||
return
|
||||
}
|
||||
if (
|
||||
frame.opcode === TerminalStreamOpcode.Output ||
|
||||
frame.opcode === TerminalStreamOpcode.OutputSpan
|
||||
|
|
|
|||
|
|
@ -106,6 +106,8 @@ describe('remote runtime terminal data subscriptions', () => {
|
|||
ackOutput?: 1
|
||||
ackOutputSourceRanges?: 1
|
||||
desktopViewportClaims?: 1
|
||||
outputPause?: 1
|
||||
writeUnavailable?: 1
|
||||
}
|
||||
}>(subscribeFrame.payload)
|
||||
expect(subscribePayload?.streamId).toEqual(expect.any(Number))
|
||||
|
|
@ -113,7 +115,8 @@ describe('remote runtime terminal data subscriptions', () => {
|
|||
ackOutput: 1,
|
||||
ackOutputSourceRanges: 1,
|
||||
desktopViewportClaims: 1,
|
||||
outputPause: 1
|
||||
outputPause: 1,
|
||||
writeUnavailable: 1
|
||||
})
|
||||
|
||||
callbacks?.onBinary?.(
|
||||
|
|
|
|||
|
|
@ -30,7 +30,9 @@ export enum TerminalStreamOpcode {
|
|||
ClaimViewport = 14,
|
||||
OutputSpan = 15,
|
||||
// Negotiated per stream; older hosts reject unknown opcodes, so clients send only after capability confirmation.
|
||||
SetOutputPaused = 16
|
||||
SetOutputPaused = 16,
|
||||
// Negotiated per stream because older clients reject unknown opcodes.
|
||||
WriteUnavailable = 17
|
||||
}
|
||||
|
||||
export type TerminalStreamFrame = {
|
||||
|
|
@ -119,6 +121,7 @@ function isTerminalStreamOpcode(value: number): value is TerminalStreamOpcode {
|
|||
value === TerminalStreamOpcode.Ack ||
|
||||
value === TerminalStreamOpcode.ClaimViewport ||
|
||||
value === TerminalStreamOpcode.OutputSpan ||
|
||||
value === TerminalStreamOpcode.SetOutputPaused
|
||||
value === TerminalStreamOpcode.SetOutputPaused ||
|
||||
value === TerminalStreamOpcode.WriteUnavailable
|
||||
)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,436 @@
|
|||
/**
|
||||
* Carries a host-rejected paired-runtime write the whole way: the real
|
||||
* terminal.multiplex dispatcher rejects the authoritative PTY write, the real
|
||||
* renderer multiplexer and remote transport decode the WriteUnavailable frame,
|
||||
* and pty-connection must turn it into an actual tab remount.
|
||||
*
|
||||
* Every other test for this signal stops at a transport callback, so the last
|
||||
* hop was unproven — and that hop is where it died: pane recovery probes
|
||||
* `pty:hasPty`, which owns no registry entry for a `remote:` id. The parametrized
|
||||
* liveness answers below cover every reply main can produce for one.
|
||||
*/
|
||||
import { beforeEach, afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { RpcDispatcher } from '../../src/main/runtime/rpc/dispatcher'
|
||||
import { TERMINAL_METHODS } from '../../src/main/runtime/rpc/methods/terminal'
|
||||
import type { OrcaRuntimeService } from '../../src/main/runtime/orca-runtime'
|
||||
import {
|
||||
TerminalStreamOpcode,
|
||||
decodeTerminalStreamFrame
|
||||
} from '../../src/shared/terminal-stream-protocol'
|
||||
|
||||
const ENVIRONMENT_ID = 'env-1'
|
||||
const TERMINAL_HANDLE = 'terminal-1'
|
||||
const REMOTE_PTY_ID = `remote:${ENVIRONMENT_ID}@@${TERMINAL_HANDLE}`
|
||||
const LEAF_ID = '11111111-1111-4111-8111-111111111111'
|
||||
|
||||
type StoreState = Record<string, unknown>
|
||||
|
||||
let mockStoreState: StoreState
|
||||
let storeSubscribers: ((state: StoreState) => void)[] = []
|
||||
const remountTerminalTabForRecovery = vi.fn<(tabId: string) => boolean>(() => true)
|
||||
|
||||
vi.mock('@/store', () => ({
|
||||
useAppStore: {
|
||||
getState: () => mockStoreState,
|
||||
subscribe: (listener: (state: StoreState) => void) => {
|
||||
storeSubscribers.push(listener)
|
||||
return () => {
|
||||
storeSubscribers = storeSubscribers.filter((candidate) => candidate !== listener)
|
||||
}
|
||||
}
|
||||
}
|
||||
}))
|
||||
|
||||
vi.mock('@/runtime/sync-runtime-graph', () => ({ scheduleRuntimeGraphSync: vi.fn() }))
|
||||
vi.mock('@/components/terminal-pane/terminal-webgl-atlas-recovery', () => ({
|
||||
scheduleTerminalWebglAtlasRecovery: vi.fn()
|
||||
}))
|
||||
vi.mock('sonner', () => ({ toast: { info: vi.fn() } }))
|
||||
vi.mock('@/lib/codex-stale-pane-sweep', () => ({ notifyCodexPaneBoundForStaleSweep: vi.fn() }))
|
||||
vi.mock('@/runtime/web-runtime-session', () => ({
|
||||
refreshWebRuntimeSessionTabsSnapshot: vi.fn(async () => {})
|
||||
}))
|
||||
|
||||
/** One live paired host: the real dispatcher, wired to a runtime that refuses the write. */
|
||||
function startHost(): {
|
||||
bridge: {
|
||||
subscribe: (
|
||||
args: { method: string },
|
||||
callbacks: {
|
||||
onResponse: (response: unknown) => void
|
||||
onBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void
|
||||
onClose?: () => void
|
||||
}
|
||||
) => Promise<{ unsubscribe: () => void; sendBinary: (bytes: Uint8Array) => void }>
|
||||
call: (request: { method: string; params?: unknown }) => Promise<unknown>
|
||||
}
|
||||
sendTerminal: ReturnType<typeof vi.fn>
|
||||
/** Opcodes the host pushed to this client, in order. */
|
||||
hostOpcodes: number[]
|
||||
} {
|
||||
const hostOpcodes: number[] = []
|
||||
// The host's whole reason to emit the opcode: the PTY refused the bytes.
|
||||
const sendTerminal = vi.fn().mockResolvedValue({ accepted: false })
|
||||
const runtime = {
|
||||
getRuntimeId: () => 'test-runtime',
|
||||
registerRemoteTerminalViewSubscriber: () => () => {},
|
||||
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
|
||||
requestRendererTerminalTabMount: vi.fn().mockReturnValue(true),
|
||||
updateRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
|
||||
unregisterRemoteDesktopViewer: vi.fn().mockResolvedValue(true),
|
||||
unregisterRemoteDesktopViewers: vi.fn().mockResolvedValue(true),
|
||||
isPtyResizeDrivenRemotely: vi.fn().mockReturnValue(false),
|
||||
getRemoteDesktopFitHold: vi.fn().mockReturnValue({ mode: 'desktop-fit', cols: 120, rows: 40 }),
|
||||
isRemoteDesktopViewerOwner: vi.fn().mockReturnValue(false),
|
||||
getPtyOutputSequence: vi.fn().mockReturnValue(0),
|
||||
attachRemoteTerminalSourceRangeConsumer: vi.fn().mockReturnValue(false),
|
||||
detachRemoteTerminalSourceRangeConsumer: vi.fn(),
|
||||
getRendererTerminalSerializerGeneration: vi.fn().mockReturnValue(0),
|
||||
sendTerminal,
|
||||
readTerminal: vi.fn().mockResolvedValue({ tail: [], truncated: false }),
|
||||
serializeTerminalBuffer: vi.fn().mockResolvedValue({ data: 'snapshot', cols: 120, rows: 40 }),
|
||||
serializeAuthoritativeTerminalBuffer: vi
|
||||
.fn()
|
||||
.mockResolvedValue({ data: 'snapshot', cols: 120, rows: 40 }),
|
||||
getTerminalSize: vi.fn().mockReturnValue({ cols: 120, rows: 40 }),
|
||||
getMobileDisplayMode: vi.fn().mockReturnValue('auto'),
|
||||
getLayout: vi.fn().mockReturnValue({ seq: 1 }),
|
||||
subscribeToTerminalData: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToTerminalResize: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToFitOverrideChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
subscribeToDriverChanges: vi.fn().mockReturnValue(vi.fn()),
|
||||
getTerminalFitOverride: vi.fn().mockReturnValue(null),
|
||||
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
|
||||
registerSubscriptionCleanup: vi.fn(),
|
||||
cleanupSubscription: vi.fn(),
|
||||
waitForTerminal: vi.fn(() => new Promise<never>(() => {}))
|
||||
} as unknown as OrcaRuntimeService
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
|
||||
|
||||
const bridge = {
|
||||
async subscribe(
|
||||
args: { method: string },
|
||||
callbacks: {
|
||||
onResponse: (response: unknown) => void
|
||||
onBinary?: (bytes: Uint8Array<ArrayBufferLike>) => void
|
||||
onClose?: () => void
|
||||
}
|
||||
) {
|
||||
const handlers = new Map<
|
||||
number,
|
||||
(frame: NonNullable<ReturnType<typeof decodeTerminalStreamFrame>>) => void
|
||||
>()
|
||||
void dispatcher.dispatchStreaming(
|
||||
{ id: 'req-1', authToken: 'tok', method: args.method, params: {} },
|
||||
(message) => callbacks.onResponse(JSON.parse(message)),
|
||||
{
|
||||
connectionId: 'conn-e2e',
|
||||
sendBinary: (bytes) => {
|
||||
const opcode = decodeTerminalStreamFrame(bytes)?.opcode
|
||||
if (opcode !== undefined) {
|
||||
hostOpcodes.push(opcode)
|
||||
}
|
||||
callbacks.onBinary?.(bytes)
|
||||
return true
|
||||
},
|
||||
registerBinaryStreamHandler: (streamId, handler) => {
|
||||
handlers.set(streamId, handler)
|
||||
return () => {
|
||||
if (handlers.get(streamId) === handler) {
|
||||
handlers.delete(streamId)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
return {
|
||||
unsubscribe: vi.fn(),
|
||||
sendBinary: (bytes: Uint8Array) => {
|
||||
const frame = decodeTerminalStreamFrame(bytes)
|
||||
if (frame) {
|
||||
handlers.get(frame.streamId)?.(frame)
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
async call(request: { method: string; params?: unknown }) {
|
||||
if (request.method === 'terminal.resolvePane') {
|
||||
const params = request.params as { paneKey: string; worktreeId: string }
|
||||
const separator = params.paneKey.indexOf(':')
|
||||
return {
|
||||
ok: true,
|
||||
result: {
|
||||
terminal: {
|
||||
handle: TERMINAL_HANDLE,
|
||||
tabId: params.paneKey.slice(0, separator),
|
||||
leafId: params.paneKey.slice(separator + 1),
|
||||
worktreeId: params.worktreeId
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return { ok: true, result: { terminal: { handle: TERMINAL_HANDLE } } }
|
||||
}
|
||||
}
|
||||
return { bridge, sendTerminal, hostOpcodes }
|
||||
}
|
||||
|
||||
function createPane() {
|
||||
const activeBuffer = { type: 'normal' as const, viewportY: 0, baseY: 0, cursorY: 0, cursorX: 0 }
|
||||
const container = new EventTarget() as HTMLElement
|
||||
Object.defineProperty(container, 'dataset', { configurable: true, value: {} })
|
||||
const terminal = {
|
||||
cols: 120,
|
||||
rows: 40,
|
||||
element: {},
|
||||
buffer: { active: activeBuffer },
|
||||
modes: { bracketedPasteMode: false, sendFocusMode: false },
|
||||
options: { scrollback: 5_000, ignoreBracketedPasteMode: false, theme: {} },
|
||||
write: vi.fn((data: string, callback?: () => void) => {
|
||||
if (data === '' || callback?.name === 'runParsedSteps') {
|
||||
callback?.()
|
||||
}
|
||||
}),
|
||||
resize: vi.fn(),
|
||||
clear: vi.fn(),
|
||||
scrollToBottom: vi.fn(),
|
||||
scrollToLine: vi.fn(),
|
||||
scrollLines: vi.fn(),
|
||||
paste: vi.fn(),
|
||||
onData: vi.fn(() => ({ dispose: vi.fn() })),
|
||||
onResize: vi.fn(() => ({ dispose: vi.fn() })),
|
||||
onRender: vi.fn(() => ({ dispose: vi.fn() })),
|
||||
onTitleChange: vi.fn(() => ({ dispose: vi.fn() })),
|
||||
hasSelection: vi.fn(() => false),
|
||||
parser: {
|
||||
registerCsiHandler: vi.fn(() => ({ dispose: vi.fn() })),
|
||||
registerOscHandler: vi.fn(() => ({ dispose: vi.fn() }))
|
||||
}
|
||||
}
|
||||
return { id: 1, leafId: LEAF_ID, stablePaneId: LEAF_ID, terminal, container }
|
||||
}
|
||||
|
||||
function createManager() {
|
||||
const panes = [{ id: 1, leafId: LEAF_ID }]
|
||||
return {
|
||||
setPaneGpuRendering: vi.fn(),
|
||||
markPaneHasComplexScriptOutput: vi.fn(),
|
||||
rebuildPaneWebgl: vi.fn(),
|
||||
hasWebglRenderer: vi.fn(() => false),
|
||||
getPanes: vi.fn(() => panes),
|
||||
closePane: vi.fn(),
|
||||
getActivePane: vi.fn(() => panes[0]),
|
||||
getNumericIdForLeaf: vi.fn(() => 1),
|
||||
setActivePane: vi.fn()
|
||||
}
|
||||
}
|
||||
|
||||
function createDeps() {
|
||||
return {
|
||||
tabId: 'tab-1',
|
||||
worktreeId: 'wt-1',
|
||||
cwd: '/tmp/wt-1',
|
||||
startup: null,
|
||||
restoredLeafId: LEAF_ID,
|
||||
restoredPtyIdByLeafId: { [LEAF_ID]: REMOTE_PTY_ID },
|
||||
paneTransportsRef: { current: new Map() },
|
||||
paneMode2031Ref: { current: new Map() },
|
||||
paneKittyKeyboardModesRef: { current: new Map() },
|
||||
paneLastThemeModeRef: { current: new Map() },
|
||||
replayingPanesRef: { current: new Map() },
|
||||
isActiveRef: { current: true },
|
||||
isVisibleRef: { current: true },
|
||||
onPtyExitRef: { current: vi.fn() },
|
||||
onAgentExitedRef: { current: vi.fn() },
|
||||
onPtyErrorRef: { current: vi.fn() },
|
||||
clearTabPtyId: vi.fn(),
|
||||
consumeSuppressedPtyExit: vi.fn(() => false),
|
||||
isPtyShutdownPending: vi.fn(() => false),
|
||||
updateTabTitle: vi.fn(),
|
||||
setRuntimePaneTitle: vi.fn(),
|
||||
clearRuntimePaneTitle: vi.fn(),
|
||||
updateTabPtyId: vi.fn(),
|
||||
markWorktreeUnread: vi.fn(),
|
||||
markTerminalTabUnread: vi.fn(),
|
||||
markTerminalPaneUnread: vi.fn(),
|
||||
clearWorktreeUnread: vi.fn(),
|
||||
clearTerminalTabUnread: vi.fn(),
|
||||
clearTerminalPaneUnread: vi.fn(),
|
||||
dispatchNotification: vi.fn(),
|
||||
onShowSessionRestoredBanner: vi.fn(),
|
||||
setCacheTimerStartedAt: vi.fn(),
|
||||
syncPanePtyLayoutBinding: vi.fn(),
|
||||
clearExitedPanePtyLayoutBinding: vi.fn()
|
||||
}
|
||||
}
|
||||
|
||||
/** Every answer main can give for a `remote:` id it owns no registry entry for. */
|
||||
const LIVENESS_ANSWERS: [string, () => Promise<boolean | null>][] = [
|
||||
['a fabricated dead answer from the local registry', async () => false],
|
||||
['an explicit unknown', async () => null],
|
||||
[
|
||||
'a failed probe',
|
||||
async () => {
|
||||
throw new Error('ipc unavailable')
|
||||
}
|
||||
]
|
||||
]
|
||||
|
||||
describe('host-rejected paired-runtime input reaches a pane remount', () => {
|
||||
beforeEach(() => {
|
||||
vi.resetModules()
|
||||
vi.clearAllMocks()
|
||||
storeSubscribers = []
|
||||
remountTerminalTabForRecovery.mockReturnValue(true)
|
||||
mockStoreState = {
|
||||
activeWorktreeId: 'wt-1',
|
||||
activeWorkspaceExecutionHostId: `runtime:${ENVIRONMENT_ID}`,
|
||||
tabsByWorktree: { 'wt-1': [{ id: 'tab-1', ptyId: REMOTE_PTY_ID }] },
|
||||
ptyIdsByTabId: { 'tab-1': [REMOTE_PTY_ID] },
|
||||
terminalLayoutsByTabId: {
|
||||
'tab-1': {
|
||||
root: { type: 'leaf', leafId: LEAF_ID },
|
||||
activeLeafId: LEAF_ID,
|
||||
expandedLeafId: null,
|
||||
ptyIdsByLeafId: { [LEAF_ID]: REMOTE_PTY_ID }
|
||||
}
|
||||
},
|
||||
unreadTerminalTabs: {},
|
||||
deleteStateByWorktreeId: {},
|
||||
worktreesByRepo: {
|
||||
repo1: [{ id: 'wt-1', repoId: 'repo1', path: '/tmp/wt-1', hostId: 'local' }]
|
||||
},
|
||||
runtimeStatusByEnvironmentId: new Map(),
|
||||
repos: [{ id: 'repo1', connectionId: null, displayName: 'orca' }],
|
||||
projects: [],
|
||||
sshConnectionStates: new Map(),
|
||||
transientClearedAgentStatusConnectionIds: {},
|
||||
cacheTimerByKey: {},
|
||||
settings: { terminalMainSideEffectAuthority: false },
|
||||
codexRestartNoticeByPtyId: {},
|
||||
deferredSshReconnectTargets: [],
|
||||
deferredSshSessionIdsByTabId: {},
|
||||
removeDeferredSshReconnectTarget: vi.fn(),
|
||||
removeDeferredSshSessionId: vi.fn(),
|
||||
consumePendingColdRestore: vi.fn(() => null),
|
||||
consumePendingSnapshot: vi.fn(() => null),
|
||||
runtimePaneTitlesByTabId: {},
|
||||
agentStatusByPaneKey: {},
|
||||
retainedAgentsByPaneKey: {},
|
||||
paneForegroundAgentByPaneKey: {},
|
||||
sleepingAgentSessionsByPaneKey: {},
|
||||
suppressedPtyExitIds: {},
|
||||
agentLaunchConfigByPaneKey: {},
|
||||
getAgentLaunchConfigForStatusEntry: vi.fn(),
|
||||
getAgentLaunchConfigForStatusMetadata: vi.fn(),
|
||||
clearSleepingAgentSession: vi.fn(),
|
||||
registerAgentLaunchConfig: vi.fn(),
|
||||
clearAgentLaunchConfig: vi.fn(),
|
||||
markWorktreeUnread: vi.fn(),
|
||||
observeTerminalGitHubPullRequestLink: vi.fn(),
|
||||
recordTerminalInput: vi.fn(),
|
||||
setAgentStatus: vi.fn(),
|
||||
removeAgentStatus: vi.fn(),
|
||||
dropAgentStatus: vi.fn(),
|
||||
retireAgentPaneAuthority: vi.fn(),
|
||||
setPaneForegroundAgent: vi.fn(),
|
||||
clearPaneForegroundAgent: vi.fn(),
|
||||
markTerminalTabUnread: vi.fn(),
|
||||
markTerminalPaneUnread: vi.fn(),
|
||||
markAgentCompletionPaneUnread: vi.fn(),
|
||||
remountTerminalTabForRecovery
|
||||
}
|
||||
globalThis.requestAnimationFrame = vi.fn((callback: FrameRequestCallback) => {
|
||||
callback(0)
|
||||
return 1
|
||||
})
|
||||
globalThis.cancelAnimationFrame = vi.fn()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
delete (globalThis as { window?: unknown }).window
|
||||
})
|
||||
|
||||
for (const [label, hasPty] of LIVENESS_ANSWERS) {
|
||||
it(`remounts the tab when the pane liveness probe returns ${label}`, async () => {
|
||||
const { bridge, sendTerminal, hostOpcodes } = startHost()
|
||||
;(globalThis as unknown as { window: unknown }).window = {
|
||||
api: {
|
||||
runtimeEnvironments: { call: bridge.call, subscribe: bridge.subscribe },
|
||||
pty: {
|
||||
hasPty: vi.fn(hasPty),
|
||||
kill: vi.fn(),
|
||||
signal: vi.fn(),
|
||||
listSessions: vi.fn().mockResolvedValue([]),
|
||||
getSize: vi.fn().mockResolvedValue(null),
|
||||
reportGeometry: vi.fn(),
|
||||
getMainBufferSnapshot: vi.fn().mockResolvedValue(null),
|
||||
getForegroundProcess: vi.fn().mockResolvedValue(null),
|
||||
inspectProcess: vi.fn().mockResolvedValue({
|
||||
foregroundProcess: null,
|
||||
hasChildProcesses: false
|
||||
}),
|
||||
confirmForegroundProcess: vi.fn().mockResolvedValue(null),
|
||||
hasChildProcesses: vi.fn().mockResolvedValue(false),
|
||||
write: vi.fn(),
|
||||
writeAccepted: vi.fn().mockResolvedValue(true),
|
||||
setHiddenRendererPty: vi.fn(),
|
||||
setPtyDeliveryInterest: vi.fn(),
|
||||
ackColdRestore: vi.fn(),
|
||||
onClearBufferRequest: vi.fn(() => vi.fn()),
|
||||
onSerializeBufferRequest: vi.fn(() => vi.fn()),
|
||||
sendSerializedBuffer: vi.fn(),
|
||||
declarePendingPaneSerializer: vi.fn().mockResolvedValue(1),
|
||||
settlePaneSerializer: vi.fn().mockResolvedValue(undefined),
|
||||
clearPendingPaneSerializer: vi.fn().mockResolvedValue(undefined),
|
||||
reportRendererSerializerReady: vi.fn().mockResolvedValue(undefined)
|
||||
},
|
||||
platform: { get: vi.fn(() => ({ platform: 'darwin', osRelease: '25.0.0' })) },
|
||||
notifications: {
|
||||
dispatch: vi.fn().mockResolvedValue({ delivered: true }),
|
||||
playSound: vi.fn().mockResolvedValue({ played: true })
|
||||
},
|
||||
runtime: { restoreTerminalFit: vi.fn().mockResolvedValue({ restored: true }) },
|
||||
agentStatus: { inferInterrupt: vi.fn().mockResolvedValue(false) },
|
||||
ssh: {
|
||||
connect: vi.fn().mockResolvedValue({ status: 'connected' }),
|
||||
needsPassphrasePrompt: vi.fn().mockResolvedValue(false)
|
||||
}
|
||||
},
|
||||
dispatchEvent: vi.fn(),
|
||||
addEventListener: vi.fn(),
|
||||
removeEventListener: vi.fn()
|
||||
}
|
||||
|
||||
const { connectPanePty } = await import('@/components/terminal-pane/pty-connection')
|
||||
const { _resetTerminalPaneRecoveryForTests } =
|
||||
await import('@/components/terminal-pane/terminal-pane-recovery')
|
||||
_resetTerminalPaneRecoveryForTests()
|
||||
|
||||
const pane = createPane()
|
||||
const deps = createDeps()
|
||||
const binding = connectPanePty(pane as never, createManager() as never, deps as never)
|
||||
await vi.waitFor(() => {
|
||||
expect(deps.paneTransportsRef.current.get(1)?.getPtyId()).toBe(REMOTE_PTY_ID)
|
||||
})
|
||||
|
||||
// The user types; the host accepts the frame and the PTY refuses the bytes.
|
||||
sendTerminalInput(pane, 'ls\r')
|
||||
await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalled())
|
||||
// Hop 1: the host turned the refusal into the negotiated frame.
|
||||
await vi.waitFor(() => expect(hostOpcodes).toContain(TerminalStreamOpcode.WriteUnavailable))
|
||||
// Hop 2 (the one that was missing): it survives pane recovery as a remount.
|
||||
await vi.waitFor(() => expect(remountTerminalTabForRecovery).toHaveBeenCalledWith('tab-1'))
|
||||
|
||||
binding.dispose()
|
||||
_resetTerminalPaneRecoveryForTests()
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
function sendTerminalInput(pane: ReturnType<typeof createPane>, data: string): void {
|
||||
const calls = pane.terminal.onData.mock.calls as unknown as [(data: string) => void][]
|
||||
const handler = calls[0]?.[0]
|
||||
expect(handler).toBeTypeOf('function')
|
||||
handler?.(data)
|
||||
}
|
||||
Loading…
Reference in New Issue