mirror of
https://github.com/stablyai/orca.git
synced 2026-10-26 16:02:42 +00:00
fix(orchestration): verify large Codex prompt submission
This commit is contained in:
Brennan Benson
@@ -5912,6 +5912,17 @@ export function registerPtyHandlers(
|
||||
return null
|
||||
}
|
||||
},
|
||||
supportsProviderBufferSnapshot: async (ptyId, signal) => {
|
||||
try {
|
||||
const provider = getProviderForPty(ptyId)
|
||||
if (provider.probeAuthoritativeBufferSnapshotSupport) {
|
||||
return await provider.probeAuthoritativeBufferSnapshotSupport({ signal })
|
||||
}
|
||||
return provider.canProvideAuthoritativeBufferSnapshot?.(ptyId) === true
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
},
|
||||
hasRendererSerializer: (ptyId) => {
|
||||
// Why: a synchronous probe lets the runtime decide whether to skip the daemon-snapshot seed (renderer will hydrate) or run it (no renderer authoritative).
|
||||
return rendererSerializerReadiness.has(ptyId)
|
||||
|
||||
@@ -29,6 +29,8 @@ export type PtyProviderBufferSnapshot = {
|
||||
cwd?: string | null
|
||||
lastTitle?: string
|
||||
seq: number
|
||||
/** Process identity that owned the provider screen at this sequence. */
|
||||
incarnationId?: PtyIncarnationId
|
||||
source: 'headless'
|
||||
oscLinks?: TerminalOscLinkRange[]
|
||||
alternateScreen?: boolean
|
||||
@@ -178,6 +180,10 @@ export type IPtyProvider = {
|
||||
) => Promise<PtyProviderBufferSnapshot | null>
|
||||
/** Whether this exact PTY can return a sequence-safe provider snapshot. */
|
||||
canProvideAuthoritativeBufferSnapshot?: (id: string) => boolean
|
||||
/** Negotiated proof that provider snapshots are available on this execution host. */
|
||||
probeAuthoritativeBufferSnapshotSupport?: (
|
||||
options?: PtyProbeOptions
|
||||
) => boolean | Promise<boolean>
|
||||
/**
|
||||
* The size the PTY has ACTUALLY applied, not the last size requested.
|
||||
* resize() is fire-and-forget for remote providers (daemon/SSH `notify`),
|
||||
|
||||
@@ -128,6 +128,10 @@ export class SshPtyProviderOutputState {
|
||||
}
|
||||
}
|
||||
|
||||
ptyIncarnation(relayPtyId: string): string | undefined {
|
||||
return this.incarnationByRelayPtyId.get(relayPtyId)
|
||||
}
|
||||
|
||||
private resolvePtyIncarnation(relayPtyId: string, incarnationId: unknown): string {
|
||||
this.rememberPtyIncarnation(relayPtyId, incarnationId)
|
||||
let resolved = this.incarnationByRelayPtyId.get(relayPtyId)
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, expect, it, vi, beforeEach } from 'vitest'
|
||||
import { SshPtyProvider } from './ssh-pty-provider'
|
||||
import { AGENT_SESSION_EXECUTION_OWNER_PROTOCOL_VERSION } from '../../shared/agent-session-host-authority'
|
||||
import { PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION } from '../../shared/pty-visible-screen-snapshot'
|
||||
import {
|
||||
createMockMux,
|
||||
expectRequest,
|
||||
@@ -25,6 +26,72 @@ describe('SshPtyProvider', () => {
|
||||
expect(provider.canProvideAuthoritativeBufferSnapshot(scopedPty1)).toBe(false)
|
||||
})
|
||||
|
||||
it('negotiates authoritative visible-screen snapshots', async () => {
|
||||
mux.request.mockResolvedValue({
|
||||
visibleScreenSnapshotVersion: PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION
|
||||
})
|
||||
|
||||
await expect(provider.probeAuthoritativeBufferSnapshotSupport()).resolves.toBe(true)
|
||||
expect(provider.canProvideAuthoritativeBufferSnapshot(scopedPty1)).toBe(true)
|
||||
expectRequest(mux.request, 'pty.getCapabilities', undefined, { timeoutMs: 5_000 })
|
||||
})
|
||||
|
||||
it('keeps old SSH relays snapshot-ineligible', async () => {
|
||||
mux.request.mockResolvedValue({})
|
||||
|
||||
await expect(provider.probeAuthoritativeBufferSnapshotSupport()).resolves.toBe(false)
|
||||
expect(provider.canProvideAuthoritativeBufferSnapshot(scopedPty1)).toBe(false)
|
||||
})
|
||||
|
||||
it('returns an authoritative snapshot for the current incarnation', async () => {
|
||||
mux.request
|
||||
.mockResolvedValueOnce({
|
||||
visibleScreenSnapshotVersion: PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION
|
||||
})
|
||||
.mockResolvedValueOnce({ replay: '', incarnationId: 'incarnation-1' })
|
||||
.mockResolvedValueOnce({
|
||||
data: '',
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
seq: 12,
|
||||
incarnationId: 'incarnation-1',
|
||||
source: 'headless'
|
||||
})
|
||||
|
||||
await provider.probeAuthoritativeBufferSnapshotSupport()
|
||||
await provider.attachForReconnect(scopedPty1)
|
||||
|
||||
await expect(provider.getBufferSnapshot(scopedPty1)).resolves.toEqual({
|
||||
data: '',
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
seq: 12,
|
||||
incarnationId: 'incarnation-1',
|
||||
source: 'headless'
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects a visible-screen snapshot from another incarnation', async () => {
|
||||
mux.request
|
||||
.mockResolvedValueOnce({
|
||||
visibleScreenSnapshotVersion: PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION
|
||||
})
|
||||
.mockResolvedValueOnce({ replay: '', incarnationId: 'incarnation-1' })
|
||||
.mockResolvedValueOnce({
|
||||
data: 'wrong owner',
|
||||
cols: 80,
|
||||
rows: 24,
|
||||
seq: 12,
|
||||
incarnationId: 'incarnation-2',
|
||||
source: 'headless'
|
||||
})
|
||||
|
||||
await provider.probeAuthoritativeBufferSnapshotSupport()
|
||||
await provider.attachForReconnect(scopedPty1)
|
||||
|
||||
await expect(provider.getBufferSnapshot(scopedPty1)).resolves.toBeNull()
|
||||
})
|
||||
|
||||
it('keeps a shared claim probe alive when one waiter disconnects', async () => {
|
||||
let finishProbe!: (result: { agentSessionClaimVersion: number }) => void
|
||||
mux.request.mockReturnValueOnce(
|
||||
|
||||
@@ -23,6 +23,7 @@ import { SshPtySpawnExitRaceTracker } from './ssh-pty-spawn-exit-race'
|
||||
import { SshAgentSessionCapabilities } from './ssh-agent-session-capabilities'
|
||||
import type { PtyProcessInspection } from './pty-process-inspection'
|
||||
import { writeToSshPty, writeToSshPtyWithSettlement } from './ssh-pty-write'
|
||||
import { SshPtyVisibleScreenSnapshots } from './ssh-pty-visible-screen-snapshot'
|
||||
|
||||
// Why: sequential relay teardown calls share one absolute budget; convert to the mux-relative timeout only at dispatch.
|
||||
function relayTimeoutOptions(deadlineMs: number | undefined): { timeoutMs: number } | undefined {
|
||||
@@ -38,6 +39,7 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
private readonly agentSessionCapabilities: SshAgentSessionCapabilities
|
||||
private spawnExitRaces = new SshPtySpawnExitRaceTracker()
|
||||
private readonly outputState: SshPtyProviderOutputState
|
||||
private readonly visibleScreenSnapshots: SshPtyVisibleScreenSnapshots
|
||||
|
||||
requestHostRpc: NonNullable<IPtyProvider['requestHostRpc']> = (method, params, options) =>
|
||||
this.mux.request(method, params as Record<string, unknown>, options)
|
||||
@@ -61,6 +63,11 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
this.spawnExitRaces.recordExit(relayPtyId, incarnationId)
|
||||
}
|
||||
})
|
||||
this.visibleScreenSnapshots = new SshPtyVisibleScreenSnapshots(
|
||||
mux,
|
||||
(id) => this.toRelayPtyId(id),
|
||||
(id) => this.outputState.ptyIncarnation(id)
|
||||
)
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
@@ -70,7 +77,18 @@ export class SshPtyProvider implements IPtyProvider {
|
||||
|
||||
getConnectionId = (): string => this.connectionId
|
||||
|
||||
canProvideAuthoritativeBufferSnapshot = (_id: string): boolean => false
|
||||
canProvideAuthoritativeBufferSnapshot = (_id: string): boolean =>
|
||||
this.visibleScreenSnapshots.canProvide()
|
||||
|
||||
async probeAuthoritativeBufferSnapshotSupport(
|
||||
options: { signal?: AbortSignal } = {}
|
||||
): Promise<boolean> {
|
||||
return await this.visibleScreenSnapshots.probe(options)
|
||||
}
|
||||
|
||||
async getBufferSnapshot(id: string, opts: { scrollbackRows?: number } = {}) {
|
||||
return await this.visibleScreenSnapshots.get(id, opts)
|
||||
}
|
||||
|
||||
private toRelayPtyId = (id: string): string => toRelaySshPtyId(this.connectionId, id)
|
||||
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
import { PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION } from '../../shared/pty-visible-screen-snapshot'
|
||||
import type { SshChannelMultiplexer } from '../ssh/ssh-channel-multiplexer'
|
||||
import { waitForSshCapabilityProbe } from './ssh-capability-probe-waiter'
|
||||
import type { PtyProviderBufferSnapshot } from './types'
|
||||
|
||||
export class SshPtyVisibleScreenSnapshots {
|
||||
private capabilityProbe: Promise<boolean> | null = null
|
||||
private supported = false
|
||||
|
||||
constructor(
|
||||
private readonly mux: SshChannelMultiplexer,
|
||||
private readonly toRelayPtyId: (id: string) => string,
|
||||
private readonly readIncarnation: (relayPtyId: string) => string | undefined
|
||||
) {}
|
||||
|
||||
canProvide(): boolean {
|
||||
return this.supported
|
||||
}
|
||||
|
||||
async probe(options: { signal?: AbortSignal } = {}): Promise<boolean> {
|
||||
const probe = this.capabilityProbe ?? this.probeCapability()
|
||||
this.capabilityProbe = probe
|
||||
try {
|
||||
const supported = await waitForSshCapabilityProbe(probe, options.signal)
|
||||
this.supported = supported
|
||||
if (!supported && this.capabilityProbe === probe) {
|
||||
this.capabilityProbe = null
|
||||
}
|
||||
return supported
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
async get(
|
||||
id: string,
|
||||
opts: { scrollbackRows?: number } = {}
|
||||
): Promise<PtyProviderBufferSnapshot | null> {
|
||||
if (!this.supported) {
|
||||
return null
|
||||
}
|
||||
const relayPtyId = this.toRelayPtyId(id)
|
||||
const expectedIncarnation = this.readIncarnation(relayPtyId)
|
||||
if (!expectedIncarnation) {
|
||||
return null
|
||||
}
|
||||
const result = await this.mux.request('pty.getBufferSnapshot', {
|
||||
id: relayPtyId,
|
||||
scrollbackRows: opts.scrollbackRows ?? 0
|
||||
})
|
||||
return isSshPtyBufferSnapshot(result, expectedIncarnation) ? result : null
|
||||
}
|
||||
|
||||
private async probeCapability(): Promise<boolean> {
|
||||
try {
|
||||
const result = (await this.mux.request('pty.getCapabilities', undefined, {
|
||||
timeoutMs: 5_000
|
||||
})) as { visibleScreenSnapshotVersion?: unknown }
|
||||
return result.visibleScreenSnapshotVersion === PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION
|
||||
} catch {
|
||||
return false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isSshPtyBufferSnapshot(
|
||||
value: unknown,
|
||||
expectedIncarnation: string
|
||||
): value is PtyProviderBufferSnapshot {
|
||||
if (!value || typeof value !== 'object') {
|
||||
return false
|
||||
}
|
||||
const snapshot = value as Partial<PtyProviderBufferSnapshot>
|
||||
return (
|
||||
snapshot.incarnationId === expectedIncarnation &&
|
||||
typeof snapshot.data === 'string' &&
|
||||
Number.isInteger(snapshot.cols) &&
|
||||
Number(snapshot.cols) > 0 &&
|
||||
Number.isInteger(snapshot.rows) &&
|
||||
Number(snapshot.rows) > 0 &&
|
||||
Number.isSafeInteger(snapshot.seq) &&
|
||||
Number(snapshot.seq) >= 0 &&
|
||||
snapshot.source === 'headless'
|
||||
)
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import {
|
||||
} from '../../shared/agent-prompt-injection'
|
||||
import { OrcaRuntimeService } from './orca-runtime'
|
||||
import { makeStore } from './runtime-rpc-worktree-store-fixtures'
|
||||
import type { TuiAgent } from '../../shared/tui-agent'
|
||||
|
||||
const WORKTREE_PATH = '/tmp/worktree-a'
|
||||
|
||||
@@ -30,7 +31,9 @@ vi.mock('../git/worktree', () => ({
|
||||
}))
|
||||
|
||||
async function createPromptRuntime(
|
||||
onWrite: (runtime: OrcaRuntimeService, data: string, writeIndex: number) => void
|
||||
onWrite: (runtime: OrcaRuntimeService, data: string, writeIndex: number) => void,
|
||||
agent: TuiAgent = 'aider',
|
||||
settleWrite?: (data: string) => Promise<boolean>
|
||||
): Promise<{ runtime: OrcaRuntimeService; handle: string; writes: string[] }> {
|
||||
const runtime = new OrcaRuntimeService(makeStore() as never)
|
||||
const writes: string[] = []
|
||||
@@ -41,11 +44,20 @@ async function createPromptRuntime(
|
||||
onWrite(runtime, data, writes.length)
|
||||
return true
|
||||
},
|
||||
...(settleWrite
|
||||
? {
|
||||
writeWithSettlement: async (_ptyId: string, data: string) => {
|
||||
writes.push(data)
|
||||
onWrite(runtime, data, writes.length)
|
||||
return await settleWrite(data)
|
||||
}
|
||||
}
|
||||
: {}),
|
||||
kill: () => true,
|
||||
getForegroundProcess: async () => null
|
||||
})
|
||||
const terminal = await runtime.createTerminal(`path:${WORKTREE_PATH}`, {
|
||||
launchAgent: 'aider'
|
||||
launchAgent: agent
|
||||
})
|
||||
return { runtime, handle: terminal.handle, writes }
|
||||
}
|
||||
@@ -68,6 +80,113 @@ describe('agent prompt submission runtime', () => {
|
||||
expect(writes.filter((data) => data === '\r')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('waits for a large Codex paste marker before Enter despite cursor output', async () => {
|
||||
vi.useFakeTimers()
|
||||
const prompt = 'x'.repeat(1_001)
|
||||
let markerVisible = false
|
||||
const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => {
|
||||
if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) {
|
||||
runtime.onPtyData('pty-prompt', '\x1b[?25h', Date.now())
|
||||
setTimeout(() => {
|
||||
markerVisible = true
|
||||
runtime.onPtyData('pty-prompt', '[Pasted Content 1001 chars]', Date.now())
|
||||
}, 500)
|
||||
} else if (data === '\r') {
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[Hworking', Date.now())
|
||||
}
|
||||
}, 'codex')
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[H› ', Date.now())
|
||||
|
||||
const submission = runtime.sendTerminalAgentPrompt(handle, prompt)
|
||||
await vi.advanceTimersByTimeAsync(499)
|
||||
expect(writes).not.toContain('\r')
|
||||
await vi.advanceTimersByTimeAsync(501)
|
||||
await submission
|
||||
|
||||
expect(markerVisible).toBe(true)
|
||||
expect(writes.filter((data) => data === '\r')).toHaveLength(1)
|
||||
})
|
||||
|
||||
it('fails a large Codex paste before Enter when no marker appears', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => {
|
||||
if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) {
|
||||
runtime.onPtyData('pty-prompt', '\x1b[?25h', Date.now())
|
||||
}
|
||||
}, 'codex')
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[H› ', Date.now())
|
||||
|
||||
const submission = runtime.sendTerminalAgentPrompt(handle, 'x'.repeat(1_001))
|
||||
const rejected = expect(submission).rejects.toThrow('agent_prompt_ingest_unverified')
|
||||
await vi.runAllTimersAsync()
|
||||
|
||||
await rejected
|
||||
expect(writes).not.toContain('\r')
|
||||
})
|
||||
|
||||
it('fails a large Codex paste closed when no visible screen is available', async () => {
|
||||
const { runtime, handle, writes } = await createPromptRuntime(() => undefined, 'codex')
|
||||
|
||||
await expect(runtime.sendTerminalAgentPrompt(handle, 'x'.repeat(1_001))).rejects.toThrow(
|
||||
'agent_prompt_unverifiable'
|
||||
)
|
||||
expect(writes).toEqual([])
|
||||
})
|
||||
|
||||
it('maps unsettled large Codex paste delivery to unverifiable', async () => {
|
||||
const { runtime, handle, writes } = await createPromptRuntime(
|
||||
() => undefined,
|
||||
'codex',
|
||||
async () => false
|
||||
)
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[H› ', Date.now())
|
||||
|
||||
await expect(runtime.sendTerminalAgentPrompt(handle, 'x'.repeat(1_001))).rejects.toThrow(
|
||||
'agent_prompt_unverifiable'
|
||||
)
|
||||
expect(writes).toHaveLength(1)
|
||||
expect(writes).not.toContain('\r')
|
||||
})
|
||||
|
||||
it('retries large Codex submit only while its marker remains', async () => {
|
||||
vi.useFakeTimers()
|
||||
let enters = 0
|
||||
const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => {
|
||||
if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) {
|
||||
runtime.onPtyData('pty-prompt', '[Pasted Content 1001 chars] #23', Date.now())
|
||||
} else if (data === '\r') {
|
||||
enters += 1
|
||||
if (enters === 2) {
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[Hworking', Date.now())
|
||||
}
|
||||
}
|
||||
}, 'codex')
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[H› ', Date.now())
|
||||
|
||||
const submission = runtime.sendTerminalAgentPrompt(handle, 'x'.repeat(1_001))
|
||||
await vi.runAllTimersAsync()
|
||||
await expect(submission).resolves.toMatchObject({ accepted: true })
|
||||
|
||||
expect(writes.filter((data) => data === '\r')).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('bounds large Codex retries when its marker never leaves', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => {
|
||||
if (data.includes(AGENT_PROMPT_BRACKETED_PASTE_END)) {
|
||||
runtime.onPtyData('pty-prompt', '[Pasted Content 1001 chars]', Date.now())
|
||||
}
|
||||
}, 'codex')
|
||||
runtime.onPtyData('pty-prompt', '\x1b[2J\x1b[H› ', Date.now())
|
||||
|
||||
const submission = runtime.sendTerminalAgentPrompt(handle, 'x'.repeat(1_001))
|
||||
const rejected = expect(submission).rejects.toThrow('agent_prompt_stalled')
|
||||
await vi.runAllTimersAsync()
|
||||
|
||||
await rejected
|
||||
expect(writes.filter((data) => data === '\r')).toHaveLength(3)
|
||||
})
|
||||
|
||||
it('accepts a working-to-idle cycle completed before the first poll', async () => {
|
||||
vi.useFakeTimers()
|
||||
const { runtime, handle, writes } = await createPromptRuntime((runtime, data) => {
|
||||
|
||||
@@ -101,6 +101,14 @@ import {
|
||||
AGENT_PROMPT_SUBMIT_DELAY_MS,
|
||||
buildAgentPromptPasteBytes
|
||||
} from '../../shared/agent-prompt-injection'
|
||||
import {
|
||||
CODEX_LARGE_PASTE_CHAR_THRESHOLD,
|
||||
codexAgentPromptCharCount,
|
||||
collectCodexPasteMarkers,
|
||||
countCodexPasteMarker,
|
||||
findCodexPasteMarkerDelta,
|
||||
type CodexPasteMarkerMultiset
|
||||
} from '../../shared/agent-prompt-codex-ingest'
|
||||
import {
|
||||
type AgentPromptActivity,
|
||||
verifyAgentPromptSubmission
|
||||
@@ -1794,6 +1802,18 @@ type RuntimeVisibleTerminalState = {
|
||||
generation: number
|
||||
}
|
||||
|
||||
type CodexPasteIngestBaseline = Readonly<{
|
||||
charCount: number
|
||||
markers: CodexPasteMarkerMultiset
|
||||
screen: RuntimeVisibleTerminalState
|
||||
}>
|
||||
|
||||
type CodexPasteIngestProof = Readonly<{
|
||||
marker: string
|
||||
visibleCount: number
|
||||
screen: RuntimeVisibleTerminalState
|
||||
}>
|
||||
|
||||
type ProviderBufferAcquisition = {
|
||||
generation: number
|
||||
scrollbackRows: number
|
||||
@@ -1955,6 +1975,7 @@ type RuntimePtyController = {
|
||||
ptyId: string,
|
||||
opts?: { scrollbackRows?: number }
|
||||
): Promise<PtyProviderBufferSnapshot | null>
|
||||
supportsProviderBufferSnapshot?(ptyId: string, signal?: AbortSignal): Promise<boolean>
|
||||
// Why: synchronous probe used by maybeHydrateHeadlessFromRenderer to skip
|
||||
// hydration when no renderer is authoritative for this PTY. See
|
||||
// docs/mobile-prefer-renderer-scrollback.md.
|
||||
@@ -2027,6 +2048,8 @@ const BRACKETED_PASTE_END = '\x1b[201~'
|
||||
const BRACKETED_PASTE_QUIET_MS = 1500
|
||||
const AGENT_PROMPT_RENDER_TIMEOUT_MS = 8000
|
||||
const AGENT_PROMPT_RENDER_QUIET_MS = 1500
|
||||
const AGENT_PROMPT_SCREEN_POLL_MS = 50
|
||||
const AGENT_PROMPT_MAX_SUBMITS = 3
|
||||
// Why: Claude and Codex emit show-cursor after accepting bracketed paste.
|
||||
const AGENT_PROMPT_RENDER_MARKER = '\x1b[?25h'
|
||||
|
||||
@@ -18681,6 +18704,7 @@ export class OrcaRuntimeService {
|
||||
} = {}
|
||||
): Promise<RuntimeTerminalSend> {
|
||||
const payload = buildAgentPromptPasteBytes(prompt)
|
||||
const codexCharCount = codexAgentPromptCharCount(prompt)
|
||||
const pty = this.getLivePtyForHandle(handle)
|
||||
if (pty) {
|
||||
if (!pty.pty.connected) {
|
||||
@@ -18699,6 +18723,7 @@ export class OrcaRuntimeService {
|
||||
pty.pty.ptyId,
|
||||
generation,
|
||||
payload,
|
||||
codexCharCount,
|
||||
options
|
||||
)
|
||||
}
|
||||
@@ -18721,7 +18746,14 @@ export class OrcaRuntimeService {
|
||||
const submits = await this.serializeAgentPromptSubmission(leaf.ptyId, generation, async () => {
|
||||
this.assertLiveTerminalHandleTargetsPty(handle, leaf.ptyId!)
|
||||
this.assertAgentPromptGeneration(leaf.ptyId!, generation)
|
||||
return await this.writeTerminalAgentPrompt(handle, leaf.ptyId!, generation, payload, options)
|
||||
return await this.writeTerminalAgentPrompt(
|
||||
handle,
|
||||
leaf.ptyId!,
|
||||
generation,
|
||||
payload,
|
||||
codexCharCount,
|
||||
options
|
||||
)
|
||||
})
|
||||
const bytesWritten = Buffer.byteLength(payload, 'utf8') + submits
|
||||
return { handle, accepted: true, bytesWritten }
|
||||
@@ -19395,6 +19427,7 @@ export class OrcaRuntimeService {
|
||||
ptyId: string,
|
||||
generation: number,
|
||||
pastePayload: string,
|
||||
codexCharCount: number,
|
||||
options: {
|
||||
beforeWrite?: (ptyId: string) => void | Promise<void>
|
||||
suffixFailureError?: string
|
||||
@@ -19405,7 +19438,18 @@ export class OrcaRuntimeService {
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
const permissionBaseline = this.getAgentPromptActivity(handle, ptyId)
|
||||
this.assertAgentPromptPermissionSafe(permissionBaseline, permissionBaseline)
|
||||
const renderGate = this.createAgentPromptRenderGate(ptyId)
|
||||
const isLargeCodexPaste =
|
||||
this.getAgentPromptTargetAgent(ptyId) === 'codex' &&
|
||||
codexCharCount > CODEX_LARGE_PASTE_CHAR_THRESHOLD
|
||||
const codexIngestBaseline = isLargeCodexPaste
|
||||
? await this.captureCodexPasteIngestBaseline(
|
||||
ptyId,
|
||||
generation,
|
||||
codexCharCount,
|
||||
options.signal
|
||||
)
|
||||
: null
|
||||
const renderGate = isLargeCodexPaste ? null : this.createAgentPromptRenderGate(ptyId)
|
||||
let wrotePasteBytes = false
|
||||
let completedPaste = false
|
||||
try {
|
||||
@@ -19413,22 +19457,18 @@ export class OrcaRuntimeService {
|
||||
let chunk = chunks.next()
|
||||
while (!chunk.done) {
|
||||
const nextChunk = chunks.next()
|
||||
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)
|
||||
)
|
||||
if (nextChunk.done) {
|
||||
renderGate?.arm()
|
||||
}
|
||||
const wrote = this.ptyController?.write(ptyId, chunk.value) ?? false
|
||||
if (!wrote) {
|
||||
throw new Error('terminal_not_writable')
|
||||
}
|
||||
await this.writeAgentPromptBytes({
|
||||
handle,
|
||||
ptyId,
|
||||
generation,
|
||||
data: chunk.value,
|
||||
permissionBaseline,
|
||||
...(isLargeCodexPaste ? { writeFailureError: 'agent_prompt_unverifiable' } : {}),
|
||||
options
|
||||
})
|
||||
wrotePasteBytes = true
|
||||
chunk = nextChunk
|
||||
if (!chunk.done) {
|
||||
@@ -19448,7 +19488,17 @@ export class OrcaRuntimeService {
|
||||
throw error
|
||||
}
|
||||
|
||||
if (renderGate) {
|
||||
let codexIngestProof: CodexPasteIngestProof | null = null
|
||||
if (codexIngestBaseline) {
|
||||
codexIngestProof = await this.waitForCodexPasteIngest(
|
||||
handle,
|
||||
ptyId,
|
||||
generation,
|
||||
permissionBaseline,
|
||||
codexIngestBaseline,
|
||||
options.signal
|
||||
)
|
||||
} else if (renderGate) {
|
||||
try {
|
||||
await waitForAgentPromptPromise(renderGate.wait(), options.signal)
|
||||
} finally {
|
||||
@@ -19457,24 +19507,26 @@ export class OrcaRuntimeService {
|
||||
} else {
|
||||
await waitForAgentPromptDelay(AGENT_PROMPT_SUBMIT_DELAY_MS, options.signal)
|
||||
}
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
try {
|
||||
await options.beforeWrite?.(ptyId)
|
||||
} catch (error) {
|
||||
if (options.suffixFailureError) {
|
||||
throw new Error(options.suffixFailureError)
|
||||
}
|
||||
throw error
|
||||
if (codexIngestProof) {
|
||||
return await this.submitVerifiedCodexPaste({
|
||||
handle,
|
||||
ptyId,
|
||||
generation,
|
||||
permissionBaseline,
|
||||
ingest: codexIngestProof,
|
||||
options
|
||||
})
|
||||
}
|
||||
assertAgentPromptRequestActive(options.signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
const baseline = this.getAgentPromptActivity(handle, ptyId)
|
||||
this.assertAgentPromptPermissionSafe(permissionBaseline, baseline)
|
||||
const suffixWrote = this.ptyController?.write(ptyId, AGENT_PROMPT_SUBMIT) ?? false
|
||||
if (!suffixWrote) {
|
||||
throw new Error(options.suffixFailureError ?? 'terminal_not_writable')
|
||||
}
|
||||
await this.writeAgentPromptBytes({
|
||||
handle,
|
||||
ptyId,
|
||||
generation,
|
||||
data: AGENT_PROMPT_SUBMIT,
|
||||
permissionBaseline,
|
||||
failureError: options.suffixFailureError,
|
||||
options
|
||||
})
|
||||
await verifyAgentPromptSubmission({
|
||||
baseline,
|
||||
readActivity: () => this.getAgentPromptActivity(handle, ptyId),
|
||||
@@ -19483,6 +19535,198 @@ export class OrcaRuntimeService {
|
||||
return 1
|
||||
}
|
||||
|
||||
private getAgentPromptTargetAgent(ptyId: string): TuiAgent | null | undefined {
|
||||
const pty = this.ptysById.get(ptyId)
|
||||
return pty?.launchAgent ?? pty?.foregroundAgent
|
||||
}
|
||||
|
||||
private async captureCodexPasteIngestBaseline(
|
||||
ptyId: string,
|
||||
generation: number,
|
||||
charCount: number,
|
||||
signal?: AbortSignal
|
||||
): Promise<CodexPasteIngestBaseline> {
|
||||
if (this.providerSnapshotPreferredPtys.has(ptyId)) {
|
||||
const supported = await this.ptyController?.supportsProviderBufferSnapshot?.(ptyId, signal)
|
||||
assertAgentPromptRequestActive(signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
if (supported !== true) {
|
||||
throw new Error('agent_prompt_unverifiable')
|
||||
}
|
||||
}
|
||||
const screen = await this.readCodexPasteScreen(ptyId, generation)
|
||||
if (!screen) {
|
||||
throw new Error('agent_prompt_unverifiable')
|
||||
}
|
||||
return {
|
||||
charCount,
|
||||
markers: collectCodexPasteMarkers(screen.lines, charCount),
|
||||
screen
|
||||
}
|
||||
}
|
||||
|
||||
private async waitForCodexPasteIngest(
|
||||
handle: string,
|
||||
ptyId: string,
|
||||
generation: number,
|
||||
permissionBaseline: AgentPromptActivity,
|
||||
baseline: CodexPasteIngestBaseline,
|
||||
signal?: AbortSignal
|
||||
): Promise<CodexPasteIngestProof> {
|
||||
const deadline = Date.now() + AGENT_PROMPT_RENDER_TIMEOUT_MS
|
||||
while (Date.now() < deadline) {
|
||||
assertAgentPromptRequestActive(signal)
|
||||
this.assertAgentPromptGeneration(ptyId, generation)
|
||||
this.assertAgentPromptPermissionSafe(
|
||||
permissionBaseline,
|
||||
this.getAgentPromptActivity(handle, ptyId)
|
||||
)
|
||||
const screen = await this.readCodexPasteScreen(ptyId, generation)
|
||||
if (screen) {
|
||||
if (screen.sequence < baseline.screen.sequence) {
|
||||
throw new Error('agent_prompt_unverifiable')
|
||||
}
|
||||
const delta = findCodexPasteMarkerDelta(
|
||||
baseline.markers,
|
||||
collectCodexPasteMarkers(screen.lines, baseline.charCount)
|
||||
)
|
||||
if (delta) {
|
||||
return { ...delta, screen }
|
||||
}
|
||||
}
|
||||
await waitForAgentPromptDelay(AGENT_PROMPT_SCREEN_POLL_MS, signal)
|
||||
}
|
||||
throw new Error('agent_prompt_ingest_unverified')
|
||||
}
|
||||
|
||||
private async submitVerifiedCodexPaste(args: {
|
||||
handle: string
|
||||
ptyId: string
|
||||
generation: number
|
||||
permissionBaseline: AgentPromptActivity
|
||||
ingest: CodexPasteIngestProof
|
||||
options: {
|
||||
beforeWrite?: (ptyId: string) => void | Promise<void>
|
||||
signal?: AbortSignal
|
||||
}
|
||||
}): Promise<number> {
|
||||
let previousScreen = args.ingest.screen
|
||||
for (let submits = 1; submits <= AGENT_PROMPT_MAX_SUBMITS; submits += 1) {
|
||||
await this.writeAgentPromptBytes({
|
||||
handle: args.handle,
|
||||
ptyId: args.ptyId,
|
||||
generation: args.generation,
|
||||
data: AGENT_PROMPT_SUBMIT,
|
||||
permissionBaseline: args.permissionBaseline,
|
||||
failureError: 'agent_prompt_unverifiable',
|
||||
options: args.options
|
||||
})
|
||||
await waitForAgentPromptDelay(AGENT_PROMPT_SUBMIT_DELAY_MS, args.options.signal)
|
||||
this.assertAgentPromptPermissionSafe(
|
||||
args.permissionBaseline,
|
||||
this.getAgentPromptActivity(args.handle, args.ptyId)
|
||||
)
|
||||
const screen = await this.readCodexPasteScreen(args.ptyId, args.generation)
|
||||
if (!screen || screen.sequence < previousScreen.sequence) {
|
||||
throw new Error('agent_prompt_unverifiable')
|
||||
}
|
||||
const visibleCount = countCodexPasteMarker(screen.lines, args.ingest.marker)
|
||||
if (visibleCount < args.ingest.visibleCount) {
|
||||
if (screen.sequence <= previousScreen.sequence) {
|
||||
throw new Error('agent_prompt_unverifiable')
|
||||
}
|
||||
return submits
|
||||
}
|
||||
if (visibleCount > args.ingest.visibleCount) {
|
||||
throw new Error('agent_prompt_unverifiable')
|
||||
}
|
||||
previousScreen = screen
|
||||
}
|
||||
throw new Error('agent_prompt_stalled')
|
||||
}
|
||||
|
||||
private async readCodexPasteScreen(
|
||||
ptyId: string,
|
||||
generation: number
|
||||
): Promise<RuntimeVisibleTerminalState | null> {
|
||||
if (!this.providerSnapshotPreferredPtys.has(ptyId)) {
|
||||
const screen = await this.readHeadlessVisibleTerminalState(ptyId)
|
||||
return screen?.generation === generation ? screen : null
|
||||
}
|
||||
const snapshot = await this.captureProviderTerminalBuffer(
|
||||
ptyId,
|
||||
{ scrollbackRows: 0 },
|
||||
generation
|
||||
)
|
||||
if (
|
||||
!snapshot ||
|
||||
this.getPtyLifecycleGeneration(ptyId) !== generation ||
|
||||
this.getPtyOutputSequence(ptyId) > snapshot.seq
|
||||
) {
|
||||
return null
|
||||
}
|
||||
return {
|
||||
lines: await this.parseVisibleSnapshotLines(snapshot),
|
||||
isAlternateScreen: snapshot.alternateScreen ?? false,
|
||||
sequence: snapshot.seq,
|
||||
generation
|
||||
}
|
||||
}
|
||||
|
||||
private async writeAgentPromptBytes(args: {
|
||||
handle: string
|
||||
ptyId: string
|
||||
generation: number
|
||||
data: string
|
||||
permissionBaseline: AgentPromptActivity
|
||||
failureError?: string
|
||||
writeFailureError?: string
|
||||
options: {
|
||||
beforeWrite?: (ptyId: string) => void | Promise<void>
|
||||
signal?: AbortSignal
|
||||
}
|
||||
}): Promise<void> {
|
||||
assertAgentPromptRequestActive(args.options.signal)
|
||||
this.assertLiveTerminalHandleTargetsPty(args.handle, args.ptyId)
|
||||
this.assertAgentPromptGeneration(args.ptyId, args.generation)
|
||||
try {
|
||||
await args.options.beforeWrite?.(args.ptyId)
|
||||
} catch (error) {
|
||||
if (args.failureError) {
|
||||
throw new Error(args.failureError)
|
||||
}
|
||||
throw error
|
||||
}
|
||||
assertAgentPromptRequestActive(args.options.signal)
|
||||
this.assertLiveTerminalHandleTargetsPty(args.handle, args.ptyId)
|
||||
this.assertAgentPromptGeneration(args.ptyId, args.generation)
|
||||
this.assertAgentPromptPermissionSafe(
|
||||
args.permissionBaseline,
|
||||
this.getAgentPromptActivity(args.handle, args.ptyId)
|
||||
)
|
||||
this.assertAgentPromptWritable(args.handle, args.ptyId)
|
||||
const wrote = this.ptyController?.writeWithSettlement
|
||||
? await this.ptyController.writeWithSettlement(args.ptyId, args.data)
|
||||
: (this.ptyController?.write(args.ptyId, args.data) ?? false)
|
||||
if (!wrote) {
|
||||
throw new Error(args.writeFailureError ?? args.failureError ?? 'terminal_not_writable')
|
||||
}
|
||||
}
|
||||
|
||||
private assertAgentPromptWritable(handle: string, ptyId: string): void {
|
||||
const runtimePty = this.getLivePtyForHandle(handle)
|
||||
if (runtimePty) {
|
||||
if (runtimePty.pty.ptyId !== ptyId || !runtimePty.pty.connected) {
|
||||
throw new Error('terminal_not_writable')
|
||||
}
|
||||
return
|
||||
}
|
||||
const { leaf } = this.getLiveLeafForHandle(handle)
|
||||
if (leaf.ptyId !== ptyId || !leaf.connected || !leaf.writable) {
|
||||
throw new Error('terminal_not_writable')
|
||||
}
|
||||
}
|
||||
|
||||
private async serializeAgentPromptSubmission<T>(
|
||||
ptyId: string,
|
||||
generation: number,
|
||||
|
||||
@@ -20,6 +20,7 @@ import {
|
||||
resolveFederatedWorkerLaunchReceipt,
|
||||
type OrchestrationWorkerLaunchReceipt
|
||||
} from './orchestration-worker-launch-preferences'
|
||||
import { assertPromptSubmitVerificationSupported } from './orchestration-prompt-submit-capability'
|
||||
import { validateFederatedWorkerStartPlacement } from './orchestration-worker-start-validation'
|
||||
|
||||
export async function startFederatedWorker(args: {
|
||||
@@ -78,6 +79,7 @@ export async function startFederatedWorker(args: {
|
||||
`Connected server ${server.name} does not support orchestration federation.`
|
||||
)
|
||||
}
|
||||
assertPromptSubmitVerificationSupported(status.capabilities, server.name)
|
||||
assertWorkerLaunchPreferencesRuntimeSupported({
|
||||
model: params.model,
|
||||
effort: params.effort,
|
||||
|
||||
@@ -1,7 +1,9 @@
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import {
|
||||
ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_FEDERATION_RUNTIME_CAPABILITY
|
||||
ORCHESTRATION_FEDERATION_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_PROMPT_SUBMIT_VERIFICATION_RUNTIME_CAPABILITY,
|
||||
RUNTIME_CAPABILITIES
|
||||
} from '../../../../shared/protocol-version'
|
||||
import { OrcaRuntimeService } from '../../orca-runtime'
|
||||
import { OrchestrationDb } from '../../orchestration/db'
|
||||
@@ -245,4 +247,54 @@ describe('orchestration migration behavior', () => {
|
||||
expect(db.getTask(task.id)?.status).toBe('ready')
|
||||
expect(db.getDispatchContext(task.id)).toBeUndefined()
|
||||
})
|
||||
|
||||
it('rejects a connected server missing prompt verification before Task mutation', async () => {
|
||||
const { db, runtime } = createRuntime()
|
||||
const run = db.createRun({
|
||||
objective: 'unverified remote worker',
|
||||
coordinatorHandle: 'term_coord',
|
||||
coordinatorPaneKey: 'tab_coord:leaf_coord'
|
||||
})
|
||||
const task = db.createTask({ spec: 'remote work', runId: run.id })
|
||||
vi.spyOn(runtime, 'resolveOrchestrationWorkerServer').mockReturnValue({
|
||||
environmentId: 'environment_windows',
|
||||
name: 'windows',
|
||||
peerFingerprint: 'windows_peer'
|
||||
})
|
||||
vi.spyOn(runtime, 'callOrchestrationWorkerServer').mockResolvedValue({
|
||||
capabilities: [
|
||||
ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_FEDERATION_RUNTIME_CAPABILITY
|
||||
]
|
||||
})
|
||||
|
||||
await expect(
|
||||
startFederatedWorker({
|
||||
params: {
|
||||
task: task.id,
|
||||
from: 'term_coord',
|
||||
on: 'windows',
|
||||
worktree: 'new-top-level',
|
||||
repo: 'id:windows-repo',
|
||||
name: 'remote-work',
|
||||
agent: 'codex'
|
||||
},
|
||||
runtime,
|
||||
db,
|
||||
runId: run.id,
|
||||
task,
|
||||
orchestrationMutation: {
|
||||
callerFingerprint: 'caller',
|
||||
requestId: 'remote_start',
|
||||
method: 'orchestration.workerStart',
|
||||
payloadHash: 'payload'
|
||||
}
|
||||
})
|
||||
).rejects.toMatchObject({ code: 'capability_unsupported' })
|
||||
expect(db.getTask(task.id)?.status).toBe('ready')
|
||||
expect(db.getDispatchContext(task.id)).toBeUndefined()
|
||||
expect(RUNTIME_CAPABILITIES).toContain(
|
||||
ORCHESTRATION_PROMPT_SUBMIT_VERIFICATION_RUNTIME_CAPABILITY
|
||||
)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -0,0 +1,15 @@
|
||||
import { ORCHESTRATION_PROMPT_SUBMIT_VERIFICATION_RUNTIME_CAPABILITY } from '../../../../shared/protocol-version'
|
||||
import { OrchestrationError } from '../../orchestration/orchestration-error'
|
||||
|
||||
export function assertPromptSubmitVerificationSupported(
|
||||
capabilities: readonly string[] | undefined,
|
||||
serverName: string
|
||||
): void {
|
||||
if (capabilities?.includes(ORCHESTRATION_PROMPT_SUBMIT_VERIFICATION_RUNTIME_CAPABILITY)) {
|
||||
return
|
||||
}
|
||||
throw new OrchestrationError(
|
||||
'capability_unsupported',
|
||||
`Connected server ${serverName} cannot verify worker prompt submission. No effects were applied.`
|
||||
)
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'
|
||||
import { PTY_STARTUP_INGRESS_VERSION } from '../shared/pty-startup-ingress'
|
||||
import { PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION } from '../shared/pty-visible-screen-snapshot'
|
||||
|
||||
const { mockPtySpawn, mockPtyInstance, mockCreateShellPromptReadinessProbe } = vi.hoisted(() => ({
|
||||
mockPtySpawn: vi.fn(),
|
||||
@@ -128,6 +129,36 @@ describe('PtyHandler', () => {
|
||||
expect(dispatcher.notify).toHaveBeenCalledWith('pty.data', { id: 'pty-1', data: 'hello world' })
|
||||
})
|
||||
|
||||
it('serves an ordered visible-screen snapshot for the current PTY incarnation', async () => {
|
||||
let dataCallback: ((data: string) => void) | undefined
|
||||
mockPtySpawn.mockReturnValue({
|
||||
...mockPtyInstance,
|
||||
onData: vi.fn((callback: (data: string) => void) => {
|
||||
dataCallback = callback
|
||||
}),
|
||||
onExit: vi.fn()
|
||||
})
|
||||
const spawned = (await dispatcher.callRequest('pty.spawn', {
|
||||
cols: 80,
|
||||
rows: 24
|
||||
})) as { id: string; incarnationId: string }
|
||||
|
||||
dataCallback?.('[Pasted Content 1001 chars]')
|
||||
const snapshot = (await dispatcher.callRequest('pty.getBufferSnapshot', {
|
||||
id: spawned.id
|
||||
})) as { data: string; seq: number; incarnationId: string }
|
||||
const capabilities = (await dispatcher.callRequest('pty.getCapabilities')) as {
|
||||
visibleScreenSnapshotVersion: number
|
||||
}
|
||||
|
||||
expect(snapshot.data).toContain('[Pasted Content 1001 chars]')
|
||||
expect(snapshot.seq).toBe('[Pasted Content 1001 chars]'.length)
|
||||
expect(snapshot.incarnationId).toBe(spawned.incarnationId)
|
||||
expect(capabilities.visibleScreenSnapshotVersion).toBe(
|
||||
PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION
|
||||
)
|
||||
})
|
||||
|
||||
it('consumes capable startup queries before relay replay and fanout', async () => {
|
||||
let dataCallback: ((data: string) => void) | undefined
|
||||
const term = {
|
||||
|
||||
@@ -69,6 +69,7 @@ describe('PtyHandler', () => {
|
||||
expect(methods).toContain('pty.sendSignal')
|
||||
expect(methods).toContain('pty.getCwd')
|
||||
expect(methods).toContain('pty.getInitialCwd')
|
||||
expect(methods).toContain('pty.getBufferSnapshot')
|
||||
expect(methods).toContain('pty.clearBuffer')
|
||||
expect(methods).toContain('pty.hasChildProcesses')
|
||||
expect(methods).toContain('pty.getForegroundProcess')
|
||||
|
||||
@@ -89,6 +89,9 @@ import {
|
||||
injectRelayFishHistoryEnv,
|
||||
injectRelayHistoryEnv
|
||||
} from './terminal-history'
|
||||
import { HeadlessEmulator } from '../main/daemon/headless-emulator'
|
||||
import type { PtyProviderBufferSnapshot } from '../main/providers/pty-provider-contract'
|
||||
import { PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION } from '../shared/pty-visible-screen-snapshot'
|
||||
|
||||
// Why: only Linux compiles node-pty (no prebuilt), so the build-tools remedy is a closable setup gap
|
||||
// there and wrong advice anywhere node-pty ships one. The relay only sees an unloadable binding, never
|
||||
@@ -149,6 +152,10 @@ type ManagedPty = {
|
||||
/** Why a chunk deque: rebuilding a rolling 100KB string per PTY chunk copied the
|
||||
* whole window on every write once saturated. Readers are attach/adopt/revive only. */
|
||||
buffered: RecentPtyOutputBuffer
|
||||
screen: HeadlessEmulator
|
||||
screenWriteChain: Promise<void>
|
||||
screenSequence: number
|
||||
screenAppliedSequence: number
|
||||
/** Timer for SIGKILL fallback after a graceful SIGTERM shutdown. */
|
||||
killTimer?: ReturnType<typeof setTimeout>
|
||||
/** True once disposeManagedPty has run; blocks double-dispose and makes post-dispose calls fail "not found" not silently. */
|
||||
@@ -232,11 +239,24 @@ function finishPtyCreationOperations(operations: readonly (() => void)[]): void
|
||||
}
|
||||
}
|
||||
|
||||
function createManagedPtyScreen(
|
||||
cols: number,
|
||||
rows: number
|
||||
): Pick<ManagedPty, 'screen' | 'screenWriteChain' | 'screenSequence' | 'screenAppliedSequence'> {
|
||||
return {
|
||||
screen: new HeadlessEmulator({ cols, rows, scrollback: 0 }),
|
||||
screenWriteChain: Promise.resolve(),
|
||||
screenSequence: 0,
|
||||
screenAppliedSequence: 0
|
||||
}
|
||||
}
|
||||
|
||||
function disposeManagedPty(managed: ManagedPty): void {
|
||||
if (managed.disposed) {
|
||||
return
|
||||
}
|
||||
managed.disposed = true
|
||||
managed.screen.dispose()
|
||||
// Why: clear the SIGKILL fallback timer so it can't fire pty.kill on an already-disposed instance.
|
||||
if (managed.killTimer) {
|
||||
clearTimeout(managed.killTimer)
|
||||
@@ -823,6 +843,13 @@ export class PtyHandler {
|
||||
})
|
||||
}
|
||||
managed.pty.onData((data: string) => {
|
||||
const screenSequence = managed.screenSequence + data.length
|
||||
managed.screenSequence = screenSequence
|
||||
const screenWrite = managed.screenWriteChain.then(async () => {
|
||||
await managed.screen.write(data)
|
||||
managed.screenAppliedSequence = screenSequence
|
||||
})
|
||||
managed.screenWriteChain = screenWrite.catch(() => {})
|
||||
const startup = managed.startupCommand
|
||||
if (startup?.waitForShellReady && startup.outputScanState && !startup.delivered) {
|
||||
const scanned = scanShellStartupOutput(startup.outputScanState, data)
|
||||
@@ -913,6 +940,7 @@ export class PtyHandler {
|
||||
this.dispatcher.onRequest('pty.getCwd', (p) => this.getCwd(p))
|
||||
this.dispatcher.onRequest('pty.getInitialCwd', (p) => this.getInitialCwd(p))
|
||||
this.dispatcher.onRequest('pty.getSize', (p) => this.getSize(p))
|
||||
this.dispatcher.onRequest('pty.getBufferSnapshot', (p) => this.getBufferSnapshot(p))
|
||||
this.dispatcher.onRequest('pty.clearBuffer', (p) => this.clearBuffer(p))
|
||||
this.dispatcher.onRequest('pty.hasChildProcesses', (p) => this.hasChildProcesses(p))
|
||||
this.dispatcher.onRequest('pty.getForegroundProcess', (p) => this.getForegroundProcess(p))
|
||||
@@ -920,7 +948,8 @@ export class PtyHandler {
|
||||
this.dispatcher.onRequest('pty.getCapabilities', async () => ({
|
||||
startupIngressVersion: PTY_STARTUP_INGRESS_VERSION,
|
||||
agentSessionClaimVersion: AGENT_SESSION_EXECUTION_OWNER_PROTOCOL_VERSION,
|
||||
agentSessionCreateOperationVersion: AGENT_SESSION_CREATE_OPERATION_PROTOCOL_VERSION
|
||||
agentSessionCreateOperationVersion: AGENT_SESSION_CREATE_OPERATION_PROTOCOL_VERSION,
|
||||
visibleScreenSnapshotVersion: PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION
|
||||
}))
|
||||
this.dispatcher.onRequest('pty.listProcesses', () => this.listProcesses())
|
||||
this.dispatcher.onRequest('pty.getDefaultShell', async () => resolveDefaultShell())
|
||||
@@ -1694,6 +1723,7 @@ export class PtyHandler {
|
||||
preserveChunkBoundaries: false,
|
||||
limit: REPLAY_BUFFER_MAX
|
||||
}),
|
||||
...createManagedPtyScreen(cols, rows),
|
||||
paneKey,
|
||||
tabId,
|
||||
...(attachIdentity.paneKey || attachIdentity.tabId ? { attachIdentity } : {}),
|
||||
@@ -1897,6 +1927,43 @@ export class PtyHandler {
|
||||
const managed = this.ptys.get(id)
|
||||
if (managed && !managed.disposed) {
|
||||
managed.pty.resize(cols, rows)
|
||||
managed.screenWriteChain = managed.screenWriteChain
|
||||
.then(() => managed.screen.resize(cols, rows))
|
||||
.catch(() => {})
|
||||
}
|
||||
}
|
||||
|
||||
private async getBufferSnapshot(
|
||||
params: Record<string, unknown>
|
||||
): Promise<PtyProviderBufferSnapshot | null> {
|
||||
const managed = this.ptys.get(params.id as string)
|
||||
if (!managed || managed.disposed) {
|
||||
return null
|
||||
}
|
||||
const incarnationId = managed.incarnationId
|
||||
await managed.screenWriteChain
|
||||
if (this.ptys.get(managed.id) !== managed || managed.disposed) {
|
||||
return null
|
||||
}
|
||||
const scrollbackRows = Math.max(0, Math.floor(Number(params.scrollbackRows) || 0))
|
||||
const snapshot = managed.screen.getSnapshot({ scrollbackRows })
|
||||
return {
|
||||
data: snapshot.rehydrateSequences + snapshot.snapshotAnsi,
|
||||
frameRestoreAnsi: snapshot.frameRestoreAnsi,
|
||||
scrollbackAnsi: snapshot.scrollbackAnsi,
|
||||
cols: snapshot.cols,
|
||||
rows: snapshot.rows,
|
||||
cwd: snapshot.cwd,
|
||||
lastTitle: snapshot.lastTitle,
|
||||
seq: managed.screenAppliedSequence,
|
||||
incarnationId,
|
||||
source: 'headless',
|
||||
oscLinks: snapshot.oscLinks,
|
||||
alternateScreen: managed.screen.isAlternateScreen,
|
||||
pendingEscapeTailAnsi: snapshot.pendingEscapeTailAnsi,
|
||||
...(typeof snapshot.modes.kittyKeyboardFlags === 'number'
|
||||
? { kittyKeyboardFlags: snapshot.modes.kittyKeyboardFlags }
|
||||
: {})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2285,6 +2352,7 @@ export class PtyHandler {
|
||||
preserveChunkBoundaries: false,
|
||||
limit: REPLAY_BUFFER_MAX
|
||||
}),
|
||||
...createManagedPtyScreen(entry.cols, entry.rows),
|
||||
paneKey: entry.paneKey,
|
||||
tabId: entry.tabId,
|
||||
attachIdentity: entry.attachIdentity,
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import {
|
||||
codexAgentPromptCharCount,
|
||||
collectCodexPasteMarkers,
|
||||
countCodexPasteMarker,
|
||||
findCodexPasteMarkerDelta
|
||||
} from './agent-prompt-codex-ingest'
|
||||
|
||||
describe('Codex agent-prompt paste ingest', () => {
|
||||
it.each([
|
||||
['astral Unicode', '😀', 1],
|
||||
['CRLF and bare CR', 'a\r\nb\rc', 5],
|
||||
['ESC literalization', 'a\x1bb', 7],
|
||||
['C0 and C1 controls', 'a\x00b\x85c', 3],
|
||||
['LF and tab retention', 'a\nb\tc', 5]
|
||||
])('matches Codex normalization for %s', (_name, prompt, expected) => {
|
||||
expect(codexAgentPromptCharCount(prompt)).toBe(expected)
|
||||
})
|
||||
|
||||
it('finds one delivery-owned member without predicting its suffix', () => {
|
||||
const baseline = collectCodexPasteMarkers(['[Pasted Content 5000 chars]'], 5_000)
|
||||
const current = collectCodexPasteMarkers(
|
||||
['[Pasted Content 5000 chars] [Pasted Content 5000 chars] #7'],
|
||||
5_000
|
||||
)
|
||||
|
||||
expect(findCodexPasteMarkerDelta(baseline, current)).toEqual({
|
||||
marker: '[Pasted Content 5000 chars] #7',
|
||||
visibleCount: 1
|
||||
})
|
||||
})
|
||||
|
||||
it('finds a marker split across rendered grid rows', () => {
|
||||
const current = collectCodexPasteMarkers(['[Pasted Content ', '5000 chars] #23'], 5_000)
|
||||
|
||||
expect(findCodexPasteMarkerDelta(new Map(), current)).toEqual({
|
||||
marker: '[Pasted Content 5000 chars] #23',
|
||||
visibleCount: 1
|
||||
})
|
||||
})
|
||||
|
||||
it('finds an unsuffixed marker added beside a pre-existing identical marker', () => {
|
||||
const baseline = collectCodexPasteMarkers(['[Pasted Content 5000 chars]'], 5_000)
|
||||
const current = collectCodexPasteMarkers(
|
||||
['[Pasted Content 5000 chars] [Pasted Content 5000 chars]'],
|
||||
5_000
|
||||
)
|
||||
|
||||
expect(findCodexPasteMarkerDelta(baseline, current)).toEqual({
|
||||
marker: '[Pasted Content 5000 chars]',
|
||||
visibleCount: 2
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects redraws, removed baselines, and multiple additions', () => {
|
||||
const baseline = collectCodexPasteMarkers(['[Pasted Content 5000 chars]'], 5_000)
|
||||
expect(findCodexPasteMarkerDelta(baseline, baseline)).toBeNull()
|
||||
expect(findCodexPasteMarkerDelta(baseline, new Map())).toBeNull()
|
||||
expect(
|
||||
findCodexPasteMarkerDelta(
|
||||
baseline,
|
||||
collectCodexPasteMarkers(
|
||||
['[Pasted Content 5000 chars] #2 [Pasted Content 5000 chars] #3'],
|
||||
5_000
|
||||
)
|
||||
)
|
||||
).toBeNull()
|
||||
})
|
||||
|
||||
it('counts an exact owned marker', () => {
|
||||
expect(
|
||||
countCodexPasteMarker(
|
||||
['[Pasted Content 5000 chars] #2', '[Pasted Content 5000 chars]'],
|
||||
'[Pasted Content 5000 chars] #2'
|
||||
)
|
||||
).toBe(1)
|
||||
})
|
||||
|
||||
it('counts an exact marker across rendered grid rows', () => {
|
||||
expect(
|
||||
countCodexPasteMarker(
|
||||
['[Pasted Content 5000', ' chars] #2'],
|
||||
'[Pasted Content 5000 chars] #2'
|
||||
)
|
||||
).toBe(1)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,78 @@
|
||||
import { sanitizeAgentPromptText } from './agent-prompt-injection'
|
||||
|
||||
export const CODEX_LARGE_PASTE_CHAR_THRESHOLD = 1_000
|
||||
|
||||
export type CodexPasteMarkerMultiset = ReadonlyMap<string, number>
|
||||
|
||||
export type CodexPasteMarkerDelta = Readonly<{
|
||||
marker: string
|
||||
visibleCount: number
|
||||
}>
|
||||
|
||||
export function codexAgentPromptCharCount(prompt: string): number {
|
||||
const normalized = sanitizeAgentPromptText(prompt).replace(/\r\n|\r/g, '\n')
|
||||
let count = 0
|
||||
for (const character of normalized) {
|
||||
const codePoint = character.codePointAt(0)!
|
||||
if (character !== '\n' && character !== '\t' && isUnicodeControl(codePoint)) {
|
||||
continue
|
||||
}
|
||||
count += 1
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
export function collectCodexPasteMarkers(
|
||||
lines: readonly string[],
|
||||
charCount: number
|
||||
): Map<string, number> {
|
||||
const markers = new Map<string, number>()
|
||||
const pattern = new RegExp(`\\[Pasted Content ${charCount} chars\\](?: #[1-9]\\d*)?`, 'g')
|
||||
for (const marker of lines.join('').matchAll(pattern)) {
|
||||
const value = marker[0]
|
||||
markers.set(value, (markers.get(value) ?? 0) + 1)
|
||||
}
|
||||
return markers
|
||||
}
|
||||
|
||||
export function findCodexPasteMarkerDelta(
|
||||
baseline: CodexPasteMarkerMultiset,
|
||||
current: CodexPasteMarkerMultiset
|
||||
): CodexPasteMarkerDelta | null {
|
||||
let delta: CodexPasteMarkerDelta | null = null
|
||||
for (const [marker, baselineCount] of baseline) {
|
||||
if ((current.get(marker) ?? 0) < baselineCount) {
|
||||
return null
|
||||
}
|
||||
}
|
||||
for (const [marker, currentCount] of current) {
|
||||
const increment = currentCount - (baseline.get(marker) ?? 0)
|
||||
if (increment === 0) {
|
||||
continue
|
||||
}
|
||||
if (increment !== 1 || delta) {
|
||||
return null
|
||||
}
|
||||
delta = { marker, visibleCount: currentCount }
|
||||
}
|
||||
return delta
|
||||
}
|
||||
|
||||
export function countCodexPasteMarker(lines: readonly string[], marker: string): number {
|
||||
let count = 0
|
||||
const visibleText = lines.join('')
|
||||
let cursor = 0
|
||||
while (cursor <= visibleText.length - marker.length) {
|
||||
const match = visibleText.indexOf(marker, cursor)
|
||||
if (match === -1) {
|
||||
break
|
||||
}
|
||||
count += 1
|
||||
cursor = match + marker.length
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
function isUnicodeControl(codePoint: number): boolean {
|
||||
return codePoint <= 0x1f || (codePoint >= 0x7f && codePoint <= 0x9f)
|
||||
}
|
||||
@@ -49,6 +49,8 @@ export const ORCHESTRATION_WORKER_STOP_VERDICT_RUNTIME_CAPABILITY =
|
||||
'orchestration.worker-stop-verdict.v1' as const
|
||||
export const ORCHESTRATION_WORKER_LAUNCH_PREFERENCES_RUNTIME_CAPABILITY =
|
||||
'orchestration.worker-launch-preferences.v1' as const
|
||||
export const ORCHESTRATION_PROMPT_SUBMIT_VERIFICATION_RUNTIME_CAPABILITY =
|
||||
'orchestration.prompt-submit-verification.v1' as const
|
||||
export const ORCHESTRATION_FEDERATION_CONTROL_MAIL_PROTOCOL_VERSION = 2 as const
|
||||
export const ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_PROTOCOL_VERSION = 3 as const
|
||||
export const ORCHESTRATION_CONTRACT_VERSION = 1 as const
|
||||
@@ -123,6 +125,7 @@ export const RUNTIME_CAPABILITIES = [
|
||||
ORCHESTRATION_FEDERATION_LIFECYCLE_SETTLEMENT_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_WORKER_STOP_VERDICT_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_WORKER_LAUNCH_PREFERENCES_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_PROMPT_SUBMIT_VERIFICATION_RUNTIME_CAPABILITY,
|
||||
ORCHESTRATION_CONTRACT_RUNTIME_CAPABILITY,
|
||||
BROWSER_SCREENCAST_RUNTIME_CAPABILITY,
|
||||
BROWSER_TAB_CREATE_KNOWN_ID_RUNTIME_CAPABILITY,
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
export const PTY_VISIBLE_SCREEN_SNAPSHOT_PROTOCOL_VERSION = 1 as const
|
||||
Reference in New Issue
Block a user