fix(pty): close provider cancellation races

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

@@ -82,6 +82,22 @@ function spawnArgv(ctx: Context, config: ResolvedConfig, spec: PtyBackendSpawnSp
   }).argv
 }
 
+async function initializeSession(session: LocalPtySession, signal?: AbortSignal): Promise<void> {
+  if (signal === undefined) {
+    await session.initialize(signal)
+    return
+  }
+  const aborted = Promise.withResolvers<never>()
+  const onAbort = (): void => { aborted.reject(signal.reason) }
+  signal.addEventListener('abort', onAbort, { once: true })
+  try {
+    signal.throwIfAborted()
+    await Promise.race([session.initialize(signal), aborted.promise])
+  } finally {
+    signal.removeEventListener('abort', onAbort)
+  }
+}
+
 /** Local shell backend registered under the configured type. */
 export class LocalPtyBackend implements PtyBackend {
   readonly type: string
@@ -116,7 +132,7 @@ export class LocalPtyBackend implements PtyBackend {
     })
     const session = this.createSession(terminal, this.config)
     try {
-      await session.initialize(spec.signal)
+      await initializeSession(session, spec.signal)
       return session
     } catch (error) {
       try {

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

@@ -6,6 +6,7 @@ import type {
   SubprocessTerminalForeground,
   SubprocessTerminalHandle,
 } from '@deepseek-ai/dsh-subprocess'
+import { PtyError } from '@deepseek-ai/dsh-pty'
 import type {
   PtyBackendSession,
   PtyReadRequest,
@@ -219,7 +220,7 @@ 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) throw new PtyError('PTY session already has an active send or draining provider operation', 'SEND_ACTIVE')
     if (request.signal?.aborted === true) throw new Error('PTY send aborted before write')
 
     const operation = new LocalSendOperation(
@@ -247,7 +248,7 @@ export class LocalPtySession implements PtyBackendSession {
   private async beginSend(operation: LocalSendOperation, request: PtySendRequest): Promise<void> {
     try {
       const foreground = await this.terminal.inspectForeground()
-      if (this.active !== operation || this.closing) return
+      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) {

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

@@ -143,6 +143,32 @@ describe('LocalPtyBackend startup rollback', () => {
     } satisfies Partial<PtyBackendCleanupError>))
   })
 
+  it('starts startup rollback when cancellation wins a stalled initialization', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SandboxPolicyService, { mode: 'danger-full-access', workspaceRoot: '/tmp' })
+    const initialization = Promise.withResolvers<undefined>()
+    const initializationStarted = Promise.withResolvers<undefined>()
+    const close = vi.fn<() => Promise<void>>().mockResolvedValue(undefined)
+    const session = {
+      initialize: () => {
+        initializationStarted.resolve(undefined)
+        return initialization.promise
+      },
+      close,
+    } as unknown as LocalPtySession
+    const backend = new LocalPtyBackend(ctx, config(), async () => terminalHandle(), () => session)
+    const controller = new AbortController()
+    const reason = new Error('cancel stalled startup')
+
+    const spawning = backend.spawn(spec(agent(ctx), controller.signal))
+    await initializationStarted.promise
+    controller.abort(reason)
+
+    await expect(spawning).rejects.toBe(reason)
+    expect(close).toHaveBeenCalledWith('PTY startup failed')
+    initialization.resolve(undefined)
+  })
+
   it('wraps confined argv, scrubs the environment, and returns initialized sessions', async () => {
     const ctx = new Context()
     await ctx.plugin(RecordingSandbox)

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

@@ -8,6 +8,7 @@ import type {
   SubprocessTerminalHandle,
   SubprocessTerminalSignal,
 } from '@deepseek-ai/dsh-subprocess'
+import { PtyError } from '@deepseek-ai/dsh-pty'
 import type {
   ProcessIdentity,
   ProcessInspector,
@@ -374,6 +375,54 @@ describe('LocalPtySession readiness and output', () => {
     expect(inspector.groups).not.toContainEqual([789, 'SIGINT'])
   })
 
+  it('does not let an in-flight readiness inspection release a canceled send before signalling settles', async () => {
+    vi.useFakeTimers()
+    const terminal = new FakeTerminal()
+    const inspector = new FakeInspector()
+    const session = makeSession(terminal, inspector, config())
+    await initialize(session, terminal)
+
+    const readiness = Promise.withResolvers<{ processGroupId: number; inputWaiting: boolean }>()
+    let inspections = 0
+    terminal.inspectForeground = async () => {
+      inspections += 1
+      if (inspections === 1) return { processGroupId: 456, inputWaiting: false }
+      if (inspections === 2) return await readiness.promise
+      return { processGroupId: 456, inputWaiting: true }
+    }
+    const signalling = Promise.withResolvers<undefined>()
+    const signalled = Promise.withResolvers<number>()
+    terminal.signalForeground = async (signal) => {
+      await signalling.promise
+      const foreground = await terminal.inspectForeground()
+      if (foreground === undefined) throw new Error('cannot resolve foreground')
+      inspector.signalGroup(foreground.processGroupId, signal)
+      signalled.resolve(foreground.processGroupId)
+      return foreground.processGroupId
+    }
+
+    const operation = session.startSend({ text: 'first', submit: true })
+    await Promise.resolve()
+    await Promise.resolve()
+    await vi.advanceTimersByTimeAsync(10)
+    expect(inspections).toBe(2)
+    expect(operation.cancel()).toBe(true)
+    let settled = false
+    void operation.done.then(() => { settled = true })
+
+    readiness.resolve({ processGroupId: 456, inputWaiting: true })
+    await Promise.resolve()
+    await Promise.resolve()
+    expect(settled).toBe(false)
+    expect(() => session.startSend({ text: 'successor', submit: true })).toThrow(PtyError)
+
+    signalling.resolve(undefined)
+    expect(await signalled.promise).toBe(456)
+    expect(inspector.groups).toContainEqual([456, 'SIGINT'])
+    await session.close('test complete')
+    expect((await operation.done).waitReason).toBe('session_exit')
+  })
+
   it('signals only after an in-flight provider write lands', async () => {
     vi.useFakeTimers()
     const terminal = new FakeTerminal()
@@ -438,7 +487,9 @@ describe('LocalPtySession readiness and output', () => {
     await vi.advanceTimersByTimeAsync(100)
 
     expect((await operation.done).waitReason).toBe('timeout')
-    expect(() => session.startSend({ text: 'must wait', submit: true })).toThrow('active send')
+    expect(() => session.startSend({ text: 'must wait', submit: true })).toThrow(expect.objectContaining({
+      code: 'SEND_ACTIVE',
+    }))
 
     writeGate.resolve(undefined)
     await vi.advanceTimersByTimeAsync(0)