Merge remote-tracking branch 'origin/master' into feat/send-unify
# Conflicts: # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl # examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/stdout.expected.jsonl # packages/cordis/tool-cordis/src/api-catalog.ts # packages/pty/pty-local/tests/index.spec.ts # packages/session-query/session-query/tests/tracing.spec.ts
This commit is contained in:
@@ -4,11 +4,16 @@ Owner-scoped persistent PTY seam. `PtyService` registers as `ctx.pty`, mints opa
|
||||
|
||||
## Contract
|
||||
|
||||
- Backends register one stable `type` and return an unpublished `PtyBackendSession`; failed or cancelled setup must clean partial resources.
|
||||
- Backends register one stable `type` and return an unpublished `PtyBackendSession`; failed or cancelled setup must clean partial resources, and a failed cleanup rejects with `PtyBackendCleanupError` so the registry can retain it across cancellation.
|
||||
- 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.
|
||||
- A rollback-close or backend-reported startup cleanup failure rejects the disposing lifecycle instead of claiming quiescence. Caller-triggered cancellation still receives its exact reason; lifecycle-triggered rollback failure also rejects the pending spawn.
|
||||
- A backend cleanup failure that follows caller cancellation remains owner activity until owner or service disposal consumes and reports it, so lifecycle policy cannot mistake failed cleanup for quiescence.
|
||||
- `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.
|
||||
- `PtySendResult.waitReason` and `sessionStatus` are independent. `session_exit` describes the top-level PTY process, not an arbitrary foreground command.
|
||||
- `kill()` and disposal resolve only after the backend's captured process tree is quiescent. A cleanup failure rejects instead of claiming success.
|
||||
- `kill()` and disposal resolve only after the backend's captured process tree is quiescent. A cleanup failure rejects instead of claiming success and clears the matching backend and registry fences so a later close can retry without disturbing a newer attempt.
|
||||
|
||||
The seam contains no `node-pty`, sandbox, tool-schema, prompt, task, or terminal-rendering policy. Implementations own terminal mechanics; consumers own model presentation and optional background-task registration.
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { PtyBackendCleanupError } from './types.ts'
|
||||
import type {
|
||||
PtyBackend,
|
||||
PtyBackendSession,
|
||||
@@ -39,6 +40,7 @@ export type {
|
||||
PtySpawnResult,
|
||||
PtyWaitReason,
|
||||
} from './types.ts'
|
||||
export { PtyBackendCleanupError } from './types.ts'
|
||||
|
||||
/** Opaque identity minted by {@link PtyService} for one live PTY session. */
|
||||
export type PtySessionId = PtySessionIdValue
|
||||
@@ -77,10 +79,6 @@ export function PtySessionId(value: string): PtySessionId {
|
||||
return value as PtySessionId
|
||||
}
|
||||
|
||||
function isAborted(signal: AbortSignal | undefined): boolean {
|
||||
return signal?.aborted === true
|
||||
}
|
||||
|
||||
interface SessionRecord {
|
||||
readonly id: PtySessionId
|
||||
readonly owner: Agent
|
||||
@@ -91,11 +89,24 @@ interface SessionRecord {
|
||||
closing: Promise<void> | undefined
|
||||
}
|
||||
|
||||
interface PendingSpawn {
|
||||
readonly owner: Agent
|
||||
readonly controller: AbortController
|
||||
readonly settled: Promise<void>
|
||||
cleanupFailure: { error: unknown } | undefined
|
||||
}
|
||||
|
||||
interface SpawnReservation {
|
||||
readonly signal: AbortSignal
|
||||
release(cleanupFailure: { error: unknown } | undefined): 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, Set<PendingSpawn>>()
|
||||
private readonly ownerCleanups = new Map<Agent, () => Promise<void> | void>()
|
||||
private readonly disposedOwners = new WeakSet<Agent>()
|
||||
private nextId = 0
|
||||
@@ -142,15 +153,19 @@ export class PtyService extends Service {
|
||||
*/
|
||||
async spawn(owner: Agent, request: PtySpawnRequest, signal?: AbortSignal): Promise<PtySpawnResult> {
|
||||
this.assertActive()
|
||||
signal?.throwIfAborted()
|
||||
this.ensureOwnerCleanup(owner)
|
||||
const backend = this.backends.get(request.type)
|
||||
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')
|
||||
if (isAborted(signal)) throw new Error('PTY spawn aborted')
|
||||
|
||||
const releaseName = this.reserveName(owner, request.name)
|
||||
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
|
||||
let cleanupFailure: { error: unknown } | undefined
|
||||
try {
|
||||
session = await backend.spawn({
|
||||
sessionId,
|
||||
@@ -158,9 +173,13 @@ 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,
|
||||
})
|
||||
if (this.disposing || isAborted(signal) || !this.isLiveOwner(owner)) {
|
||||
signal?.throwIfAborted()
|
||||
if (this.disposing) {
|
||||
throw new PtyError('PTY service is disposing', 'SERVICE_DISPOSING')
|
||||
}
|
||||
if (!this.isLiveOwner(owner)) {
|
||||
throw new PtyError('PTY owner is no longer live', 'OWNER_NOT_LIVE')
|
||||
}
|
||||
const record: SessionRecord = {
|
||||
@@ -175,19 +194,45 @@ export class PtyService extends Service {
|
||||
this.sessions.set(sessionId, record)
|
||||
return this.snapshot(record, session.motd)
|
||||
} catch (error) {
|
||||
if (error instanceof PtyBackendCleanupError) {
|
||||
cleanupFailure = { error: error.cleanupError }
|
||||
}
|
||||
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 }
|
||||
cleanupFailure = rollbackFailure
|
||||
}
|
||||
}
|
||||
throw error
|
||||
let failure: unknown = error
|
||||
try {
|
||||
signal?.throwIfAborted()
|
||||
spawnReservation.signal.throwIfAborted()
|
||||
} catch (cancellation: unknown) {
|
||||
failure = cancellation
|
||||
}
|
||||
if (rollbackFailure !== undefined && signal?.aborted !== true) {
|
||||
throw new AggregateError([failure, rollbackFailure.error], 'PTY spawn and rollback both failed')
|
||||
}
|
||||
throw failure
|
||||
} finally {
|
||||
spawnReservation.release(cleanupFailure)
|
||||
releaseName()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Test whether an exact owner has a published session or unpublished spawn.
|
||||
* @param owner - exact live owner to inspect.
|
||||
* @returns true across the entire spawn-to-close interval, with no publication gap.
|
||||
*/
|
||||
hasOwnerActivity(owner: Agent): boolean {
|
||||
return (this.pendingSpawns.get(owner)?.size ?? 0) > 0
|
||||
|| [...this.sessions.values()].some(record => record.owner === owner)
|
||||
}
|
||||
|
||||
/**
|
||||
* Start one exclusive interactive send.
|
||||
* @param owner - exact session owner.
|
||||
@@ -302,6 +347,43 @@ export class PtyService extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
private reserveSpawn(owner: Agent): SpawnReservation {
|
||||
const controller = new AbortController()
|
||||
const settlement = Promise.withResolvers<void>()
|
||||
const pending: PendingSpawn = { owner, controller, settled: settlement.promise, cleanupFailure: undefined }
|
||||
const owned = this.pendingSpawns.get(owner) ?? new Set<PendingSpawn>()
|
||||
owned.add(pending)
|
||||
this.pendingSpawns.set(owner, owned)
|
||||
return {
|
||||
signal: controller.signal,
|
||||
release: (cleanupFailure) => {
|
||||
pending.cleanupFailure = cleanupFailure
|
||||
if (cleanupFailure === undefined) this.removePendingSpawn(pending)
|
||||
settlement.resolve()
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
private removePendingSpawn(pending: PendingSpawn): void {
|
||||
const owned = this.pendingSpawns.get(pending.owner)
|
||||
if (owned === undefined) return
|
||||
owned.delete(pending)
|
||||
if (owned.size === 0) this.pendingSpawns.delete(pending.owner)
|
||||
}
|
||||
|
||||
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))
|
||||
const failures = pending.flatMap(spawn => spawn.cleanupFailure === undefined ? [] : [spawn.cleanupFailure.error])
|
||||
for (const spawn of pending) this.removePendingSpawn(spawn)
|
||||
if (failures.length > 0) {
|
||||
throw new AggregateError(failures, 'failed to roll back unpublished PTY setup')
|
||||
}
|
||||
}
|
||||
|
||||
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')
|
||||
@@ -322,23 +404,49 @@ export class PtyService extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
private async abortAndClose(owner: Agent | undefined, abortReason: PtyError, closeReason: string): Promise<void> {
|
||||
const failures: unknown[] = []
|
||||
try {
|
||||
await this.abortPendingSpawns(owner, abortReason)
|
||||
} catch (error: unknown) {
|
||||
failures.push(error)
|
||||
}
|
||||
const records = [...this.sessions.values()].filter(record => owner === undefined || record.owner === owner)
|
||||
try {
|
||||
await this.closeRecords(records, closeReason)
|
||||
} catch (error: unknown) {
|
||||
failures.push(error)
|
||||
}
|
||||
if (failures.length > 0) throw new AggregateError(failures, 'failed to clean up PTY lifecycle')
|
||||
}
|
||||
|
||||
private async disposeOwned(owner: Agent): Promise<void> {
|
||||
const owned = [...this.sessions.values()].filter(record => record.owner === owner)
|
||||
await this.closeRecords(owned, 'PTY owner disposed')
|
||||
this.reservedNames.delete(owner)
|
||||
try {
|
||||
await this.abortAndClose(
|
||||
owner,
|
||||
new PtyError('PTY owner is no longer live', 'OWNER_NOT_LIVE'),
|
||||
'PTY owner disposed',
|
||||
)
|
||||
} finally {
|
||||
this.reservedNames.delete(owner)
|
||||
}
|
||||
}
|
||||
|
||||
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.closeRecords(records, 'PTY service disposed')
|
||||
await this.abortAndClose(
|
||||
undefined,
|
||||
new PtyError('PTY service is disposing', 'SERVICE_DISPOSING'),
|
||||
'PTY service disposed',
|
||||
)
|
||||
} finally {
|
||||
this.backends.clear()
|
||||
this.reservedNames.clear()
|
||||
this.pendingSpawns.clear()
|
||||
const cleanups = [...this.ownerCleanups.values()]
|
||||
this.ownerCleanups.clear()
|
||||
await Promise.all(cleanups.map(cleanup => Promise.resolve(cleanup())))
|
||||
@@ -349,8 +457,14 @@ export class PtyService extends Service {
|
||||
const results = await Promise.allSettled(records.map(async (record) => {
|
||||
const closing = record.closing ?? record.session.close(reason)
|
||||
record.closing = closing
|
||||
await closing
|
||||
this.sessions.delete(record.id)
|
||||
try {
|
||||
await closing
|
||||
this.sessions.delete(record.id)
|
||||
} catch (error: unknown) {
|
||||
// A concurrent retry may already own a newer fence; never clear it.
|
||||
if (record.closing === closing) record.closing = undefined
|
||||
throw error
|
||||
}
|
||||
}))
|
||||
const failures = results
|
||||
.filter((result): result is PromiseRejectedResult => result.status === 'rejected')
|
||||
|
||||
@@ -10,6 +10,21 @@ import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
/** Internal exported basis for the public `PtySessionId` type/value pair. */
|
||||
export type PtySessionIdValue = Branded<'PtySessionId'>
|
||||
|
||||
/**
|
||||
* Backend-reported failure to clean partial resources after unpublished setup failed.
|
||||
* @param spawnError - original setup or cancellation failure.
|
||||
* @param cleanupError - failure that may leave backend-owned resources alive.
|
||||
*/
|
||||
export class PtyBackendCleanupError extends AggregateError {
|
||||
constructor(
|
||||
readonly spawnError: unknown,
|
||||
readonly cleanupError: unknown,
|
||||
) {
|
||||
super([spawnError, cleanupError], 'PTY backend startup and cleanup both failed')
|
||||
this.name = 'PtyBackendCleanupError'
|
||||
}
|
||||
}
|
||||
|
||||
/** Why one interactive send returned control to its caller. */
|
||||
export type PtyWaitReason = 'stdin_read' | 'inferred_idle' | 'timeout' | 'session_exit'
|
||||
|
||||
@@ -147,7 +162,7 @@ export interface PtyBackendSession {
|
||||
export interface PtyBackend {
|
||||
/** Stable type selected by {@link PtySpawnRequest.type}. */
|
||||
readonly type: string
|
||||
/** Create an unpublished session or reject after cleaning partial resources. */
|
||||
/** Create an unpublished session or reject after cleaning partial resources; cleanup failure uses {@link PtyBackendCleanupError}. */
|
||||
spawn(spec: PtyBackendSpawnSpec): Promise<PtyBackendSession>
|
||||
}
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ import { Context } from 'cordis'
|
||||
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import AgentRegistry, { AgentMessageId } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import PtyService, { PtyError, PtySessionId } from '@deepseek-ai/dsh-pty'
|
||||
import PtyService, { PtyBackendCleanupError, PtyError, PtySessionId } from '@deepseek-ai/dsh-pty'
|
||||
import type {
|
||||
PtyBackend,
|
||||
PtyBackendSession,
|
||||
@@ -161,6 +161,7 @@ describe('PtyService ownership and lifecycle', () => {
|
||||
|
||||
const created = await ctx.pty.spawn(owner, { type: 'stub', name: 'main', cwd: '/tmp' })
|
||||
expect(created).toMatchObject({ sessionId: 'pty-1', name: 'main', type: 'stub', pid: 123, motd: 'stub ready', status: { kind: 'running' } })
|
||||
expect(ctx.pty.hasOwnerActivity(owner)).toBe(true)
|
||||
expect(ctx.pty.list(owner)).toHaveLength(1)
|
||||
expect(ctx.pty.list(foreign)).toEqual([])
|
||||
expect(() => ctx.pty.read(foreign, created.sessionId)).toThrow('belongs to another agent')
|
||||
@@ -179,8 +180,9 @@ describe('PtyService ownership and lifecycle', () => {
|
||||
const created = await ctx.pty.spawn(owner, { type: 'stub', name: 'main' })
|
||||
await expect(ctx.pty.spawn(owner, { type: 'stub', name: '' })).rejects.toThrow('must be non-empty')
|
||||
const aborted = new AbortController()
|
||||
aborted.abort()
|
||||
await expect(ctx.pty.spawn(owner, { type: 'stub' }, aborted.signal)).rejects.toThrow('spawn aborted')
|
||||
const abortReason = new Error('spawn aborted')
|
||||
aborted.abort(abortReason)
|
||||
await expect(ctx.pty.spawn(owner, { type: 'stub' }, aborted.signal)).rejects.toBe(abortReason)
|
||||
await expect(ctx.pty.spawn(owner, { type: 'stub', name: 'main' })).rejects.toMatchObject({ code: 'DUPLICATE_NAME' })
|
||||
|
||||
const operation = ctx.pty.startSend(owner, created.sessionId, { text: 'echo hi', submit: true })
|
||||
@@ -206,12 +208,222 @@ 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'])
|
||||
})
|
||||
|
||||
it('preserves caller cancellation when a pending backend spawn completes', async () => {
|
||||
const ctx = await harness()
|
||||
const gate = Promise.withResolvers<PtyBackendSession>()
|
||||
const session = new StubSession()
|
||||
ctx.pty.registerBackend({ type: 'slow', spawn: () => gate.promise })
|
||||
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: 'slow' }, controller.signal)
|
||||
controller.abort(reason)
|
||||
gate.resolve(session)
|
||||
|
||||
await expect(pending).rejects.toBe(reason)
|
||||
expect(session.closed).toEqual(['PTY spawn rolled back'])
|
||||
expect(ctx.agents.get(owner.id)).toBe(owner)
|
||||
})
|
||||
|
||||
it('preserves caller cancellation when unpublished rollback fails', async () => {
|
||||
const ctx = await harness()
|
||||
const gate = Promise.withResolvers<PtyBackendSession>()
|
||||
const session = new StubSession()
|
||||
session.rejectClose = true
|
||||
ctx.pty.registerBackend({ type: 'slow', spawn: () => gate.promise })
|
||||
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: 'slow' }, controller.signal)
|
||||
controller.abort(reason)
|
||||
gate.resolve(session)
|
||||
|
||||
await expect(pending).rejects.toBe(reason)
|
||||
expect(ctx.pty.hasOwnerActivity(owner)).toBe(true)
|
||||
const internal = ctx.pty as unknown as { disposeAll(): Promise<void> }
|
||||
await expect(internal.disposeAll()).rejects.toThrow('failed to clean up PTY lifecycle')
|
||||
expect(ctx.pty.hasOwnerActivity(owner)).toBe(false)
|
||||
expect(session.closed).toEqual(['PTY spawn rolled back'])
|
||||
})
|
||||
|
||||
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(['owner', 'service'] as const)('retains caller-triggered backend cleanup failure until %s disposal', async (scope) => {
|
||||
const ctx = await harness()
|
||||
const started = Promise.withResolvers<undefined>()
|
||||
const cleanupFailure = new Error('backend cleanup failed')
|
||||
ctx.pty.registerBackend({
|
||||
type: 'cleanup-failing',
|
||||
spawn: ({ signal }) => new Promise((_resolve, reject) => {
|
||||
if (signal === undefined) throw new Error('missing spawn signal')
|
||||
started.resolve(undefined)
|
||||
signal.addEventListener('abort', () => {
|
||||
reject(new PtyBackendCleanupError(signal.reason, cleanupFailure))
|
||||
}, { 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: 'cleanup-failing' }, controller.signal)
|
||||
await started.promise
|
||||
controller.abort(reason)
|
||||
|
||||
await expect(pending).rejects.toBe(reason)
|
||||
expect(ctx.pty.hasOwnerActivity(owner)).toBe(true)
|
||||
const internal = ctx.pty as unknown as {
|
||||
disposeOwned(owner: Agent): Promise<void>
|
||||
disposeAll(): Promise<void>
|
||||
}
|
||||
const disposal = scope === 'owner' ? internal.disposeOwned(owner) : internal.disposeAll()
|
||||
await expect(disposal).rejects.toThrow('failed to clean up PTY lifecycle')
|
||||
expect(ctx.pty.hasOwnerActivity(owner)).toBe(false)
|
||||
})
|
||||
|
||||
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()
|
||||
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' })
|
||||
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)
|
||||
|
||||
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'])
|
||||
})
|
||||
|
||||
it('reports unpublished rollback failure through service disposal', async () => {
|
||||
const ctx = await harness()
|
||||
const gate = Promise.withResolvers<PtyBackendSession>()
|
||||
const session = new StubSession()
|
||||
session.rejectClose = true
|
||||
ctx.pty.registerBackend({ type: 'slow', spawn: () => gate.promise })
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
|
||||
const pending = ctx.pty.spawn(owner, { type: 'slow' })
|
||||
const pendingFailure = expect(pending).rejects.toThrow('PTY spawn and rollback both failed')
|
||||
const internal = ctx.pty as unknown as { disposeAll(): Promise<void> }
|
||||
const disposalFailure = expect(internal.disposeAll()).rejects.toThrow('failed to clean up PTY lifecycle')
|
||||
gate.resolve(session)
|
||||
|
||||
await pendingFailure
|
||||
await disposalFailure
|
||||
expect(session.closed).toEqual(['PTY spawn rolled back'])
|
||||
})
|
||||
|
||||
it.each([
|
||||
{ scope: 'owner', code: 'OWNER_NOT_LIVE' },
|
||||
{ scope: 'service', code: 'SERVICE_DISPOSING' },
|
||||
] as const)('$scope disposal retains backend-side startup cleanup failure', async ({ scope, code }) => {
|
||||
const ctx = await harness()
|
||||
const started = Promise.withResolvers<undefined>()
|
||||
const cleanupFailure = new Error('backend cleanup failed')
|
||||
let backendAbortReason: unknown
|
||||
ctx.pty.registerBackend({
|
||||
type: 'cleanup-failing',
|
||||
spawn: ({ signal }) => new Promise((_resolve, reject) => {
|
||||
if (signal === undefined) throw new Error('missing spawn signal')
|
||||
started.resolve(undefined)
|
||||
signal.addEventListener('abort', () => {
|
||||
backendAbortReason = signal.reason
|
||||
reject(new PtyBackendCleanupError(signal.reason, cleanupFailure))
|
||||
}, { once: true })
|
||||
}),
|
||||
})
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
|
||||
const pending = ctx.pty.spawn(owner, { type: 'cleanup-failing' })
|
||||
await started.promise
|
||||
const internal = ctx.pty as unknown as {
|
||||
disposeOwned(owner: Agent): Promise<void>
|
||||
disposeAll(): Promise<void>
|
||||
}
|
||||
const disposal = scope === 'owner' ? internal.disposeOwned(owner) : internal.disposeAll()
|
||||
const pendingError = await pending.then(
|
||||
() => { throw new Error('pending spawn unexpectedly succeeded') },
|
||||
(error: unknown) => error,
|
||||
)
|
||||
|
||||
expect(pendingError).toBe(backendAbortReason)
|
||||
expect(pendingError).toMatchObject({ code })
|
||||
const disposalError = await disposal.then(
|
||||
() => { throw new Error('disposal unexpectedly succeeded') },
|
||||
(error: unknown) => error,
|
||||
)
|
||||
expect(disposalError).toMatchObject({ message: 'failed to clean up PTY lifecycle' })
|
||||
const rollbackError = (disposalError as AggregateError).errors[0] as unknown
|
||||
const cleanupErrors = (rollbackError as AggregateError).errors as unknown[]
|
||||
expect(cleanupErrors).toEqual([cleanupFailure])
|
||||
})
|
||||
|
||||
it('keeps independent reservations and handles provider failure before publication', async () => {
|
||||
const ctx = await harness()
|
||||
const firstGate = Promise.withResolvers<PtyBackendSession>()
|
||||
@@ -255,14 +467,27 @@ describe('PtyService ownership and lifecycle', () => {
|
||||
ctx.agents.register(owner)
|
||||
const failedSpawn = new StubSession()
|
||||
failedSpawn.rejectClose = true
|
||||
let ownerDisposal = Promise.resolve()
|
||||
const internal = ctx.pty as unknown as {
|
||||
disposedOwners: WeakSet<Agent>
|
||||
disposeOwned(owner: Agent): Promise<void>
|
||||
}
|
||||
ctx.pty.registerBackend({
|
||||
type: 'bad-spawn',
|
||||
async spawn() {
|
||||
await disposeAgentScope(owner)
|
||||
async spawn({ signal }) {
|
||||
if (signal === undefined) throw new Error('missing spawn signal')
|
||||
internal.disposedOwners.add(owner)
|
||||
ownerDisposal = internal.disposeOwned(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 expect(ownerDisposal).rejects.toThrow('failed to clean up PTY lifecycle')
|
||||
|
||||
const nextOwner = stubAgent(ctx, 'next')
|
||||
ctx.agents.register(nextOwner)
|
||||
@@ -339,8 +564,15 @@ describe('PtyService ownership and lifecycle', () => {
|
||||
sessions: Map<PtySessionIdType, unknown>
|
||||
closeRecords(records: unknown[], reason: string): Promise<void>
|
||||
}
|
||||
await expect(internal.closeRecords([...internal.sessions.values()], 'test failure')).rejects.toThrow('failed to close 1 PTY session')
|
||||
const records = [...internal.sessions.values()]
|
||||
const firstFailure = expect(internal.closeRecords(records, 'test failure')).rejects.toThrow('failed to close 1 PTY session')
|
||||
const joinedFailure = expect(internal.closeRecords(records, 'joined failure')).rejects.toThrow('failed to close 1 PTY session')
|
||||
await firstFailure
|
||||
await joinedFailure
|
||||
b.sessions[0]!.rejectClose = false
|
||||
await expect(internal.closeRecords([...internal.sessions.values()], 'retry')).resolves.toBeUndefined()
|
||||
expect(b.sessions[0]!.closed).toEqual(['test failure', 'retry'])
|
||||
expect(internal.sessions.size).toBe(0)
|
||||
await disposePtyService(ctx)
|
||||
await expect(service.spawn(owner, { type: 'stub' })).rejects.toMatchObject({ code: 'SERVICE_DISPOSING' })
|
||||
})
|
||||
@@ -361,7 +593,7 @@ describe('PtyService ownership and lifecycle', () => {
|
||||
}
|
||||
// Teardown surfaces the close failure, but its finally still clears the
|
||||
// backend and owner-cleanup registries instead of orphaning them.
|
||||
await expect(internal.disposeAll()).rejects.toThrow('failed to close 1 PTY session')
|
||||
await expect(internal.disposeAll()).rejects.toThrow('failed to clean up PTY lifecycle')
|
||||
expect(internal.backends.size).toBe(0)
|
||||
expect(internal.ownerCleanups.size).toBe(0)
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user