mirror of
https://github.com/stablyai/orca.git
synced 2026-10-01 08:01:56 +00:00
fix(terminal): preserve large agent prompt pastes (#17718)
* fix(terminal): preserve large agent prompt pastes * fix(terminal): guard oversized SSH PTY writes * fix(terminal): avoid timer delay for generic sends * fix(pty): propagate provider write refusals
This commit is contained in:
@@ -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())
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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 => {
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<T>(promise: Promise<T>, 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<void> {
|
||||
@@ -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
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user