diff --git a/src/main/ipc/pty-controller-ownership-routing.test.ts b/src/main/ipc/pty-controller-ownership-routing.test.ts index ac4beb018d0..0d236e63cb2 100644 --- a/src/main/ipc/pty-controller-ownership-routing.test.ts +++ b/src/main/ipc/pty-controller-ownership-routing.test.ts @@ -94,6 +94,15 @@ describe('registerPtyHandlers', () => { unregisterSshPtyProvider(connectionId) clearProviderPtyState(ptyId) }) + it('preserves a provider write refusal for callers that gate follow-up input', () => { + const provider = createAgentClaimProvider({}) + provider.write.mockReturnValue(false) + setLocalPtyProvider(provider as never) + const controller = registerAgentClaimController() + + expect(controller.write('pty-refused', 'input')).toBe(false) + expect(provider.write).toHaveBeenCalledWith('pty-refused', 'input') + }) describe('controller probePtyLiveness routing', () => { it('proves absence for an id the in-process local provider never owned', async () => { setLocalPtyProvider(new LocalPtyProvider()) diff --git a/src/main/ipc/pty/runtime/operations.ts b/src/main/ipc/pty/runtime/operations.ts index 00db3380309..ae1dcc33cf0 100644 --- a/src/main/ipc/pty/runtime/operations.ts +++ b/src/main/ipc/pty/runtime/operations.ts @@ -34,8 +34,7 @@ export function writePtyFromRuntimeController( return false } try { - getProviderForPty(ptyId).write(ptyId, data) - return true + return getProviderForPty(ptyId).write(ptyId, data) !== false } catch { return false } diff --git a/src/main/providers/ssh-pty-write.test.ts b/src/main/providers/ssh-pty-write.test.ts index 7da03677b3e..77cd529ba63 100644 --- a/src/main/providers/ssh-pty-write.test.ts +++ b/src/main/providers/ssh-pty-write.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { SshPtyProvider } from './ssh-pty-provider' import { SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS } from './ssh-pty-write' +import { MULTIPLEXER_ORDINARY_QUEUE_MAX_BYTES } from '../ssh/ssh-multiplexer-transport-writer' describe('SSH PTY writes', () => { afterEach(() => { @@ -42,6 +43,37 @@ describe('SSH PTY writes', () => { await expect(pending).resolves.toBe(false) }) + it('rejects an atomic write that cannot fit in one ordinary relay frame', () => { + const mux = { + isDisposed: vi.fn().mockReturnValue(false), + notify: vi.fn(), + onNotification: vi.fn() + } + const provider = new SshPtyProvider('conn-1', mux as never) + + expect( + provider.write('ssh:conn-1@@pty-1', 'x'.repeat(MULTIPLEXER_ORDINARY_QUEUE_MAX_BYTES)) + ).toBe(false) + expect(mux.notify).not.toHaveBeenCalled() + }) + + it('rejects an oversized settled write before touching the mux', async () => { + const mux = { + isDisposed: vi.fn().mockReturnValue(false), + notifyWithSettlement: vi.fn(), + onNotification: vi.fn() + } + const provider = new SshPtyProvider('conn-1', mux as never) + + await expect( + provider.writeWithSettlement( + 'ssh:conn-1@@pty-1', + 'x'.repeat(MULTIPLEXER_ORDINARY_QUEUE_MAX_BYTES) + ) + ).resolves.toBe(false) + expect(mux.notifyWithSettlement).not.toHaveBeenCalled() + }) + it('rejects settled writes immediately after the transport is disposed', async () => { const mux = { isDisposed: vi.fn().mockReturnValue(true), diff --git a/src/main/providers/ssh-pty-write.ts b/src/main/providers/ssh-pty-write.ts index 08adae7e833..6f5b65d20d0 100644 --- a/src/main/providers/ssh-pty-write.ts +++ b/src/main/providers/ssh-pty-write.ts @@ -1,9 +1,23 @@ import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer' -import { TIMEOUT_MS } from '../ssh/relay-protocol' +import { encodeJsonRpcFrame, TIMEOUT_MS } from '../ssh/relay-protocol' +import { MULTIPLEXER_ORDINARY_QUEUE_MAX_BYTES } from '../ssh/ssh-multiplexer-transport-writer' // Allow ordinary-lane backpressure to clear well beyond the mux health window. export const SSH_PTY_WRITE_SETTLEMENT_TIMEOUT_MS = TIMEOUT_MS * 3 +export function assertSshPtyWriteFitsTransport(relayPtyId: string, data: string): void { + const frame = encodeJsonRpcFrame( + { jsonrpc: '2.0', method: 'pty.data', params: { id: relayPtyId, data } }, + 0, + 0 + ) + if (frame.length > MULTIPLEXER_ORDINARY_QUEUE_MAX_BYTES) { + throw new Error( + `SSH PTY input exceeds the ${MULTIPLEXER_ORDINARY_QUEUE_MAX_BYTES}-byte transport limit` + ) + } +} + export function writeToSshPty( mux: SshChannelMultiplexer, relayPtyId: string, @@ -12,6 +26,11 @@ export function writeToSshPty( if (mux.isDisposed()) { return false } + try { + assertSshPtyWriteFitsTransport(relayPtyId, data) + } catch { + return false + } mux.notify('pty.data', { id: relayPtyId, data }) return !mux.isDisposed() } @@ -24,6 +43,11 @@ export function writeToSshPtyWithSettlement( if (mux.isDisposed()) { return Promise.resolve(false) } + try { + assertSshPtyWriteFitsTransport(relayPtyId, data) + } catch { + return Promise.resolve(false) + } return new Promise((resolve) => { let settled = false const finish = (accepted: boolean): void => { diff --git a/src/main/runtime/agent-prompt-submission-runtime.test.ts b/src/main/runtime/agent-prompt-submission-runtime.test.ts index 3b8fe67bea7..14ac4cf4fb9 100644 --- a/src/main/runtime/agent-prompt-submission-runtime.test.ts +++ b/src/main/runtime/agent-prompt-submission-runtime.test.ts @@ -270,7 +270,7 @@ describe('agent prompt submission runtime', () => { expect(writes).not.toContain('\r') }) - it('stops a chunked paste when permission appears between chunks', async () => { + it('does not submit an atomic paste after permission appears', async () => { const { runtime, handle, writes } = await createPromptRuntime(() => undefined) let writeChecks = 0 @@ -284,12 +284,12 @@ describe('agent prompt submission runtime', () => { }) await expect(submission).rejects.toThrow('agent_prompt_blocked') - expect(writes).toHaveLength(2) - expect(writes[1]).toBe(AGENT_PROMPT_BRACKETED_PASTE_END) + expect(writes).toHaveLength(1) + expect(writes[0]).toContain(AGENT_PROMPT_BRACKETED_PASTE_END) expect(writes).not.toContain('\r') }) - it('stops a chunked paste after transient output-only permission', async () => { + it('does not submit an atomic paste after transient output-only permission', async () => { const { runtime, handle, writes } = await createPromptRuntime(() => undefined) runtime.onPtyData('pty-prompt', 'initial output\n', Date.now()) let writeChecks = 0 @@ -309,8 +309,8 @@ describe('agent prompt submission runtime', () => { }) await expect(submission).rejects.toThrow('agent_prompt_blocked') - expect(writes).toHaveLength(2) - expect(writes[1]).toBe(AGENT_PROMPT_BRACKETED_PASTE_END) + expect(writes).toHaveLength(1) + expect(writes[0]).toContain(AGENT_PROMPT_BRACKETED_PASTE_END) expect(writes).not.toContain('\r') }) @@ -736,7 +736,7 @@ describe('agent prompt submission runtime', () => { await expect(submission).rejects.toThrow('terminal_handle_stale') expect(writes).toHaveLength(1) - expect(writes[0]).not.toContain(AGENT_PROMPT_BRACKETED_PASTE_END) + expect(writes[0]).toContain(AGENT_PROMPT_BRACKETED_PASTE_END) }) it('does not send delayed Enter after cancellation during settlement', async () => { diff --git a/src/main/runtime/orca-runtime.test.ts b/src/main/runtime/orca-runtime.test.ts index 661706e9316..d3f130a231c 100644 --- a/src/main/runtime/orca-runtime.test.ts +++ b/src/main/runtime/orca-runtime.test.ts @@ -17988,7 +17988,7 @@ describe('OrcaRuntimeService', () => { } }) - it('chunks large agent prompt paste frames before delayed submit', async () => { + it('writes large agent prompt paste frames atomically before delayed submit', async () => { vi.useFakeTimers() try { const writes: string[] = [] @@ -18017,7 +18017,7 @@ describe('OrcaRuntimeService', () => { Buffer.byteLength(`${buildAgentPromptPasteBytes(prompt)}\r`, 'utf8') ) expect(writes.at(-1)).toBe('\r') - expect(pasteWrites.length).toBeGreaterThan(1) + expect(pasteWrites).toHaveLength(1) expect(pasteWrites.join('')).toBe(buildAgentPromptPasteBytes(prompt)) expect(pasteWrites[0]).toContain(AGENT_PROMPT_BRACKETED_PASTE_START) expect(pasteWrites.at(-1)).toContain(AGENT_PROMPT_BRACKETED_PASTE_END) @@ -18026,18 +18026,16 @@ describe('OrcaRuntimeService', () => { } }) - it('closes an incomplete agent prompt paste when a later chunk write fails', async () => { + it('rejects an agent prompt when the atomic paste write fails', async () => { vi.useFakeTimers() try { const writes: string[] = [] - let writeCount = 0 const runtime = new OrcaRuntimeService(store) runtime.setPtyController({ spawn: vi.fn().mockResolvedValue({ id: 'pty-bg' }), write: (_ptyId, data) => { - writeCount += 1 writes.push(data) - return writeCount !== 2 + return false }, kill: () => true, getForegroundProcess: async () => null @@ -18053,7 +18051,7 @@ describe('OrcaRuntimeService', () => { await sendRejection expect(writes[0]).toContain(AGENT_PROMPT_BRACKETED_PASTE_START) - expect(writes.at(-1)).toBe(AGENT_PROMPT_BRACKETED_PASTE_END) + expect(writes).toHaveLength(1) expect(writes).not.toContain('\r') } finally { vi.useRealTimers() @@ -18085,6 +18083,33 @@ describe('OrcaRuntimeService', () => { expect(writes).toEqual(['x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES), 'tail']) }) + it('yields chunked terminal input through immediates between writes', async () => { + const immediate = vi.spyOn(globalThis, 'setImmediate') + try { + const writes: string[] = [] + const runtime = new OrcaRuntimeService(store) + runtime.setPtyController({ + spawn: vi.fn().mockResolvedValue({ id: 'pty-bg' }), + write: (_ptyId, data) => { + writes.push(data) + return true + }, + kill: () => true, + getForegroundProcess: async () => null + }) + const { handle } = await runtime.createTerminal(`path:${TEST_WORKTREE_PATH}`) + + const text = `${'x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES)}\nline two\nline three` + await runtime.sendTerminal(handle, { text, enter: true }) + + expect(writes.at(-1)).toBe('\r') + expect(writes.slice(0, -1).join('')).toBe(text) + expect(immediate).toHaveBeenCalled() + } finally { + immediate.mockRestore() + } + }) + it('yields while validating accepted large terminal.send text before provider writes', async () => { const writes: string[] = [] const runtime = new OrcaRuntimeService(store) diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts index 46c3c60565d..20da83122ad 100644 --- a/src/main/runtime/orca-runtime.ts +++ b/src/main/runtime/orca-runtime.ts @@ -148,7 +148,6 @@ import { iterateTerminalInputChunks } from '../../shared/terminal-input' import { - AGENT_PROMPT_BRACKETED_PASTE_END, AGENT_PROMPT_SUBMIT, buildAgentPromptPasteBytes, getAgentPromptSubmitDelayMs, @@ -2297,11 +2296,9 @@ async function waitForAgentPromptPromise(promise: Promise, signal?: AbortS }) } -// Why not setTimeout(0): it costs a full ~15.19 ms Windows timer tick per chunk (~0.95 s/MB) -// and never bought backpressure -- 16 KiB per tick paces ~1.07 MB/s, 11x above ConPTY's -// ~96 KB/s drain, so the in-flight buffer grew regardless. setImmediate keeps the only thing -// the yield actually did (let abort/permission/data callbacks run between chunks) at ~0.01 ms, -// and TERMINAL_INPUT_MAX_BYTES still bounds what can be in flight either way. +// Generic terminal.send uses setImmediate to let abort/permission/data callbacks run between +// chunks without paying a full Windows timer tick for every 16 KiB write. Agent prompts use an +// atomic bracketed-paste write below, so they do not rely on this scheduler. // Why the global and not node:timers/promises: only the global is intercepted by fake timers, // so a chunked paste stays observable on the test clock. function yieldBetweenTerminalInputChunks(): Promise { @@ -21948,60 +21945,25 @@ export class OrcaRuntimeService { const pasteByteLength = Buffer.byteLength(pastePayload, 'utf8') const pasteIngestMs = getTerminalPasteIngestMs(writeHostPlatform, pasteByteLength) const renderGate = this.createAgentPromptRenderGate(ptyId, pasteIngestMs) - let wrotePasteBytes = false - let completedPaste = false try { - const chunks = iterateTerminalInputChunks(pastePayload) - let chunk = chunks.next() - let firstChunk = true - while (!chunk.done) { - const nextChunk = chunks.next() - assertAgentPromptRequestActive(options.signal) - this.assertAgentPromptGeneration(ptyId, generation) - // Why: the first chunk was just admitted above; re-checking the lease there would only - // re-read what `assertAdmitted` established. - if (!firstChunk) { - agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted) - } - firstChunk = false - await options.beforeWrite?.(ptyId) - assertAgentPromptRequestActive(options.signal) - this.assertAgentPromptGeneration(ptyId, generation) - this.assertAgentPromptPermissionSafe( - permissionBaseline, - this.getAgentPromptActivity(handle, ptyId) - ) - agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted) - if (nextChunk.done) { - renderGate?.arm() - } - const wrote = this.ptyController?.write(ptyId, chunk.value) ?? false - if (!wrote) { - throw new Error('terminal_not_writable') - } - wrotePasteBytes = true - chunk = nextChunk - if (!chunk.done) { - await yieldBetweenTerminalInputChunks() - } + assertAgentPromptRequestActive(options.signal) + this.assertAgentPromptGeneration(ptyId, generation) + await options.beforeWrite?.(ptyId) + assertAgentPromptRequestActive(options.signal) + this.assertAgentPromptGeneration(ptyId, generation) + this.assertAgentPromptPermissionSafe( + permissionBaseline, + this.getAgentPromptActivity(handle, ptyId) + ) + agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted) + // Keep the bracketed paste frame in one PTY write; Claude's composer can drop the + // beginning when a large frame is split into independently processed chunks. + renderGate?.arm() + const wrote = this.ptyController?.write(ptyId, pastePayload) ?? false + if (!wrote) { + throw new Error('terminal_not_writable') } - completedPaste = true } catch (error) { - if ( - wrotePasteBytes && - !completedPaste && - this.getPtyLifecycleGeneration(ptyId) === generation - ) { - // Why: a lease that moved mid-paste also refuses this terminator, leaving the TUI in paste - // mode — the incoming owner re-establishes the mode, and feeding a session we no longer own - // is the worse outcome. - try { - agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted) - this.ptyController?.write(ptyId, AGENT_PROMPT_BRACKETED_PASTE_END) - } catch { - // The original refusal is the actionable error. - } - } renderGate?.dispose() throw error }