fix(rebase): migrate the replayed E2B branch onto the rebased parent

The linear replay carried old-lineage content over parent-owned files;
this checkpoint restores them and adapts the branch to the parent's
post-rebase seam:

- restore all pty/lsp/subprocess/code-runtime surfaces to the parent's
  exact content (this branch claims none of them) and drop the net-zero
  code-runtime-e2b/pty-e2b/lsp-e2b residue and its registrations
- widen serializeRemoteEnvironment to the seam's NodeJS.ProcessEnv
  tombstone contract: an explicit undefined removes an ambient entry
- migrate the two E2B fixture Agent stubs to the Inbox-model interface
  and Session.create
- re-apply the branch's gen-doc-graphs roles, THIRD_PARTY_NOTICES e2b
  row, and packages/README group row (trimmed to the doc budget);
  regenerate catalogs and re-record bilingual pairings

packages/pty/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/pty/README.md
-README.md: a6706ed653bc60909a23b3598c41bb10ef499cbe
-README.zh.md: 01fecf8d57d933168e91331c4d3c3e4666c13cdc
+README.md: a4f743056b4a524be9623b0f700f37e0534b463f
+README.zh.md: c84ad3f1b59afcdbbd111f1b82c57c56aa24fdcf

packages/pty/README.md

@@ -7,8 +7,7 @@ English | [中文](README.zh.md)
 | Package | Role | ctx key |
 |---|---|---|
 | [`pty`](pty/README.md) (`@deepseek-ai/dsh-pty`) | Backend registry, branded ids, exact-Agent ownership, session operations, and awaited cleanup | `ctx.pty` |
-| [`pty-local`](pty-local/README.md) (`@deepseek-ai/dsh-pty-local`) | Local `node-pty` backend, readiness detection, bounded terminal state, sandboxing, and process-session supervision | registers on `ctx.pty` |
-| [`e2b/pty-e2b`](../e2b/pty-e2b/README.md) (`@deepseek-ai/dsh-pty-e2b`) | E2B byte-PTY backend, remote foreground signaling, bounded terminal state, and awaited remote cleanup | registers on `ctx.pty` |
+| `pty-local` (`@deepseek-ai/dsh-pty-local`) | Shell backend over `ctx.subprocess.spawnTerminal`: readiness detection, bounded terminal state, sandbox policy, and session operations | registers on `ctx.pty` |
 | `tool-pty` (`@deepseek-ai/dsh-tool-pty`) | Six model-facing tools and generic task integration for background sends | registers on `ctx.tools` |
 
-The core design lives in the [persistent PTY Agent Note](../../.agents/notes/implemented/feature/2026-07-16-persistent-pty-sessions.md); the remote ownership boundary lives in the [shared E2B runtime note](../../.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md).
+The design and deferred boundaries live in the [persistent PTY Agent Note](../../.agents/notes/implemented/feature/2026-07-16-persistent-pty-sessions.md).

packages/pty/README.zh.md

@@ -7,8 +7,7 @@
 | 包 | 职责 | ctx 键 |
 |---|---|---|
 | [`pty`](pty/README.md)（`@deepseek-ai/dsh-pty`） | 后端注册表、品牌化 id、精确的 Agent 所有权、会话操作与等待完成的清理 | `ctx.pty` |
-| [`pty-local`](pty-local/README.md)（`@deepseek-ai/dsh-pty-local`） | 本地 `node-pty` 后端、就绪检测、有界终端状态、沙箱与进程会话监管 | 注册到 `ctx.pty` |
-| [`e2b/pty-e2b`](../e2b/pty-e2b/README.md)（`@deepseek-ai/dsh-pty-e2b`） | E2B 字节 PTY 后端、远程前台信号传递、有界终端状态与等待完成的远程清理 | 注册到 `ctx.pty` |
+| `pty-local`（`@deepseek-ai/dsh-pty-local`） | `ctx.subprocess.spawnTerminal` 之上的 shell 后端：就绪检测、有界终端状态、沙箱策略与会话操作 | 注册到 `ctx.pty` |
 | `tool-pty`（`@deepseek-ai/dsh-tool-pty`） | 6 个面向模型的工具，并为后台发送集成通用任务 | 注册到 `ctx.tools` |
 
-核心设计记录在[持久 PTY Agent Note](../../.agents/notes/implemented/feature/2026-07-16-persistent-pty-sessions.md) 中；远程所有权边界记录在 [共享 E2B 运行时 Agent Note](../../.agents/notes/implemented/feature/2026-07-27-e2b-remote-runtime-poc.md) 中。
+设计与暂缓边界记录在[持久 PTY Agent Note](../../.agents/notes/implemented/feature/2026-07-16-persistent-pty-sessions.md) 中。

packages/pty/pty-local/src/sanitize.ts

@@ -0,0 +1,188 @@
+/** Streaming terminal-control sanitizer for the line-oriented first release. */
+
+import { Buffer } from 'node:buffer'
+
+/** OSC marker emitted by the controlled bash before each prompt. */
+export const PROMPT_MARKER_PREFIX = '133;D;'
+
+/** Exact printable prompt emitted after the private marker. */
+export const CONTROLLED_PROMPT = 'dsh> '
+
+/** One sanitized chunk plus whether it contained the owned prompt marker. */
+export interface SanitizedChunk {
+  text: string
+  prompt: boolean
+  /** Printable text after the latest owned marker in this chunk. */
+  promptTail?: string
+}
+
+/**
+ * Remove CSI/OSC/short escape sequences while preserving split-sequence carry.
+ * Full terminal emulation is deliberately deferred; ordinary line output and
+ * the private prompt marker are the supported contract.
+ */
+export class TerminalSanitizer {
+  private pending = ''
+  private discardMode: 'osc' | 'csi' | undefined
+  private discardOscEscape = false
+  private trailingCarriageReturn = false
+  private trackingPromptTail = false
+
+  constructor(private readonly maxPendingBytes: number) {}
+
+  /**
+   * Consume one decoded `node-pty` data chunk.
+   * @param chunk - decoded terminal data.
+   * @returns Printable text and whether the private prompt marker completed.
+   */
+  push(chunk: string): SanitizedChunk {
+    this.pending += this.discardPrefix(chunk)
+    let text = ''
+    let prompt = false
+    let includePromptTail = this.trackingPromptTail
+    let promptTail = ''
+    let index = 0
+    const appendText = (value: string): void => {
+      text += value
+      if (this.trackingPromptTail) promptTail += value
+    }
+    while (index < this.pending.length) {
+      const escape = this.pending.indexOf('\x1b', index)
+      if (escape < 0) {
+        appendText(this.pending.slice(index))
+        index = this.pending.length
+        break
+      }
+      appendText(this.pending.slice(index, escape))
+      if (escape + 1 >= this.pending.length) {
+        index = escape
+        break
+      }
+      const kind = this.pending[escape + 1]
+      if (kind === ']') {
+        const bel = this.pending.indexOf('\x07', escape + 2)
+        const stringTerminator = this.pending.indexOf('\x1b\\', escape + 2)
+        let end = -1
+        if (bel >= 0 && stringTerminator >= 0) end = Math.min(bel + 1, stringTerminator + 2)
+        else if (bel >= 0) end = bel + 1
+        else if (stringTerminator >= 0) end = stringTerminator + 2
+        if (end < 0) {
+          index = escape
+          break
+        }
+        const terminatorBytes = this.pending[end - 1] === '\x07' ? 1 : 2
+        const content = this.pending.slice(escape + 2, end - terminatorBytes)
+        if (content.startsWith(PROMPT_MARKER_PREFIX)) {
+          prompt = true
+          this.trackingPromptTail = true
+          includePromptTail = true
+          promptTail = ''
+        }
+        index = end
+        continue
+      }
+      if (kind === '[') {
+        let end = escape + 2
+        while (end < this.pending.length) {
+          const code = this.pending.charCodeAt(end)
+          if (code >= 0x40 && code <= 0x7e) break
+          end += 1
+        }
+        if (end >= this.pending.length) {
+          index = escape
+          break
+        }
+        index = end + 1
+        continue
+      }
+      // Two-byte escape family (save/restore cursor and similar).
+      index = escape + 2
+    }
+    this.pending = this.pending.slice(index)
+    this.enforcePendingBound()
+    return {
+      text: this.normalizeText(text),
+      prompt,
+      ...includePromptTail ? { promptTail } : {},
+    }
+  }
+
+  /**
+   * Flush a trailing printable fragment when the PTY exits.
+   * @returns Remaining printable text; incomplete escapes are discarded.
+   */
+  flush(): string {
+    const text = this.pending.startsWith('\x1b') ? '' : this.pending
+    this.pending = ''
+    this.discardMode = undefined
+    this.discardOscEscape = false
+    this.trackingPromptTail = false
+    const normalized = this.normalizeText(text)
+    if (!this.trailingCarriageReturn) return normalized
+    this.trailingCarriageReturn = false
+    return `${normalized}\n`
+  }
+
+  private normalizeText(text: string): string {
+    let complete = this.trailingCarriageReturn ? `\r${text}` : text
+    this.trailingCarriageReturn = false
+    if (complete.endsWith('\r')) {
+      complete = complete.slice(0, -1)
+      this.trailingCarriageReturn = true
+    }
+    return normalizeTerminalText(complete)
+  }
+
+  private enforcePendingBound(): void {
+    if (Buffer.byteLength(this.pending) <= this.maxPendingBytes) return
+    this.discardMode = this.pending[1] === ']' ? 'osc' : 'csi'
+    this.pending = ''
+  }
+
+  private discardPrefix(chunk: string): string {
+    if (this.discardMode === undefined) return chunk
+    if (this.discardMode === 'csi') {
+      for (let index = 0; index < chunk.length; index += 1) {
+        const code = chunk.charCodeAt(index)
+        if (code >= 0x40 && code <= 0x7e) {
+          this.discardMode = undefined
+          return chunk.slice(index + 1)
+        }
+      }
+      return ''
+    }
+
+    let index = 0
+    if (this.discardOscEscape) {
+      this.discardOscEscape = false
+      if (chunk.startsWith('\\')) {
+        this.discardMode = undefined
+        return chunk.slice(1)
+      }
+    }
+    while (index < chunk.length) {
+      if (chunk[index] === '\x07') {
+        this.discardMode = undefined
+        return chunk.slice(index + 1)
+      }
+      if (chunk[index] === '\x1b') {
+        if (chunk[index + 1] === '\\') {
+          this.discardMode = undefined
+          return chunk.slice(index + 2)
+        }
+        if (index + 1 === chunk.length) this.discardOscEscape = true
+      }
+      index += 1
+    }
+    return ''
+  }
+}
+
+/**
+ * Normalize CRLF and standalone carriage returns for line-oriented rendering.
+ * @param text - sanitized terminal text.
+ * @returns Line-normalized text with BEL removed.
+ */
+export function normalizeTerminalText(text: string): string {
+  return text.replaceAll('\r\n', '\n').replaceAll('\r', '\n').replaceAll('\x07', '')
+}

packages/pty/pty-local/src/session.ts

@@ -1,7 +1,12 @@
-/** Local `node-pty` session: bounded output, readiness, signals, and teardown. */
+/** Persistent PTY session over the subprocess seam's terminal primitive. */
 
-import type { IDisposable, IPty } from 'node-pty'
-import { PtyTerminalSanitizer, PtyTextBuffer, ptySignalName, ptyUtf8Tail } from '@deepseek-ai/dsh-pty'
+import { Buffer } from 'node:buffer'
+import type {
+  SubprocessOutcome,
+  SubprocessTerminalForeground,
+  SubprocessTerminalHandle,
+} from '@deepseek-ai/dsh-subprocess'
+import { PtyError } from '@deepseek-ai/dsh-pty'
 import type {
   PtyBackendSession,
   PtyReadRequest,
@@ -16,30 +21,89 @@ import type {
   PtyWaitReason,
 } from '@deepseek-ai/dsh-pty'
 import type { ResolvedConfig } from './config.ts'
-import type { ProcessIdentity, ProcessInspector } from './process-inspector.ts'
+import { CONTROLLED_PROMPT, TerminalSanitizer } from './sanitize.ts'
 
-function delay(ms: number): Promise<void> {
-  return new Promise(resolve => setTimeout(resolve, ms))
+function utf8Tail(text: string, maxBytes: number): { text: string; truncated: boolean } {
+  if (Buffer.byteLength(text) <= maxBytes) return { text, truncated: false }
+  const chars = Array.from(text)
+  let bytes = 0
+  let start = chars.length
+  while (start > 0) {
+    const next = Buffer.byteLength(chars[start - 1] as string)
+    if (bytes + next > maxBytes) break
+    bytes += next
+    start -= 1
+  }
+  return { text: chars.slice(start).join(''), truncated: true }
+}
+
+class BoundedTextBuffer {
+  private value = ''
+  private dropped = false
+
+  constructor(
+    private readonly maxBytes: number,
+    private readonly maxLines?: number,
+  ) {}
+
+  append(text: string): void {
+    if (text.length === 0) return
+    this.value += text
+    if (this.maxLines !== undefined) {
+      const lines = this.value.split('\n')
+      if (lines.length > this.maxLines) {
+        this.value = lines.slice(lines.length - this.maxLines).join('\n')
+        this.dropped = true
+      }
+    }
+    const tail = utf8Tail(this.value, this.maxBytes)
+    this.value = tail.text
+    this.dropped ||= tail.truncated
+  }
+
+  consume(): PtySendRead {
+    const delta = this.value
+    const truncated = this.dropped
+    this.value = ''
+    this.dropped = false
+    return { delta, truncated }
+  }
+
+  snapshot(): { text: string; truncated: boolean } {
+    return { text: this.value, truncated: this.dropped }
+  }
 }
 
 class LocalSendOperation implements PtySendOperation {
-  private readonly output: PtyTextBuffer
+  private readonly output: BoundedTextBuffer
   private readonly promise: PromiseWithResolvers<PtySendResult>
   private finished = false
+  private cancellationRequested = false
+  private initialForegroundLeftWait: boolean
+  private initialForegroundPgid: number | undefined
 
   constructor(
     maxBytes: number,
     readonly startedAt: number,
     private readonly onCancel: () => void,
   ) {
-    this.output = new PtyTextBuffer(maxBytes)
+    this.output = new BoundedTextBuffer(maxBytes)
     this.promise = Promise.withResolvers<PtySendResult>()
+    this.initialForegroundLeftWait = true
   }
 
   get done(): Promise<PtySendResult> {
     return this.promise.promise
   }
 
+  get settled(): boolean {
+    return this.finished
+  }
+
+  get cancelRequested(): boolean {
+    return this.cancellationRequested
+  }
+
   append(text: string): void {
     if (!this.finished) this.output.append(text)
   }
@@ -66,50 +130,75 @@ class LocalSendOperation implements PtySendOperation {
     return this.output.consume()
   }
 
+  setInitialForeground(foreground: SubprocessTerminalForeground | undefined): void {
+    this.initialForegroundPgid = foreground?.processGroupId
+    this.initialForegroundLeftWait = foreground?.inputWaiting !== true
+  }
+
+  acceptsStdinWait(pgid: number, waiting: boolean): boolean {
+    // The same group may still expose the wait that existed before terminal.write.
+    // Observe every poll so a departure before the exact-settlement threshold
+    // still makes a later return to that wait post-write evidence.
+    if (pgid !== this.initialForegroundPgid) return waiting
+    if (!waiting) this.initialForegroundLeftWait = true
+    return waiting && this.initialForegroundLeftWait
+  }
+
   cancel(): boolean {
     if (this.finished) return false
+    this.cancellationRequested = true
     this.onCancel()
     return true
   }
 }
 
-/** Backend session wrapping one `node-pty` process and its captured process tree. */
+/** Backend session wrapping one provider-owned terminal process. */
 export class LocalPtySession implements PtyBackendSession {
   motd = ''
   readonly pid: number
-  private readonly sanitizer: PtyTerminalSanitizer
-  private readonly scrollback: PtyTextBuffer
-  private readonly exitPromise: PromiseWithResolvers<void> = Promise.withResolvers<void>()
-  private readonly dataDisposable: IDisposable
-  private readonly exitDisposable: IDisposable
+  private readonly decoder = new TextDecoder()
+  private readonly sanitizer: TerminalSanitizer
+  private readonly scrollback: BoundedTextBuffer
+  private readonly outputEnded = Promise.withResolvers<void>()
+  private readonly completion: Promise<void>
   private statusValue: PtySessionStatus = { kind: 'running' }
+  // TODO(pty-send-state-consolidation): Fold the per-send fields below
+  // (active/activeTimer/activeDeadlineTimer/activeAbort/interrupting/
+  // activeWrite/pollingReady/polling) into one send-lifecycle owner; the
+  // cancellation/readiness interplay now has enough pinned tests to carry
+  // that refactor safely.
   private active: LocalSendOperation | undefined
   private activeTimer: NodeJS.Timeout | undefined
+  private activeDeadlineTimer: NodeJS.Timeout | undefined
   private activeAbort: (() => void) | undefined
+  private interrupting: LocalSendOperation | undefined
+  private activeWrite: Promise<boolean> | undefined
+  private pollingReady: LocalSendOperation | undefined
+  private polling = false
   private promptSeen = false
   private promptTextSeen = false
+  private promptTail = ''
   private shellPgid: number | undefined
   private initializing = false
   private lastOutputAt = Date.now()
   private closing = false
   private closePromise: Promise<void> | undefined
+  private transportFailure: Error | undefined
 
   constructor(
-    private readonly terminal: IPty,
-    private readonly inspector: ProcessInspector,
+    private readonly terminal: SubprocessTerminalHandle,
     private readonly config: ResolvedConfig,
   ) {
     this.pid = terminal.pid
-    this.sanitizer = new PtyTerminalSanitizer(config.maxReadBytes)
-    this.scrollback = new PtyTextBuffer(config.scrollbackMaxBytes, config.scrollbackLines)
-    this.dataDisposable = terminal.onData((data) => { this.onData(data) })
-    this.exitDisposable = terminal.onExit(({ exitCode, signal }) => {
-      const tail = this.sanitizer.flush()
-      this.appendOutput(tail)
-      this.statusValue = { kind: 'exited', exitCode, signal: ptySignalName(signal) }
-      this.settleActive('session_exit')
-      this.exitPromise.resolve()
-    })
+    this.sanitizer = new TerminalSanitizer(config.maxReadBytes)
+    this.scrollback = new BoundedTextBuffer(config.scrollbackMaxBytes, config.scrollbackLines)
+    terminal.output.on('data', this.onTerminalData)
+    terminal.output.once('end', this.onTerminalEnd)
+    terminal.output.once('error', this.onTerminalError)
+    this.completion = terminal.done.then(
+      outcome => this.onExit(outcome),
+      (error: unknown) => { this.onTransportFailure(error) },
+    )
   }
 
   /**
@@ -136,7 +225,14 @@ export class LocalPtySession implements PtyBackendSession {
   startSend(request: PtySendRequest): PtySendOperation {
     if (this.closing) throw new Error('PTY session is closing')
     if (this.statusValue.kind === 'exited') throw new Error('PTY session has exited')
-    if (this.active !== undefined) throw new Error('PTY session already has an active send')
+    if (this.active !== undefined) {
+      const draining = this.activeWrite !== undefined
+        ? ' or draining provider write'
+        : this.interrupting !== undefined
+          ? ' or draining foreground interrupt'
+          : ''
+      throw new PtyError(`PTY session already has an active send${draining}`, 'SEND_ACTIVE')
+    }
     if (request.signal?.aborted === true) throw new Error('PTY send aborted before write')
 
     const operation = new LocalSendOperation(
@@ -145,29 +241,79 @@ export class LocalPtySession implements PtyBackendSession {
       () => { this.interrupt(operation) },
     )
     this.active = operation
-    this.lastOutputAt = Date.now()
-    this.promptSeen = false
-    this.promptTextSeen = false
+    this.resetReadinessEvidence()
 
     if (request.signal !== undefined) {
       const onAbort = (): void => { operation.cancel() }
       request.signal.addEventListener('abort', onAbort, { once: true })
       this.activeAbort = () => request.signal?.removeEventListener('abort', onAbort)
     }
-
-    try {
-      if (request.text.length > 0) this.terminal.write(request.text)
-      if (request.submit) this.terminal.write('\r')
-    } catch (error: unknown) {
-      this.clearActive()
-      operation.fail(error)
-      return operation
-    }
-
-    this.activeTimer = setInterval(() => { this.pollReadiness(operation) }, this.config.pollIntervalMs)
+    this.activeDeadlineTimer = setTimeout(() => {
+      if (this.active === operation) {
+        this.settleActive('timeout', this.activeWrite !== undefined || this.interrupting === operation)
+      }
+    }, this.config.timeoutMs)
+    void this.beginSend(operation, request)
     return operation
   }
 
+  private async beginSend(operation: LocalSendOperation, request: PtySendRequest): Promise<void> {
+    let foreground: SubprocessTerminalForeground | undefined
+    try {
+      foreground = await this.terminal.inspectForeground()
+    } catch (error: unknown) {
+      // A pre-write inspection failure while cancellation owns the slot must not
+      // release it: interruptOnce's in-flight foreground signal could land on a
+      // successor's foreground group. The interrupt path's post-signal tail
+      // resumes polling, whose guarded catch propagates a persistent failure.
+      // A retained settled operation implies that same in-flight interrupt, so
+      // this guard admits only an unsettled active send.
+      if (this.active === operation && !this.closing && this.interrupting !== operation) {
+        this.failActive(error)
+      }
+      return
+    }
+    try {
+      if (this.active !== operation || this.closing || this.interrupting === operation) return
+      operation.setInitialForeground(foreground)
+      const input = `${request.text}${request.submit ? '\r' : ''}`
+      if (input.length > 0 && !operation.cancelRequested) {
+        this.resetReadinessEvidence()
+        const write = this.terminal.write(input)
+        this.activeWrite = write.then(() => true, () => false)
+        try {
+          await write
+        } finally {
+          this.activeWrite = undefined
+        }
+      }
+      // Cancellation owns post-write signalling and reservation release.
+      if (operation.cancelRequested) return
+      if (this.active === operation && operation.settled) {
+        this.clearActive()
+        return
+      }
+      // Closing can race the awaited provider write even though static analysis sees only local assignments.
+      // oxlint-disable-next-line typescript/no-unnecessary-condition -- awaited provider writes can close the session.
+      if (this.active === operation && !this.closing) {
+        this.pollingReady = operation
+        this.schedulePoll(operation)
+      }
+    } catch (error: unknown) {
+      if (this.active === operation && !this.closing) {
+        if (operation.settled) this.clearActive()
+        else this.failActive(error)
+      }
+    }
+  }
+
+  private resetReadinessEvidence(): void {
+    this.lastOutputAt = Date.now()
+    this.promptSeen = false
+    this.promptTextSeen = false
+    this.promptTail = ''
+  }
+
   read(request: PtyReadRequest): PtyReadResult {
     const snapshot = this.scrollback.snapshot()
     const lines = snapshot.text.split('\n')
@@ -182,7 +328,7 @@ export class LocalPtySession implements PtyBackendSession {
     const end = totalLines - offset
     const start = Math.max(0, end - count)
     const requested = lines.slice(start, end).join('\n')
-    const bounded = ptyUtf8Tail(requested, this.config.maxReadBytes)
+    const bounded = utf8Tail(requested, this.config.maxReadBytes)
     const returnedLines = bounded.text.length === 0 ? 0 : bounded.text.split('\n').length
     return {
       text: bounded.text,
@@ -193,16 +339,10 @@ export class LocalPtySession implements PtyBackendSession {
     }
   }
 
-  signal(signal: PtySignal): Promise<PtySignalResult> {
-    return Promise.resolve().then(() => {
-      const pgid = this.inspector.foregroundPgid(this.pid)
-      if (pgid === undefined) throw new Error(`cannot resolve foreground process group for PTY ${this.pid}`)
-      if (signal === 'SIGKILL' && pgid === this.pid) {
-        throw new Error('refusing to SIGKILL the PTY shell; use terminal_close')
-      }
-      this.inspector.signalGroup(pgid, signal)
-      return { delivered: true, targetPgid: pgid }
-    })
+  async signal(signal: PtySignal): Promise<PtySignalResult> {
+    if (this.closing) throw new Error('PTY session is closing')
+    const targetPgid = await this.terminal.signalForeground(signal)
+    return { delivered: true, targetPgid }
   }
 
   status(): PtySessionStatus {
@@ -221,21 +361,56 @@ export class LocalPtySession implements PtyBackendSession {
     return closing
   }
 
+  private readonly onTerminalData = (chunk: Buffer | Uint8Array | string): void => {
+    const bytes = typeof chunk === 'string' ? Buffer.from(chunk, 'utf8') : chunk
+    this.onData(this.decoder.decode(bytes, { stream: true }))
+  }
+
+  private readonly onTerminalEnd = (): void => {
+    this.onData(this.decoder.decode())
+    this.appendOutput(this.sanitizer.flush())
+    this.outputEnded.resolve()
+  }
+
+  private readonly onTerminalError = (error: Error): void => {
+    this.onTransportFailure(error)
+    this.outputEnded.resolve()
+  }
+
   private onData(data: string): void {
     const sanitized = this.sanitizer.push(data)
     this.appendOutput(sanitized.text)
     if (sanitized.prompt) {
-      const foregroundPgid = this.inspector.foregroundPgid(this.pid)
-      if (this.shellPgid === undefined) this.shellPgid = foregroundPgid
+      // TODO(pty-delayed-signal-prompt): With a reproducer, define a marker-generation boundary
+      // before attributing a signal-delayed prompt to a later send.
       // Bash can print PROMPT_COMMAND before the kernel publishes its return
       // to the foreground process group. Retain the marker; polling below is
       // the authority that accepts it only after bash owns the foreground.
       this.promptSeen = true
-      this.promptTextSeen = sanitized.promptText === true
+      this.promptTail = ''
       this.lastOutputAt = Date.now()
-    } else if (this.promptSeen && sanitized.promptText === true) {
-      this.promptTextSeen = true
     }
+    if (this.promptSeen && sanitized.promptTail !== undefined) {
+      const remaining = Math.max(0, CONTROLLED_PROMPT.length + 1 - this.promptTail.length)
+      this.promptTail += sanitized.promptTail.slice(0, remaining)
+      if (sanitized.promptTail.length > remaining) this.promptTail = `${CONTROLLED_PROMPT}\0`
+      this.promptTextSeen = this.promptTail === CONTROLLED_PROMPT
+    }
+  }
+
+  private async onExit(outcome: SubprocessOutcome): Promise<void> {
+    await this.outputEnded.promise
+    if (this.transportFailure !== undefined) return
+    this.statusValue = { kind: 'exited', exitCode: outcome.exitCode, signal: outcome.signal }
+    this.settleActive('session_exit')
+  }
+
+  private onTransportFailure(error: unknown): void {
+    const failure = error instanceof Error ? error : new Error(String(error))
+    this.transportFailure ??= failure
+    this.statusValue = { kind: 'exited', exitCode: null, signal: null }
+    this.failActive(failure)
+    void this.terminal.terminate().catch(() => {})
   }
 
   private appendOutput(text: string): void {
@@ -245,60 +420,94 @@ export class LocalPtySession implements PtyBackendSession {
     this.active?.append(text)
   }
 
-  private pollReadiness(operation: LocalSendOperation): void {
-    if (this.active !== operation) return
-    if (this.statusValue.kind === 'exited') {
-      this.settleActive('session_exit')
-      return
-    }
-    if (this.promptSeen && this.promptTextSeen && Date.now() - this.lastOutputAt >= this.config.pollIntervalMs) {
-      const pgid = this.inspector.foregroundPgid(this.pid)
-      if (this.shellPgid !== undefined && pgid === this.shellPgid) {
-        this.settleActive('stdin_read')
-        return
-      }
-    }
-    const elapsed = Date.now() - operation.startedAt
-    const startupHasOutput = !this.initializing || this.scrollback.snapshot().text.length > 0
-    if (startupHasOutput && elapsed >= this.config.exactProbeAfterMs) {
-      const pgid = this.inspector.foregroundPgid(this.pid)
-      if (pgid !== undefined && this.inspector.isStdinWaiting(pgid)) {
-        this.settleActive('stdin_read')
-        return
-      }
-    }
-    // A prompt candidate can race bash's foreground handoff, but an interactive
-    // child also inherits PROMPT_COMMAND. Silence therefore remains the bound
-    // on waiting for shell ownership instead of letting a child marker suppress
-    // readiness until the absolute timeout. When a prompt marker was seen, the
-    // configured grace holds the fallback past the silence bound so polls in
-    // that window can observe the foreground handoff and settle as stdin_read.
-    const idleFor = Date.now() - this.lastOutputAt
-    const handoffGrace = this.promptSeen ? this.config.handoffGraceMs : 0
-    if (startupHasOutput && idleFor >= this.config.idleSilenceMs + handoffGrace) {
-      this.settleActive('inferred_idle')
-      return
-    }
-    if (elapsed >= this.config.timeoutMs) this.settleActive('timeout')
+  private schedulePoll(operation: LocalSendOperation, delayMs = this.config.pollIntervalMs): void {
+    if (this.active !== operation || this.interrupting === operation || this.polling) return
+    if (this.activeTimer !== undefined) clearTimeout(this.activeTimer)
+    this.activeTimer = setTimeout(() => {
+      this.activeTimer = undefined
+      void this.pollReadiness(operation)
+    }, delayMs)
   }
 
-  private settleActive(waitReason: PtyWaitReason): void {
+  private async pollReadiness(operation: LocalSendOperation): Promise<void> {
+    if (this.active !== operation || this.polling) return
+    this.polling = true
+    try {
+      if (this.statusValue.kind === 'exited') {
+        this.settleActive('session_exit')
+        return
+      }
+      const foreground = await this.terminal.inspectForeground()
+      if (this.active !== operation || this.closing || this.interrupting === operation) return
+      const idleFor = Date.now() - this.lastOutputAt
+      if (this.promptSeen && foreground !== undefined && this.shellPgid === undefined) {
+        this.shellPgid = foreground.processGroupId
+      }
+      if (this.promptSeen && this.promptTextSeen && idleFor >= this.config.pollIntervalMs
+        && foreground?.processGroupId === this.shellPgid) {
+        this.settleActive('stdin_read')
+        return
+      }
+      const elapsed = Date.now() - operation.startedAt
+      const startupHasOutput = !this.initializing || this.scrollback.snapshot().text.length > 0
+      const acceptsStdinWait = startupHasOutput && foreground !== undefined
+        && operation.acceptsStdinWait(foreground.processGroupId, foreground.inputWaiting)
+      if (elapsed >= this.config.exactProbeAfterMs && acceptsStdinWait) {
+        this.settleActive('stdin_read')
+        return
+      }
+      // A prompt candidate can race bash's foreground handoff, but an interactive
+      // child also inherits PROMPT_COMMAND. Silence therefore remains the bound
+      // on waiting for shell ownership instead of letting a child marker suppress
+      // readiness until the absolute timeout.
+      const handoffGrace = this.promptSeen ? this.config.handoffGraceMs : 0
+      if (startupHasOutput && idleFor >= this.config.idleSilenceMs + handoffGrace) {
+        this.settleActive('inferred_idle')
+      }
+    } catch (error: unknown) {
+      if (this.active === operation && !this.closing && this.interrupting !== operation) this.failActive(error)
+    } finally {
+      this.polling = false
+      const active = this.active
+      // Awaited provider inspection can clear or replace the active send despite static analysis.
+      // oxlint-disable-next-line typescript/no-unnecessary-condition -- awaited inspection can replace the active send.
+      if (active !== undefined && this.pollingReady === active) this.schedulePoll(active)
+    }
+  }
+
+  private settleActive(waitReason: PtyWaitReason, retainOwnership = false): void {
     const operation = this.active
     if (operation === undefined) return
     const scrollbackTruncated = this.scrollback.snapshot().truncated
-    this.clearActive()
+    if (retainOwnership) {
+      this.stopPolling()
+      this.activeAbort?.()
+      this.activeAbort = undefined
+    } else {
+      this.clearActive()
+    }
     operation.settle(waitReason, this.statusValue, scrollbackTruncated)
   }
 
   private stopPolling(): void {
-    if (this.activeTimer !== undefined) clearInterval(this.activeTimer)
+    this.stopReadinessPolling()
+    if (this.activeDeadlineTimer !== undefined) clearTimeout(this.activeDeadlineTimer)
+    this.activeDeadlineTimer = undefined
+  }
+
+  private stopReadinessPolling(): void {
+    if (this.activeTimer !== undefined) clearTimeout(this.activeTimer)
     this.activeTimer = undefined
+    this.pollingReady = undefined
   }
 
   private clearActive(): void {
+    const operation = this.active
     this.stopPolling()
     this.activeAbort?.()
     this.activeAbort = undefined
+    if (this.interrupting === operation) this.interrupting = undefined
+    this.pollingReady = undefined
     this.active = undefined
   }
 
@@ -311,104 +520,46 @@ export class LocalPtySession implements PtyBackendSession {
 
   private interrupt(operation: LocalSendOperation): void {
     if (this.active !== operation) return
+    this.interrupting = operation
+    this.stopReadinessPolling()
+    void this.interruptOnce(operation)
+  }
+
+  private async interruptOnce(operation: LocalSendOperation): Promise<void> {
     try {
-      const pgid = this.inspector.foregroundPgid(this.pid)
-      if (pgid === undefined) throw new Error(`cannot resolve foreground process group for PTY ${this.pid}`)
-      this.inspector.signalGroup(pgid, 'SIGINT')
+      const activeWrite = this.activeWrite
+      if (activeWrite !== undefined && !await activeWrite) return
+      await this.terminal.signalForeground('SIGINT')
     } catch (error: unknown) {
-      this.failActive(error)
+      if (this.active === operation && !this.closing) this.onTransportFailure(error)
+      return
+    } finally {
+      if (this.interrupting === operation) this.interrupting = undefined
     }
-  }
-
-  private survivors(members: ProcessIdentity[]): ProcessIdentity[] {
-    return members.filter(member => this.inspector.isAlive(member))
-  }
-
-  private descendants(): ProcessIdentity[] {
-    return this.inspector.processTree(this.pid).filter(member => member.pid !== this.pid)
-  }
-
-  private async waitForExit(members: ProcessIdentity[]): Promise<ProcessIdentity[]> {
-    const deadline = Date.now() + this.config.disposeGraceMs
-    let survivors = this.survivors(members)
-    while (survivors.length > 0 && Date.now() < deadline) {
-      await delay(Math.min(25, Math.max(1, deadline - Date.now())))
-      survivors = this.survivors(members)
-    }
-    return survivors
-  }
-
-  private signalMembers(members: ProcessIdentity[], signal: 'SIGTERM' | 'SIGKILL'): void {
-    for (const member of members) {
-      try {
-        this.inspector.signalProcess(member, signal)
-      } catch (_alreadyExitedDuringSignal) {
-        // Identity is rechecked by the inspector; a same-tick exit is success.
-      }
-    }
-  }
-
-  private unionMembers(...groups: ProcessIdentity[][]): ProcessIdentity[] {
-    const members: ProcessIdentity[] = []
-    const seen = new Set<string>()
-    for (const group of groups) {
-      for (const member of group) {
-        const key = JSON.stringify([member.pid, member.started])
-        if (seen.has(key)) continue
-        seen.add(key)
-        members.push(member)
-      }
-    }
-    return members
-  }
-
-  private async stopDescendants(): Promise<ProcessIdentity[]> {
-    const captured = this.descendants()
-    this.signalMembers(captured, 'SIGTERM')
-    const capturedSurvivors = await this.waitForExit(captured)
-    // A TERM-handling descendant may have forked while winding down. Rescan
-    // while the shell can still reap every member, then kill both the fresh
-    // tree and captured survivors that were reparented out of that tree.
-    const members = this.unionMembers(capturedSurvivors, this.descendants())
-    this.signalMembers(members, 'SIGKILL')
-    const survivors = await this.waitForExit(members)
-    return this.survivors(this.unionMembers(survivors, this.descendants()))
-  }
-
-  private async stopShell(): Promise<void> {
-    try {
-      this.terminal.kill('SIGTERM')
-    } catch (_topLevelAlreadyExitedDuringTerm) {
-      // The exit notification remains authoritative.
-    }
-    if (this.statusValue.kind === 'running') {
-      await Promise.race([this.exitPromise.promise, delay(this.config.disposeGraceMs)])
-    }
-    if (this.statusValue.kind === 'running') {
-      try {
-        this.terminal.kill('SIGKILL')
-      } catch (_topLevelAlreadyExitedDuringKill) {
-        // The exit notification remains authoritative.
-      }
-      await Promise.race([this.exitPromise.promise, delay(this.config.disposeGraceMs)])
-    }
-    if (this.statusValue.kind === 'running') {
-      throw new Error(`PTY cleanup failed; surviving pids: ${this.pid}`)
+    if (this.active === operation && operation.settled) {
+      this.clearActive()
+    } else if (this.active === operation && !this.closing) {
+      this.pollingReady = operation
+      this.schedulePoll(operation, 0)
     }
   }
 
   private async closeOnce(reason: string): Promise<void> {
-    this.dataDisposable.dispose()
     // Stop readiness polling but retain the active operation: teardown settles
     // it as session_exit below, so an in-flight send is never mis-settled as
     // stdin_read/inferred_idle/timeout during the grace period.
     this.stopPolling()
-    const survivors = await this.stopDescendants()
-    if (survivors.length > 0) {
-      throw new Error(`PTY cleanup failed (${reason}); surviving pids: ${survivors.map(member => member.pid).join(', ')}`)
+    try {
+      await this.terminal.terminate()
+    } catch (error: unknown) {
+      throw new Error(`PTY cleanup failed (${reason})`, { cause: error })
     }
-    await this.stopShell()
+    // Quiescence is the active send's terminal outcome.
     this.settleActive('session_exit')
-    this.exitDisposable.dispose()
+    await this.completion
+    this.terminal.output.off('data', this.onTerminalData)
+    this.terminal.output.off('end', this.onTerminalEnd)
+    this.terminal.output.off('error', this.onTerminalError)
+    if (this.transportFailure !== undefined) throw this.transportFailure
   }
 }

packages/pty/pty-local/tests/sanitize.spec.ts

@@ -1,17 +1,17 @@
 import { describe, expect, it } from 'vitest'
-import { normalizePtyTerminalText, PtyTerminalSanitizer } from '@deepseek-ai/dsh-pty'
+import { normalizeTerminalText, TerminalSanitizer } from '@deepseek-ai/dsh-pty-local/src/sanitize.ts'
 
-describe('PtyTerminalSanitizer', () => {
+describe('TerminalSanitizer', () => {
   it('removes split CSI and owned OSC prompt markers', () => {
-    const sanitizer = new PtyTerminalSanitizer(64)
+    const sanitizer = new TerminalSanitizer(64)
     expect(sanitizer.push('red\x1b[3')).toEqual({ text: 'red', prompt: false })
     expect(sanitizer.push('1m text\x1b[0m\r\n')).toEqual({ text: ' text\n', prompt: false })
     expect(sanitizer.push('\x1b]133;')).toEqual({ text: '', prompt: false })
-    expect(sanitizer.push('D;0\x07dsh> ')).toEqual({ text: 'dsh> ', prompt: true, promptText: true })
+    expect(sanitizer.push('D;0\x07dsh> ')).toEqual({ text: 'dsh> ', prompt: true, promptTail: 'dsh> ' })
   })
 
   it('drops unrelated OSC, short escapes, BEL, and incomplete trailing escape', () => {
-    const sanitizer = new PtyTerminalSanitizer(64)
+    const sanitizer = new TerminalSanitizer(64)
     expect(sanitizer.push('a\x1b]0;title\x1b\\b\x1b7c\x07')).toEqual({ text: 'abc', prompt: false })
     expect(sanitizer.push('tail\x1b')).toEqual({ text: 'tail', prompt: false })
     expect(sanitizer.flush()).toBe('')
@@ -22,11 +22,11 @@ describe('PtyTerminalSanitizer', () => {
   })
 
   it('normalizes CRLF and standalone carriage returns', () => {
-    expect(normalizePtyTerminalText('a\r\nb\rc\x07')).toBe('a\nb\nc')
+    expect(normalizeTerminalText('a\r\nb\rc\x07')).toBe('a\nb\nc')
   })
 
   it('carries a trailing carriage return across data chunks and flushes standalone CR', () => {
-    const sanitizer = new PtyTerminalSanitizer(64)
+    const sanitizer = new TerminalSanitizer(64)
     expect(sanitizer.push('a\r')).toEqual({ text: 'a', prompt: false })
     expect(sanitizer.push('\nb')).toEqual({ text: '\nb', prompt: false })
     expect(sanitizer.push('\r')).toEqual({ text: '', prompt: false })
@@ -34,41 +34,41 @@ describe('PtyTerminalSanitizer', () => {
   })
 
   it('reports printable prompt text that follows a marker in a later chunk', () => {
-    const sanitizer = new PtyTerminalSanitizer(64)
-    expect(sanitizer.push('\x1b]133;D;0\x07')).toEqual({ text: '', prompt: true })
-    expect(sanitizer.push('dsh> ')).toEqual({ text: 'dsh> ', prompt: false, promptText: true })
+    const sanitizer = new TerminalSanitizer(64)
+    expect(sanitizer.push('\x1b]133;D;0\x07')).toEqual({ text: '', prompt: true, promptTail: '' })
+    expect(sanitizer.push('dsh> ')).toEqual({ text: 'dsh> ', prompt: false, promptTail: 'dsh> ' })
   })
 
   it('bounds and discards unterminated control sequences through their terminators', () => {
-    const oscBel = new PtyTerminalSanitizer(8)
+    const oscBel = new TerminalSanitizer(8)
     expect(oscBel.push(`\x1b]0;${'x'.repeat(16)}`)).toEqual({ text: '', prompt: false })
     expect(oscBel.push('more\x07tail')).toEqual({ text: 'tail', prompt: false })
 
-    const oscSt = new PtyTerminalSanitizer(8)
+    const oscSt = new TerminalSanitizer(8)
     oscSt.push(`\x1b]0;${'x'.repeat(16)}`)
     expect(oscSt.push('more\x1b')).toEqual({ text: '', prompt: false })
     expect(oscSt.push('\\tail')).toEqual({ text: 'tail', prompt: false })
 
-    const oscDirectSt = new PtyTerminalSanitizer(8)
+    const oscDirectSt = new TerminalSanitizer(8)
     oscDirectSt.push(`\x1b]0;${'x'.repeat(16)}`)
     expect(oscDirectSt.push('more\x1b\\tail')).toEqual({ text: 'tail', prompt: false })
 
-    const oscFalseSt = new PtyTerminalSanitizer(8)
+    const oscFalseSt = new TerminalSanitizer(8)
     oscFalseSt.push(`\x1b]0;${'x'.repeat(16)}`)
     oscFalseSt.push('\x1b')
     expect(oscFalseSt.push('more')).toEqual({ text: '', prompt: false })
     expect(oscFalseSt.push('\x07tail')).toEqual({ text: 'tail', prompt: false })
 
-    const oscNonTerminatingEscape = new PtyTerminalSanitizer(8)
+    const oscNonTerminatingEscape = new TerminalSanitizer(8)
     oscNonTerminatingEscape.push(`\x1b]0;${'x'.repeat(16)}`)
     expect(oscNonTerminatingEscape.push('more\x1bxmore\x07tail')).toEqual({ text: 'tail', prompt: false })
 
-    const csi = new PtyTerminalSanitizer(8)
+    const csi = new TerminalSanitizer(8)
     expect(csi.push(`\x1b[${'1'.repeat(16)}`)).toEqual({ text: '', prompt: false })
     expect(csi.push('123')).toEqual({ text: '', prompt: false })
     expect(csi.push('mtext')).toEqual({ text: 'text', prompt: false })
 
-    const flushed = new PtyTerminalSanitizer(8)
+    const flushed = new TerminalSanitizer(8)
     flushed.push(`\x1b]0;${'x'.repeat(16)}`)
     expect(flushed.flush()).toBe('')
     expect(flushed.push('text')).toEqual({ text: 'text', prompt: false })