mirror of
https://github.com/stablyai/orca.git
synced 2026-10-30 16:47:34 +00:00
fix: finish admitted terminal prompts after sender disconnects (#34373)
* fix: finish admitted terminal prompts after sender disconnects * test: consolidate prompt input contract assertions * Keep first-submit ownership independent of sender observation * Recheck prompt ownership after awaited guards * Cover abandoned prompt admission during restored pane startup * Use typed provider write spy in prompt socket regression
This commit is contained in:
@@ -150,7 +150,7 @@ describe('agent prompt writer: onPermission wait', () => {
|
||||
const sent = runtime.sendTerminalAgentPrompt(handle, 'held prompt', {
|
||||
inputKind: 'driving',
|
||||
onPermission: 'wait',
|
||||
signal: controller.signal
|
||||
observationSignal: controller.signal
|
||||
})
|
||||
const rejected = expect(sent).rejects.toThrow('request_aborted')
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
@@ -215,4 +215,40 @@ describe('agent prompt writer: expectedBinding', () => {
|
||||
).rejects.toThrow('terminal_handle_stale')
|
||||
expect(writes).toEqual([])
|
||||
})
|
||||
it.each(['enrich', 'replace', 'explicit'] as const)('suffix identity %s', async (change) => {
|
||||
vi.useFakeTimers()
|
||||
const { runtime, handle, writes } = await createRuntime()
|
||||
const [pane] = (await runtime.listTerminals()).terminals
|
||||
const binding = runtime.getTerminalPromptRequestBinding(handle)
|
||||
const register = (incarnationId: string) =>
|
||||
runtime.registerPty('pty-prompt', pane.worktreeId, null, { ...pane, incarnationId })
|
||||
let checks = 0
|
||||
const sent = runtime.sendTerminalAgentPrompt(handle, 'identity', {
|
||||
inputKind: 'driving',
|
||||
acceptQueued: true,
|
||||
requestId: 'identity',
|
||||
observationTimeoutMs: 0,
|
||||
...(change === 'explicit' ? { expectedBinding: binding } : {}),
|
||||
beforeWrite: async () => {
|
||||
if (++checks === 1 && change === 'replace') {
|
||||
register('old')
|
||||
}
|
||||
if (checks === 2) {
|
||||
register('learned')
|
||||
}
|
||||
}
|
||||
})
|
||||
void sent.catch(() => undefined)
|
||||
await vi.runAllTimersAsync()
|
||||
expect(writes).toHaveLength(change === 'enrich' ? 2 : 1)
|
||||
if (change !== 'enrich') {
|
||||
await expect(sent).rejects.toThrow()
|
||||
return
|
||||
}
|
||||
const receipt = await sent
|
||||
expect(receipt.prompt?.processIncarnation).toBe('learned')
|
||||
expect(await runtime.observeTerminalAgentPrompt(handle, receipt.prompt!, 0)).not.toMatchObject({
|
||||
observation: 'incarnation_replaced'
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import { deliverFollowUp } from '../external-conversations/external-conversation-send'
|
||||
import { ConversationChain } from '../external-conversations/external-conversation-chain'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
AGENT_PROMPT_BRACKETED_PASTE_END,
|
||||
@@ -418,7 +420,7 @@ describe('agent prompt submission runtime hook and generation cases', () => {
|
||||
})
|
||||
})
|
||||
|
||||
it('does not queue a replacement generation behind an obsolete submission', async () => {
|
||||
it.each(['reset', 'exit'] as const)('retires pending %s', async (retirement) => {
|
||||
vi.useFakeTimers()
|
||||
let releaseFirst!: () => void
|
||||
let firstWriteReached!: () => void
|
||||
@@ -441,13 +443,23 @@ describe('agent prompt submission runtime hook and generation cases', () => {
|
||||
await firstGate
|
||||
}
|
||||
})
|
||||
const obsolete = expect(first).rejects.toThrow('terminal_handle_stale')
|
||||
await firstWrite
|
||||
runtime.synchronizePtyOutputSequenceFromProvider(
|
||||
'pty-prompt',
|
||||
{ value: 0, generation: 'reset' },
|
||||
0
|
||||
)
|
||||
if (retirement === 'exit') {
|
||||
runtime.onPtyExit('pty-prompt', 0)
|
||||
} else {
|
||||
runtime.synchronizePtyOutputSequenceFromProvider(
|
||||
'pty-prompt',
|
||||
{ value: 0, generation: 'reset' },
|
||||
0
|
||||
)
|
||||
}
|
||||
|
||||
await obsolete
|
||||
if (retirement === 'exit') {
|
||||
releaseFirst()
|
||||
return
|
||||
}
|
||||
const replacement = runtime.sendTerminalAgentPrompt(handle, 'replacement prompt', {
|
||||
inputKind: 'driving'
|
||||
})
|
||||
@@ -456,17 +468,21 @@ describe('agent prompt submission runtime hook and generation cases', () => {
|
||||
expect(writes.some((data) => data.includes('replacement prompt'))).toBe(true)
|
||||
|
||||
releaseFirst()
|
||||
await expect(first).rejects.toThrow('terminal_handle_stale')
|
||||
})
|
||||
|
||||
it('does not close a partial paste after the PTY generation changes', async () => {
|
||||
it.each(['generation', 'signal'] as const)('suffix stop %s', async (stop) => {
|
||||
const { runtime, handle, writes } = await createPromptRuntime(() => undefined)
|
||||
let writeChecks = 0
|
||||
const controller = new AbortController()
|
||||
|
||||
const submission = runtime.sendTerminalAgentPrompt(handle, 'x'.repeat(20_000), {
|
||||
inputKind: 'driving',
|
||||
signal: controller.signal,
|
||||
beforeWrite: () => {
|
||||
writeChecks += 1
|
||||
if (writeChecks === 2 && stop === 'signal') {
|
||||
return controller.abort()
|
||||
}
|
||||
if (writeChecks === 2) {
|
||||
runtime.synchronizePtyOutputSequenceFromProvider(
|
||||
'pty-prompt',
|
||||
@@ -477,7 +493,9 @@ describe('agent prompt submission runtime hook and generation cases', () => {
|
||||
}
|
||||
})
|
||||
|
||||
await expect(submission).rejects.toThrow('terminal_handle_stale')
|
||||
await expect(submission).rejects.toThrow(
|
||||
stop === 'signal' ? 'request_aborted' : 'terminal_handle_stale'
|
||||
)
|
||||
expect(writes).toHaveLength(1)
|
||||
expect(writes[0]).toContain(AGENT_PROMPT_BRACKETED_PASTE_END)
|
||||
})
|
||||
@@ -500,6 +518,41 @@ describe('agent prompt submission runtime hook and generation cases', () => {
|
||||
expect(writes.filter((data) => data === '\r')).toHaveLength(0)
|
||||
})
|
||||
|
||||
it.each(['agent', 'reject'] as const)('follow-up cancel before late %s', async (late) => {
|
||||
const { runtime, handle, writes } = await createPromptRuntime(() => undefined)
|
||||
const probe = Promise.withResolvers<'agent'>(),
|
||||
reached = Promise.withResolvers<void>()
|
||||
const unsubscribe = vi.fn()
|
||||
vi.spyOn(runtime, 'subscribeToTerminalData').mockReturnValue(unsubscribe)
|
||||
vi.mocked(runtime.readLaunchedAgentForeground).mockImplementation(() => {
|
||||
reached.resolve()
|
||||
return probe.promise
|
||||
})
|
||||
const chain = new ConversationChain(new AbortController().signal)
|
||||
const delivered = chain.enqueue('prompt', (signal) =>
|
||||
deliverFollowUp(runtime, {
|
||||
handle,
|
||||
agent: 'claude',
|
||||
expectedBinding: runtime.getTerminalPromptRequestBinding(handle),
|
||||
text: 'follow-up',
|
||||
requestId: 'follow-up',
|
||||
signal
|
||||
})
|
||||
)
|
||||
await reached.promise
|
||||
void chain.cancel('prompt', 'conversation_cancelled')
|
||||
await expect(delivered).resolves.toMatchObject({ outcome: 'refused' })
|
||||
expect(unsubscribe).toHaveBeenCalledTimes(1)
|
||||
if (late === 'agent') {
|
||||
probe.resolve('agent')
|
||||
} else {
|
||||
probe.reject(new Error('late failure'))
|
||||
}
|
||||
await Promise.resolve()
|
||||
expect(unsubscribe).toHaveBeenCalledTimes(1)
|
||||
expect(writes).toEqual([])
|
||||
})
|
||||
|
||||
it('does not send another Enter after cancellation during verification', async () => {
|
||||
vi.useFakeTimers()
|
||||
const controller = new AbortController()
|
||||
|
||||
@@ -8,7 +8,8 @@ export const AGENT_PROMPT_TEST_WORKTREE_ID = 'repo-1::/tmp/worktree-a'
|
||||
|
||||
export async function createAgentPromptSubmissionRuntime(
|
||||
onWrite: (runtime: OrcaRuntimeService, data: string, writeIndex: number) => void,
|
||||
launchAgent: TuiAgent = 'aider'
|
||||
launchAgent: TuiAgent = 'aider',
|
||||
write?: (runtime: OrcaRuntimeService, ptyId: string, data: string) => boolean
|
||||
): Promise<{ runtime: OrcaRuntimeService; handle: string; writes: string[] }> {
|
||||
const runtime = new OrcaRuntimeService(makeStore() as never)
|
||||
const writes: string[] = []
|
||||
@@ -17,10 +18,10 @@ export async function createAgentPromptSubmissionRuntime(
|
||||
write: (_ptyId, data) => {
|
||||
writes.push(data)
|
||||
onWrite(runtime, data, writes.length)
|
||||
return true
|
||||
return write?.(runtime, _ptyId, data) ?? true
|
||||
},
|
||||
kill: () => true,
|
||||
getForegroundProcess: async () => null
|
||||
getForegroundProcess: async () => (write ? launchAgent : null)
|
||||
})
|
||||
const terminal = await runtime.createTerminal(`path:${AGENT_PROMPT_TEST_WORKTREE_PATH}`, {
|
||||
launchAgent
|
||||
|
||||
@@ -83,7 +83,10 @@ describe('the check before each write of a launch prompt', () => {
|
||||
|
||||
const cleared = guardHarness(['agent'])
|
||||
await cleared.guard.beforeWrite('pty-1')
|
||||
const scheduled = Promise.resolve().then(() => cleared.guard.beforeWrite('pty-1'))
|
||||
cleared.guard.dispose()
|
||||
await expect(scheduled).rejects.toThrow('request_aborted')
|
||||
expect(cleared.runtime.subscribeToTerminalData).toHaveBeenCalledTimes(1)
|
||||
expect(cleared.unsubscribe).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -29,19 +29,33 @@ export function createLaunchedAgentWriteGuard(
|
||||
agent: TuiAgent,
|
||||
{ unprovableHost = 'refuse' }: { unprovableHost?: UnprovableHostPolicy } = {}
|
||||
): LaunchedAgentWriteGuard {
|
||||
let cleared: { ptyId: string; shellMayHaveReturned: boolean; unsubscribe: () => void } | null =
|
||||
null
|
||||
let disposed = false
|
||||
let current: {
|
||||
ptyId: string
|
||||
proven: boolean
|
||||
shellMayHaveReturned: boolean
|
||||
unsubscribe: () => void
|
||||
} | null = null
|
||||
const clearWatch = (): void => {
|
||||
const previous = current
|
||||
current = null
|
||||
previous?.unsubscribe()
|
||||
}
|
||||
const dispose = (): void => {
|
||||
cleared?.unsubscribe()
|
||||
cleared = null
|
||||
disposed = true
|
||||
clearWatch()
|
||||
}
|
||||
const beforeWrite = async (ptyId: string): Promise<void> => {
|
||||
if (cleared?.ptyId === ptyId && !cleared.shellMayHaveReturned) {
|
||||
if (disposed) {
|
||||
throw new Error('request_aborted')
|
||||
}
|
||||
if (current?.ptyId === ptyId && current.proven && !current.shellMayHaveReturned) {
|
||||
return
|
||||
}
|
||||
dispose()
|
||||
clearWatch()
|
||||
let carry = ''
|
||||
const watch = { ptyId, shellMayHaveReturned: false, unsubscribe: (): void => {} }
|
||||
const watch = { ptyId, proven: false, shellMayHaveReturned: false, unsubscribe: (): void => {} }
|
||||
current = watch
|
||||
// Subscribe before probing so shell return during the read cannot be missed.
|
||||
watch.unsubscribe = runtime.subscribeToTerminalData(ptyId, (data) => {
|
||||
const window = carry + data
|
||||
@@ -50,27 +64,34 @@ export function createLaunchedAgentWriteGuard(
|
||||
watch.shellMayHaveReturned = true
|
||||
}
|
||||
})
|
||||
let foreground: LaunchedAgentForeground
|
||||
try {
|
||||
foreground = await runtime.readLaunchedAgentForeground(ptyId, agent)
|
||||
const foreground: LaunchedAgentForeground = await runtime.readLaunchedAgentForeground(
|
||||
ptyId,
|
||||
agent
|
||||
)
|
||||
if (disposed || current !== watch) {
|
||||
throw new Error('request_aborted')
|
||||
}
|
||||
if (foreground === 'agent') {
|
||||
watch.proven = true
|
||||
return
|
||||
}
|
||||
clearWatch()
|
||||
if (
|
||||
foreground === 'shell' ||
|
||||
unprovableHost === 'refuse' ||
|
||||
runtime.launchedAgentHostProvesAgent?.(ptyId) !== false ||
|
||||
(unprovableHost === 'write-if-shell-provable' &&
|
||||
runtime.launchedAgentHostProvesShellAlone?.(ptyId) !== true)
|
||||
) {
|
||||
throw new Error('agent_not_in_foreground')
|
||||
}
|
||||
} catch (error) {
|
||||
watch.unsubscribe()
|
||||
if (current === watch) {
|
||||
clearWatch()
|
||||
}
|
||||
throw error
|
||||
}
|
||||
if (foreground === 'agent') {
|
||||
cleared = watch
|
||||
return
|
||||
}
|
||||
watch.unsubscribe()
|
||||
if (
|
||||
foreground === 'shell' ||
|
||||
unprovableHost === 'refuse' ||
|
||||
runtime.launchedAgentHostProvesAgent?.(ptyId) !== false ||
|
||||
(unprovableHost === 'write-if-shell-provable' &&
|
||||
runtime.launchedAgentHostProvesShellAlone?.(ptyId) !== true)
|
||||
) {
|
||||
throw new Error('agent_not_in_foreground')
|
||||
}
|
||||
}
|
||||
return { beforeWrite, dispose }
|
||||
}
|
||||
|
||||
@@ -85,6 +85,16 @@ const DRIVING = { inputKind: 'driving' } as const
|
||||
describe('a CLI trigger on a saved pane with no terminal (layout-owner build)', () => {
|
||||
it('a send starts the pane once and delivers to it, and the handle keeps naming it', async () => {
|
||||
const { runtime, handle, spawn, started, write } = await restoredPane()
|
||||
const observer = new AbortController()
|
||||
const abandoned = runtime.sendTerminalAgentPrompt(handle, 'abandoned', {
|
||||
...DRIVING,
|
||||
observationSignal: observer.signal
|
||||
})
|
||||
const refused = expect(abandoned).rejects.toThrow('request_aborted')
|
||||
await vi.waitFor(() => expect(spawn).toHaveBeenCalled())
|
||||
observer.abort()
|
||||
await refused
|
||||
expect(write).not.toHaveBeenCalled()
|
||||
|
||||
const send = runtime.sendTerminal(handle, { text: 'echo hi', enter: true }, DRIVING)
|
||||
await vi.waitFor(() => expect(spawn).toHaveBeenCalled())
|
||||
@@ -96,6 +106,7 @@ describe('a CLI trigger on a saved pane with no terminal (layout-owner build)',
|
||||
runtime.sendTerminal(handle, { text: 'echo again', enter: true }, DRIVING)
|
||||
).resolves.toMatchObject({ handle, accepted: true })
|
||||
expect(write).toHaveBeenCalledWith('pty-c', 'echo again', 'driving')
|
||||
expect(write).toHaveBeenCalledTimes(4)
|
||||
expect(spawn).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
|
||||
@@ -197,7 +197,7 @@ export const TERMINAL_SEND_METHODS = [
|
||||
? await runtime.sendTerminalAgentPrompt(params.terminal, params.text!, {
|
||||
inputKind: 'driving',
|
||||
beforeWrite,
|
||||
signal,
|
||||
observationSignal: signal,
|
||||
...(orchestrationMutation
|
||||
? {
|
||||
acceptQueued: true,
|
||||
|
||||
@@ -48,7 +48,7 @@ describe('terminal agent prompt send RPC', () => {
|
||||
expect(sendTerminalAgentPrompt).toHaveBeenCalledWith('terminal-1', 'review this change', {
|
||||
inputKind: 'driving',
|
||||
beforeWrite: undefined,
|
||||
signal: undefined
|
||||
observationSignal: undefined
|
||||
})
|
||||
expect(sendTerminal).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
@@ -0,0 +1,126 @@
|
||||
import './unused-default-rpc-methods.test-fixture'
|
||||
import { spawn } from 'node:child_process'
|
||||
import { once } from 'node:events'
|
||||
import { mkdtempSync, writeFileSync, rmSync } from 'node:fs'
|
||||
import { createRequire } from 'node:module'
|
||||
import { tmpdir } from 'node:os'
|
||||
import { join, resolve } from 'node:path'
|
||||
import { expect, it, vi } from 'vitest'
|
||||
import { createAgentPromptSubmissionRuntime } from '../agent-prompt-submission-runtime-test-fixture'
|
||||
import { UnixSocketTransport } from './unix-socket-transport'
|
||||
import { RpcDispatcher } from './dispatcher'
|
||||
import type { RpcRequest } from './core'
|
||||
import { TERMINAL_METHODS } from './methods/terminal'
|
||||
import { STATUS_METHODS } from './methods/status'
|
||||
import { OrchestrationDb } from '../orchestration/db'
|
||||
import { writePtyFromRuntimeController } from '../../ipc/pty/runtime/operations'
|
||||
import { getLocalPtyProvider } from '../../ipc/pty/provider/registry'
|
||||
|
||||
vi.mock('../../git/worktree', () => {
|
||||
const read = async () => [
|
||||
{ path: '/tmp/worktree-a', head: 'abc', branch: 'prompt', isBare: false, isMainWorktree: false }
|
||||
]
|
||||
return { listWorktrees: read, listWorktreesStrict: read }
|
||||
})
|
||||
|
||||
it.each(['before', 'after'] as const)(
|
||||
'real CLI close %s paste finishes admitted input and drops a cancelled follower',
|
||||
async (boundary) => {
|
||||
const latch = () => Promise.withResolvers<void>()
|
||||
const [pasted, release, closed, received] = [latch(), latch(), latch(), latch()]
|
||||
const root = mkdtempSync(join(tmpdir(), 'orca-prompt-socket-'))
|
||||
const kind = process.platform === 'win32' ? 'named-pipe' : 'unix'
|
||||
const endpoint =
|
||||
kind === 'named-pipe' ? `\\\\.\\pipe\\${root.split(/[/\\]/).at(-1)}` : join(root, 'rpc')
|
||||
const transport = new UnixSocketTransport({ endpoint, kind })
|
||||
const providerWrite = vi.spyOn(getLocalPtyProvider(), 'write').mockReturnValue(true)
|
||||
const { runtime, handle, writes } = await createAgentPromptSubmissionRuntime(
|
||||
() => pasted.resolve(),
|
||||
'claude',
|
||||
(host, id, data) => writePtyFromRuntimeController({ runtime: host }, id, data, 'driving')
|
||||
)
|
||||
const db = new OrchestrationDb(':memory:')
|
||||
runtime.setOrchestrationDb(db)
|
||||
const methods = [...STATUS_METHODS, ...TERMINAL_METHODS]
|
||||
const dispatcher = new RpcDispatcher({ runtime, methods })
|
||||
Object.defineProperty(runtime, 'createAgentPromptRenderGate', {
|
||||
value: () => ({ arm: () => {}, wait: () => release.promise, dispose: () => {} })
|
||||
})
|
||||
const done: Promise<unknown>[] = []
|
||||
transport.onMessage((raw, reply, context) => {
|
||||
const request: RpcRequest = JSON.parse(raw)
|
||||
const prompt = request.method === 'terminal.send'
|
||||
if (prompt) {
|
||||
received.resolve()
|
||||
context!.signal.addEventListener('abort', () => closed.resolve(), { once: true })
|
||||
}
|
||||
const admission = prompt && boundary === 'before' ? release.promise : Promise.resolve()
|
||||
done.push(
|
||||
admission.then(async () =>
|
||||
reply(JSON.stringify(await dispatcher.dispatch(request, context)))
|
||||
)
|
||||
)
|
||||
})
|
||||
await transport.start()
|
||||
const discovery = {
|
||||
runtimeId: runtime.getRuntimeId(),
|
||||
pid: process.pid,
|
||||
startedAt: 1,
|
||||
authToken: 'fixture-token',
|
||||
transports: [{ kind, endpoint }]
|
||||
}
|
||||
writeFileSync(join(root, 'orca-runtime.json'), JSON.stringify(discovery))
|
||||
const preload = join(root, 'exit-on-input.ts')
|
||||
writeFileSync(preload, "process.stdin.once('data', () => process.exit(0))")
|
||||
const program = createRequire(resolve('package.json')).resolve('bun/bin/bun.exe')
|
||||
const env = {
|
||||
...Object.fromEntries(
|
||||
'HOME USERPROFILE APPDATA LOCALAPPDATA XDG_CONFIG_HOME XDG_CACHE_HOME XDG_DATA_HOME XDG_STATE_HOME ORCA_E2E_HOME_DIR ORCA_E2E_USER_DATA_DIR ORCA_USER_DATA_PATH TEMP TMP TMPDIR CODEX_HOME CLAUDE_CONFIG_DIR'
|
||||
.split(' ')
|
||||
.map((key) => [key, root])
|
||||
),
|
||||
ORCA_BACKGROUND_LAUNCH: '1',
|
||||
ORCA_DISABLE_MACOS_LOGIN_SHELL: '1',
|
||||
PATH: process.env.PATH,
|
||||
SystemRoot: process.env.SystemRoot
|
||||
}
|
||||
const args = ['terminal', 'send', '--terminal', handle, '--text', 'first', '--enter', '--json']
|
||||
const command = ['--preload', preload, resolve('src/cli/index.ts'), ...args]
|
||||
const child = spawn(program, command, { cwd: root, env, stdio: 'pipe' })
|
||||
const exited = once(child, 'close')
|
||||
child.stderr.resume()
|
||||
try {
|
||||
await Promise.race([
|
||||
received.promise,
|
||||
exited.then(() => expect.fail('CLI exited before sending'))
|
||||
])
|
||||
await (boundary === 'after' ? pasted.promise : received.promise)
|
||||
child.stdin.end('exit') // The test-owned child exits itself; no process-tree termination.
|
||||
await exited
|
||||
await closed.promise
|
||||
const secondObserver = new AbortController()
|
||||
const second = runtime.sendTerminalAgentPrompt(handle, 'second', {
|
||||
inputKind: 'driving',
|
||||
observationSignal: secondObserver.signal
|
||||
})
|
||||
void second.catch(() => undefined)
|
||||
secondObserver.abort()
|
||||
release.resolve()
|
||||
await Promise.all(done)
|
||||
await expect(second).rejects.toThrow('request_aborted')
|
||||
expect(writes).toEqual(boundary === 'after' ? [expect.stringContaining('first'), '\r'] : [])
|
||||
} finally {
|
||||
release.resolve()
|
||||
if (child.exitCode === null) {
|
||||
child.stdin.end('exit')
|
||||
}
|
||||
await exited
|
||||
await transport.stop()
|
||||
await Promise.allSettled(done)
|
||||
db.close()
|
||||
providerWrite.mockRestore()
|
||||
rmSync(root, { recursive: true, force: true })
|
||||
}
|
||||
},
|
||||
20_000
|
||||
)
|
||||
@@ -262,6 +262,8 @@ export type RuntimeProviderSnapshotReadOptions = {
|
||||
export type RuntimeAgentPromptWriteOptions = Omit<RuntimeTerminalWriteOptions, 'inputKind'> & {
|
||||
/** `launch` for the prompt an agent starts with; `driving` for any prompt sent to a running one. */
|
||||
inputKind: Exclude<TerminalInputKind, 'query-reply'>
|
||||
/** Caller observation may end without cancelling a paste already accepted by the host. */
|
||||
observationSignal?: AbortSignal
|
||||
/** Raw prompt text for submit scheduling; not written, only used for line-aware delays. */
|
||||
promptForSchedule?: string
|
||||
/** Caller observed composer input acceptance; schedule Enter using desktop draft-paste timing instead of render settlement. */
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
import type { RuntimeAgentPromptWriteOptions } from '../../runtime-terminal-contracts'
|
||||
|
||||
export function agentPromptObservationSignal(
|
||||
options: RuntimeAgentPromptWriteOptions
|
||||
): AbortSignal | undefined {
|
||||
return options.signal && options.observationSignal
|
||||
? AbortSignal.any([options.signal, options.observationSignal])
|
||||
: (options.signal ?? options.observationSignal)
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { agentPromptObservationSignal } from './agent-prompt-signals'
|
||||
import type { RuntimeAgentPromptWriteOptions } from '../../runtime-terminal-contracts'
|
||||
import type { RuntimeTerminalSend } from '../../../../shared/runtime-types'
|
||||
import type { AgentTransitionDetector } from '../../../agent-hooks/agent-transition-detector'
|
||||
@@ -74,7 +75,7 @@ export abstract class OrcaRuntimeWithSubmitTerminalAgentPrompt extends OrcaRunti
|
||||
}
|
||||
: {}),
|
||||
subscribeExit: (exited) => this.subscribeToPtyExit(ptyId, exited),
|
||||
signal: options.signal
|
||||
signal: agentPromptObservationSignal(options)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { agentPromptObservationSignal } from './agent-prompt-signals'
|
||||
import type {
|
||||
AgentPromptExpectedBinding,
|
||||
RuntimeAgentPromptWriteOptions
|
||||
@@ -71,7 +72,8 @@ export abstract class OrcaRuntimeWithWriteTerminalAgentPrompt extends OrcaRuntim
|
||||
pastePayload: string,
|
||||
options: RuntimeAgentPromptWriteOptions
|
||||
): Promise<{ submits: number; prompt?: RuntimeTerminalPromptDelivery }> {
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
const observationSignal = agentPromptObservationSignal(options)
|
||||
assertAgentPromptRequestActive(observationSignal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
const permissionBaseline = this.getAgentPromptActivity(handle, ptyId)
|
||||
if (options.onPermission === 'wait') {
|
||||
@@ -94,13 +96,41 @@ export abstract class OrcaRuntimeWithWriteTerminalAgentPrompt extends OrcaRuntim
|
||||
const preSubmitBaseline = submitWithPaste
|
||||
? this.getAgentPromptActivity(handle, ptyId, waitTextCache)
|
||||
: undefined
|
||||
try {
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
await options.beforeWrite?.(ptyId)
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
const incarnationId = pty?.incarnationId
|
||||
const lifecycleSignal = this.getPtyLifecycleSignal(ptyId)
|
||||
const deliverySignal = options.signal
|
||||
? AbortSignal.any([options.signal, lifecycleSignal])
|
||||
: lifecycleSignal
|
||||
const admissionSignal = observationSignal
|
||||
? AbortSignal.any([observationSignal, deliverySignal])
|
||||
: deliverySignal
|
||||
const assertWritable = (signal: AbortSignal): void => {
|
||||
assertAgentPromptRequestActive(signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
this.assertLiveTerminalHandleTargetsPty(handle, ptyId)
|
||||
if (incarnationId != null && this.ptysById.get(ptyId)?.incarnationId !== incarnationId) {
|
||||
throw new Error('terminal_handle_stale')
|
||||
}
|
||||
this.assertAgentPromptExpectedBinding(handle, options.expectedBinding)
|
||||
if (this.isPtyStopRequested(ptyId)) {
|
||||
throw new Error('terminal_not_writable')
|
||||
}
|
||||
}
|
||||
const beforeWrite = async (signal: AbortSignal): Promise<void> => {
|
||||
assertWritable(signal)
|
||||
await waitForAgentPromptPromise(
|
||||
Promise.resolve().then(() => {
|
||||
assertWritable(signal)
|
||||
return options.beforeWrite?.(ptyId)
|
||||
}),
|
||||
signal
|
||||
)
|
||||
assertWritable(signal)
|
||||
}
|
||||
let baseline: AgentPromptActivity
|
||||
try {
|
||||
await beforeWrite(admissionSignal)
|
||||
assertWritable(admissionSignal)
|
||||
if (options.onPermission === 'wait') {
|
||||
this.throwIfAgentPromptPermissionReturned(
|
||||
permissionBaseline,
|
||||
@@ -120,74 +150,82 @@ export abstract class OrcaRuntimeWithWriteTerminalAgentPrompt extends OrcaRuntim
|
||||
if (submitWithPaste) {
|
||||
options.onSubmitWritten?.()
|
||||
}
|
||||
} catch (error) {
|
||||
renderGate?.dispose()
|
||||
throw error
|
||||
}
|
||||
|
||||
if (submitWithPaste) {
|
||||
// The Enter was part of the paste frame; waiting here would only delay receipt settlement.
|
||||
renderGate?.dispose()
|
||||
} else if (renderGate) {
|
||||
try {
|
||||
await waitForAgentPromptPromise(renderGate.wait(), options.signal)
|
||||
} finally {
|
||||
renderGate.dispose()
|
||||
}
|
||||
} else if (options.composerReady) {
|
||||
await waitForAgentPromptDelay(
|
||||
AGENT_PROMPT_POST_PASTE_SUBMIT_DELAY_MS + pasteIngestMs,
|
||||
options.signal
|
||||
)
|
||||
} else {
|
||||
const agent = this.getPtyAgent(ptyId)
|
||||
const submitDelayMs = options.promptForSchedule
|
||||
? resolveAgentPromptSubmitDelayForAgent(writeHostPlatform, options.promptForSchedule, agent)
|
||||
: getAgentPromptSubmitDelayMs(writeHostPlatform, pasteByteLength)
|
||||
await waitForAgentPromptDelay(submitDelayMs, options.signal)
|
||||
}
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
if (!submitWithPaste) {
|
||||
try {
|
||||
await options.beforeWrite?.(ptyId)
|
||||
} catch (error) {
|
||||
if (options.suffixFailureError) {
|
||||
throw new Error(options.suffixFailureError)
|
||||
if (!submitWithPaste) {
|
||||
// Once paste is accepted, only host cancellation can suppress its first Enter.
|
||||
if (renderGate) {
|
||||
await waitForAgentPromptPromise(renderGate.wait(), deliverySignal)
|
||||
} else if (options.composerReady) {
|
||||
await waitForAgentPromptDelay(
|
||||
AGENT_PROMPT_POST_PASTE_SUBMIT_DELAY_MS + pasteIngestMs,
|
||||
deliverySignal
|
||||
)
|
||||
} else {
|
||||
const agent = this.getPtyAgent(ptyId)
|
||||
const submitDelayMs = options.promptForSchedule
|
||||
? resolveAgentPromptSubmitDelayForAgent(
|
||||
writeHostPlatform,
|
||||
options.promptForSchedule,
|
||||
agent
|
||||
)
|
||||
: getAgentPromptSubmitDelayMs(writeHostPlatform, pasteByteLength)
|
||||
await waitForAgentPromptDelay(submitDelayMs, deliverySignal)
|
||||
}
|
||||
try {
|
||||
await beforeWrite(deliverySignal)
|
||||
} catch (error) {
|
||||
if (options.suffixFailureError) {
|
||||
throw new Error(options.suffixFailureError)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
throw error
|
||||
}
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
}
|
||||
const baseline = preSubmitBaseline ?? this.getAgentPromptActivity(handle, ptyId, waitTextCache)
|
||||
this.assertAgentPromptPermissionSafe(permissionBaseline, baseline)
|
||||
if (!submitWithPaste) {
|
||||
if (!this.ptyController?.write(ptyId, AGENT_PROMPT_SUBMIT, options.inputKind)) {
|
||||
assertWritable(deliverySignal)
|
||||
baseline = preSubmitBaseline ?? this.getAgentPromptActivity(handle, ptyId, waitTextCache)
|
||||
this.assertAgentPromptPermissionSafe(permissionBaseline, baseline)
|
||||
if (
|
||||
!submitWithPaste &&
|
||||
!this.ptyController?.write(ptyId, AGENT_PROMPT_SUBMIT, options.inputKind)
|
||||
) {
|
||||
throw new Error(options.suffixFailureError ?? 'terminal_not_writable')
|
||||
}
|
||||
options.onSubmitWritten?.()
|
||||
if (!submitWithPaste) {
|
||||
options.onSubmitWritten?.()
|
||||
}
|
||||
} catch (error) {
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
throw error
|
||||
} finally {
|
||||
renderGate?.dispose()
|
||||
}
|
||||
const submits = submitWithPaste
|
||||
? 1
|
||||
: 1 +
|
||||
(await this.resubmitHeldAgentPrompt(ptyId, generation, options, {
|
||||
baseline,
|
||||
readActivity: () => this.getAgentPromptActivity(handle, ptyId, waitTextCache),
|
||||
prompt: options.promptForSchedule,
|
||||
heldBudgetMs: getAgentPromptHeldSubmitBudgetMs(
|
||||
writeHostPlatform,
|
||||
this.getPtyAgent(ptyId),
|
||||
pastePayload
|
||||
)
|
||||
}))
|
||||
(await this.resubmitHeldAgentPrompt(
|
||||
ptyId,
|
||||
generation,
|
||||
{
|
||||
...options,
|
||||
signal: admissionSignal,
|
||||
beforeWrite: () => beforeWrite(admissionSignal)
|
||||
},
|
||||
{
|
||||
baseline,
|
||||
readActivity: () => this.getAgentPromptActivity(handle, ptyId, waitTextCache),
|
||||
prompt: options.promptForSchedule,
|
||||
heldBudgetMs: getAgentPromptHeldSubmitBudgetMs(
|
||||
writeHostPlatform,
|
||||
this.getPtyAgent(ptyId),
|
||||
pastePayload
|
||||
)
|
||||
}
|
||||
))
|
||||
const effectTimeoutMs = resolveAgentPromptEffectTimeoutMs(this.getPtyAgent(ptyId))
|
||||
if (!options.acceptQueued || !options.requestId) {
|
||||
await verifyAgentPromptSubmission({
|
||||
baseline,
|
||||
readActivity: () => this.getAgentPromptActivity(handle, ptyId, waitTextCache),
|
||||
timeoutMs: effectTimeoutMs,
|
||||
signal: options.signal
|
||||
signal: observationSignal
|
||||
})
|
||||
return { submits }
|
||||
}
|
||||
@@ -242,7 +280,7 @@ export abstract class OrcaRuntimeWithWriteTerminalAgentPrompt extends OrcaRuntim
|
||||
evidence
|
||||
),
|
||||
allowOutputEvidence: false,
|
||||
signal: options.signal,
|
||||
signal: observationSignal,
|
||||
timeoutMs: options.observationTimeoutMs ?? effectTimeoutMs
|
||||
})
|
||||
this.forgetAgentPromptRequest(ptyId, generation, options.requestId)
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { agentPromptObservationSignal } from './agent-prompt-signals'
|
||||
import type { RuntimeTerminalWriteOptions } from '../../runtime-terminal-writer'
|
||||
import type { WriteSettlement } from '../../../../shared/pty-write-settlement'
|
||||
import type { RuntimeLeafRecord } from '../../runtime-terminal-state-records'
|
||||
@@ -226,7 +227,7 @@ export abstract class OrcaRuntimeWithControllerKnowsPtyIsLive extends OrcaRuntim
|
||||
}
|
||||
// Switch point: a saved pane main holds with no terminal yet starts first (once, shared).
|
||||
const starting = ORCA_LAYOUT_OWNER_IS_MAIN
|
||||
? this.startRestoredHandlePane(handle, { signal: options.signal })
|
||||
? this.startRestoredHandlePane(handle, { signal: agentPromptObservationSignal(options) })
|
||||
: null
|
||||
if (starting) {
|
||||
const target = await starting
|
||||
|
||||
@@ -6,6 +6,17 @@ import { extractLastOscTitle } from '../../../../shared/osc-title-extraction'
|
||||
import type { RuntimeTerminalDataMeta } from '../../runtime-terminal-stream-consumers'
|
||||
|
||||
export abstract class OrcaRuntimeWithRecordAgentPromptLifecycleState extends OrcaRuntimeWithCreateTerminalSideEffectCommandCodeDetector {
|
||||
private readonly ptyLifecycleCancellation = new Map<string, AbortController>()
|
||||
|
||||
protected getPtyLifecycleSignal(ptyId: string): AbortSignal {
|
||||
let controller = this.ptyLifecycleCancellation.get(ptyId)
|
||||
if (!controller) {
|
||||
controller = new AbortController()
|
||||
this.ptyLifecycleCancellation.set(ptyId, controller)
|
||||
}
|
||||
return controller.signal
|
||||
}
|
||||
|
||||
protected recordAgentPromptLifecycleState(ptyId: string, status: AgentStatus | null): void {
|
||||
if (status === 'permission') {
|
||||
this.recordAgentPromptPermissionObservation(ptyId)
|
||||
@@ -87,6 +98,9 @@ export abstract class OrcaRuntimeWithRecordAgentPromptLifecycleState extends Orc
|
||||
|
||||
protected advancePtyLifecycleGeneration(ptyId: string): void {
|
||||
this.ptyLifecycleGenerationById.set(ptyId, this.nextPtyLifecycleGeneration++)
|
||||
const cancellation = this.ptyLifecycleCancellation.get(ptyId)
|
||||
this.ptyLifecycleCancellation.delete(ptyId)
|
||||
cancellation?.abort()
|
||||
this.clearAgentPromptCorrelationForPty(ptyId)
|
||||
// A stop intent belongs to one process incarnation; never let it label a
|
||||
// replacement process when the provider reports a generation reset.
|
||||
|
||||
Reference in New Issue
Block a user