Enforce the agent-session lease at every PTY write choke point

Route every byte that can reach a PTY through the single-writer lease
adjudicator, so a session owned by one runtime cannot be typed into by
another. Nothing binds a PTY to a durable session record yet, so the gate
short-circuits and today's shells, legacy agent terminals, and unbound
sessions are untouched.

Gated paths: terminal.send RPC, the renderer IPC write paths
(pty:write / pty:writeAccepted), plugin sendTerminalText, the agent-prompt
paste, and ptyController.write as a backstop for runtime writes that never
passed a typed gate. Every yield inside a write - inter-chunk, the
text/suffix pause, the agent-prompt submit delay, deferred size
measurement - re-checks the fence, so a handoff mid-paste stops the
remaining chunks instead of interleaving two owners.

Refusals are typed and additive on the wire: an optional
agentSessionRefusal on RuntimeTerminalSend and on the existing
pty:writeUnavailable payload, refusal codes drawn only from the existing
agent-session error codes, and stream-input refusals mapped onto the
WriteUnavailable frame old clients already decode.
This commit is contained in:
Brennan Benson
2026-08-09 13:00:56 -07:00
parent a2b20acfcc
commit c69b12b3dd
15 changed files with 1596 additions and 15 deletions
@@ -0,0 +1,280 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
const { handleMock, onMock, removeHandlerMock, removeAllListenersMock } = vi.hoisted(() => ({
handleMock: vi.fn(),
onMock: vi.fn(),
removeHandlerMock: vi.fn(),
removeAllListenersMock: vi.fn()
}))
vi.mock('electron', () => ({
app: {
isPackaged: true,
getPath: vi.fn().mockReturnValue('/tmp/orca-test-userdata')
},
ipcMain: {
handle: handleMock,
on: onMock,
removeHandler: removeHandlerMock,
removeAllListeners: removeAllListenersMock
},
powerMonitor: { on: vi.fn() }
}))
vi.mock('fs', () => ({
existsSync: () => true,
statSync: () => ({ isDirectory: () => true, mode: 0o755 }),
accessSync: () => undefined,
mkdirSync: vi.fn(),
readFileSync: vi.fn(() => ''),
writeFileSync: vi.fn(),
chmodSync: vi.fn(),
constants: { X_OK: 1 }
}))
vi.mock('node-pty', () => ({
spawn: vi.fn().mockReturnValue({
onData: vi.fn(),
onExit: vi.fn(),
write: vi.fn(),
resize: vi.fn(),
kill: vi.fn(),
process: 'zsh',
pid: 12345
})
}))
vi.mock('../opencode/hook-service', () => ({
openCodeHookService: { buildPtyEnv: () => ({}), clearPty: vi.fn() }
}))
vi.mock('../pi/titlebar-extension-service', () => ({
piTitlebarExtensionService: { buildPtyEnv: () => ({}), clearPty: vi.fn() }
}))
import {
deletePtyOwnership,
registerPtyHandlers,
registerSshPtyProvider,
setPtyOwnership,
unregisterSshPtyProvider
} from './pty'
import { agentSessionPtyWriteGate } from '../runtime/agent-session-pty-write-gate'
import {
agentSessionLeaseFixture,
agentSessionRecordFixture
} from '../../shared/agent-session-record.test-fixture'
import { TERMINAL_INPUT_CHUNK_MAX_BYTES } from '../../shared/terminal-input'
import type { AgentSessionLease, AgentSessionRecord } from '../../shared/agent-session-record'
import type { IPtyProvider } from '../providers/types'
// The renderer IPC path and the runtime controller are the two byte entry points this module owns;
// both are proved to consult the lease, and both are proved to leave an unbound PTY alone.
const CONNECTION_ID = 'conn-lease'
const PTY_ID = 'ssh:conn-lease@@pty-1'
const SESSION_ID = 'session-alpha-1'
const handlers = new Map<string, (...args: unknown[]) => unknown>()
const records = new Map<string, AgentSessionRecord>()
const mainWindow = {
isDestroyed: () => false,
webContents: { on: vi.fn(), send: vi.fn(), removeListener: vi.fn() }
}
const mainWindowIpcEvent = { sender: mainWindow.webContents }
let ptyController: { write: (ptyId: string, data: string) => boolean } | null = null
function createMockProvider(): IPtyProvider {
return {
spawn: vi.fn().mockResolvedValue({ id: PTY_ID }),
attach: vi.fn(),
write: vi.fn(),
resize: vi.fn(),
shutdown: vi.fn(),
sendSignal: vi.fn(),
getCwd: vi.fn(),
getInitialCwd: vi.fn(),
clearBuffer: vi.fn(),
acknowledgeDataEvent: vi.fn(),
hasChildProcesses: vi.fn(),
getForegroundProcess: vi.fn(),
serialize: vi.fn(),
revive: vi.fn(),
listProcesses: vi.fn(),
getDefaultShell: vi.fn(),
getProfiles: vi.fn(),
onData: vi.fn().mockReturnValue(() => {}),
onReplay: vi.fn().mockReturnValue(() => {}),
onExit: vi.fn().mockReturnValue(() => {})
} as unknown as IPtyProvider
}
let provider: IPtyProvider
function publish(lease: AgentSessionLease): void {
records.set(lease.sessionId, agentSessionRecordFixture(lease))
}
function enforce(lease: AgentSessionLease = agentSessionLeaseFixture()): void {
publish(lease)
agentSessionPtyWriteGate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
agentSessionPtyWriteGate.bindPty(PTY_ID, SESSION_ID)
}
function writeFromRenderer(data: string): void {
;(handlers.get('pty:write') as (event: unknown, args: unknown) => void)(mainWindowIpcEvent, {
id: PTY_ID,
data
})
}
async function writeAcceptedFromRenderer(data: string): Promise<boolean> {
return await (
handlers.get('pty:writeAccepted') as (
event: unknown,
args: unknown
) => boolean | Promise<boolean>
)(mainWindowIpcEvent, { id: PTY_ID, data })
}
function lastRefusal(): { id: string; agentSessionRefusal?: { code: string } } | null {
const call = mainWindow.webContents.send.mock.calls.findLast(
([channel]) => channel === 'pty:writeUnavailable'
)
return (call?.[1] as { id: string; agentSessionRefusal?: { code: string } }) ?? null
}
beforeEach(() => {
handlers.clear()
records.clear()
handleMock.mockReset()
onMock.mockReset()
mainWindow.webContents.send.mockReset()
ptyController = null
handleMock.mockImplementation((channel: string, handler: (...a: unknown[]) => unknown) => {
handlers.set(channel, handler)
})
onMock.mockImplementation((channel: string, handler: (...a: unknown[]) => unknown) => {
handlers.set(channel, handler)
})
const runtime = {
setPtyController: (controller: { write: (ptyId: string, data: string) => boolean }) => {
ptyController = controller
},
getDriver: () => ({ kind: 'desktop' })
}
registerPtyHandlers(mainWindow as never, runtime as never)
provider = createMockProvider()
registerSshPtyProvider(CONNECTION_ID, provider)
setPtyOwnership(PTY_ID, CONNECTION_ID)
})
afterEach(() => {
agentSessionPtyWriteGate.detachRecordLookup()
deletePtyOwnership(PTY_ID)
unregisterSshPtyProvider(CONNECTION_ID)
})
describe('renderer IPC write path', () => {
it('writes an unbound pty unchanged, which is every shell that exists today', () => {
writeFromRenderer('ls')
expect(provider.write).toHaveBeenCalledWith(PTY_ID, 'ls')
expect(lastRefusal()).toBeNull()
})
it('writes when the TUI owner holds a proven-live lease', () => {
enforce()
writeFromRenderer('ls')
expect(provider.write).toHaveBeenCalledWith(PTY_ID, 'ls')
})
it('refuses and reports the owner when native chat holds the session', () => {
enforce(agentSessionLeaseFixture({ runtimeKind: 'native' }))
writeFromRenderer('ls')
expect(provider.write).not.toHaveBeenCalled()
const refusal = lastRefusal()
expect(refusal?.id).toBe(PTY_ID)
expect(refusal?.agentSessionRefusal).toMatchObject({
code: 'agent_session_conflict',
sessionId: SESSION_ID,
ownerRuntimeKind: 'native',
ownerPid: 4242
})
})
it('refuses while the lease is unreconciled after a host restart', () => {
enforce(agentSessionLeaseFixture({ unreconciled: true }))
writeFromRenderer('ls')
expect(provider.write).not.toHaveBeenCalled()
expect(lastRefusal()?.agentSessionRefusal?.code).toBe('execution_owner_reconciling')
})
it('refuses the acknowledged write path too', async () => {
enforce(agentSessionLeaseFixture({ handoffStage: 'preparing' }))
await expect(writeAcceptedFromRenderer('')).resolves.toBe(false)
expect(provider.write).not.toHaveBeenCalled()
expect(lastRefusal()?.agentSessionRefusal?.code).toBe('agent_session_conflict')
})
it('leaves the acknowledged write path silent for an unbound pty', async () => {
await writeAcceptedFromRenderer('')
expect(lastRefusal()).toBeNull()
})
it('stops a chunked paste once the fence advances mid-flight', async () => {
enforce(agentSessionLeaseFixture({ runtimeFence: 7 }))
vi.mocked(provider.write).mockImplementation(() => {
publish(agentSessionLeaseFixture({ runtimeFence: 8 }))
})
writeFromRenderer('x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES * 2 + 8))
await vi.waitFor(() => expect(lastRefusal()).not.toBeNull())
expect(provider.write).toHaveBeenCalledTimes(1)
expect(lastRefusal()?.agentSessionRefusal?.code).toBe('agent_session_checkpoint_stale')
})
it('lets a chunked paste finish while the same owner holds the fence', async () => {
enforce()
writeFromRenderer('x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES * 2 + 8))
await vi.waitFor(() => expect(provider.write).toHaveBeenCalledTimes(3))
expect(lastRefusal()).toBeNull()
})
})
describe('runtime controller backstop', () => {
it('lets a runtime write through on an unbound pty', () => {
expect(ptyController?.write(PTY_ID, 'reply')).toBe(true)
expect(provider.write).toHaveBeenCalledWith(PTY_ID, 'reply')
})
it('blocks a runtime write that never passed a typed gate', () => {
// Query replies, followups, and deliveries reach the provider through this one function.
enforce(agentSessionLeaseFixture({ runtimeKind: 'native' }))
expect(ptyController?.write(PTY_ID, 'reply')).toBe(false)
expect(provider.write).not.toHaveBeenCalled()
expect(lastRefusal()?.agentSessionRefusal?.code).toBe('agent_session_conflict')
})
it('lets a runtime write through while the TUI owner holds the lease', () => {
enforce()
expect(ptyController?.write(PTY_ID, 'reply')).toBe(true)
expect(provider.write).toHaveBeenCalledWith(PTY_ID, 'reply')
})
})
+79 -8
View File
@@ -15,6 +15,11 @@ export { getBashShellReadyRcfileContent } from '../providers/local-pty-shell-rea
import type { OrcaRuntimeService } from '../runtime/orca-runtime'
import type { Store } from '../persistence'
import { retireTerminalSurfaceFromPersistence } from '../runtime/mobile-session-terminal-persistence-retirement'
import {
agentSessionPtyWriteGate,
type AgentSessionPtyWriteAdmittance
} from '../runtime/agent-session-pty-write-gate'
import type { AgentSessionPtyWriteRefusal } from '../../shared/agent-session-pty-write-admission'
import type { GlobalSettings, TuiAgent } from '../../shared/types'
import { toSshExecutionHostId } from '../../shared/execution-host'
import { normalizeRuntimePathForComparison } from '../../shared/cross-platform-path'
@@ -5273,6 +5278,11 @@ export function registerPtyHandlers(
}
},
write: (ptyId, data) => {
// Why: the backstop for every runtime write path — query replies, followups, deliveries —
// so a caller that forgets the typed gate still cannot reach a provider.
if (!admitAgentSessionPtyWrite(ptyId)) {
return false
}
try {
getProviderForPty(ptyId).write(ptyId, data)
return true
@@ -6765,10 +6775,51 @@ export function registerPtyHandlers(
mainWindow.webContents.send('pty:writeUnavailable', { id })
}
// Why: a lease refusal is never a silent drop — it rides the existing write-unavailable channel
// with an additive field, so old renderers keep their current behavior and new ones can name the
// owner. See docs/reference/remote-wire-compatibility.md.
const reportAgentSessionWriteRefusal = (
id: string,
refusal: AgentSessionPtyWriteRefusal
): void => {
if (
mainWindow.isDestroyed() ||
(typeof mainWindow.webContents.isDestroyed === 'function' &&
mainWindow.webContents.isDestroyed())
) {
return
}
mainWindow.webContents.send('pty:writeUnavailable', { id, agentSessionRefusal: refusal })
}
/** Single lease check for every byte-entry point this module owns. */
const admitAgentSessionPtyWrite = (id: string): AgentSessionPtyWriteAdmittance | null => {
const admission = agentSessionPtyWriteGate.admit(id)
if (admission.admitted) {
return { sessionId: admission.sessionId, runtimeFence: admission.runtimeFence }
}
reportAgentSessionWriteRefusal(id, admission.refusal)
return null
}
/** Re-check after a yield: the lease can move to another owner between chunks. */
const readmitAgentSessionPtyWrite = (
id: string,
admitted: AgentSessionPtyWriteAdmittance
): boolean => {
const admission = agentSessionPtyWriteGate.readmit(id, admitted)
if (admission.admitted) {
return true
}
reportAgentSessionWriteRefusal(id, admission.refusal)
return false
}
const writePtyProviderInputWithinLimit = (
provider: IPtyProvider,
id: string,
data: string
data: string,
admitted: AgentSessionPtyWriteAdmittance
): boolean | Promise<boolean> => {
const chunks = iterateTerminalInputChunks(data)
const first = chunks.next()
@@ -6781,21 +6832,27 @@ export function registerPtyHandlers(
provider.write(id, first.value)
return true
}
return writePtyProviderInputChunks(provider, id, chunks, first.value, second.value)
return writePtyProviderInputChunks(provider, id, chunks, first.value, second.value, admitted)
}
const writePtyProviderInput = (
provider: IPtyProvider,
id: string,
data: string
data: string,
admitted: AgentSessionPtyWriteAdmittance
): boolean | Promise<boolean> => {
try {
const tooLarge = isTerminalInputTooLargeWithDeferredMeasurement(data)
if (typeof tooLarge === 'boolean') {
return tooLarge ? false : writePtyProviderInputWithinLimit(provider, id, data)
return tooLarge ? false : writePtyProviderInputWithinLimit(provider, id, data, admitted)
}
return tooLarge
.then((result) => (result ? false : writePtyProviderInputWithinLimit(provider, id, data)))
.then((result) => {
if (result || !readmitAgentSessionPtyWrite(id, admitted)) {
return false
}
return writePtyProviderInputWithinLimit(provider, id, data, admitted)
})
.catch((error) => {
reportUnavailablePtyWrite(id, error)
return false
@@ -6811,12 +6868,18 @@ export function registerPtyHandlers(
id: string,
chunks: Iterator<string>,
firstChunk: string,
secondChunk: string
secondChunk: string,
admitted: AgentSessionPtyWriteAdmittance
): Promise<boolean> => {
try {
let chunk: IteratorResult<string> = { done: false, value: firstChunk }
let nextChunk: IteratorResult<string> = { done: false, value: secondChunk }
let first = true
while (!chunk.done) {
if (!first && !readmitAgentSessionPtyWrite(id, admitted)) {
return false
}
first = false
provider.write(id, chunk.value)
if (!nextChunk.done) {
await new Promise((resolve) => setTimeout(resolve, 0))
@@ -6866,6 +6929,10 @@ export function registerPtyHandlers(
if (runtime?.getDriver(args.id).kind === 'mobile') {
return false
}
const admitted = admitAgentSessionPtyWrite(args.id)
if (!admitted) {
return false
}
const provider = ptyOwnership.has(args.id) ? tryGetProviderForPty(args.id) : undefined
if (!provider) {
return false
@@ -6877,7 +6944,7 @@ export function registerPtyHandlers(
if (visibleRendererPtys.has(args.id)) {
clearHiddenRendererResizeOutput(args.id)
}
return writePtyProviderInput(provider, args.id, args.data)
return writePtyProviderInput(provider, args.id, args.data, admitted)
} catch {
return false
}
@@ -6887,6 +6954,10 @@ export function registerPtyHandlers(
if (runtime?.getDriver(args.id).kind === 'mobile') {
return false
}
const admitted = admitAgentSessionPtyWrite(args.id)
if (!admitted) {
return false
}
// Why: the ack infers Ctrl+C/Escape reached the local PTY; SSH providers are fire-and-forget relay notifications and can't truthfully acknowledge yet.
if (ptyOwnership.get(args.id) !== null) {
return false
@@ -6902,7 +6973,7 @@ export function registerPtyHandlers(
if (visibleRendererPtys.has(args.id)) {
clearHiddenRendererResizeOutput(args.id)
}
return writePtyProviderInput(provider, args.id, args.data)
return writePtyProviderInput(provider, args.id, args.data, admitted)
} catch {
return false
}
@@ -4,6 +4,7 @@ import { describe, expect, it, vi } from 'vitest'
import { PLUGIN_WORKSPACE_TERMINAL_LIMIT } from '../../shared/plugins/plugin-host-api'
import { bindPluginHostServices, type PluginRuntimeDelegate } from './plugin-host-service-bindings'
import { executePluginHostCall, type PluginHostServices } from './plugin-host-methods'
import { AgentSessionPtyWriteRefusedError } from '../../shared/agent-session-pty-write-admission'
function createServices(storageSet: PluginHostServices['storage']['set']): PluginHostServices {
return {
@@ -232,3 +233,34 @@ describe('terminal.sendText explicit worktree routing', () => {
expect(delegate.listTerminals).toHaveBeenCalledTimes(1)
})
})
describe('terminal.sendText under a refusing agent-session lease', () => {
it('reports who holds the session instead of an accepted-looking result', async () => {
const { delegate, services } = createTerminalHarness(['terminal:local:one'])
vi.mocked(delegate.sendTerminal).mockRejectedValue(
new AgentSessionPtyWriteRefusedError({
code: 'agent_session_conflict',
sessionId: 'session-alpha-1',
ownerRuntimeKind: 'native',
handoffStage: null,
ownerPid: 4242,
runtimeFence: 7
})
)
const outcome = await sendTerminalText(services, 'terminal:local:one')
expect(outcome).toMatchObject({ ok: false, code: 'action_failed' })
expect(outcome.ok ? '' : outcome.error).toContain('session-alpha-1')
expect(outcome.ok ? '' : outcome.error).toContain('native chat')
})
it('sends unchanged when no lease refuses, which is every plugin send today', async () => {
const { delegate, services } = createTerminalHarness(['terminal:local:one'])
const outcome = await sendTerminalText(services, 'terminal:local:one')
expect(outcome).toEqual({ ok: true, value: { accepted: true } })
expect(delegate.sendTerminal).toHaveBeenCalledTimes(1)
})
})
@@ -3,6 +3,10 @@ import { PLUGIN_WORKSPACE_TERMINAL_LIMIT } from '../../shared/plugins/plugin-hos
import type { PluginHostServices } from './plugin-host-methods'
import { PluginSecretsStore } from './plugin-secrets-store'
import { PluginKvStore } from './plugin-storage-store'
import {
describeAgentSessionPtyWriteRefusal,
isAgentSessionPtyWriteRefusedError
} from '../../shared/agent-session-pty-write-admission'
/** Structural subset of OrcaRuntimeService exposed to plugin facade bindings. */
export type PluginRuntimeDelegate = {
@@ -59,8 +63,17 @@ export function bindPluginHostServices(input: {
.map((terminal) => ({ id: terminal.handle }))
},
sendTerminalText: async (terminalId, action) => {
const result = await delegate.sendTerminal(terminalId, action)
return { accepted: result.accepted }
try {
const result = await delegate.sendTerminal(terminalId, action)
return { accepted: result.accepted }
} catch (error) {
// Why: the plugin API carries only `accepted`, so a lease refusal would read as a silent
// drop; restate it as the message idiom plugin methods already surface to callers.
if (isAgentSessionPtyWriteRefusedError(error)) {
throw new Error(describeAgentSessionPtyWriteRefusal(error.refusal))
}
throw error
}
},
dispatchPluginNotification: (notification) => delegate.dispatchPluginNotification(notification),
storage: {
@@ -0,0 +1,275 @@
import { afterEach, describe, expect, it, vi } from 'vitest'
import { OrcaRuntimeService } from './orca-runtime'
import { agentSessionPtyWriteGate } from './agent-session-pty-write-gate'
import { getDefaultWorkspaceSession } from '../../shared/constants'
import {
agentSessionLeaseFixture,
agentSessionRecordFixture
} from '../../shared/agent-session-record.test-fixture'
import { TERMINAL_INPUT_CHUNK_MAX_BYTES } from '../../shared/terminal-input'
import type { AgentSessionLease, AgentSessionRecord } from '../../shared/agent-session-record'
import type { WorkspaceSessionState } from '../../shared/types'
// The runtime send paths are the choke point every RPC, plugin, and orchestration write funnels
// through, so each one is proved to consult the lease and to leave unbound PTYs untouched.
const WORKTREE_ID = 'repo-1::/tmp/lease-worktree'
const LEAF_ID = '22222222-2222-4222-8222-222222222222'
const PTY_ID = 'pty-agent-session'
const SESSION_ID = 'session-alpha-1'
function makeStore() {
const session: WorkspaceSessionState = getDefaultWorkspaceSession()
return {
getWorkspaceSession: vi.fn(() => session),
setWorkspaceSession: vi.fn(),
getRepos: vi.fn(() => [
{
id: 'repo-1',
path: '/tmp/lease-worktree',
displayName: 'lease',
badgeColor: '#000000',
addedAt: 0
}
]),
getAllWorktreeMeta: vi.fn(() => ({})),
getWorktreeMeta: vi.fn(() => undefined),
setWorktreeMeta: vi.fn(),
removeWorktreeMeta: vi.fn(),
getSettings: vi.fn(() => ({ workspaceDir: '/tmp/workspaces' })),
getProjects: vi.fn(() => [])
}
}
const records = new Map<string, AgentSessionRecord>()
function publish(lease: AgentSessionLease): void {
records.set(lease.sessionId, agentSessionRecordFixture(lease))
}
async function makeRuntime(options: { onWrite?: (ptyId: string, data: string) => void } = {}) {
const runtime = new OrcaRuntimeService(makeStore() as never)
const write = vi.fn((ptyId: string, data: string) => {
options.onWrite?.(ptyId, data)
return true
})
runtime.setPtyController({
spawn: vi.fn(async () => ({ id: 'never' })),
write,
kill: () => true,
getForegroundProcess: async () => null,
listProcesses: vi.fn(async () => []),
hasPty: () => true
} as never)
runtime.attachWindow(1)
runtime.syncWindowGraph(1, {
tabs: [
{
tabId: 'tab-1',
worktreeId: WORKTREE_ID,
title: 'Agent',
activeLeafId: LEAF_ID,
layout: null
}
],
leaves: [
{
tabId: 'tab-1',
worktreeId: WORKTREE_ID,
leafId: LEAF_ID,
paneRuntimeId: 1,
ptyId: PTY_ID,
paneTitle: null,
title: ''
}
]
})
const { terminals } = await runtime.listTerminals(`id:${WORKTREE_ID}`)
return { runtime, handle: terminals[0].handle, write }
}
function enforce(lease: AgentSessionLease = agentSessionLeaseFixture()): void {
publish(lease)
agentSessionPtyWriteGate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
agentSessionPtyWriteGate.bindPty(PTY_ID, SESSION_ID)
}
afterEach(() => {
agentSessionPtyWriteGate.detachRecordLookup()
records.clear()
})
describe('exemption: nothing that exists today is enforced', () => {
it('sends normally when no session record is bound to the pty', async () => {
const { runtime, handle, write } = await makeRuntime()
await expect(runtime.sendTerminal(handle, { text: 'ls' })).resolves.toMatchObject({
accepted: true
})
expect(write).toHaveBeenCalledWith(PTY_ID, 'ls')
})
it('sends normally when the store is attached but this pty is a legacy agent terminal', async () => {
const { runtime, handle, write } = await makeRuntime()
publish(agentSessionLeaseFixture({ runtimeKind: 'native' }))
agentSessionPtyWriteGate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
agentSessionPtyWriteGate.bindPty('some-other-pty', SESSION_ID)
await expect(runtime.sendTerminal(handle, { text: 'ls' })).resolves.toMatchObject({
accepted: true
})
expect(write).toHaveBeenCalledWith(PTY_ID, 'ls')
})
it('sends agent prompts normally on an unbound pty', async () => {
const { runtime, handle, write } = await makeRuntime()
await expect(runtime.sendTerminalAgentPrompt(handle, 'go')).resolves.toBeDefined()
expect(write).toHaveBeenCalled()
})
})
describe('terminal.send path', () => {
it('writes when the TUI owner holds a proven-live lease', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce()
await expect(runtime.sendTerminal(handle, { text: 'ls' })).resolves.toMatchObject({
accepted: true
})
expect(write).toHaveBeenCalledWith(PTY_ID, 'ls')
})
it('refuses and writes nothing when native chat owns the session', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce(agentSessionLeaseFixture({ runtimeKind: 'native' }))
await expect(runtime.sendTerminal(handle, { text: 'ls' })).rejects.toThrow(
'agent_session_conflict'
)
expect(write).not.toHaveBeenCalled()
})
it('refuses while a handoff is in flight', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce(agentSessionLeaseFixture({ handoffStage: 'new-owner-proving' }))
await expect(runtime.sendTerminal(handle, { text: 'ls' })).rejects.toThrow(
'agent_session_conflict'
)
expect(write).not.toHaveBeenCalled()
})
it('refuses while the lease is unreconciled after a host restart', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce(agentSessionLeaseFixture({ unreconciled: true }))
await expect(runtime.sendTerminal(handle, { text: 'ls' })).rejects.toThrow(
'execution_owner_reconciling'
)
expect(write).not.toHaveBeenCalled()
})
it('refuses an interrupt, which carries no text of its own', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce(agentSessionLeaseFixture({ runtimeKind: 'native' }))
await expect(runtime.sendTerminal(handle, { interrupt: true })).rejects.toThrow(
'agent_session_conflict'
)
expect(write).not.toHaveBeenCalled()
})
it('refuses before the mobile floor is reserved', async () => {
const { runtime, handle } = await makeRuntime()
enforce(agentSessionLeaseFixture({ runtimeKind: 'native' }))
const reserveWrite = vi.fn()
await expect(runtime.sendTerminal(handle, { text: 'ls' }, { reserveWrite })).rejects.toThrow()
expect(reserveWrite).not.toHaveBeenCalled()
})
})
describe('agent prompt path', () => {
it('refuses the whole paste when another runtime owns the session', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce(agentSessionLeaseFixture({ runtimeKind: 'native' }))
await expect(runtime.sendTerminalAgentPrompt(handle, 'do the thing')).rejects.toThrow(
'agent_session_conflict'
)
expect(write).not.toHaveBeenCalled()
})
it('writes the prompt when the TUI owner still holds the lease', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce()
await expect(runtime.sendTerminalAgentPrompt(handle, 'do the thing')).resolves.toBeDefined()
expect(write).toHaveBeenCalled()
})
})
describe('lease transition against an in-flight write', () => {
const CHUNKED_TEXT = 'x'.repeat(TERMINAL_INPUT_CHUNK_MAX_BYTES * 2 + 8)
it('stops a paste mid-flight once the fence advances under it', async () => {
let written = 0
const { runtime, handle, write } = await makeRuntime({
onWrite: () => {
written += 1
if (written === 1) {
// A handoff completed between the first and second chunk.
publish(agentSessionLeaseFixture({ runtimeFence: 8 }))
}
}
})
enforce(agentSessionLeaseFixture({ runtimeFence: 7 }))
await expect(runtime.sendTerminal(handle, { text: CHUNKED_TEXT })).rejects.toThrow(
'agent_session_checkpoint_stale'
)
expect(write).toHaveBeenCalledTimes(1)
})
it('lets a paste finish while the same owner holds the fence', async () => {
const { runtime, handle, write } = await makeRuntime()
enforce()
await expect(runtime.sendTerminal(handle, { text: CHUNKED_TEXT })).resolves.toMatchObject({
accepted: true
})
expect(write).toHaveBeenCalledTimes(3)
})
it('withholds the submit when the lease moves during the text/suffix pause', async () => {
const { runtime, handle, write } = await makeRuntime({
onWrite: (_ptyId, data) => {
if (data === 'ls') {
publish(agentSessionLeaseFixture({ runtimeFence: 8 }))
}
}
})
enforce(agentSessionLeaseFixture({ runtimeFence: 7 }))
await expect(runtime.sendTerminal(handle, { text: 'ls', enter: true })).rejects.toThrow(
'agent_session_checkpoint_stale'
)
expect(write).toHaveBeenCalledTimes(1)
expect(write).not.toHaveBeenCalledWith(PTY_ID, '\r')
})
})
@@ -0,0 +1,147 @@
import { beforeEach, describe, expect, it } from 'vitest'
import { AgentSessionPtyWriteGate } from './agent-session-pty-write-gate'
import {
agentSessionLeaseFixture,
agentSessionRecordFixture
} from '../../shared/agent-session-record.test-fixture'
import { isAgentSessionPtyWriteRefusedError } from '../../shared/agent-session-pty-write-admission'
import type { AgentSessionLease, AgentSessionRecord } from '../../shared/agent-session-record'
const PTY_ID = 'pty-1'
const SESSION_ID = 'session-alpha-1'
let gate: AgentSessionPtyWriteGate
let records: Map<string, AgentSessionRecord>
function publish(lease: AgentSessionLease): void {
records.set(lease.sessionId, agentSessionRecordFixture(lease))
}
beforeEach(() => {
gate = new AgentSessionPtyWriteGate()
records = new Map()
})
describe('capability invisibility', () => {
it('admits every PTY while nothing is bound, which is the shape of today builds', () => {
gate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
expect(gate.enforcing).toBe(false)
expect(gate.admit('any-shell')).toEqual({
admitted: true,
sessionId: null,
runtimeFence: null
})
})
it('admits an unbound PTY even once another PTY is bound and refusing', () => {
gate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
publish(agentSessionLeaseFixture({ runtimeKind: 'native' }))
gate.bindPty(PTY_ID, SESSION_ID)
expect(gate.admit(PTY_ID).admitted).toBe(false)
expect(gate.admit('ordinary-shell').admitted).toBe(true)
})
it('admits a bound PTY while no store is attached, so a half-wired host cannot refuse', () => {
gate.bindPty(PTY_ID, SESSION_ID)
expect(gate.enforcing).toBe(false)
expect(gate.admit(PTY_ID).admitted).toBe(true)
})
})
describe('binding lifecycle', () => {
beforeEach(() => {
gate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
})
it('reports the session a PTY is bound to', () => {
gate.bindPty(PTY_ID, SESSION_ID)
expect(gate.boundSessionId(PTY_ID)).toBe(SESSION_ID)
expect(gate.boundSessionId('other')).toBeNull()
})
it('returns an unbound PTY to the exempt path when it is unbound', () => {
publish(agentSessionLeaseFixture({ runtimeKind: 'native' }))
gate.bindPty(PTY_ID, SESSION_ID)
expect(gate.admit(PTY_ID).admitted).toBe(false)
gate.unbindPty(PTY_ID)
expect(gate.admit(PTY_ID).admitted).toBe(true)
})
it('drops every binding when the store detaches', () => {
publish(agentSessionLeaseFixture({ runtimeKind: 'native' }))
gate.bindPty(PTY_ID, SESSION_ID)
gate.detachRecordLookup()
expect(gate.enforcing).toBe(false)
expect(gate.admit(PTY_ID).admitted).toBe(true)
})
})
describe('admission through the store', () => {
beforeEach(() => {
gate.attachRecordLookup((sessionId) => records.get(sessionId) ?? null)
gate.bindPty(PTY_ID, SESSION_ID)
})
it('admits a proven-live TUI owner and reports the fence it was admitted under', () => {
publish(agentSessionLeaseFixture({ runtimeFence: 11 }))
expect(gate.admit(PTY_ID)).toEqual({
admitted: true,
sessionId: SESSION_ID,
runtimeFence: 11
})
})
it('refuses when the record vanished from the store', () => {
const admission = gate.admit(PTY_ID)
expect(admission.admitted).toBe(false)
})
it('throws the typed refusal from assertAdmitted', () => {
publish(agentSessionLeaseFixture({ runtimeKind: 'native' }))
let thrown: unknown = null
try {
gate.assertAdmitted(PTY_ID)
} catch (error) {
thrown = error
}
expect(isAgentSessionPtyWriteRefusedError(thrown)).toBe(true)
if (!isAgentSessionPtyWriteRefusedError(thrown)) {
return
}
expect(thrown.refusal.code).toBe('agent_session_conflict')
expect(thrown.refusal.ownerRuntimeKind).toBe('native')
})
it('returns the admittance from assertAdmitted so later chunks can be fenced', () => {
publish(agentSessionLeaseFixture({ runtimeFence: 3 }))
expect(gate.assertAdmitted(PTY_ID)).toEqual({ sessionId: SESSION_ID, runtimeFence: 3 })
})
it('throws from assertReadmitted once the lease moved under an in-flight write', () => {
publish(agentSessionLeaseFixture({ runtimeFence: 3 }))
const admitted = gate.assertAdmitted(PTY_ID)
publish(agentSessionLeaseFixture({ runtimeFence: 4 }))
expect(() => gate.assertReadmitted(PTY_ID, admitted)).toThrowError(
'agent_session_checkpoint_stale'
)
})
it('lets an in-flight write finish while the lease is unchanged', () => {
publish(agentSessionLeaseFixture({ runtimeFence: 3 }))
const admitted = gate.assertAdmitted(PTY_ID)
expect(() => gate.assertReadmitted(PTY_ID, admitted)).not.toThrow()
})
it('leaves an in-flight write on an exempt PTY alone', () => {
const admitted = gate.assertAdmitted('ordinary-shell')
expect(admitted).toEqual({ sessionId: null, runtimeFence: null })
expect(() => gate.assertReadmitted('ordinary-shell', admitted)).not.toThrow()
})
it('refuses an exempt write that acquired a refusing binding mid-flight', () => {
const admitted = gate.assertAdmitted('late-bound')
publish(agentSessionLeaseFixture({ sessionId: 'session-beta-2', runtimeKind: 'native' }))
gate.bindPty('late-bound', 'session-beta-2')
expect(() => gate.assertReadmitted('late-bound', admitted)).toThrow()
})
})
@@ -0,0 +1,109 @@
/**
* Host-side registry that maps a live PTY to the durable agent session it belongs to, and answers
* whether bytes may enter that PTY right now.
*
* It lives on the execution host that owns the process, never in a client: a paired desktop, a
* mobile client, and an SSH-attached Orca all reach the same gate through the host that spawned
* the PTY. With nothing bound — the state of every build until a later part registers records —
* `admit` short-circuits to admitted, so no existing terminal path changes behavior or cost.
*/
import type { AgentSessionRecord } from '../../shared/agent-session-record'
import {
AgentSessionPtyWriteRefusedError,
evaluateAgentSessionPtyWriteAdmission,
reevaluateAgentSessionPtyWriteAdmission,
type AgentSessionPtyBinding,
type AgentSessionPtyWriteAdmission
} from '../../shared/agent-session-pty-write-admission'
export type AgentSessionRecordLookup = (sessionId: string) => AgentSessionRecord | null
/** What an admitted write carries forward so its later chunks can be fenced against the same lease. */
export type AgentSessionPtyWriteAdmittance = {
sessionId: string | null
runtimeFence: number | null
}
const ADMITTED_UNBOUND: AgentSessionPtyWriteAdmission = {
admitted: true,
sessionId: null,
runtimeFence: null
}
export class AgentSessionPtyWriteGate {
private readonly sessionIdByPtyId = new Map<string, string>()
private lookup: AgentSessionRecordLookup | null = null
/** Point the gate at the durable store. Until this is called nothing can be enforced. */
attachRecordLookup(lookup: AgentSessionRecordLookup): void {
this.lookup = lookup
}
detachRecordLookup(): void {
this.lookup = null
this.sessionIdByPtyId.clear()
}
bindPty(ptyId: string, sessionId: string): void {
this.sessionIdByPtyId.set(ptyId, sessionId)
}
unbindPty(ptyId: string): void {
this.sessionIdByPtyId.delete(ptyId)
}
boundSessionId(ptyId: string): string | null {
return this.sessionIdByPtyId.get(ptyId) ?? null
}
/** False while no PTY is bound, which is every write path in today's builds. */
get enforcing(): boolean {
return this.lookup !== null && this.sessionIdByPtyId.size > 0
}
admit(ptyId: string): AgentSessionPtyWriteAdmission {
if (!this.enforcing) {
return ADMITTED_UNBOUND
}
return evaluateAgentSessionPtyWriteAdmission(this.binding(ptyId))
}
/** Re-check a write already in flight against the fence it was admitted under. */
readmit(ptyId: string, admitted: AgentSessionPtyWriteAdmittance): AgentSessionPtyWriteAdmission {
if (admitted.sessionId === null && !this.enforcing) {
return ADMITTED_UNBOUND
}
return reevaluateAgentSessionPtyWriteAdmission({ admitted, binding: this.binding(ptyId) })
}
/** Admit or throw the typed refusal. Used where the caller already reports errors to a client. */
assertAdmitted(ptyId: string): AgentSessionPtyWriteAdmittance {
const admission = this.admit(ptyId)
if (!admission.admitted) {
throw new AgentSessionPtyWriteRefusedError(admission.refusal)
}
return { sessionId: admission.sessionId, runtimeFence: admission.runtimeFence }
}
assertReadmitted(ptyId: string, admitted: AgentSessionPtyWriteAdmittance): void {
const admission = this.readmit(ptyId, admitted)
if (!admission.admitted) {
throw new AgentSessionPtyWriteRefusedError(admission.refusal)
}
}
private binding(ptyId: string): AgentSessionPtyBinding | null {
const sessionId = this.sessionIdByPtyId.get(ptyId)
if (sessionId === undefined) {
return null
}
return { sessionId, record: this.lookup?.(sessionId) ?? null }
}
}
/**
* One gate per host process. Both the runtime's send paths and the PTY IPC layer consult this same
* instance, so a write cannot reach a provider by entering through the other door.
*/
export const agentSessionPtyWriteGate = new AgentSessionPtyWriteGate()
+30 -2
View File
@@ -69,6 +69,10 @@ import {
createEphemeralAgentSessionClaimSigner,
type AgentSessionClaimSigner
} from './agent-session-claim-identity'
import {
agentSessionPtyWriteGate,
type AgentSessionPtyWriteAdmittance
} from './agent-session-pty-write-gate'
import {
hasCompatibleAgentTitleIdentity,
normalizeCompatibleAgentStatusEntryForOwner,
@@ -17242,18 +17246,24 @@ export class OrcaRuntimeService {
suffixFailureError?: string
} = {}
): Promise<void> {
// Why: the lease is checked before the mobile floor is reserved, so a refused send never takes
// a claim it will not use.
const admitted = agentSessionPtyWriteGate.assertAdmitted(ptyId)
// Why: direct terminal.send can carry paste-sized text from RPC/mobile
// clients; chunk text before PTY/ConPTY while preserving suffix separation.
const hasText = typeof action.text === 'string' && action.text.length > 0
const hasSuffix = action.enter || action.interrupt
if (hasText) {
await this.writeTerminalInputChunks(ptyId, action.text!, options)
await this.writeTerminalInputChunks(ptyId, action.text!, options, admitted)
}
if (hasSuffix) {
const suffix = (action.enter ? '\r' : '') + (action.interrupt ? '\x03' : '')
if (hasText) {
await new Promise((resolve) => setTimeout(resolve, 500))
}
// Why: the 500ms text/suffix pause is long enough for a handoff to complete, so the submit
// is re-checked against the fence the text was admitted under.
agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted)
try {
await options.beforeWrite?.(ptyId)
options.reserveWrite?.(ptyId)
@@ -17290,11 +17300,19 @@ export class OrcaRuntimeService {
beforeWrite?: (ptyId: string) => void | Promise<void>
reserveWrite?: (ptyId: string) => void
afterWrite?: (ptyId: string) => void | Promise<void>
} = {}
} = {},
admitted: AgentSessionPtyWriteAdmittance = { sessionId: null, runtimeFence: null }
): Promise<void> {
const chunks = iterateTerminalInputChunks(text)
let chunk = chunks.next()
let firstChunk = true
while (!chunk.done) {
// Why: every inter-chunk yield is a window for a handoff to take the lease; the rest of a
// paste must not land in a session this runtime no longer owns.
if (!firstChunk) {
agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted)
}
firstChunk = false
await options.beforeWrite?.(ptyId)
options.reserveWrite?.(ptyId)
const wrote = this.ptyController?.write(ptyId, chunk.value) ?? false
@@ -17317,12 +17335,18 @@ export class OrcaRuntimeService {
suffixFailureError?: string
} = {}
): Promise<void> {
const admitted = agentSessionPtyWriteGate.assertAdmitted(ptyId)
let wrotePasteBytes = false
let completedPaste = false
try {
const chunks = iterateTerminalInputChunks(pastePayload)
let chunk = chunks.next()
let firstChunk = true
while (!chunk.done) {
if (!firstChunk) {
agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted)
}
firstChunk = false
await options.beforeWrite?.(ptyId)
const wrote = this.ptyController?.write(ptyId, chunk.value) ?? false
if (!wrote) {
@@ -17337,12 +17361,16 @@ export class OrcaRuntimeService {
completedPaste = true
} catch (error) {
if (wrotePasteBytes && !completedPaste) {
// 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.
this.ptyController?.write(ptyId, AGENT_PROMPT_BRACKETED_PASTE_END)
}
throw error
}
await new Promise((resolve) => setTimeout(resolve, AGENT_PROMPT_SUBMIT_DELAY_MS))
agentSessionPtyWriteGate.assertReadmitted(ptyId, admitted)
try {
await options.beforeWrite?.(ptyId)
} catch (error) {
+18
View File
@@ -74,6 +74,7 @@ import {
import type { TerminalSnapshotUnavailableReason } from '../../../../shared/terminal-snapshot-unavailability'
import type { RemoteTerminalSourceRangeReplacementReservation } from '../../remote-terminal-source-range-consumer'
import { withTerminalCloseAttribution } from '../terminal-close-attribution'
import { isAgentSessionPtyWriteRefusedError } from '../../../../shared/agent-session-pty-write-admission'
const REQUESTED_SNAPSHOT_BYTE_BUDGET = 2 * 1024 * 1024
const TERMINAL_OUTPUT_FLUSH_MS = 5
@@ -319,6 +320,11 @@ function resolveMobileFloorClientId(
type TerminalStreamInputOutcome = 'delivered' | 'rejected' | 'failed'
function isTerminalStreamInputRejection(error: unknown): boolean {
// Why: a lease refusal is a deliberate rejection, not a transport failure, so the stream reports
// it through the WriteUnavailable frame old clients already decode rather than a new opcode.
if (isAgentSessionPtyWriteRefusedError(error)) {
return true
}
const message = error instanceof Error ? error.message : String(error)
return message.includes('terminal_not_writable') || message.includes('terminal_handle_stale')
}
@@ -1368,6 +1374,18 @@ export const TERMINAL_METHODS: RpcAnyMethod[] = [
)
} catch (error) {
mobileFloorClaim.current?.rollback()
if (isAgentSessionPtyWriteRefusedError(error)) {
// Why: name the owner and the stage instead of a bare not-writable, so a client can say
// who holds the session rather than retrying into a lease it will never win.
return {
send: {
handle: params.terminal,
accepted: false,
bytesWritten: 0,
agentSessionRefusal: error.refusal
}
}
}
const refusedReason = getTerminalSendGuardRefusedReason(error)
if (refusedReason) {
return {
@@ -0,0 +1,106 @@
import { describe, expect, it, vi } from 'vitest'
import { RpcDispatcher } from './dispatcher'
import { TERMINAL_METHODS } from './methods/terminal'
import type { RpcRequest } from './core'
import type { OrcaRuntimeService } from '../orca-runtime'
import {
AgentSessionPtyWriteRefusedError,
type AgentSessionPtyWriteRefusal
} from '../../../shared/agent-session-pty-write-admission'
// terminal.send is the one write path a paired client can reach, so a lease refusal has to arrive
// as a result it can render — not as a transport error and not as a silent `accepted: false`.
const REFUSAL: AgentSessionPtyWriteRefusal = {
code: 'agent_session_conflict',
sessionId: 'session-alpha-1',
ownerRuntimeKind: 'native',
handoffStage: 'preparing',
ownerPid: 4242,
runtimeFence: 7
}
const rollback = vi.fn()
function stubRuntime(overrides: Partial<OrcaRuntimeService> = {}): OrcaRuntimeService {
return {
getRuntimeId: () => 'test-runtime',
resolveLiveLeafForHandle: vi.fn().mockReturnValue({ ptyId: 'pty-1' }),
getDriver: vi.fn().mockReturnValue({ kind: 'idle' }),
beginMobileInputFloor: vi.fn(() => ({ commit: async () => {}, rollback })),
...overrides
} as OrcaRuntimeService
}
function makeRequest(params: unknown): RpcRequest {
return { id: 'req-1', authToken: 'tok', method: 'terminal.send', params }
}
async function send(runtime: OrcaRuntimeService, client: { id: string; type: string }) {
const dispatcher = new RpcDispatcher({ runtime, methods: TERMINAL_METHODS })
return await dispatcher.dispatch(makeRequest({ terminal: 'terminal-1', text: 'hello', client }))
}
describe('terminal.send under a refusing lease', () => {
it('returns the typed refusal instead of failing the call', async () => {
const runtime = stubRuntime({
sendTerminal: vi.fn().mockRejectedValue(new AgentSessionPtyWriteRefusedError(REFUSAL))
})
const response = await send(runtime, { id: 'desktop-1', type: 'desktop' })
expect(response.ok).toBe(true)
if (!response.ok) {
throw new Error(response.error.message)
}
expect(response.result).toEqual({
send: {
handle: 'terminal-1',
accepted: false,
bytesWritten: 0,
agentSessionRefusal: REFUSAL
}
})
})
it('keeps the refusal additive so an old client still reads accepted: false', async () => {
const runtime = stubRuntime({
sendTerminal: vi.fn().mockRejectedValue(new AgentSessionPtyWriteRefusedError(REFUSAL))
})
const response = await send(runtime, { id: 'desktop-1', type: 'desktop' })
expect(response.ok).toBe(true)
if (!response.ok) {
throw new Error(response.error.message)
}
const result = response.result as { send: { accepted: boolean; refusedReason?: string } }
expect(result.send.accepted).toBe(false)
// A new `refusedReason` value would reach old clients as an unknown enum member.
expect(result.send.refusedReason).toBeUndefined()
})
it('releases the mobile input floor a refused send never used', async () => {
rollback.mockClear()
const runtime = stubRuntime({
sendTerminal: vi.fn().mockImplementation(async (_handle, _action, options) => {
options?.reserveWrite?.('pty-1')
throw new AgentSessionPtyWriteRefusedError(REFUSAL)
})
})
await send(runtime, { id: 'mobile-1', type: 'mobile' })
expect(rollback).toHaveBeenCalled()
})
it('still surfaces unrelated send failures as errors', async () => {
const runtime = stubRuntime({
sendTerminal: vi.fn().mockRejectedValue(new Error('terminal_not_writable'))
})
const response = await send(runtime, { id: 'desktop-1', type: 'desktop' })
expect(response.ok).toBe(false)
})
})
+12 -3
View File
@@ -14,6 +14,7 @@ import type {
TerminalPreviewConnectResult,
TerminalPreviewDataPayload
} from '../shared/terminal-preview'
import type { AgentSessionPtyWriteRefusal } from '../shared/agent-session-pty-write-admission'
import type { CliInstallStatus } from '../shared/cli-install-types'
import type { AgentHookInstallStatus } from '../shared/agent-hook-types'
import type { CodexConfigSyncStatus } from '../shared/codex-config-sync-types'
@@ -974,9 +975,17 @@ const api = {
},
writeAccepted: (id: string, data: string): Promise<boolean> =>
ipcRenderer.invoke('pty:writeAccepted', { id, data }),
onWriteUnavailable: (callback: (payload: { id: string }) => void): (() => void) => {
const handler = (_event: Electron.IpcRendererEvent, payload: { id: string }): void =>
callback(payload)
onWriteUnavailable: (
callback: (payload: {
id: string
/** Set only when a durable agent-session lease refused the write; absent otherwise. */
agentSessionRefusal?: AgentSessionPtyWriteRefusal
}) => void
): (() => void) => {
const handler = (
_event: Electron.IpcRendererEvent,
payload: { id: string; agentSessionRefusal?: AgentSessionPtyWriteRefusal }
): void => callback(payload)
ipcRenderer.on('pty:writeUnavailable', handler)
return () => ipcRenderer.removeListener('pty:writeUnavailable', handler)
},
@@ -0,0 +1,247 @@
import { describe, expect, it } from 'vitest'
import {
AgentSessionPtyWriteRefusedError,
describeAgentSessionPtyWriteRefusal,
evaluateAgentSessionPtyWriteAdmission,
isAgentSessionPtyWriteRefusedError,
reevaluateAgentSessionPtyWriteAdmission
} from './agent-session-pty-write-admission'
import {
agentSessionLeaseFixture,
agentSessionRecordFixture
} from './agent-session-record.test-fixture'
import type { AgentSessionLease } from './agent-session-record'
import { AGENT_SESSION_RPC_ERROR_CODES } from './agent-session-host-authority'
function bindingFor(lease: AgentSessionLease) {
return { sessionId: lease.sessionId, record: agentSessionRecordFixture(lease) }
}
describe('exemptions', () => {
it('admits a PTY with no binding, which is every ordinary shell and legacy agent terminal', () => {
const admission = evaluateAgentSessionPtyWriteAdmission(null)
expect(admission).toEqual({ admitted: true, sessionId: null, runtimeFence: null })
})
it('admits a proven-live TUI owner', () => {
const lease = agentSessionLeaseFixture()
expect(evaluateAgentSessionPtyWriteAdmission(bindingFor(lease))).toEqual({
admitted: true,
sessionId: lease.sessionId,
runtimeFence: lease.runtimeFence
})
})
})
describe('refusal matrix', () => {
const cases: { name: string; lease: Partial<AgentSessionLease>; code: string }[] = [
{
name: 'native chat owns the session',
lease: { runtimeKind: 'native' },
code: 'agent_session_conflict'
},
{
name: 'a handoff is preparing',
lease: { handoffStage: 'preparing' },
code: 'agent_session_conflict'
},
{
name: 'the old owner stopped mid-handoff',
lease: { handoffStage: 'old-owner-stopped' },
code: 'agent_session_conflict'
},
{
name: 'the new owner has not proved itself',
lease: { handoffStage: 'new-owner-proving' },
code: 'agent_session_conflict'
},
{
name: 'two claims collided',
lease: { claimStatus: 'conflicted' },
code: 'agent_session_conflict'
},
{
name: 'the lease is unreconciled after a host restart',
lease: { unreconciled: true },
code: 'execution_owner_reconciling'
},
{
name: 'recovery is running',
lease: { handoffStage: 'recovering' },
code: 'execution_owner_reconciling'
},
{
name: 'a human must finish recovery',
lease: { handoffStage: 'manual-recovery' },
code: 'execution_owner_reconciling'
},
{
name: 'the claim is reserved but no process exists yet',
lease: { claimStatus: 'reserved', ownerProcess: null },
code: 'agent_session_ownership_unknown'
},
{
name: 'the owner released the session',
lease: { claimStatus: 'released' },
code: 'agent_session_ownership_unknown'
}
]
for (const testCase of cases) {
it(`refuses when ${testCase.name}`, () => {
const lease = agentSessionLeaseFixture(testCase.lease)
const admission = evaluateAgentSessionPtyWriteAdmission(bindingFor(lease))
expect(admission.admitted).toBe(false)
if (admission.admitted) {
return
}
expect(admission.refusal.code).toBe(testCase.code)
expect(admission.refusal.sessionId).toBe(lease.sessionId)
expect(admission.refusal.ownerRuntimeKind).toBe(lease.runtimeKind)
expect(admission.refusal.handoffStage).toBe(lease.handoffStage)
expect(admission.refusal.runtimeFence).toBe(lease.runtimeFence)
})
}
it('distinguishes a reconciling window from a session another runtime owns', () => {
// Phase-2 clients render "recovering, retry shortly" only when these two do not collapse.
const reconciling = evaluateAgentSessionPtyWriteAdmission(
bindingFor(agentSessionLeaseFixture({ unreconciled: true }))
)
const owned = evaluateAgentSessionPtyWriteAdmission(
bindingFor(agentSessionLeaseFixture({ runtimeKind: 'native' }))
)
expect(reconciling.admitted).toBe(false)
expect(owned.admitted).toBe(false)
if (reconciling.admitted || owned.admitted) {
return
}
expect(reconciling.refusal.code).not.toBe(owned.refusal.code)
})
it('fails closed when a bound PTY has no readable record', () => {
const admission = evaluateAgentSessionPtyWriteAdmission({
sessionId: 'session-alpha-1',
record: null
})
expect(admission.admitted).toBe(false)
if (admission.admitted) {
return
}
expect(admission.refusal.code).toBe('execution_owner_reconciling')
expect(admission.refusal.ownerRuntimeKind).toBeNull()
})
it('fails closed when the record answers for a different session', () => {
const admission = evaluateAgentSessionPtyWriteAdmission({
sessionId: 'session-beta-2',
record: agentSessionRecordFixture()
})
expect(admission.admitted).toBe(false)
if (admission.admitted) {
return
}
expect(admission.refusal.code).toBe('agent_session_ownership_unknown')
expect(admission.refusal.sessionId).toBe('session-beta-2')
})
it('only emits codes old clients already decode', () => {
const emitted = new Set(
cases.map((testCase) => {
const admission = evaluateAgentSessionPtyWriteAdmission(
bindingFor(agentSessionLeaseFixture(testCase.lease))
)
return admission.admitted ? 'admitted' : admission.refusal.code
})
)
for (const code of emitted) {
expect(AGENT_SESSION_RPC_ERROR_CODES).toContain(code)
}
})
})
describe('in-flight fence race', () => {
const admittedLease = agentSessionLeaseFixture()
const admitted = { sessionId: admittedLease.sessionId, runtimeFence: admittedLease.runtimeFence }
it('lets the rest of a write land while the same fence still holds', () => {
const next = reevaluateAgentSessionPtyWriteAdmission({
admitted,
binding: bindingFor(admittedLease)
})
expect(next.admitted).toBe(true)
})
it('refuses the rest of a write once the fence advanced under it', () => {
// A handoff completed between two chunks: the new owner's lease would otherwise admit them.
const moved = agentSessionLeaseFixture({ runtimeFence: admittedLease.runtimeFence + 1 })
const next = reevaluateAgentSessionPtyWriteAdmission({ admitted, binding: bindingFor(moved) })
expect(next.admitted).toBe(false)
if (next.admitted) {
return
}
expect(next.refusal.code).toBe('agent_session_checkpoint_stale')
expect(next.refusal.runtimeFence).toBe(moved.runtimeFence)
})
it('refuses the rest of a write when the PTY was rebound to another session', () => {
const other = agentSessionLeaseFixture({ sessionId: 'session-beta-2' })
const next = reevaluateAgentSessionPtyWriteAdmission({ admitted, binding: bindingFor(other) })
expect(next.admitted).toBe(false)
})
it('refuses the rest of a write when the binding disappeared mid-flight', () => {
const next = reevaluateAgentSessionPtyWriteAdmission({ admitted, binding: null })
expect(next.admitted).toBe(false)
})
it('refuses the rest of a write when the lease stopped admitting a writer', () => {
const next = reevaluateAgentSessionPtyWriteAdmission({
admitted,
binding: bindingFor(agentSessionLeaseFixture({ handoffStage: 'preparing' }))
})
expect(next.admitted).toBe(false)
if (next.admitted) {
return
}
expect(next.refusal.code).toBe('agent_session_conflict')
})
it('judges a write admitted while unbound on whatever binding appeared', () => {
const next = reevaluateAgentSessionPtyWriteAdmission({
admitted: { sessionId: null, runtimeFence: null },
binding: bindingFor(agentSessionLeaseFixture({ runtimeKind: 'native' }))
})
expect(next.admitted).toBe(false)
})
})
describe('typed error', () => {
it('carries the refusal and reports its code as the message', () => {
const admission = evaluateAgentSessionPtyWriteAdmission(
bindingFor(agentSessionLeaseFixture({ runtimeKind: 'native' }))
)
if (admission.admitted) {
throw new Error('expected a refusal')
}
const error = new AgentSessionPtyWriteRefusedError(admission.refusal)
expect(isAgentSessionPtyWriteRefusedError(error)).toBe(true)
expect(isAgentSessionPtyWriteRefusedError(new Error('agent_session_conflict'))).toBe(false)
expect(error.message).toBe('agent_session_conflict')
expect(error.refusal).toEqual(admission.refusal)
})
it('describes who holds the session and what stage it is in', () => {
const admission = evaluateAgentSessionPtyWriteAdmission(
bindingFor(agentSessionLeaseFixture({ runtimeKind: 'native', handoffStage: 'preparing' }))
)
if (admission.admitted) {
throw new Error('expected a refusal')
}
const described = describeAgentSessionPtyWriteRefusal(admission.refusal)
expect(described).toContain('session-alpha-1')
expect(described).toContain('native chat')
expect(described).toContain('pid 4242')
expect(described).toContain('preparing')
})
})
@@ -0,0 +1,177 @@
/**
* PTY-write admission for agent sessions that have a durable record.
*
* A PTY is the TUI runtime's write surface, so bytes may enter it only while the session's lease
* admits a writer and that writer is the TUI. The admit decision is `agentSessionLeaseAdmitsWriter`
* verbatim; everything here only decides whether the lease is even the TUI's to hold, and — once
* that helper has already refused — which refusal a client should be shown.
*
* A PTY with no binding is a session this host knows nothing about: every ordinary shell and every
* legacy agent terminal. Those are never consulted and never refused. A binding whose record is
* missing is the opposite case — the record was lost, not absent — and fails closed.
*/
import {
agentSessionLeaseAdmitsWriter,
isAgentSessionFenceCurrent
} from './agent-session-lease-adjudication'
import type {
AgentSessionHandoffStage,
AgentSessionLease,
AgentSessionOwnerRuntimeKind,
AgentSessionRecord
} from './agent-session-record'
/**
* Every code is already in `AGENT_SESSION_RPC_ERROR_CODES`, so a refusal reaching an old client
* carries a code it has seen since the host-authority release rather than a new one.
*/
export type AgentSessionPtyWriteRefusalCode =
| 'agent_session_conflict'
| 'agent_session_ownership_unknown'
| 'agent_session_checkpoint_stale'
| 'execution_owner_reconciling'
export type AgentSessionPtyWriteRefusal = {
code: AgentSessionPtyWriteRefusalCode
sessionId: string
/** Runtime the lease names as owner; null once the record is gone. */
ownerRuntimeKind: AgentSessionOwnerRuntimeKind | null
handoffStage: AgentSessionHandoffStage | null
/** Pid the lease names, so the refusal can say who holds the session. */
ownerPid: number | null
runtimeFence: number | null
}
export type AgentSessionPtyWriteAdmission =
| { admitted: true; sessionId: string | null; runtimeFence: number | null }
| { admitted: false; refusal: AgentSessionPtyWriteRefusal }
/** One PTY's durable binding: the session it belongs to and that session's record, if readable. */
export type AgentSessionPtyBinding = {
sessionId: string
record: AgentSessionRecord | null
}
const UNBOUND_ADMISSION: AgentSessionPtyWriteAdmission = {
admitted: true,
sessionId: null,
runtimeFence: null
}
/** Runs only after `agentSessionLeaseAdmitsWriter` (or the runtime-kind test) already refused. */
function classifyRefusal(lease: AgentSessionLease): AgentSessionPtyWriteRefusalCode {
if (lease.unreconciled) {
return 'execution_owner_reconciling'
}
if (lease.claimStatus === 'conflicted') {
return 'agent_session_conflict'
}
if (lease.handoffStage === 'recovering' || lease.handoffStage === 'manual-recovery') {
return 'execution_owner_reconciling'
}
if (lease.handoffStage !== null || lease.runtimeKind !== 'tui') {
// Why: a live native owner and a mid-flight handoff are both "someone else holds it", which is
// actionable in a way "we cannot tell" is not.
return 'agent_session_conflict'
}
return 'agent_session_ownership_unknown'
}
function refuse(
code: AgentSessionPtyWriteRefusalCode,
sessionId: string,
lease: AgentSessionLease | null
): AgentSessionPtyWriteAdmission {
return {
admitted: false,
refusal: {
code,
sessionId,
ownerRuntimeKind: lease?.runtimeKind ?? null,
handoffStage: lease?.handoffStage ?? null,
ownerPid: lease?.ownerProcess?.pid ?? null,
runtimeFence: lease?.runtimeFence ?? null
}
}
}
export function evaluateAgentSessionPtyWriteAdmission(
binding: AgentSessionPtyBinding | null
): AgentSessionPtyWriteAdmission {
if (!binding) {
return UNBOUND_ADMISSION
}
const record = binding.record
if (!record) {
// Why: a bound PTY whose record cannot be read is a lost lease, not an unmanaged shell.
return refuse('execution_owner_reconciling', binding.sessionId, null)
}
if (record.sessionId !== binding.sessionId) {
return refuse('agent_session_ownership_unknown', binding.sessionId, record.lease)
}
const lease = record.lease
if (lease.runtimeKind === 'tui' && agentSessionLeaseAdmitsWriter(lease)) {
return { admitted: true, sessionId: record.sessionId, runtimeFence: lease.runtimeFence }
}
return refuse(classifyRefusal(lease), binding.sessionId, lease)
}
/**
* Re-admit a write that already began. Chunked input and the text/suffix pause both yield, so a
* lease transition can land between two writes of one logical send; the fence observed at
* admission is what the remaining bytes are checked against.
*/
export function reevaluateAgentSessionPtyWriteAdmission(args: {
admitted: { sessionId: string | null; runtimeFence: number | null }
binding: AgentSessionPtyBinding | null
}): AgentSessionPtyWriteAdmission {
const { admitted, binding } = args
const next = evaluateAgentSessionPtyWriteAdmission(binding)
if (admitted.sessionId === null || admitted.runtimeFence === null) {
// Why: an unbound write that acquires a binding mid-flight is judged on the new binding alone.
return next
}
if (!next.admitted) {
return next
}
const lease = binding?.record?.lease ?? null
if (
next.sessionId !== admitted.sessionId ||
!lease ||
!isAgentSessionFenceCurrent(lease, admitted.runtimeFence)
) {
return refuse('agent_session_checkpoint_stale', admitted.sessionId, lease)
}
return next
}
export class AgentSessionPtyWriteRefusedError extends Error {
readonly refusal: AgentSessionPtyWriteRefusal
constructor(refusal: AgentSessionPtyWriteRefusal) {
// Why: callers that already switch on `error.message` as an RPC code keep working unchanged.
super(refusal.code)
this.name = 'AgentSessionPtyWriteRefusedError'
this.refusal = refusal
}
}
export function isAgentSessionPtyWriteRefusedError(
error: unknown
): error is AgentSessionPtyWriteRefusedError {
return error instanceof AgentSessionPtyWriteRefusedError
}
/** Human-readable refusal, so a client that only surfaces a message still names the owner. */
export function describeAgentSessionPtyWriteRefusal(refusal: AgentSessionPtyWriteRefusal): string {
const owner =
refusal.ownerRuntimeKind === null
? 'no recorded owner'
: `${refusal.ownerRuntimeKind === 'native' ? 'native chat' : 'the agent TUI'}${
refusal.ownerPid === null ? '' : ` (pid ${refusal.ownerPid})`
}`
const stage =
refusal.handoffStage === null ? 'no handoff in progress' : `handoff ${refusal.handoffStage}`
return `Agent session ${refusal.sessionId} is held by ${owner}; ${stage} (${refusal.code}).`
}
@@ -0,0 +1,63 @@
/** Durable-record fixtures shared by the write-admission tests across shared, runtime, and IPC. */
import type { AgentSessionLease, AgentSessionRecord } from './agent-session-record'
const OWNER_PROCESS = {
hostId: 'local',
pid: 4242,
processStartTimeMs: 1_700_000_000_000,
spawnToken: 'spawn-tui'
}
/** A proven-live TUI owner: the only lease state that admits a PTY write. */
export function agentSessionLeaseFixture(
overrides: Partial<AgentSessionLease> = {}
): AgentSessionLease {
return {
sessionId: 'session-alpha-1',
runtimeKind: 'tui',
runtimeFence: 7,
handoffStage: null,
provenHandleLinkId: 'link-1',
ownerProcess: OWNER_PROCESS,
reservedSpawnToken: 'spawn-tui',
leaseDeadlineAt: 60_000,
lastRenewedAt: 30_000,
handoffOperationId: null,
journalCheckpoint: null,
claimKeyId: 'key-1',
claimStatus: 'live',
unreconciled: false,
deathEvidence: null,
...overrides
}
}
export function agentSessionRecordFixture(
lease: AgentSessionLease = agentSessionLeaseFixture()
): AgentSessionRecord {
return {
schemaVersion: 1,
sessionId: lease.sessionId,
location: {
executionHostId: 'local',
wslDistro: null,
workspaceId: 'workspace-1',
workspaceKind: 'git-worktree'
},
provider: 'claude',
providerHandleChain: [
{
linkId: 'link-1',
origin: 'created',
mintedAtFence: lease.runtimeFence,
observedAt: 1_000,
handle: { provider: 'claude', sessionId: lease.sessionId, leafUuid: null }
}
],
accountHome: { variable: 'CLAUDE_CONFIG_DIR', path: '/home/user/.claude' },
lease,
createdAt: 1_000,
updatedAt: 2_000
}
}
+6
View File
@@ -5,6 +5,7 @@ import type {
AgentStatusState,
AgentType
} from './agent-status-types'
import type { AgentSessionPtyWriteRefusal } from './agent-session-pty-write-admission'
import type {
BaseRefSearchResult,
BrowserCookieImportResult,
@@ -605,6 +606,11 @@ export type RuntimeTerminalSend = {
accepted: boolean
bytesWritten: number
refusedReason?: 'no-agent' | 'permission'
/**
* Present only when a durable agent-session lease refused the write. Additive and optional: an
* old client sees the `accepted: false` it already handles and ignores this field.
*/
agentSessionRefusal?: AgentSessionPtyWriteRefusal
}
export type RuntimeTerminalAgentStatusState = 'working' | 'permission' | 'idle' | null