fix(pty): retain cleanup evidence through policy
This commit is contained in:
@@ -8,6 +8,7 @@ Owner-scoped persistent PTY seam. `PtyService` registers as `ctx.pty`, mints opa
|
||||
- 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 service rollback or backend-reported startup cleanup failure rejects the disposing lifecycle instead of claiming quiescence; the spawn caller still receives its exact cancellation reason.
|
||||
- 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.
|
||||
|
||||
@@ -90,6 +90,7 @@ interface SessionRecord {
|
||||
}
|
||||
|
||||
interface PendingSpawn {
|
||||
readonly owner: Agent
|
||||
readonly controller: AbortController
|
||||
readonly settled: Promise<void>
|
||||
cleanupFailure: { error: unknown } | undefined
|
||||
@@ -349,7 +350,7 @@ export class PtyService extends Service {
|
||||
private reserveSpawn(owner: Agent): SpawnReservation {
|
||||
const controller = new AbortController()
|
||||
const settlement = Promise.withResolvers<void>()
|
||||
const pending: PendingSpawn = { controller, settled: settlement.promise, cleanupFailure: undefined }
|
||||
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)
|
||||
@@ -357,13 +358,19 @@ export class PtyService extends Service {
|
||||
signal: controller.signal,
|
||||
release: (cleanupFailure) => {
|
||||
pending.cleanupFailure = cleanupFailure
|
||||
owned.delete(pending)
|
||||
if (owned.size === 0) this.pendingSpawns.delete(owner)
|
||||
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])
|
||||
@@ -371,6 +378,7 @@ export class PtyService extends Service {
|
||||
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')
|
||||
}
|
||||
|
||||
@@ -160,6 +160,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')
|
||||
@@ -256,6 +257,40 @@ describe('PtyService ownership and lifecycle', () => {
|
||||
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' },
|
||||
|
||||
@@ -11,7 +11,7 @@ Six model-facing tools over `ctx.pty`: `terminal_open`, `terminal_send`, `termin
|
||||
| `enableRunInBackground` | `true` | expose and accept `run_in_background`; false omits the schema field and rejects a forced undeclared argument |
|
||||
| `maxResultBytes` | `262144` | UTF-8 cap (minimum `64`) for each complete terminal result or PTY task output after wait, session, pagination, truncation, and task-status metadata |
|
||||
|
||||
Both values are validated at load. The minimum result cap keeps every registry-issued session or task id visible in its creation acknowledgement. When a result exceeds `maxResultBytes`, rendering reserves space for control metadata and a truncation marker when they fit; cuts preserve UTF-8 boundaries.
|
||||
Both values are validated at load. The minimum result cap keeps every registry-issued session or task id visible in its creation acknowledgement. When a result exceeds `maxResultBytes`, rendering reserves space for control metadata and a truncation marker when they fit; cuts preserve UTF-8 boundaries. An outer `tools/post-execute` wrapper applies the same cap after a terminal pre-execute denial or single-text post-execute replacement/block; a structured multi-block policy result retains its shape.
|
||||
|
||||
## Model Experience
|
||||
|
||||
@@ -53,11 +53,11 @@ Prefix-stable while tool visibility and definitions are unchanged.
|
||||
|
||||
#### What the model sees
|
||||
|
||||
Spawn returns the id and bounded MOTD. Send/read return bounded terminal text plus readiness/history markers. Background mode returns a generic task id. Every complete result is capped by `maxResultBytes`, including normalized error text and generic task status text. Results remain in session history until compaction; incremental task reads do not repeat consumed output.
|
||||
Spawn returns the id and bounded MOTD. Send/read return bounded terminal text plus readiness/history markers. Background mode returns a generic task id. Every terminal-owned or policy-produced single-text result is capped by `maxResultBytes` after normalized errors, denials, replacements, blocks, and generic task status text. Structured multi-block policy results retain their shape. Results remain in session history until compaction; incremental task reads do not repeat consumed output.
|
||||
|
||||
#### Token effect
|
||||
|
||||
Data-dependent and bounded by `maxResultBytes`; each returned result remains in history until compaction.
|
||||
Terminal-owned and policy-produced single-text results are data-dependent and bounded by `maxResultBytes`; a policy that deliberately substitutes structured multi-block content owns that content's bound. Each returned result remains in history until compaction.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@ import { PtySessionId } from '@deepseek-ai/dsh-pty'
|
||||
import type { PtySendResult, PtySessionId as PtySessionIdType, PtySignal } from '@deepseek-ai/dsh-pty'
|
||||
import type {} from '@deepseek-ai/dsh-tasks'
|
||||
import { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import type { ToolExecutionResult, ToolResult } from '@deepseek-ai/dsh-tools'
|
||||
import type { PostToolDecision, ToolExecutionResult } from '@deepseek-ai/dsh-tools'
|
||||
import { boundTerminalText, renderList, renderRead, renderSend, renderSendRead, renderSpawn } from './render.ts'
|
||||
|
||||
declare module '@deepseek-ai/dsh-tasks' {
|
||||
@@ -95,9 +95,9 @@ function textResult(text: string, maxBytes: number): ContentBlock[] {
|
||||
return [{ type: 'text', text: boundTerminalText(text, maxBytes) }]
|
||||
}
|
||||
|
||||
function rawResultText(result: ToolResult): string | undefined {
|
||||
if (result.content.length !== 1) return undefined
|
||||
const block = result.content[0]
|
||||
function rawContentText(content: readonly ContentBlock[]): string | undefined {
|
||||
if (content.length !== 1) return undefined
|
||||
const block = content[0]
|
||||
return block?.type === 'text' ? block.text : undefined
|
||||
}
|
||||
|
||||
@@ -114,12 +114,17 @@ export function apply(ctx: Context, config: Config = {}): void {
|
||||
if (!Number.isSafeInteger(maxResultBytes) || maxResultBytes < MIN_MAX_RESULT_BYTES) {
|
||||
throw new Error(`tool-pty: maxResultBytes must be a safe integer of at least ${MIN_MAX_RESULT_BYTES}`)
|
||||
}
|
||||
ctx.on('tools/execute', async (exec, next): Promise<ToolExecutionResult> => {
|
||||
const result = await next()
|
||||
if (!TOOL_NAMES.has(exec.name)) return result
|
||||
const raw = rawResultText(result)
|
||||
return raw === undefined ? result : { ...result, content: textResult(raw, maxResultBytes) }
|
||||
})
|
||||
ctx.on('tools/post-execute', async (exec, result, next): Promise<PostToolDecision> => {
|
||||
const decision = await next()
|
||||
if (!TOOL_NAMES.has(exec.name)) return decision
|
||||
const content = decision.kind === 'block' ? decision.feedback : decision.content ?? result.content
|
||||
const raw = rawContentText(content)
|
||||
if (raw === undefined) return decision
|
||||
const bounded = textResult(raw, maxResultBytes)
|
||||
return decision.kind === 'block'
|
||||
? { ...decision, feedback: bounded }
|
||||
: { ...decision, content: bounded }
|
||||
}, { prepend: true })
|
||||
ctx.systemPrompt.section({
|
||||
name: 'tool:pty',
|
||||
order: 106,
|
||||
@@ -206,7 +211,7 @@ export function apply(ctx: Context, config: Config = {}): void {
|
||||
},
|
||||
presentResult(args, result) {
|
||||
if ((args as Partial<SendArgs>).run_in_background === true || result.isError) return undefined
|
||||
const raw = rawResultText(result)
|
||||
const raw = rawContentText(result.content)
|
||||
return raw === undefined ? undefined : { card: 'terminal', output: raw }
|
||||
},
|
||||
}))
|
||||
|
||||
@@ -217,6 +217,37 @@ describe('tool-pty foreground surface', () => {
|
||||
expect(Buffer.byteLength(text(background))).toBeLessThanOrEqual(64)
|
||||
})
|
||||
|
||||
it('bounds terminal results after pre- and post-execute policy', async () => {
|
||||
const { ctx, agent } = await setup(false, { maxResultBytes: 64 })
|
||||
ctx.on('tools/pre-execute', async (exec, next) => exec.name === 'terminal_list'
|
||||
? { kind: 'deny', reason: 'd'.repeat(1_000) }
|
||||
: next())
|
||||
ctx.on('tools/post-execute', async (exec, _result, next) => {
|
||||
if (exec.name === 'terminal_open') {
|
||||
return { kind: 'accept', content: [{ type: 'text', text: 'a'.repeat(1_000) }] }
|
||||
}
|
||||
if (exec.name === 'terminal_read') {
|
||||
return { kind: 'block', feedback: [{ type: 'text', text: 'b'.repeat(1_000) }] }
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
const denied = await call(ctx, 'terminal_list', {}, agent)
|
||||
expect(denied.isError).toBe(true)
|
||||
expect(Buffer.byteLength(text(denied))).toBeLessThanOrEqual(64)
|
||||
expect(text(denied)).toContain('[output truncated]')
|
||||
|
||||
const replaced = await call(ctx, 'terminal_open', { type: 'stub' }, agent)
|
||||
expect(replaced.isError).toBe(false)
|
||||
expect(Buffer.byteLength(text(replaced))).toBeLessThanOrEqual(64)
|
||||
expect(text(replaced)).toContain('[output truncated]')
|
||||
|
||||
const blocked = await call(ctx, 'terminal_read', { sessionId: 'pty-1' }, agent)
|
||||
expect(blocked.isError).toBe(true)
|
||||
expect(Buffer.byteLength(text(blocked))).toBeLessThanOrEqual(64)
|
||||
expect(text(blocked)).toContain('[output truncated]')
|
||||
})
|
||||
|
||||
it('leaves a structured around-dispatch replacement unchanged', async () => {
|
||||
const { ctx, agent } = await setup(false, { maxResultBytes: 64 })
|
||||
ctx.on('tools/execute', async (exec, next) => exec.name === 'terminal_list'
|
||||
|
||||
Reference in New Issue
Block a user