From c69b12b3dd5bda44330082fddff1bec5ed80ab68 Mon Sep 17 00:00:00 2001
From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com>
Date: Sun, 9 Aug 2026 13:00:42 -0700
Subject: [PATCH] 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.
---
 .../ipc/pty-agent-session-write-gate.test.ts  | 280 ++++++++++++++++++
 src/main/ipc/pty.ts                           |  87 +++++-
 src/main/plugins/plugin-host-methods.test.ts  |  32 ++
 .../plugins/plugin-host-service-bindings.ts   |  17 +-
 ...gent-session-pty-write-enforcement.test.ts | 275 +++++++++++++++++
 .../agent-session-pty-write-gate.test.ts      | 147 +++++++++
 .../runtime/agent-session-pty-write-gate.ts   | 109 +++++++
 src/main/runtime/orca-runtime.ts              |  32 +-
 src/main/runtime/rpc/methods/terminal.ts      |  18 ++
 .../terminal-send-agent-session-lease.test.ts | 106 +++++++
 src/preload/index.ts                          |  15 +-
 .../agent-session-pty-write-admission.test.ts | 247 +++++++++++++++
 .../agent-session-pty-write-admission.ts      | 177 +++++++++++
 .../agent-session-record.test-fixture.ts      |  63 ++++
 src/shared/runtime-types.ts                   |   6 +
 15 files changed, 1596 insertions(+), 15 deletions(-)
 create mode 100644 src/main/ipc/pty-agent-session-write-gate.test.ts
 create mode 100644 src/main/runtime/agent-session-pty-write-enforcement.test.ts
 create mode 100644 src/main/runtime/agent-session-pty-write-gate.test.ts
 create mode 100644 src/main/runtime/agent-session-pty-write-gate.ts
 create mode 100644 src/main/runtime/rpc/terminal-send-agent-session-lease.test.ts
 create mode 100644 src/shared/agent-session-pty-write-admission.test.ts
 create mode 100644 src/shared/agent-session-pty-write-admission.ts
 create mode 100644 src/shared/agent-session-record.test-fixture.ts

diff --git a/src/main/ipc/pty-agent-session-write-gate.test.ts b/src/main/ipc/pty-agent-session-write-gate.test.ts
new file mode 100644
index 00000000000..0fac4101af9
--- /dev/null
+++ b/src/main/ipc/pty-agent-session-write-gate.test.ts
@@ -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')
+  })
+})
diff --git a/src/main/ipc/pty.ts b/src/main/ipc/pty.ts
index 7084b5e4b8c..566195565d0 100644
--- a/src/main/ipc/pty.ts
+++ b/src/main/ipc/pty.ts
@@ -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
     }
diff --git a/src/main/plugins/plugin-host-methods.test.ts b/src/main/plugins/plugin-host-methods.test.ts
index c6a5ff3adda..a7a86662809 100644
--- a/src/main/plugins/plugin-host-methods.test.ts
+++ b/src/main/plugins/plugin-host-methods.test.ts
@@ -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)
+  })
+})
diff --git a/src/main/plugins/plugin-host-service-bindings.ts b/src/main/plugins/plugin-host-service-bindings.ts
index 266b819dbae..10dd2b87e56 100644
--- a/src/main/plugins/plugin-host-service-bindings.ts
+++ b/src/main/plugins/plugin-host-service-bindings.ts
@@ -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: {
diff --git a/src/main/runtime/agent-session-pty-write-enforcement.test.ts b/src/main/runtime/agent-session-pty-write-enforcement.test.ts
new file mode 100644
index 00000000000..8eebc80f448
--- /dev/null
+++ b/src/main/runtime/agent-session-pty-write-enforcement.test.ts
@@ -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')
+  })
+})
diff --git a/src/main/runtime/agent-session-pty-write-gate.test.ts b/src/main/runtime/agent-session-pty-write-gate.test.ts
new file mode 100644
index 00000000000..35294d01fbf
--- /dev/null
+++ b/src/main/runtime/agent-session-pty-write-gate.test.ts
@@ -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()
+  })
+})
diff --git a/src/main/runtime/agent-session-pty-write-gate.ts b/src/main/runtime/agent-session-pty-write-gate.ts
new file mode 100644
index 00000000000..95faf8cb2b3
--- /dev/null
+++ b/src/main/runtime/agent-session-pty-write-gate.ts
@@ -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()
diff --git a/src/main/runtime/orca-runtime.ts b/src/main/runtime/orca-runtime.ts
index 543bb8c15b4..453bead8fea 100644
--- a/src/main/runtime/orca-runtime.ts
+++ b/src/main/runtime/orca-runtime.ts
@@ -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) {
diff --git a/src/main/runtime/rpc/methods/terminal.ts b/src/main/runtime/rpc/methods/terminal.ts
index 1cdec021e55..8d45ad5c0bc 100644
--- a/src/main/runtime/rpc/methods/terminal.ts
+++ b/src/main/runtime/rpc/methods/terminal.ts
@@ -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 {
diff --git a/src/main/runtime/rpc/terminal-send-agent-session-lease.test.ts b/src/main/runtime/rpc/terminal-send-agent-session-lease.test.ts
new file mode 100644
index 00000000000..94c6731f384
--- /dev/null
+++ b/src/main/runtime/rpc/terminal-send-agent-session-lease.test.ts
@@ -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)
+  })
+})
diff --git a/src/preload/index.ts b/src/preload/index.ts
index 5f66714f45d..cf5991744d7 100644
--- a/src/preload/index.ts
+++ b/src/preload/index.ts
@@ -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)
     },
diff --git a/src/shared/agent-session-pty-write-admission.test.ts b/src/shared/agent-session-pty-write-admission.test.ts
new file mode 100644
index 00000000000..9718fc8d5dc
--- /dev/null
+++ b/src/shared/agent-session-pty-write-admission.test.ts
@@ -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')
+  })
+})
diff --git a/src/shared/agent-session-pty-write-admission.ts b/src/shared/agent-session-pty-write-admission.ts
new file mode 100644
index 00000000000..78f9b915f69
--- /dev/null
+++ b/src/shared/agent-session-pty-write-admission.ts
@@ -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}).`
+}
diff --git a/src/shared/agent-session-record.test-fixture.ts b/src/shared/agent-session-record.test-fixture.ts
new file mode 100644
index 00000000000..31e633f0541
--- /dev/null
+++ b/src/shared/agent-session-record.test-fixture.ts
@@ -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
+  }
+}
diff --git a/src/shared/runtime-types.ts b/src/shared/runtime-types.ts
index 6e012e60b42..a7da46fe53b 100644
--- a/src/shared/runtime-types.ts
+++ b/src/shared/runtime-types.ts
@@ -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