fix(pty): await unpublished spawn teardown

packages/pty/pty/README.md

@@ -6,6 +6,7 @@ Owner-scoped persistent PTY seam. `PtyService` registers as `ctx.pty`, mints opa
 
 - Backends register one stable `type` and return an unpublished `PtyBackendSession`; failed or cancelled setup must clean partial resources.
 - Spawn cancellation preserves the caller's exact abort reason. Service disposal and owner loss remain distinct machine-routable failures after backend setup.
+- Owner and service disposal abort unpublished setup through a service-owned signal and await backend settlement plus rollback before returning.
 - `hasOwnerActivity(owner)` spans unpublished setup through final close, so lifecycle policy can fence the exact owner without a publication race.
 - A successful spawn publishes one `PtySessionId`. The optional `name` is owner-local display metadata, never authority.
 - One session accepts at most one live send operation. Reads and signals may observe it; another send fails until the operation settles.

packages/pty/pty/src/index.ts

@@ -87,12 +87,22 @@ interface SessionRecord {
   closing: Promise<void> | undefined
 }
 
+interface PendingSpawn {
+  readonly controller: AbortController
+  readonly settled: Promise<void>
+}
+
+interface SpawnReservation {
+  readonly signal: AbortSignal
+  release(): void
+}
+
 /** In-process registry for replaceable PTY backends and exact-Agent sessions. */
 export class PtyService extends Service {
   private readonly backends = new Map<string, PtyBackend>()
   private readonly sessions = new Map<PtySessionId, SessionRecord>()
   private readonly reservedNames = new Map<Agent, Set<string>>()
-  private readonly pendingSpawns = new Map<Agent, number>()
+  private readonly pendingSpawns = new Map<Agent, Set<PendingSpawn>>()
   private readonly ownerCleanups = new Map<Agent, () => Promise<void> | void>()
   private readonly disposedOwners = new WeakSet<Agent>()
   private nextId = 0
@@ -145,7 +155,10 @@ export class PtyService extends Service {
     if (backend === undefined) throw new PtyError(`no PTY backend registered for "${request.type}"`, 'NO_BACKEND')
     if (request.name !== undefined && request.name.length === 0) throw new Error('PTY session name must be non-empty')
     const releaseName = this.reserveName(owner, request.name)
-    const releaseSpawn = this.reserveSpawn(owner)
+    const spawnReservation = this.reserveSpawn(owner)
+    const backendSignal = signal === undefined
+      ? spawnReservation.signal
+      : AbortSignal.any([signal, spawnReservation.signal])
     const sessionId = PtySessionId(`pty-${++this.nextId}`)
     let session: PtyBackendSession | undefined
     try {
@@ -155,7 +168,7 @@ export class PtyService extends Service {
         type: request.type,
         ...request.name !== undefined ? { name: request.name } : {},
         ...request.cwd !== undefined ? { cwd: request.cwd } : {},
-        ...signal !== undefined ? { signal } : {},
+        signal: backendSignal,
       })
       signal?.throwIfAborted()
       if (this.disposing) {
@@ -176,16 +189,27 @@ export class PtyService extends Service {
       this.sessions.set(sessionId, record)
       return this.snapshot(record, session.motd)
     } catch (error) {
+      let rollbackFailure: { error: unknown } | undefined
       if (session !== undefined && !this.sessions.has(sessionId)) {
         try {
           await session.close('PTY spawn rolled back')
         } catch (closeError: unknown) {
-          throw new AggregateError([error, closeError], 'PTY spawn and rollback both failed')
+          rollbackFailure = { error: closeError }
         }
       }
-      throw error
+      let failure: unknown = error
+      try {
+        signal?.throwIfAborted()
+        spawnReservation.signal.throwIfAborted()
+      } catch (cancellation: unknown) {
+        failure = cancellation
+      }
+      if (rollbackFailure !== undefined) {
+        throw new AggregateError([failure, rollbackFailure.error], 'PTY spawn and rollback both failed')
+      }
+      throw failure
     } finally {
-      releaseSpawn()
+      spawnReservation.release()
       releaseName()
     }
   }
@@ -196,7 +220,7 @@ export class PtyService extends Service {
    * @returns true across the entire spawn-to-close interval, with no publication gap.
    */
   hasOwnerActivity(owner: Agent): boolean {
-    return (this.pendingSpawns.get(owner) ?? 0) > 0
+    return (this.pendingSpawns.get(owner)?.size ?? 0) > 0
       || [...this.sessions.values()].some(record => record.owner === owner)
   }
 
@@ -314,15 +338,31 @@ export class PtyService extends Service {
     }
   }
 
-  private reserveSpawn(owner: Agent): () => void {
-    this.pendingSpawns.set(owner, (this.pendingSpawns.get(owner) ?? 0) + 1)
-    return () => {
-      const remaining = (this.pendingSpawns.get(owner) ?? 1) - 1
-      if (remaining === 0) this.pendingSpawns.delete(owner)
-      else this.pendingSpawns.set(owner, remaining)
+  private reserveSpawn(owner: Agent): SpawnReservation {
+    const controller = new AbortController()
+    const settlement = Promise.withResolvers<void>()
+    const pending: PendingSpawn = { controller, settled: settlement.promise }
+    const owned = this.pendingSpawns.get(owner) ?? new Set<PendingSpawn>()
+    owned.add(pending)
+    this.pendingSpawns.set(owner, owned)
+    return {
+      signal: controller.signal,
+      release: () => {
+        owned.delete(pending)
+        if (owned.size === 0) this.pendingSpawns.delete(owner)
+        settlement.resolve()
+      },
     }
   }
 
+  private async abortPendingSpawns(owner: Agent | undefined, reason: PtyError): Promise<void> {
+    const pending = owner === undefined
+      ? [...this.pendingSpawns.values()].flatMap(owned => [...owned])
+      : [...(this.pendingSpawns.get(owner) ?? [])]
+    for (const spawn of pending) spawn.controller.abort(reason)
+    await Promise.all(pending.map(spawn => spawn.settled))
+  }
+
   private expectOwned(owner: Agent, id: PtySessionId): SessionRecord {
     const record = this.sessions.get(id)
     if (record === undefined) throw new PtyError(`unknown PTY session ${id}`, 'NO_SESSION')
@@ -344,6 +384,7 @@ export class PtyService extends Service {
   }
 
   private async disposeOwned(owner: Agent): Promise<void> {
+    await this.abortPendingSpawns(owner, new PtyError('PTY owner is no longer live', 'OWNER_NOT_LIVE'))
     const owned = [...this.sessions.values()].filter(record => record.owner === owner)
     await this.closeRecords(owned, 'PTY owner disposed')
     this.reservedNames.delete(owner)
@@ -351,11 +392,12 @@ export class PtyService extends Service {
 
   private async disposeAll(): Promise<void> {
     this.disposing = true
-    const records = [...this.sessions.values()]
     // Teardown is best-effort: a close failure still clears registries and runs
     // owner cleanups before the aggregated error propagates, so one stuck
     // session cannot orphan backends, reservations, or owner detachers.
     try {
+      await this.abortPendingSpawns(undefined, new PtyError('PTY service is disposing', 'SERVICE_DISPOSING'))
+      const records = [...this.sessions.values()]
       await this.closeRecords(records, 'PTY service disposed')
     } finally {
       this.backends.clear()

packages/pty/pty/tests/service.spec.ts

@@ -206,9 +206,10 @@ describe('PtyService ownership and lifecycle', () => {
     ctx.agents.register(owner)
     const pending = ctx.pty.spawn(owner, { type: 'slow', name: 'main' })
     await expect(ctx.pty.spawn(owner, { type: 'slow', name: 'main' })).rejects.toMatchObject({ code: 'DUPLICATE_NAME' })
-    await disposeAgentScope(owner)
+    const disposal = disposeAgentScope(owner)
     gate.resolve(session)
     await expect(pending).rejects.toMatchObject({ code: 'OWNER_NOT_LIVE' })
+    await disposal
     expect(session.closed).toEqual(['PTY spawn rolled back'])
   })
 
@@ -231,19 +232,70 @@ describe('PtyService ownership and lifecycle', () => {
     expect(ctx.agents.get(owner.id)).toBe(owner)
   })
 
-  it('rolls back an unpublished backend session when service disposal wins', async () => {
+  it('preserves caller cancellation when a backend rejects in response to it', async () => {
+    const ctx = await harness()
+    const started = Promise.withResolvers<undefined>()
+    const backendFailure = new Error('backend observed cancellation')
+    ctx.pty.registerBackend({
+      type: 'abortable',
+      spawn: ({ signal }) => new Promise((_resolve, reject) => {
+        if (signal === undefined) throw new Error('missing spawn signal')
+        started.resolve(undefined)
+        signal.addEventListener('abort', () => { reject(backendFailure) }, { once: true })
+      }),
+    })
+    const owner = stubAgent(ctx, 'owner')
+    ctx.agents.register(owner)
+    const controller = new AbortController()
+    const reason = new Error('cancelled by caller')
+
+    const pending = ctx.pty.spawn(owner, { type: 'abortable' }, controller.signal)
+    await started.promise
+    controller.abort(reason)
+
+    await expect(pending).rejects.toBe(reason)
+  })
+
+  it.each([
+    { scope: 'owner', code: 'OWNER_NOT_LIVE' },
+    { scope: 'service', code: 'SERVICE_DISPOSING' },
+  ] as const)('$scope disposal aborts and awaits unpublished backend setup', async ({ scope, code }) => {
     const ctx = await harness()
     const gate = Promise.withResolvers<PtyBackendSession>()
+    const started = Promise.withResolvers<undefined>()
     const session = new StubSession()
-    ctx.pty.registerBackend({ type: 'slow', spawn: () => gate.promise })
+    let backendSignal: AbortSignal | undefined
+    ctx.pty.registerBackend({
+      type: 'slow',
+      spawn: (spec) => {
+        backendSignal = spec.signal
+        started.resolve(undefined)
+        return gate.promise
+      },
+    })
     const owner = stubAgent(ctx, 'owner')
     ctx.agents.register(owner)
 
     const pending = ctx.pty.spawn(owner, { type: 'slow' })
-    await disposePtyService(ctx)
+    const pendingFailure = pending.then(
+      () => { throw new Error('pending spawn unexpectedly succeeded') },
+      (error: unknown) => error,
+    )
+    await started.promise
+    let disposalSettled = false
+    const disposal = (scope === 'owner' ? disposeAgentScope(owner) : disposePtyService(ctx))
+      .then(() => { disposalSettled = true })
+    await new Promise(resolve => setTimeout(resolve, 0))
+    const signalAbortedBeforeRelease = backendSignal?.aborted ?? false
+    const signalReasonBeforeRelease = backendSignal?.reason as unknown
+    const disposalSettledBeforeRelease = disposalSettled
     gate.resolve(session)
 
-    await expect(pending).rejects.toMatchObject({ code: 'SERVICE_DISPOSING' })
+    expect(await pendingFailure).toMatchObject({ code })
+    await disposal
+    expect(signalAbortedBeforeRelease).toBe(true)
+    expect(signalReasonBeforeRelease).toMatchObject({ code })
+    expect(disposalSettledBeforeRelease).toBe(false)
     expect(session.closed).toEqual(['PTY spawn rolled back'])
   })
 
@@ -290,14 +342,22 @@ describe('PtyService ownership and lifecycle', () => {
     ctx.agents.register(owner)
     const failedSpawn = new StubSession()
     failedSpawn.rejectClose = true
+    let ownerDisposal = Promise.resolve()
     ctx.pty.registerBackend({
       type: 'bad-spawn',
-      async spawn() {
-        await disposeAgentScope(owner)
+      async spawn({ signal }) {
+        if (signal === undefined) throw new Error('missing spawn signal')
+        ownerDisposal = disposeAgentScope(owner)
+        if (!signal.aborted) {
+          await new Promise<undefined>((resolve) => {
+            signal.addEventListener('abort', () => { resolve(undefined) }, { once: true })
+          })
+        }
         return failedSpawn
       },
     })
     await expect(ctx.pty.spawn(owner, { type: 'bad-spawn' })).rejects.toThrow('spawn and rollback both failed')
+    await ownerDisposal
 
     const nextOwner = stubAgent(ctx, 'next')
     ctx.agents.register(nextOwner)