From 5e43ae67f2efadd3882cba04289570204cba8095 Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 8 Oct 2026 03:44:56 -0700 Subject: [PATCH 1/3] Warn when remote terminal input delivery cannot be confirmed --- config/reliability-gates.jsonc | 107 +++++++++++++ .../rpc/methods/terminal/stream-schemas.ts | 1 + .../terminal/terminal-input-delivery.ts | 25 ++- .../terminal-multiplex-initial-snapshot.ts | 3 +- .../terminal-multiplex-input-delivery.ts | 51 ++++++ .../terminal-multiplex-slot-frames.ts | 25 +-- ...erminal-multiplex-stream-initialization.ts | 1 + .../methods/terminal/terminal-stream-types.ts | 1 + .../terminal-multiplex-input-receipts.test.ts | 112 +++++++++++++ ...ote-runtime-input-delivery-warning.test.ts | 86 ++++++++++ ...-pty-transport-attach-subscription.test.ts | 1 + ...-pty-transport-host-session-launch.test.ts | 10 +- ...mote-runtime-pty-transport-test-harness.ts | 26 +++ .../remote-runtime-pty-transport.ts | 25 ++- .../remote-terminal-input-delivery-warning.ts | 17 ++ src/renderer/src/i18n/locales/en.json | 4 +- ...remote-runtime-terminal-flow-controller.ts | 19 ++- ...te-runtime-terminal-input-receipts.test.ts | 144 +++++++++++++++++ ...emote-runtime-terminal-multiplexer-base.ts | 6 +- ...ime-terminal-multiplexer-implementation.ts | 8 +- ...mote-runtime-terminal-multiplexer-types.ts | 8 +- ...te-runtime-terminal-response-controller.ts | 30 +++- ...te-runtime-terminal-snapshot-controller.ts | 1 + .../remote-terminal-input-receipts.test.ts | 82 ++++++++++ .../runtime/remote-terminal-input-receipts.ts | 75 +++++++++ ...untime-rejected-input-remount.unit.test.ts | 7 +- tests/e2e/ssh-orcad-half-open-input.spec.ts | 148 ++++++++++++++++++ 27 files changed, 977 insertions(+), 46 deletions(-) create mode 100644 src/main/runtime/rpc/methods/terminal/terminal-multiplex-input-delivery.ts create mode 100644 src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts create mode 100644 src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts create mode 100644 src/renderer/src/components/terminal-pane/remote-terminal-input-delivery-warning.ts create mode 100644 src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts create mode 100644 src/renderer/src/runtime/remote-terminal-input-receipts.test.ts create mode 100644 src/renderer/src/runtime/remote-terminal-input-receipts.ts create mode 100644 tests/e2e/ssh-orcad-half-open-input.spec.ts diff --git a/config/reliability-gates.jsonc b/config/reliability-gates.jsonc index 3b415f60d580..7a8d41b0b860 100644 --- a/config/reliability-gates.jsonc +++ b/config/reliability-gates.jsonc @@ -10,6 +10,113 @@ } }, "gates": [ + { + "id": "terminal-input.remote-delivery-uncertainty", + "title": "Unanswered remote terminal input remains visibly uncertain", + "maturity": "experimental", + "protection": "partial", + "owner": "terminal-runtime", + "layer": "host-stream-renderer-and-real-ssh", + "surfaces": [ + "paired terminal keyboard input", + "binary input receipts", + "older-host RPC input fallback", + "reconnect warning" + ], + "platforms": [ + "macos", + "linux", + "windows" + ], + "providers": [ + "remote-runtime", + "ssh" + ], + "coveredPlatforms": [ + "macos" + ], + "coveredProviders": [ + "remote-runtime", + "ssh" + ], + "coverageNotes": "Hidden macOS Electron drives a Linux arm64 Docker SSH host through a controlled TCP proxy. Source-built server and installed pre-receipt server exercise negotiation and fallback. Native Windows and WSL are gaps; no physical packet outage claim.", + "motivatingLinks": [ + "https://github.com/stablyai/orca/pull/26483" + ], + "invariant": "Only an execution-host receipt settles an input write. Missing receipts warn without replaying uncertain bytes, replacing the remote process, or treating contact loss as process exit. Parser replies do not create delivery warnings.", + "oracle": "Real keyboard input reaches an independent host ledger. Before and after lines arrive once in the same process; a two-way stall loses the during line and warns, while a reply-only stall delivers it exactly once and still warns. The warning survives reconnect. Unit contracts prove negotiated settlement, older-host fallback, unrelated output, out-of-order receipts, bounded state and timer cleanup.", + "commands": [ + "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/renderer/src/runtime/remote-terminal-input-receipts.test.ts src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts", + "ORCA_BACKGROUND_LAUNCH=1 ORCA_E2E_SSH_DOCKER=1 pnpm run test:e2e tests/e2e/ssh-orcad-half-open-input.spec.ts --workers=1" + ], + "testFiles": [ + "src/renderer/src/runtime/remote-terminal-input-receipts.test.ts", + "src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts", + "src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts", + "src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts", + "tests/e2e/ssh-orcad-half-open-input.spec.ts" + ], + "assertionRefs": [ + { + "file": "src/renderer/src/runtime/remote-terminal-input-receipts.test.ts", + "assertions": [ + "does not mistake a later write receipt for receipt of an earlier write", + "bounds pending receipts and uses one timer during an input flood" + ] + }, + { + "file": "src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts", + "assertions": [ + "does not treat unrelated output as an input receipt", + "warns on a lost connection and never replays pending input into its replacement" + ] + }, + { + "file": "tests/e2e/ssh-orcad-half-open-input.spec.ts", + "assertions": [ + "managed host warns about uncertain input when both traffic stalls", + "managed host warns about uncertain input when to-client traffic stalls" + ] + } + ], + "evidenceRuns": [ + { + "date": "2026-10-08", + "runner": "local", + "platform": "macos", + "command": "ORCA_BACKGROUND_LAUNCH=1 pnpm test src/renderer/src/runtime/remote-terminal-input-receipts.test.ts src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts", + "result": "passed", + "durationSeconds": 1.0, + "summary": "Sixteen focused tests pass: receipt settlement, mixed-version fallback, unrelated output, exact-once transport behavior and bounded timer state." + } + ], + "runtimeBudget": { + "p95Seconds": 300, + "scope": "Local Docker SSH exercise; long-term CI p95 has not been established." + }, + "flakeHistory": { + "status": "not-started", + "evidence": "Two source-built server fault cases passed locally; no soak claim." + }, + "redGreenEvidence": { + "status": "complete", + "evidence": "Unchanged app loses the during line without a warning and fails the new rendered-warning assertion. Fixed source-built server warns around 15 seconds in both fault directions, preserves one host process and never duplicates the delivered line." + }, + "performanceBudget": { + "required": true, + "evidence": "Existing 8 ms keyboard batching remains. Negotiated binary writes do not wait for prior round trips. At most 1024 numeric receipts and one deadline timer per active stream; no command text retained, polling, inventory reads, startup waits or subprocesses. Older hosts use existing bounded RPC request deadlines." + }, + "knownGaps": [ + "Native Windows, WSL and Linux desktop clients are not exercised. Local and daemon transports are unchanged; the Linux daemon is exercised behind the managed host.", + "Older hosts without PTY settlements prove request receipt only. This prevents silent network loss but cannot prove an old provider write or command execution.", + "A warning cannot determine whether unacknowledged input ran. The user must inspect the terminal before retrying.", + "Long-term soak and wide-area typing latency measurements remain unproved." + ], + "promotionCriteria": [ + "Collect policy soak and native Windows/WSL coverage while retaining exact-once process-ledger assertions." + ], + "demotionRule": "Keep experimental until soak. Investigate missing warnings, duplicate writes or retained timers without retries or relaxed assertions." + }, { "id": "native-chat.transcript-consumed-boundary", "title": "Transcript replacement remains visible when it races message publication", diff --git a/src/main/runtime/rpc/methods/terminal/stream-schemas.ts b/src/main/runtime/rpc/methods/terminal/stream-schemas.ts index f3e6e3a7870b..a6aa7a0861bb 100644 --- a/src/main/runtime/rpc/methods/terminal/stream-schemas.ts +++ b/src/main/runtime/rpc/methods/terminal/stream-schemas.ts @@ -20,6 +20,7 @@ export const TerminalMultiplexSubscribeFrame = TerminalHandle.extend({ capabilities: z .object({ ackOutput: z.literal(1).optional(), + ackInput: z.literal(1).optional(), ackOutputSourceRanges: z.literal(1).optional(), desktopViewportClaims: z.literal(1).optional(), outputPause: z.literal(1).optional(), diff --git a/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts b/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts index 59fd827f0ce1..79aa0f9aa746 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-input-delivery.ts @@ -59,6 +59,7 @@ export async function sendTerminalStreamInput( text: string client: TerminalViewportClient | undefined isMobile: boolean + requireWriteSettlement?: true } ): Promise { const action = { text: args.text, enter: false, interrupt: false } @@ -66,11 +67,15 @@ export async function sendTerminalStreamInput( const floorClaim: MobileInputFloorClaimHolder = { current: null } try { if (!clientId) { - const result = await runtime.sendTerminal(args.terminal, action, { inputKind: 'driving' }) - return result.accepted ? 'delivered' : 'rejected' + const result = await runtime.sendTerminal(args.terminal, action, { + inputKind: 'driving', + ...(args.requireWriteSettlement ? { requireWriteSettlement: true as const } : {}) + }) + return streamInputOutcome(result, args.requireWriteSettlement) } const result = await runtime.sendTerminal(args.terminal, action, { inputKind: 'driving', + ...(args.requireWriteSettlement ? { requireWriteSettlement: true as const } : {}), reserveWrite: (writePtyId) => { const claim = runtime.beginMobileInputFloor(writePtyId, clientId) if (!claim) { @@ -82,15 +87,27 @@ export async function sendTerminalStreamInput( }) if (!result.accepted) { floorClaim.current?.rollback() - return 'rejected' } - return 'delivered' + return streamInputOutcome(result, args.requireWriteSettlement) } catch (error) { floorClaim.current?.rollback() return isTerminalStreamInputRejection(error) ? 'rejected' : 'failed' } } +function streamInputOutcome( + result: Awaited>, + requireSettlement: true | undefined +): TerminalStreamInputOutcome { + if ( + requireSettlement && + (!result.writeSettlement || result.writeSettlement.outcome === 'unverifiable') + ) { + return 'failed' + } + return result.accepted ? 'delivered' : 'rejected' +} + export type MobileInputFloorClaimHolder = { current: ReturnType } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts index e0644abd1a51..1293204c2da5 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-initial-snapshot.ts @@ -60,8 +60,9 @@ export async function publishMultiplexInitialSnapshot( rows: serialized?.rows ?? size?.rows, displayMode, seq: layoutSeq, - ...((stream.ackOutputSourceRanges || stream.supportsOutputPause) && { + ...((stream.ackOutputSourceRanges || stream.supportsOutputPause || stream.acknowledgeInput) && { capabilities: { + ...(stream.acknowledgeInput ? { ackInput: 1 as const } : {}), ...(stream.ackOutputSourceRanges ? { ackOutputSourceRanges: 1 as const } : {}), ...(stream.supportsOutputPause ? { outputPause: 1 as const } : {}) } diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-input-delivery.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-input-delivery.ts new file mode 100644 index 000000000000..bdd882d39f52 --- /dev/null +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-input-delivery.ts @@ -0,0 +1,51 @@ +import { + decodeTerminalStreamText, + type TerminalStreamFrame +} from '../../../../../shared/terminal-stream-protocol' +import { isTerminalInputLockedForClient, sendTerminalStreamInput } from './terminal-input-delivery' +import type { TerminalMultiplexConnection } from './terminal-multiplex-connection' +import type { TerminalMultiplexStream } from './terminal-stream-types' + +export function receiveMultiplexInput( + state: TerminalMultiplexConnection, + stream: TerminalMultiplexStream, + frame: TerminalStreamFrame +): void { + const { runtime, streams } = state + const text = decodeTerminalStreamText(frame.payload) + if (!text) { + return + } + const acknowledge = stream.acknowledgeInput && Number.isSafeInteger(frame.seq) && frame.seq > 0 + const publishReceipt = (outcome: 'accepted' | 'refused' | 'unverifiable'): void => { + if (acknowledge && !state.closed && streams.get(stream.streamId) === stream) { + state.emit({ type: 'input-ack', streamId: stream.streamId, seq: frame.seq, outcome }) + } + } + if (isTerminalInputLockedForClient(runtime, stream.ptyId, stream.client)) { + publishReceipt('refused') + return + } + // 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(async (claimed) => { + if (state.closed || streams.get(stream.streamId) !== stream) { + return + } + if (!claimed || isTerminalInputLockedForClient(runtime, stream.ptyId, stream.client)) { + publishReceipt('refused') + return + } + const outcome = await sendTerminalStreamInput(runtime, { + terminal: stream.terminal, + text, + client: stream.client, + isMobile: stream.isMobile, + ...(acknowledge ? { requireWriteSettlement: true as const } : {}) + }) + publishReceipt( + outcome === 'delivered' ? 'accepted' : outcome === 'rejected' ? 'refused' : 'unverifiable' + ) + state.notifyStreamWriteUnavailable(stream, outcome) + }) +} diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts index bad459b44cdf..73cae632ed48 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-slot-frames.ts @@ -1,7 +1,6 @@ import { TerminalStreamOpcode, decodeTerminalStreamJson, - decodeTerminalStreamText, type TerminalStreamFrame } from '../../../../../shared/terminal-stream-protocol' import { @@ -9,7 +8,7 @@ import { TerminalMultiplexSnapshotRequestFrame, TerminalMultiplexSourceRangeAckFrame } from './stream-schemas' -import { isTerminalInputLockedForClient, sendTerminalStreamInput } from './terminal-input-delivery' +import { receiveMultiplexInput } from './terminal-multiplex-input-delivery' import { getOutputAfterSnapshotSeq, normalizeMultiplexSnapshotScrollbackRows @@ -61,27 +60,7 @@ export function installMultiplexSlotFrames( return } if (frame.opcode === TerminalStreamOpcode.Input) { - const text = decodeTerminalStreamText(frame.payload) - if (!text) { - return - } - if (isTerminalInputLockedForClient(runtime, stream.ptyId, stream.client)) { - return - } - // 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(async (claimed) => { - if (!claimed || isTerminalInputLockedForClient(runtime, stream.ptyId, stream.client)) { - return - } - const outcome = await sendTerminalStreamInput(runtime, { - terminal: stream.terminal, - text, - client: stream.client, - isMobile: stream.isMobile - }) - state.notifyStreamWriteUnavailable(stream, outcome) - }) + receiveMultiplexInput(state, stream, frame) return } if (frame.opcode === TerminalStreamOpcode.SetOutputPaused && stream.supportsOutputPause) { diff --git a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts index cc3ff90b2d41..5aac5d7af44f 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-multiplex-stream-initialization.ts @@ -45,6 +45,7 @@ export async function initializeMultiplexStream( ptyId, client: request.client, isMobile, + acknowledgeInput: request.capabilities?.ackInput === 1, ackOutput: request.capabilities?.ackOutput === 1, ackOutputSourceRanges: sourceRangeConsumerAttached, streamGeneration, diff --git a/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts b/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts index 379a25897954..c9a62533dbb6 100644 --- a/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts +++ b/src/main/runtime/rpc/methods/terminal/terminal-stream-types.ts @@ -61,6 +61,7 @@ export type TerminalMultiplexStream = { ptyId: string client: TerminalViewportClient | undefined isMobile: boolean + acknowledgeInput: boolean ackOutput: boolean ackOutputSourceRanges: boolean streamGeneration: string diff --git a/src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts b/src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts new file mode 100644 index 000000000000..fa292883363b --- /dev/null +++ b/src/main/runtime/rpc/terminal-multiplex-input-receipts.test.ts @@ -0,0 +1,112 @@ +import './unused-default-rpc-methods.test-fixture' +import { describe, expect, it, vi } from 'vitest' +import type { OrcaRuntimeService } from '../orca-runtime' +import type { RuntimeTerminalSend } from '../../../shared/runtime-types' +import { WRITE_ACCEPTED, writeUnverifiable } from '../../../shared/pty-write-settlement' +import { + TerminalStreamOpcode, + encodeTerminalStreamText +} from '../../../shared/terminal-stream-protocol' +import { + sendDesktopMultiplexSubscribe, + startDesktopMultiplexSubscribe +} from './terminal-multiplex-test-harness' + +describe('terminal multiplex input receipts', () => { + it.each([true, false])( + 'publishes receipts only after negotiated owner settlement (negotiated=%s)', + async (negotiated) => { + let settle: ((result: RuntimeTerminalSend) => void) | undefined + const sendTerminal = vi.fn( + () => + new Promise((resolve) => { + settle = resolve + }) + ) + const harness = startDesktopMultiplexSubscribe({ sendTerminal }) + try { + await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true)) + sendDesktopMultiplexSubscribe(harness.handlers, negotiated ? { ackInput: 1 } : {}) + await vi.waitFor(() => + expect( + harness.messages.some((message) => JSON.parse(message).result?.type === 'subscribed') + ).toBe(true) + ) + harness.handlers.get(7)?.({ + opcode: TerminalStreamOpcode.Input, + streamId: 7, + seq: 42, + payload: encodeTerminalStreamText('x') + }) + await vi.waitFor(() => expect(sendTerminal).toHaveBeenCalledOnce()) + const receipts = () => + harness.messages + .map((message) => JSON.parse(message).result) + .filter((event) => event?.type === 'input-ack') + expect(receipts()).toEqual([]) + expect(sendTerminal.mock.calls[0]?.[2]).toEqual( + negotiated + ? { inputKind: 'driving', requireWriteSettlement: true } + : { inputKind: 'driving' } + ) + settle?.({ + handle: 'terminal-1', + accepted: true, + bytesWritten: 1, + writeSettlement: WRITE_ACCEPTED + }) + if (negotiated) { + await vi.waitFor(() => + expect(receipts()).toEqual([ + { type: 'input-ack', streamId: 7, seq: 42, outcome: 'accepted' } + ]) + ) + } else { + await vi.waitFor(() => + expect(sendTerminal.mock.results[0]?.value).resolves.toMatchObject({ accepted: true }) + ) + expect(receipts()).toEqual([]) + } + } finally { + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') + await harness.dispatchPromise + } + } + ) + + it('preserves unverifiable delivery instead of acknowledging acceptance or refusal', async () => { + const sendTerminal = vi.fn().mockResolvedValue({ + handle: 'terminal-1', + accepted: false, + bytesWritten: 0, + writeSettlement: writeUnverifiable('transport_settlement_lost', true) + }) + const harness = startDesktopMultiplexSubscribe({ sendTerminal }) + try { + await vi.waitFor(() => expect(harness.handlers.has(0)).toBe(true)) + sendDesktopMultiplexSubscribe(harness.handlers, { ackInput: 1 }) + await vi.waitFor(() => + expect( + harness.messages.some((message) => JSON.parse(message).result?.type === 'subscribed') + ).toBe(true) + ) + harness.handlers.get(7)?.({ + opcode: TerminalStreamOpcode.Input, + streamId: 7, + seq: 42, + payload: encodeTerminalStreamText('x') + }) + await vi.waitFor(() => + expect(harness.messages.map((message) => JSON.parse(message).result)).toContainEqual({ + type: 'input-ack', + streamId: 7, + seq: 42, + outcome: 'unverifiable' + }) + ) + } finally { + harness.registry.cleanupSubscription('terminal-multiplex:conn-desktop-first-paint') + await harness.dispatchPromise + } + }) +}) diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts b/src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts new file mode 100644 index 000000000000..d43851545d71 --- /dev/null +++ b/src/renderer/src/components/terminal-pane/remote-runtime-input-delivery-warning.test.ts @@ -0,0 +1,86 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { warnRemoteTerminalInputDelivery } from './remote-terminal-input-delivery-warning' +import { + createRemoteRuntimeTransportMocks, + type MultiplexSubscriptionCallbacks +} from './remote-runtime-pty-transport-test-harness' + +vi.mock('./remote-terminal-input-delivery-warning', () => ({ + warnRemoteTerminalInputDelivery: vi.fn() +})) + +let callbacks: MultiplexSubscriptionCallbacks = null +let paneHandle = 'terminal-1' +const harness = createRemoteRuntimeTransportMocks({ + getCallbacks: () => callbacks, + setCallbacks: (next) => { + callbacks = next + }, + getResolvedPaneHandle: () => paneHandle, + setResolvedPaneHandle: (next) => { + paneHandle = next + } +}) + +describe('legacy remote host input delivery warnings', () => { + beforeEach(() => harness.resetRemoteRuntimeTransport()) + + async function openLegacy() { + const { createRemoteRuntimePtyTransport } = await import('./remote-runtime-pty-transport') + const transport = createRemoteRuntimePtyTransport('env-1', { + worktreeId: 'wt-1', + tabId: 'tab-1', + leafId: 'pane:1' + }) + await transport.connect({ url: '', callbacks: {} }) + await vi.waitFor(() => expect(harness.subscriptionSendBinary).toHaveBeenCalled()) + callbacks?.onResponse({ + ok: true, + result: { type: 'subscribed', streamId: harness.latestSubscribePayload().streamId } + }) + return transport + } + + it.each(['accepted', 'unverifiable', 'timeout'] as const)( + 'uses host request/reply receipts and warns only for uncertainty (%s)', + async (outcome) => { + const transport = await openLegacy() + try { + const previous = harness.runtimeCall.getMockImplementation() + harness.runtimeCall.mockImplementation((request: { method: string }) => { + if (request.method !== 'terminal.send') { + return previous?.(request) + } + if (outcome === 'timeout') { + return Promise.reject(new Error('Remote request timed out')) + } + return Promise.resolve({ + ok: true, + result: { send: { accepted: outcome === 'accepted', writeSettlement: { outcome } } } + }) + }) + expect(transport.sendInput('a', 'driving')).toBe(true) + await vi.waitFor(() => + expect(harness.runtimeCall).toHaveBeenCalledWith( + expect.objectContaining({ + method: 'terminal.send', + params: expect.objectContaining({ text: 'a', requireWriteSettlement: true }) + }) + ) + ) + if (outcome === 'accepted') { + expect(warnRemoteTerminalInputDelivery).not.toHaveBeenCalled() + } else { + await vi.waitFor(() => + expect(warnRemoteTerminalInputDelivery).toHaveBeenCalledWith('env-1', 'terminal-1') + ) + } + expect( + harness.runtimeCall.mock.calls.filter(([request]) => request.method === 'terminal.send') + ).toHaveLength(1) + } finally { + transport.destroy?.() + } + } + ) +}) diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts index 07ff299ee7cd..3c1270d84c7e 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-attach-subscription.test.ts @@ -61,6 +61,7 @@ describe('createRemoteRuntimePtyTransport', () => { await vi.waitFor(() => expect(latestSubscribePayload().capabilities).toEqual({ ackOutput: 1, + ackInput: 1, ackOutputSourceRanges: 1, desktopViewportClaims: 1, outputPause: 1, diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts index f9e95fb7c8c9..d66365893929 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-host-session-launch.test.ts @@ -519,6 +519,10 @@ describe('createRemoteRuntimePtyTransport', () => { await transport.connect({ url: '', callbacks: {} }) const { streamId } = latestRemoteSubscribePayload() + subscriptionCallbacks?.onResponse({ + ok: true, + result: { type: 'subscribed', streamId, capabilities: { ackInput: 1 } } + }) runtimeCall.mockClear() const send = unsubscribe?.sendBinary as unknown as { mockClear: () => void @@ -527,12 +531,16 @@ describe('createRemoteRuntimePtyTransport', () => { send.mockClear() expect(transport.sendInput('ls\r', 'driving')).toBe(true) - await vi.runOnlyPendingTimersAsync() + await vi.advanceTimersByTimeAsync(8) expect(runtimeCall).not.toHaveBeenCalled() const inputFrame = decodeTerminalStreamFrame(send.mock.calls[0][0]) expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input) expect(inputFrame?.streamId).toBe(streamId) + subscriptionCallbacks?.onResponse({ + ok: true, + result: { type: 'input-ack', streamId, seq: inputFrame?.seq, outcome: 'accepted' } + }) transport.disconnect() expect(unsubscribeFn).toHaveBeenCalled() expect(runtimeCall).not.toHaveBeenCalledWith( diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-test-harness.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-test-harness.ts index 0cc62569b5f0..ec9c6a56c7e8 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-test-harness.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport-test-harness.ts @@ -2,6 +2,10 @@ falls outside the *.test / *.spec / tests glob set. refreshWebRuntimeSessionTabsSnapshot is imported directly by several renderer runtime modules, so an injected seam would have to be threaded through all of them. */ import { vi } from 'vitest' +import { + decodeTerminalStreamFrame, + TerminalStreamOpcode +} from '../../../../shared/terminal-stream-protocol' import type { Mock } from 'vitest' import { createTerminalStreamFixtures, @@ -83,6 +87,28 @@ export function createRemoteRuntimeTransportMocks( bindings.setCallbacks(null) bindings.setResolvedPaneHandle('terminal-1') subscriptionSendBinary.mockReset() + subscriptionSendBinary.mockImplementation((bytes) => { + const frame = decodeTerminalStreamFrame(bytes) + if (frame?.opcode === TerminalStreamOpcode.Subscribe) { + const payload = JSON.parse(new TextDecoder().decode(frame.payload)) + bindings.getCallbacks()?.onResponse({ + ok: true, + result: { type: 'subscribed', streamId: payload.streamId, capabilities: { ackInput: 1 } } + }) + } else if (frame?.opcode === TerminalStreamOpcode.Input && frame.seq > 0) { + queueMicrotask(() => + bindings.getCallbacks()?.onResponse({ + ok: true, + result: { + type: 'input-ack', + streamId: frame.streamId, + seq: frame.seq, + outcome: 'accepted' + } + }) + ) + } + }) refreshSessionTabsSnapshot.mockClear() runtimeCall.mockImplementation(async (request: { method: string; params?: unknown }) => { if (request.method === 'session.tabs.activate') { diff --git a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts index e91e35790741..9f832a8d3260 100644 --- a/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts +++ b/src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts @@ -1,3 +1,4 @@ +import { warnRemoteTerminalInputDelivery } from './remote-terminal-input-delivery-warning' import { createAgentSessionKeyboardOptions } from '@/runtime/agent-session-keyboard-capability' import { withRemoteReattachInputBuffer } from './remote-reattach-input-buffer' import { @@ -448,7 +449,9 @@ export function createRemoteRuntimePtyTransport( pendingClaimInput = [] pendingClaimQueryReplyCount = 0 for (const segment of queued) { - stream.sendInput(segment.text) + if (!stream.sendInput(segment.text, { requireReceipt: !segment.queryReply })) { + sendUnacknowledgedInput(segment.text, segment.queryReply) + } } for (const resolve of viewportClaimReadyWaiters) { resolve(true) @@ -1576,7 +1579,7 @@ export function createRemoteRuntimePtyTransport( return false } const stream = getCurrentMultiplexedStream(targetHandle) - if (stream?.sendInput(text)) { + if (stream?.sendInput(text, { requireReceipt: !queryReply })) { return true } if (pendingViewportClaim) { @@ -1587,10 +1590,20 @@ export function createRemoteRuntimePtyTransport( void callRuntime<{ send: RuntimeTerminalSend }>('terminal.send', { terminal: targetHandle, text, + ...(queryReply ? { inputKind: 'query-reply' } : { requireWriteSettlement: true as const }), client: { id: clientId, type: 'desktop' }, ...(desiredViewport ? { viewport: desiredViewport, claimViewport: true as const } : {}) }) .then((result) => { + if ( + connected && + lifecycleEpoch === targetLifecycleEpoch && + handle === targetHandle && + !queryReply && + result.send.writeSettlement?.outcome === 'unverifiable' + ) { + warnRemoteTerminalInputDelivery(currentRuntimeEnvironmentId, targetHandle) + } if ( connected && lifecycleEpoch === targetLifecycleEpoch && @@ -1607,6 +1620,9 @@ export function createRemoteRuntimePtyTransport( if (runtimeTerminalErrorMessage(error).includes('terminal_not_writable')) { notifyWriteUnavailable() } else { + if (!queryReply) { + warnRemoteTerminalInputDelivery(currentRuntimeEnvironmentId, targetHandle) + } handleRemoteTerminalError(error) } }) @@ -2248,6 +2264,11 @@ export function createRemoteRuntimePtyTransport( setDriverForPty(subscribedPtyId, driver) } }, + onInputUnverifiable: () => { + if (isCurrentSubscription()) { + warnRemoteTerminalInputDelivery(currentRuntimeEnvironmentId, subscribedHandle) + } + }, onWriteUnavailable: () => { if (isCurrentSubscription()) { notifyWriteUnavailable() diff --git a/src/renderer/src/components/terminal-pane/remote-terminal-input-delivery-warning.ts b/src/renderer/src/components/terminal-pane/remote-terminal-input-delivery-warning.ts new file mode 100644 index 000000000000..361fea5996b7 --- /dev/null +++ b/src/renderer/src/components/terminal-pane/remote-terminal-input-delivery-warning.ts @@ -0,0 +1,17 @@ +import { toast } from 'sonner' +import { translate } from '@/i18n/i18n' + +export function warnRemoteTerminalInputDelivery(environmentId: string, terminal: string): void { + toast.warning( + translate('terminal.inputDeliveryUncertain', 'Remote terminal input delivery is uncertain'), + { + id: `remote-terminal-input:${environmentId}:${terminal}`, + description: translate( + 'terminal.inputDeliveryUncertainDescription', + 'Some input may not have reached the host. Check the terminal before retrying a command.' + ), + duration: Infinity, + closeButton: true + } + ) +} diff --git a/src/renderer/src/i18n/locales/en.json b/src/renderer/src/i18n/locales/en.json index 78cd45f59d88..913aa96c3016 100644 --- a/src/renderer/src/i18n/locales/en.json +++ b/src/renderer/src/i18n/locales/en.json @@ -19354,7 +19354,9 @@ "learnMore": "Learn more", "dontShowAgain": "Don't show again", "dismiss": "Dismiss" - } + }, + "inputDeliveryUncertain": "Remote terminal input delivery is uncertain", + "inputDeliveryUncertainDescription": "Some input may not have reached the host. Check the terminal before retrying a command." }, "fileExplorer": { "root": { diff --git a/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts index 0488dd323cf2..4eb3b8fa93f0 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-flow-controller.ts @@ -40,16 +40,29 @@ export abstract class RemoteRuntimeTerminalFlowController extends RemoteRuntimeT return this.streams.get(stream.streamId) === stream } - protected sendInput(stream: RemoteRuntimeMultiplexedTerminalState, text: string): boolean { + protected sendInput( + stream: RemoteRuntimeMultiplexedTerminalState, + text: string, + options?: { requireReceipt?: boolean } + ): boolean { + if (options?.requireReceipt && !stream.acknowledgeInput) { + return false + } + if (!this.matchesCurrentEnvironmentRevision() || !this.ready || !this.subscription) { + return false + } + const sequence = options?.requireReceipt ? stream.inputReceipts.begin() : 0 const sent = this.sendFrame( stream.streamId, TerminalStreamOpcode.Input, - encodeTerminalStreamText(text) + encodeTerminalStreamText(text), + sequence ) if (sent && !stream.outputPaused) { stream.watchdog.recordCommandInput(text) } - return sent + // A thrown handoff may already have written; do not invite the RPC fallback to replay it. + return sent || sequence > 0 } protected setOutputPaused( diff --git a/src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts b/src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts new file mode 100644 index 000000000000..c665c13ab23f --- /dev/null +++ b/src/renderer/src/runtime/remote-runtime-terminal-input-receipts.test.ts @@ -0,0 +1,144 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + TerminalStreamOpcode, + decodeTerminalStreamFrame, + encodeTerminalStreamFrame, + encodeTerminalStreamText +} from '../../../shared/terminal-stream-protocol' +import { + getRemoteRuntimeTerminalMultiplexer, + resetRemoteRuntimeTerminalMultiplexersForTests +} from './remote-runtime-terminal-multiplexer' +import { REMOTE_TERMINAL_INPUT_RECEIPT_TIMEOUT_MS } from './remote-terminal-input-receipts' + +type SubscriptionCallbacks = { + onResponse: (response: unknown) => void + onBinary: (bytes: Uint8Array) => void + onClose: () => void +} + +describe('remote terminal input receipt negotiation', () => { + let callbacks: SubscriptionCallbacks | undefined + const sendBinary = vi.fn() + + beforeEach(() => { + vi.useFakeTimers() + sendBinary.mockReset() + resetRemoteRuntimeTerminalMultiplexersForTests() + callbacks = undefined + vi.stubGlobal('window', { + api: { + runtimeEnvironments: { + subscribe: vi.fn(async (_args: unknown, nextCallbacks: SubscriptionCallbacks) => { + callbacks = nextCallbacks + queueMicrotask(() => callbacks?.onResponse({ ok: true, result: { type: 'ready' } })) + return { unsubscribe: vi.fn(), sendBinary } + }) + } + } + }) + }) + + afterEach(() => { + resetRemoteRuntimeTerminalMultiplexersForTests() + vi.useRealTimers() + vi.unstubAllGlobals() + }) + + async function open(ackInput = true) { + const onInputUnverifiable = vi.fn() + const stream = await getRemoteRuntimeTerminalMultiplexer('env-receipts').subscribeTerminal({ + terminal: 'term-1', + client: { id: 'desktop-1', type: 'desktop' }, + callbacks: { onData: vi.fn(), onSnapshot: vi.fn(), onInputUnverifiable } + }) + callbacks?.onResponse({ + ok: true, + result: { + type: 'subscribed', + streamId: stream.streamId, + ...(ackInput ? { capabilities: { ackInput: 1 } } : {}) + } + }) + sendBinary.mockClear() + return { stream, onInputUnverifiable } + } + + it('keeps the binary fast path without waiting for a round trip between writes', async () => { + const { stream, onInputUnverifiable } = await open() + expect(stream.sendInput('a', { requireReceipt: true })).toBe(true) + expect(stream.sendInput('b', { requireReceipt: true })).toBe(true) + const frames = sendBinary.mock.calls.map(([bytes]) => decodeTerminalStreamFrame(bytes)) + expect(frames.map((frame) => frame?.seq)).toEqual([1, 2]) + for (const frame of frames) { + callbacks?.onResponse({ + ok: true, + result: { + type: 'input-ack', + streamId: stream.streamId, + seq: frame?.seq, + outcome: 'accepted' + } + }) + } + callbacks?.onClose() + expect(onInputUnverifiable).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + }) + + it('does not invite a fallback resend after a handoff throws', async () => { + const { stream, onInputUnverifiable } = await open() + sendBinary.mockImplementationOnce(() => { + throw new Error('handoff outcome unknown') + }) + expect(stream.sendInput('possibly-executed', { requireReceipt: true })).toBe(true) + expect(onInputUnverifiable).toHaveBeenCalledOnce() + expect(sendBinary).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + }) + + it('does not treat unrelated output as an input receipt', async () => { + const { stream, onInputUnverifiable } = await open() + stream.sendInput('a', { requireReceipt: true }) + callbacks?.onBinary( + encodeTerminalStreamFrame({ + opcode: TerminalStreamOpcode.Output, + streamId: stream.streamId, + seq: 0, + payload: encodeTerminalStreamText('unrelated host output') + }) + ) + await vi.advanceTimersByTimeAsync(REMOTE_TERMINAL_INPUT_RECEIPT_TIMEOUT_MS) + expect(onInputUnverifiable).toHaveBeenCalledOnce() + const inputs = sendBinary.mock.calls.filter( + ([bytes]) => decodeTerminalStreamFrame(bytes)?.opcode === TerminalStreamOpcode.Input + ) + expect(inputs).toHaveLength(1) + stream.close() + }) + + it('leaves receipt-requiring input to the RPC fallback on an older host', async () => { + const { stream, onInputUnverifiable } = await open(false) + expect(stream.sendInput('a', { requireReceipt: true })).toBe(false) + expect(sendBinary).not.toHaveBeenCalled() + expect(stream.sendInput('\x1b[?1;2c')).toBe(true) + expect(decodeTerminalStreamFrame(sendBinary.mock.calls[0][0])?.seq).toBe(0) + callbacks?.onClose() + expect(onInputUnverifiable).not.toHaveBeenCalled() + }) + + it('warns on a lost connection and never replays pending input into its replacement', async () => { + const { stream, onInputUnverifiable } = await open() + stream.sendInput('possibly-executed', { requireReceipt: true }) + callbacks?.onClose() + expect(onInputUnverifiable).toHaveBeenCalledOnce() + const replacement = await open() + callbacks?.onResponse({ + ok: true, + result: { type: 'input-ack', streamId: stream.streamId, seq: 1, outcome: 'accepted' } + }) + expect(sendBinary).not.toHaveBeenCalled() + replacement.stream.close() + expect(replacement.onInputUnverifiable).not.toHaveBeenCalled() + }) +}) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts index 5d53f5e76fdb..9c6aa1f9634d 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-base.ts @@ -137,13 +137,14 @@ export abstract class RemoteRuntimeTerminalMultiplexerBase { protected sendFrame( streamId: number, opcode: TerminalStreamOpcode, - payload: Uint8Array = new Uint8Array() + payload: Uint8Array = new Uint8Array(), + seq = 0 ): boolean { if (!this.matchesCurrentEnvironmentRevision() || !this.ready || !this.subscription) { return false } try { - this.subscription.sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq: 0, payload })) + this.subscription.sendBinary(encodeTerminalStreamFrame({ opcode, streamId, seq, payload })) recordE2eRemoteStreamFrame(opcode) return true } catch (error) { @@ -192,6 +193,7 @@ export abstract class RemoteRuntimeTerminalMultiplexerBase { // Why: close callbacks may resubscribe synchronously; release first so every replacement shares the new environment multiplexer. this.releaseIfCurrent(this.environmentId, this) for (const stream of streams) { + stream.inputReceipts.dispose() discardOutputAcknowledgements(stream) stream.watchdog.dispose() clearSnapshot(stream) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts index d72de381ea7b..cc6f59fe8d42 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-implementation.ts @@ -16,6 +16,7 @@ import type { RemoteRuntimeMultiplexedTerminalCallbacks, RemoteRuntimeMultiplexedTerminalState } from './remote-runtime-terminal-multiplexer-types' +import { RemoteTerminalInputReceipts } from './remote-terminal-input-receipts' import { createRemoteTerminalStreamWatchdog } from './remote-terminal-stream-watchdog' export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinaryController { @@ -34,6 +35,8 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar acknowledgeOutput: true, acknowledgeOutputSourceRanges: false, supportsOutputPause: false, + acknowledgeInput: false, + inputReceipts: new RemoteTerminalInputReceipts(() => args.callbacks.onInputUnverifiable?.()), outputPaused: false, streamGeneration: null, sourceAckedEndByte: 0, @@ -83,7 +86,8 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar const stream: RemoteRuntimeMultiplexedTerminal = { streamId, - sendInput: (text) => this.isRegisteredStream(state) && this.sendInput(state, text), + sendInput: (text, options) => + this.isRegisteredStream(state) && this.sendInput(state, text, options), resize: (cols, rows) => this.isRegisteredStream(state) && this.sendFrame( @@ -115,6 +119,7 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar serializeBufferOutcome: (opts) => this.requestSnapshotOutcome(state, opts), close: () => { if (this.streams.get(streamId) === state) { + state.inputReceipts.dispose() discardOutputAcknowledgements(state) state.watchdog.dispose() this.sendFrame(streamId, TerminalStreamOpcode.Unsubscribe) @@ -144,6 +149,7 @@ export class RemoteRuntimeTerminalMultiplexer extends RemoteRuntimeTerminalBinar ackOutputSourceRanges: 1, outputPause: 1, writeUnavailable: 1, + ackInput: 1, ...(args.client.type === 'desktop' ? { desktopViewportClaims: 1 } : {}) } }) diff --git a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts index c072e315796e..e1ffb5f721d4 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-multiplexer-types.ts @@ -1,5 +1,6 @@ import type { TerminalSnapshotUnavailableReason } from '../../../shared/terminal-snapshot-unavailability' import type { TerminalStreamEndVerdict } from '../../../shared/terminal-stream-end-verdict' +import type { RemoteTerminalInputReceipts } from './remote-terminal-input-receipts' import type { RemoteTerminalStreamWatchdog } from './remote-terminal-stream-watchdog' export type RuntimeEnvironmentSubscriptionHandle = { @@ -13,7 +14,7 @@ export type TerminalMultiplexEvent = type: 'subscribed' streamId: number streamGeneration?: string - capabilities?: { ackOutputSourceRanges?: 1; outputPause?: 1 } + capabilities?: { ackOutputSourceRanges?: 1; outputPause?: 1; ackInput?: 1 } } | { type: 'end'; streamId: number; verdict?: TerminalStreamEndVerdict } | { type: 'error'; streamId: number; message?: string } @@ -61,6 +62,7 @@ export type RemoteRuntimeMultiplexedTerminalCallbacks = { onDriverChanged?: ( driver: { kind: 'idle' } | { kind: 'desktop' } | { kind: 'mobile'; clientId: string } ) => void + onInputUnverifiable?: () => void onWriteUnavailable?: () => void onTransportClose?: (event: { recoverable: boolean; retryWithBackoff?: boolean }) => void } @@ -133,7 +135,7 @@ export type RemoteRuntimeSnapshotOutcome = { export type RemoteRuntimeMultiplexedTerminal = { streamId: number - sendInput: (text: string) => boolean + sendInput: (text: string, options?: { requireReceipt?: boolean }) => boolean resize: (cols: number, rows: number) => boolean claimViewport: (cols: number, rows: number) => boolean setOutputPaused: (paused: boolean) => boolean @@ -155,6 +157,8 @@ export type RemoteRuntimeMultiplexedTerminalState = { acknowledgeOutput: boolean acknowledgeOutputSourceRanges: boolean supportsOutputPause: boolean + acknowledgeInput: boolean + inputReceipts: RemoteTerminalInputReceipts outputPaused: boolean streamGeneration: string | null sourceAckedEndByte: number diff --git a/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts index 7453a85fa1d5..9d9958ca5910 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-response-controller.ts @@ -45,10 +45,9 @@ export abstract class RemoteRuntimeTerminalResponseController extends RemoteRunt return } if (event.type === 'subscribed') { - const capabilities = - typeof event.capabilities === 'object' && event.capabilities !== null - ? (event.capabilities as { ackOutputSourceRanges?: unknown; outputPause?: unknown }) - : null + const capabilities = isTerminalMultiplexCapabilities(event.capabilities) + ? event.capabilities + : null if ( capabilities?.ackOutputSourceRanges === 1 && typeof event.streamGeneration === 'string' && @@ -57,11 +56,24 @@ export abstract class RemoteRuntimeTerminalResponseController extends RemoteRunt stream.acknowledgeOutputSourceRanges = true stream.streamGeneration = event.streamGeneration } + stream.acknowledgeInput = capabilities?.ackInput === 1 stream.supportsOutputPause = capabilities?.outputPause === 1 if (stream.supportsOutputPause) { stream.callbacks.onOutputPauseCapability?.() } + } else if (event.type === 'input-ack') { + if ( + stream.acknowledgeInput && + typeof event.seq === 'number' && + Number.isSafeInteger(event.seq) && + (event.outcome === 'accepted' || + event.outcome === 'refused' || + event.outcome === 'unverifiable') + ) { + stream.inputReceipts.settle(event.seq, event.outcome) + } } else if (event.type === 'end') { + stream.inputReceipts.dispose() discardOutputAcknowledgements(stream) stream.watchdog.dispose() clearSnapshot(stream) @@ -123,3 +135,13 @@ export abstract class RemoteRuntimeTerminalResponseController extends RemoteRunt } } } + +type TerminalMultiplexCapabilities = { + ackOutputSourceRanges?: unknown + outputPause?: unknown + ackInput?: unknown +} + +function isTerminalMultiplexCapabilities(value: unknown): value is TerminalMultiplexCapabilities { + return typeof value === 'object' && value !== null +} diff --git a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts index 10fb18dff916..39cd2922f885 100644 --- a/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts +++ b/src/renderer/src/runtime/remote-runtime-terminal-snapshot-controller.ts @@ -252,6 +252,7 @@ export abstract class RemoteRuntimeTerminalSnapshotController extends RemoteRunt if (this.streams.get(stream.streamId) !== stream) { return } + stream.inputReceipts.dispose() stream.watchdog.dispose() discardOutputAcknowledgements(stream) clearSnapshot(stream) diff --git a/src/renderer/src/runtime/remote-terminal-input-receipts.test.ts b/src/renderer/src/runtime/remote-terminal-input-receipts.test.ts new file mode 100644 index 000000000000..b6fdf334aad5 --- /dev/null +++ b/src/renderer/src/runtime/remote-terminal-input-receipts.test.ts @@ -0,0 +1,82 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { + RemoteTerminalInputReceipts, + REMOTE_TERMINAL_INPUT_RECEIPT_LIMIT, + REMOTE_TERMINAL_INPUT_RECEIPT_TIMEOUT_MS +} from './remote-terminal-input-receipts' + +describe('remote terminal input receipts', () => { + beforeEach(() => vi.useFakeTimers()) + afterEach(() => vi.useRealTimers()) + + it('does not mistake a later write receipt for receipt of an earlier write', () => { + const warn = vi.fn() + const receipts = new RemoteTerminalInputReceipts(warn) + const first = receipts.begin() + const second = receipts.begin() + receipts.settle(second, 'accepted') + receipts.dispose() + expect(warn).toHaveBeenCalledOnce() + receipts.settle(first, 'accepted') + expect(vi.getTimerCount()).toBe(0) + }) + + it('warns once when a deadline or teardown leaves delivery uncertain', () => { + const warn = vi.fn() + const receipts = new RemoteTerminalInputReceipts(warn) + receipts.begin() + vi.advanceTimersByTime(REMOTE_TERMINAL_INPUT_RECEIPT_TIMEOUT_MS) + expect(warn).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + receipts.begin() + receipts.dispose() + expect(warn).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + }) + + it('keeps the deadline anchored to the oldest unanswered write', () => { + const warn = vi.fn() + const receipts = new RemoteTerminalInputReceipts(warn) + const first = receipts.begin() + vi.advanceTimersByTime(10_000) + const second = receipts.begin() + receipts.settle(second, 'accepted') + vi.advanceTimersByTime(5_000) + expect(warn).toHaveBeenCalledOnce() + receipts.settle(first, 'accepted') + receipts.dispose() + }) + + it('cleans up successful and refused writes without an uncertainty warning', () => { + const warn = vi.fn() + const receipts = new RemoteTerminalInputReceipts(warn) + receipts.settle(receipts.begin(), 'accepted') + receipts.settle(receipts.begin(), 'refused') + receipts.dispose() + expect(warn).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + }) + + it('reports an owner-provided unverifiable settlement', () => { + const warn = vi.fn() + const receipts = new RemoteTerminalInputReceipts(warn) + receipts.settle(receipts.begin(), 'unverifiable') + expect(warn).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(0) + }) + + it('bounds pending receipts and uses one timer during an input flood', () => { + const warn = vi.fn() + const receipts = new RemoteTerminalInputReceipts(warn) + for (let i = 0; i < REMOTE_TERMINAL_INPUT_RECEIPT_LIMIT; i++) { + receipts.begin() + } + expect(vi.getTimerCount()).toBe(1) + expect(warn).not.toHaveBeenCalled() + receipts.begin() + expect(warn).toHaveBeenCalledOnce() + expect(vi.getTimerCount()).toBe(1) + receipts.dispose() + expect(vi.getTimerCount()).toBe(0) + }) +}) diff --git a/src/renderer/src/runtime/remote-terminal-input-receipts.ts b/src/renderer/src/runtime/remote-terminal-input-receipts.ts new file mode 100644 index 000000000000..24f7970537b8 --- /dev/null +++ b/src/renderer/src/runtime/remote-terminal-input-receipts.ts @@ -0,0 +1,75 @@ +export const REMOTE_TERMINAL_INPUT_RECEIPT_TIMEOUT_MS = 15_000 +export const REMOTE_TERMINAL_INPUT_RECEIPT_LIMIT = 1_024 + +/** Tracks receipts, never input text: uncertain writes must not be replayed. */ +export class RemoteTerminalInputReceipts { + private readonly pending = new Map() + private nextSequence = 1 + private timer: ReturnType | null = null + private warned = false + + constructor(private readonly onUnverifiable: () => void) {} + + begin(): number { + if (this.pending.size >= REMOTE_TERMINAL_INPUT_RECEIPT_LIMIT) { + this.reportUnverifiable() + this.clear() + } + const sequence = this.nextSequence++ + this.pending.set(sequence, Date.now()) + this.scheduleDeadline() + return sequence + } + + settle(sequence: number, outcome: 'accepted' | 'refused' | 'unverifiable'): void { + if (!this.pending.delete(sequence)) { + return + } + if (outcome === 'unverifiable') { + this.reportUnverifiable() + } + this.cancelDeadline() + this.scheduleDeadline() + } + + dispose(): void { + if (this.pending.size > 0) { + this.reportUnverifiable() + } + this.clear() + } + + private reportUnverifiable(): void { + if (!this.warned) { + this.warned = true + this.onUnverifiable() + } + } + + private scheduleDeadline(): void { + const oldest = this.pending.values().next().value + if (this.timer !== null || oldest === undefined) { + return + } + this.timer = setTimeout( + () => { + this.timer = null + this.reportUnverifiable() + this.clear() + }, + Math.max(0, oldest + REMOTE_TERMINAL_INPUT_RECEIPT_TIMEOUT_MS - Date.now()) + ) + } + + private cancelDeadline(): void { + if (this.timer !== null) { + clearTimeout(this.timer) + this.timer = null + } + } + + private clear(): void { + this.cancelDeadline() + this.pending.clear() + } +} diff --git a/tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts b/tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts index 5c8506d35683..e9ca9d1c5e52 100644 --- a/tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts +++ b/tests/e2e/paired-runtime-rejected-input-remount.unit.test.ts @@ -46,7 +46,7 @@ vi.mock('@/store', () => ({ })) vi.mock('@/runtime/sync-runtime-graph', () => ({ scheduleRuntimeGraphSync: vi.fn() })) -vi.mock('sonner', () => ({ toast: { info: vi.fn() } })) +vi.mock('sonner', () => ({ toast: { info: vi.fn(), warning: vi.fn() } })) vi.mock('@/lib/codex-stale-pane-sweep', () => ({ notifyCodexPaneBoundForStaleSweep: vi.fn() })) vi.mock('@/runtime/web-runtime-session', () => ({ refreshWebRuntimeSessionTabsSnapshot: vi.fn(async () => {}) @@ -71,7 +71,10 @@ function startHost(): { } { 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 sendTerminal = vi.fn().mockResolvedValue({ + accepted: false, + writeSettlement: { outcome: 'refused', reason: 'provider_refused_write' } + }) const runtime = { getRuntimeId: () => 'test-runtime', registerRemoteTerminalViewSubscriber: () => () => {}, diff --git a/tests/e2e/ssh-orcad-half-open-input.spec.ts b/tests/e2e/ssh-orcad-half-open-input.spec.ts new file mode 100644 index 000000000000..e871a768fc96 --- /dev/null +++ b/tests/e2e/ssh-orcad-half-open-input.spec.ts @@ -0,0 +1,148 @@ +import { readFileSync, rmSync } from 'node:fs' +import type { ElectronApplication } from '@stablyai/playwright-test' +import { expect, test } from './helpers/orca-app' +import { startOrcadConvertHost } from './helpers/orcad-convert-host' +import { createRestartSession } from './helpers/orca-restart' +import { seedRelayEraProfile } from './helpers/orcad-upgrade-profile' +import { convertAndRetain } from './helpers/orcad-convert-flow' +import { createPairedHostTerminal, openPairedClientTab } from './helpers/paired-host-terminal' +import { focusActiveTerminalInput } from './helpers/terminal' +import { expectTerminalAccessibilityText } from './helpers/terminal-accessibility-tree' +import { + createRetentionFixtureDirectory, + writeRetentionFixture +} from './helpers/host-created-terminal-retention-oracle' +import { shellEscape } from '../../src/main/ssh/ssh-connection-utils' +import { startFreezableTcpProxy } from './helpers/freezable-tcp-proxy' + +const TEMPLATE = process.env.ORCA_E2E_ORCAD_CONVERT_TEMPLATE +test.skip( + !TEMPLATE || process.env.ORCA_E2E_SSH_DOCKER !== '1', + 'Needs the Docker SSH host and server template' +) + +for (const direction of ['both', 'to-client'] as const) { + test(`managed host warns about uncertain input when ${direction} traffic stalls`, async (// oxlint-disable-next-line no-empty-pattern -- This test owns both app launches and uses only testInfo. + {}, testInfo) => { + test.setTimeout(5 * 60_000) + const host = startOrcadConvertHost('docker', testInfo) + const proxy = await startFreezableTcpProxy(host.input.host!, host.input.port ?? 22) + const input = { ...host.input, port: proxy.port } + const session = createRestartSession(testInfo, { ORCA_ORCAD_TEMPLATE_PATH: TEMPLATE! }) + const fixtureDirectory = createRetentionFixtureDirectory() + let app: ElectronApplication | null = null + let stalled = false + try { + if (!host.exec) { + throw new Error('Docker controls are unavailable') + } + const first = await session.launch() + app = first.app + await session.close(app) + app = null + const seeded = seedRelayEraProfile(session.userDataDir, input, { + repoPath: host.remoteRepoPath, + folderPath: host.remoteFolderPath + }) + const current = await session.launch() + app = current.app + const page = current.page + await convertAndRetain(page, session.userDataDir, seeded) + const environment = (await page.evaluate(() => window.api.runtimeEnvironments.list())).find( + (entry) => entry.orcadDeployment?.sshTargetId === seeded.targetId + ) + if (!environment) { + throw new Error('Managed host was not registered') + } + const fixture = writeRetentionFixture(fixtureDirectory) + const remoteFixture = '/tmp/orca-half-open-input.mjs' + const sink = '/tmp/orca-half-open-input.log' + host.exec( + `printf '%s' ${shellEscape(readFileSync(fixture, 'utf8'))} > ${shellEscape(remoteFixture)}` + ) + const terminal = await createPairedHostTerminal( + page, + environment.id, + seeded.worktreeId, + `node ${shellEscape(remoteFixture)} ${shellEscape(sink)}` + ) + await openPairedClientTab(page, seeded.worktreeId, terminal.webTabId) + await expectTerminalAccessibilityText(page, terminal.webTabId, 'READY:') + await expect( + page.getByText('Remote terminal input delivery is uncertain', { exact: true }) + ).toBeHidden() + const runId = Date.now() + const before = `BEFORE_${runId}` + const during = `DURING_${runId}` + const after = `AFTER_${runId}` + const typeLine = async (line: string): Promise => { + await focusActiveTerminalInput(page) + await page.keyboard.type(line) + await page.keyboard.press('Enter') + } + await typeLine(before) + await expectTerminalAccessibilityText(page, terminal.webTabId, `LINE:${before}`) + const readState = () => + page.evaluate( + (id) => ({ + ssh: window.__store?.getState().sshConnectionStates.get(id.targetId)?.status, + runtime: window.__store?.getState().runtimeStatusByEnvironmentId.get(id.environmentId) + ?.snapshot?.transport + }), + { targetId: seeded.targetId, environmentId: environment.id } + ) + const initially = await readState() + const start = Date.now() + if (direction === 'both') { + proxy.cut('silent') + } else { + expect(proxy.freezeExisting('to-client')).toBeGreaterThan(0) + } + stalled = true + await typeLine(during) + const typedState = await readState() + await page.screenshot({ path: testInfo.outputPath('half-open-input-typed.png') }) + const warning = page.getByText('Remote terminal input delivery is uncertain', { exact: true }) + await expect(warning).toBeVisible({ timeout: 60_000 }) + const warnedMs = Date.now() - start + proxy.restore() + stalled = false + await expect.poll(readState, { timeout: 120_000 }).toEqual(initially) + await typeLine(after) + await expectTerminalAccessibilityText(page, terminal.webTabId, `LINE:${after}`, 60_000) + const received = host.exec(`cat ${shellEscape(sink)}`) + const report = { + warnedMs, + initially, + typedState, + received, + duringReceived: received.includes(`LINE:${during}`), + proxyEvents: proxy?.events + } + console.log('[half-open-input]', JSON.stringify(report)) + await testInfo.attach('half-open-input-report.json', { + body: JSON.stringify(report, null, 2), + contentType: 'application/json' + }) + await page.screenshot({ path: testInfo.outputPath('half-open-input-recovered.png') }) + expect(received.split('\n').filter((line) => line.startsWith('READY:'))).toHaveLength(1) + expect(received.split('\n').filter((line) => line === `LINE:${before}`)).toHaveLength(1) + expect(received.split('\n').filter((line) => line === `LINE:${after}`)).toHaveLength(1) + expect(received.split('\n').filter((line) => line === `LINE:${during}`)).toHaveLength( + direction === 'both' ? 0 : 1 + ) + await expect(warning).toBeVisible() + } finally { + if (stalled) { + proxy.restore() + } + if (app) { + await session.close(app) + } + await session.dispose() + await proxy?.close() + host.cleanup() + rmSync(fixtureDirectory, { recursive: true, force: true }) + } + }) +} From 7fa212989b1cc920dd843e83934feaa81429bf4d Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 8 Oct 2026 04:05:11 -0700 Subject: [PATCH 2/3] test(remote): check negotiated input receipt capability across versions --- .../cross-version-terminal-wire.unit.test.ts | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts b/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts index 22efc31bd3cc..bf22fa8428f5 100644 --- a/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts +++ b/tests/e2e/cross-version-wire/cross-version-terminal-wire.unit.test.ts @@ -100,7 +100,7 @@ function expectSnapshotStartFieldsRemainPublished(args: { } } -function expectWireCompatible(record: JourneyRecord): void { +function expectWireCompatible(record: JourneyRecord, ackInput = false): void { // Rule 2 — no frame may be refused by the receiving build's decoder. An opcode // the peer does not know is dropped silently, so this is the only signal. expect(record.rejected).toEqual([]) @@ -109,7 +109,9 @@ function expectWireCompatible(record: JourneyRecord): void { // The subscribe handshake still negotiates the optional output-pause opcode, // which is what keeps opcode 16 legal to send on this pairing. for (const event of record.subscribedEvents) { - expect(event.capabilities).toEqual({ outputPause: 1 }) + expect(event.capabilities).toEqual( + ackInput ? { outputPause: 1, ackInput: 1 } : { outputPause: 1 } + ) } // Input reached the process, before and after the reconnect. @@ -152,7 +154,7 @@ describe('cross-version remote terminal wire', () => { it('current client against current server completes the journey, and is the reference for a current host', () => { expectJourneyActuallyRan(currentReference) - expectWireCompatible(currentReference) + expectWireCompatible(currentReference, true) expect(currentReference.snapshotStarts).toEqual([ expect.objectContaining({ alternateScreen: false, terminalOwner: 'shell' }), expect.objectContaining({ alternateScreen: false, terminalOwner: 'shell' }), From 597c563952a80d0ae1a2ae0e20635624e53a40b0 Mon Sep 17 00:00:00 2001 From: Neil Date: Thu, 8 Oct 2026 04:46:13 -0700 Subject: [PATCH 3/3] Cover input receipt capabilities and run the SSH fault oracle in CI --- .github/workflows/e2e.yml | 7 +- config/scripts/ci-e2e-job-selection.mjs | 7 +- config/scripts/ci-e2e-job-selection.test.mjs | 8 ++ .../pr-e2e-docker-lane-contract.test.mjs | 76 +++++++++++++++++++ config/scripts/pr-e2e-gate-contract.test.mjs | 70 +---------------- config/scripts/pr-e2e-source-routing.mjs | 9 +++ .../runtime/runtime-terminal-stream.test.ts | 2 + 7 files changed, 108 insertions(+), 71 deletions(-) create mode 100644 config/scripts/pr-e2e-docker-lane-contract.test.mjs diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 2ac4365dcfb6..e8e31832a626 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -348,7 +348,8 @@ jobs: . != "tests/e2e/orcad-serve-mode-switch.spec.ts" and . != "tests/e2e/ssh-orcad-auto-convert.spec.ts" and . != "tests/e2e/windows-missing-appdata-startup.spec.ts" and - . != "tests/e2e/ssh-orcad-idle-exit.spec.ts" + . != "tests/e2e/ssh-orcad-idle-exit.spec.ts" and + . != "tests/e2e/ssh-orcad-half-open-input.spec.ts" )' <<<"$TEST_FILES_JSON" > "$RUNNER_TEMP/general-e2e-specs" fi mapfile -t TEST_FILES < "$RUNNER_TEMP/general-e2e-specs" @@ -568,7 +569,7 @@ jobs: orcad-auto-convert-docker: name: ssh host auto-converts to managed orcad (Docker) needs: [build, prepare-native-cache] - if: inputs.test_files == '' || contains(inputs.test_files, 'tests/e2e/ssh-orcad-auto-convert.spec.ts') || contains(inputs.test_files, 'tests/e2e/ssh-orcad-idle-exit.spec.ts') + if: inputs.test_files == '' || contains(inputs.test_files, 'tests/e2e/ssh-orcad-auto-convert.spec.ts') || contains(inputs.test_files, 'tests/e2e/ssh-orcad-idle-exit.spec.ts') || contains(inputs.test_files, 'tests/e2e/ssh-orcad-half-open-input.spec.ts') runs-on: ubuntu-latest timeout-minutes: 45 env: @@ -604,7 +605,7 @@ jobs: ORCA_RELAY_PATH: ${{ github.workspace }}/out/relay run: | export ORCA_E2E_ORCAD_CONVERT_TEMPLATE="$RUNNER_TEMP/orcad-convert-template" - xvfb-run --auto-servernum bash .github/scripts/e2e-with-window-manager.sh pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/ssh-orcad-auto-convert.spec.ts tests/e2e/ssh-orcad-idle-exit.spec.ts --project=electron-headless --workers=1 + xvfb-run --auto-servernum bash .github/scripts/e2e-with-window-manager.sh pnpm exec playwright test --config tests/playwright.config.ts tests/e2e/ssh-orcad-auto-convert.spec.ts tests/e2e/ssh-orcad-idle-exit.spec.ts tests/e2e/ssh-orcad-half-open-input.spec.ts --project=electron-headless --workers=1 - uses: actions/upload-artifact@v7 if: failure() with: diff --git a/config/scripts/ci-e2e-job-selection.mjs b/config/scripts/ci-e2e-job-selection.mjs index 53747f9d496d..a6c48a57cfec 100644 --- a/config/scripts/ci-e2e-job-selection.mjs +++ b/config/scripts/ci-e2e-job-selection.mjs @@ -48,6 +48,7 @@ export const ORCAD_AUTO_CONVERT_E2E_SPEC = 'tests/e2e/ssh-orcad-auto-convert.spe export const WINDOWS_MISSING_APPDATA_E2E_SPEC = 'tests/e2e/windows-missing-appdata-startup.spec.ts' // Runs in the auto-convert job, which builds the template it needs. export const ORCAD_IDLE_EXIT_E2E_SPEC = 'tests/e2e/ssh-orcad-idle-exit.spec.ts' +export const ORCAD_HALF_OPEN_INPUT_E2E_SPEC = 'tests/e2e/ssh-orcad-half-open-input.spec.ts' export const DEDICATED_E2E_SPECS = [ ...DOCKER_SSH_E2E_SPECS, NODE_NETWORK_E2E_SPEC, @@ -56,7 +57,8 @@ export const DEDICATED_E2E_SPECS = [ ORCAD_SERVE_MODE_SWITCH_E2E_SPEC, ORCAD_AUTO_CONVERT_E2E_SPEC, WINDOWS_MISSING_APPDATA_E2E_SPEC, - ORCAD_IDLE_EXIT_E2E_SPEC + ORCAD_IDLE_EXIT_E2E_SPEC, + ORCAD_HALF_OPEN_INPUT_E2E_SPEC ] const dedicatedSpecs = new Set(DEDICATED_E2E_SPECS) const dockerSpecs = new Set(DOCKER_SSH_E2E_SPECS) @@ -97,7 +99,8 @@ export function classifyE2eJobs(input, sshSourceChanged = 'false') { spec === LOCALHOST_SSH_E2E_SPEC || spec === ORCAD_SERVE_MODE_SWITCH_E2E_SPEC || spec === ORCAD_AUTO_CONVERT_E2E_SPEC || - spec === ORCAD_IDLE_EXIT_E2E_SPEC + spec === ORCAD_IDLE_EXIT_E2E_SPEC || + spec === ORCAD_HALF_OPEN_INPUT_E2E_SPEC ) } } diff --git a/config/scripts/ci-e2e-job-selection.test.mjs b/config/scripts/ci-e2e-job-selection.test.mjs index cfe71813409b..1e97a1625b5d 100644 --- a/config/scripts/ci-e2e-job-selection.test.mjs +++ b/config/scripts/ci-e2e-job-selection.test.mjs @@ -145,3 +145,11 @@ it('runs remaining SSH tests after real failures and stops them when a run is ca expect(step.if).toContain('always()') } }) + +it.each([ + 'src/main/runtime/rpc/methods/terminal/terminal-multiplex-input-delivery.ts', + 'src/renderer/src/runtime/remote-terminal-input-receipts.ts', + 'src/renderer/src/components/terminal-pane/remote-runtime-pty-transport.ts' +])('routes input delivery faults from %s', (path) => { + expect(selectPrE2eSpecs([path])).toContain('tests/e2e/ssh-orcad-half-open-input.spec.ts') +}) diff --git a/config/scripts/pr-e2e-docker-lane-contract.test.mjs b/config/scripts/pr-e2e-docker-lane-contract.test.mjs new file mode 100644 index 000000000000..e20654eeaec5 --- /dev/null +++ b/config/scripts/pr-e2e-docker-lane-contract.test.mjs @@ -0,0 +1,76 @@ +import { readdirSync, readFileSync } from 'node:fs' +import { join, resolve } from 'node:path' +import { expect, it } from 'vitest' +import { parse } from 'yaml' + +const projectDir = resolve(import.meta.dirname, '../..') +const e2eWorkflow = parse(readFileSync(join(projectDir, '.github/workflows/e2e.yml'), 'utf8')) + +it('gives every Docker-gated SSH spec a lane that runs it', () => { + // Why this shape: the sharded lanes set no ORCA_E2E_SSH_DOCKER, so a Docker-gated spec + // that no runner names runs nowhere and still reports green — the silent skip this file + // exists to prevent. Asserting reachability rather than a literal keeps that true when + // the lanes move. + // The remaining exemption needs performance validation before routine CI, recorded in + // run-ssh-docker-e2e.mjs so the gap stays legible rather than looking like coverage. + const unreachableSpecs = new Set(['tests/e2e/ssh-docker-relay-perf.spec.ts']) + // Why comments are stripped: the runner documents the exempt spec by name in a + // prose comment. A substring scan over raw text would count any spec merely *discussed* in a + // runner as claimed by it -- the silent skip this assertion exists to catch, re-entering + // through the documentation. + const stripComments = (text) => text.replace(/\/\*[\s\S]*?\*\//g, '').replace(/^\s*\/\/.*$/gm, '') + const laneRunners = [ + 'run-ssh-docker-e2e.mjs', + 'run-ssh-docker-watcher-isolation-e2e.mjs', + 'run-ssh-docker-terminal-parking-e2e.mjs' + ].map((file) => stripComments(readFileSync(join(projectDir, 'config/scripts', file), 'utf8'))) + const managedLane = e2eWorkflow.jobs['orcad-auto-convert-docker'] + const managedRun = managedLane.steps.find( + (step) => step.name === 'Convert a relay-era Docker host' + ) + expect(managedRun.env.ORCA_E2E_SSH_DOCKER).toBe('1') + expect(managedRun.run).toContain('ORCA_E2E_ORCAD_CONVERT_TEMPLATE=') + laneRunners.push(stripComments(managedRun.run)) + + // Why a comparison and not the bare name: preview and demo specs cite the flag in a + // "how to run me" comment without gating on it. Why a regex rather than one literal: an + // equally-valid spelling (double quotes, or a `!==` guard) would escape a fixed-string scan + // and the spec would silently leave the contract. + const dockerGateExpression = /ORCA_E2E_SSH_DOCKER\s*[!=]==\s*['"]1['"]/ + const dockerGatedSpecs = readdirSync(join(projectDir, 'tests/e2e')) + .filter((file) => file.endsWith('.spec.ts')) + .map((file) => `tests/e2e/${file}`) + .filter((spec) => dockerGateExpression.test(readFileSync(join(projectDir, spec), 'utf8'))) + expect(dockerGatedSpecs.length).toBeGreaterThan(0) + + const unclaimed = dockerGatedSpecs.filter( + (spec) => !unreachableSpecs.has(spec) && !laneRunners.some((runner) => runner.includes(spec)) + ) + expect( + unclaimed, + `Docker-gated specs claimed by no lane runner: ${unclaimed.join(', ')}` + ).toEqual([]) + + // Why: an exemption that outlives its spec would quietly excuse a real gap. + for (const spec of unreachableSpecs) { + expect(dockerGatedSpecs, spec).toContain(spec) + // Why also assert absence from every runner: `unreachableSpecs` short-circuits the + // unclaimed check above, so a spec could be documented as exempt while a runner still + // invokes it -- an exemption that reads as coverage removal but changes nothing, and a + // lane that stays red for a reason the file says it excluded. + for (const runner of laneRunners) { + expect(runner.includes(spec), `${spec} is exempt but still invoked by a lane runner`).toBe( + false + ) + } + } + + const laneStep = e2eWorkflow.jobs['ssh-docker-watcher-isolation'].steps.find( + (step) => step.name === 'Run remaining Docker SSH E2E' + ) + expect(laneStep.run).toContain('test:e2e:ssh-docker') + // Why: the added serial tests, several budgeting 4-10 minutes each, do not fit the old 35. + expect( + e2eWorkflow.jobs['ssh-docker-watcher-isolation']['timeout-minutes'] + ).toBeGreaterThanOrEqual(60) +}) diff --git a/config/scripts/pr-e2e-gate-contract.test.mjs b/config/scripts/pr-e2e-gate-contract.test.mjs index f9fe2d7d7327..04e0cff2da5c 100644 --- a/config/scripts/pr-e2e-gate-contract.test.mjs +++ b/config/scripts/pr-e2e-gate-contract.test.mjs @@ -385,69 +385,6 @@ describe('PR E2E gate contract', () => { expect(filterStep.run).toContain('ssh_source_changed=$SSH_SOURCE_CHANGED') }) - it('gives every Docker-gated SSH spec a lane that runs it', () => { - // Why this shape: the sharded lanes set no ORCA_E2E_SSH_DOCKER, so a Docker-gated spec - // that no runner names runs nowhere and still reports green — the silent skip this file - // exists to prevent. Asserting reachability rather than a literal keeps that true when - // the lanes move. - // The remaining exemption needs performance validation before routine CI, recorded in - // run-ssh-docker-e2e.mjs so the gap stays legible rather than looking like coverage. - const unreachableSpecs = new Set(['tests/e2e/ssh-docker-relay-perf.spec.ts']) - // Why comments are stripped: the runner documents the exempt spec by name in a - // prose comment. A substring scan over raw text would count any spec merely *discussed* in a - // runner as claimed by it -- the silent skip this assertion exists to catch, re-entering - // through the documentation. - const stripComments = (text) => - text.replace(/\/\*[\s\S]*?\*\//g, '').replace(/^\s*\/\/.*$/gm, '') - const laneRunners = [ - 'run-ssh-docker-e2e.mjs', - 'run-ssh-docker-watcher-isolation-e2e.mjs', - 'run-ssh-docker-terminal-parking-e2e.mjs' - ].map((file) => stripComments(readFileSync(join(projectDir, 'config/scripts', file), 'utf8'))) - - // Why a comparison and not the bare name: preview and demo specs cite the flag in a - // "how to run me" comment without gating on it. Why a regex rather than one literal: an - // equally-valid spelling (double quotes, or a `!==` guard) would escape a fixed-string scan - // and the spec would silently leave the contract. - const dockerGateExpression = /ORCA_E2E_SSH_DOCKER\s*[!=]==\s*['"]1['"]/ - const dockerGatedSpecs = readdirSync(join(projectDir, 'tests/e2e')) - .filter((file) => file.endsWith('.spec.ts')) - .map((file) => `tests/e2e/${file}`) - .filter((spec) => dockerGateExpression.test(readFileSync(join(projectDir, spec), 'utf8'))) - expect(dockerGatedSpecs.length).toBeGreaterThan(0) - - const unclaimed = dockerGatedSpecs.filter( - (spec) => !unreachableSpecs.has(spec) && !laneRunners.some((runner) => runner.includes(spec)) - ) - expect( - unclaimed, - `Docker-gated specs claimed by no lane runner: ${unclaimed.join(', ')}` - ).toEqual([]) - - // Why: an exemption that outlives its spec would quietly excuse a real gap. - for (const spec of unreachableSpecs) { - expect(dockerGatedSpecs, spec).toContain(spec) - // Why also assert absence from every runner: `unreachableSpecs` short-circuits the - // unclaimed check above, so a spec could be documented as exempt while a runner still - // invokes it -- an exemption that reads as coverage removal but changes nothing, and a - // lane that stays red for a reason the file says it excluded. - for (const runner of laneRunners) { - expect(runner.includes(spec), `${spec} is exempt but still invoked by a lane runner`).toBe( - false - ) - } - } - - const laneStep = e2eWorkflow.jobs['ssh-docker-watcher-isolation'].steps.find( - (step) => step.name === 'Run remaining Docker SSH E2E' - ) - expect(laneStep.run).toContain('test:e2e:ssh-docker') - // Why: the added serial tests, several budgeting 4-10 minutes each, do not fit the old 35. - expect( - e2eWorkflow.jobs['ssh-docker-watcher-isolation']['timeout-minutes'] - ).toBeGreaterThanOrEqual(60) - }) - it('scopes the VM rollback oracle to the PR range and recipe schema authorities', () => { expect(rollbackStep.run).toMatch(/diff-base\.mjs "\$BASE_SHA"[\s\S]*"\$DIFF_BASE" HEAD --/) expect(rollbackStep.run).toContain('src/shared/ephemeral-vm-recipes.ts') @@ -470,15 +407,16 @@ describe('PR E2E gate contract', () => { ], [ 'src/renderer/src/runtime/remote-runtime-terminal-multiplexer.ts', - 'tests/e2e/paired-remote-terminal-materialization-reconnect.spec.ts' + 'tests/e2e/paired-remote-terminal-materialization-reconnect.spec.ts', + ['tests/e2e/ssh-orcad-half-open-input.spec.ts'] ], [ 'src/renderer/src/components/terminal-pane/remote-pane-layout-push.ts', 'tests/e2e/paired-remote-pane-layout-retry.spec.ts' ] ] - for (const [source, spec] of cases) { - expect(selectPrE2eSpecs([source]), source).toEqual([spec]) + for (const [source, spec, additionalSpecs = []] of cases) { + expect(selectPrE2eSpecs([source]), source).toEqual([spec, ...additionalSpecs]) expect(selectPrE2eSpecs([source.replace(/\.tsx?$/, '.test.ts')]), source).toEqual([]) expect(existsSync(join(projectDir, spec)), spec).toBe(true) } diff --git a/config/scripts/pr-e2e-source-routing.mjs b/config/scripts/pr-e2e-source-routing.mjs index f080b28cb8d6..b17f961bb467 100644 --- a/config/scripts/pr-e2e-source-routing.mjs +++ b/config/scripts/pr-e2e-source-routing.mjs @@ -45,6 +45,15 @@ export const PR_E2E_SOURCE_ROUTES = [ file )) }, + { + id: 'ssh.orcad-input-delivery', + specs: ['tests/e2e/ssh-orcad-half-open-input.spec.ts'], + matches: (file) => + isProductSource(file) && + /^src\/(?:main\/runtime\/rpc\/methods\/terminal\/(?:terminal-(?:multiplex|input-delivery|stream)|stream-schemas)|renderer\/src\/(?:runtime\/(?:remote-runtime-terminal-(?:multiplexer|flow-controller|response-controller)|remote-terminal-input-receipts)|components\/terminal-pane\/(?:remote-runtime-pty-transport|remote-terminal-input-delivery-warning)))/.test( + file + ) + }, { id: 'ssh.orcad-idle-exit', specs: ['tests/e2e/ssh-orcad-idle-exit.spec.ts'], diff --git a/src/renderer/src/runtime/runtime-terminal-stream.test.ts b/src/renderer/src/runtime/runtime-terminal-stream.test.ts index cfaa771af49c..d6a2fe18644f 100644 --- a/src/renderer/src/runtime/runtime-terminal-stream.test.ts +++ b/src/renderer/src/runtime/runtime-terminal-stream.test.ts @@ -103,6 +103,7 @@ describe('remote runtime terminal data subscriptions', () => { decodeTerminalStreamJson<{ streamId: number capabilities?: { + ackInput?: 1 ackOutput?: 1 ackOutputSourceRanges?: 1 desktopViewportClaims?: 1 @@ -112,6 +113,7 @@ describe('remote runtime terminal data subscriptions', () => { }>(subscribeFrame.payload) expect(subscribePayload?.streamId).toEqual(expect.any(Number)) expect(subscribePayload?.capabilities).toEqual({ + ackInput: 1, ackOutput: 1, ackOutputSourceRanges: 1, desktopViewportClaims: 1,