mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(subagent): address codex review round 1
All five findings were real: - The terminal lifecycle edge derived its stop reason from teardown success, so a child that errored, hit its token ceiling, or was cancelled reported as completed once its checkpoint and disposal succeeded. It now reads the child's own last message turn/end, which is authoritative. - Live delivery never rechecked the caller signal after authorization yielded, so an abort that won before acceptance still enqueued the message and returned an id. Admission now re-checks at the boundary that owns the decision. - Drain flushed before cancelling, letting a running turn keep appending events the checkpoint could not cover and letting model work continue through a slow flush. It now cancels to quiescence first. - subagent/end fired after AgentHandle.dispose() unregistered the child, so the hooks bridge could not resolve it for the child's cwd and scope. The edge now publishes while the child is still registered. - activationState() read Agent.status alone, which stays idle between an accepted waking send and the microtask that admits it, so a synchronous inbox observer could see settled with a queued turn. Residency now also counts messages this manager admitted but has not seen leave the inbox.
This commit is contained in:
Dudu-0223
Tianyi Cui
@@ -794,7 +794,7 @@ A ready child settled. Scope-filtered dispatch uses the same delegating parent c
|
||||
|
||||
Types: [Scoped](../core-data-structures/scope.md) · [SubagentService](../core-data-structures/subagent.md)
|
||||
|
||||
Source: [`packages/subagent/subagent/src/index.ts:140`](../../packages/subagent/subagent/src/index.ts)
|
||||
Source: [`packages/subagent/subagent/src/index.ts:141`](../../packages/subagent/subagent/src/index.ts)
|
||||
|
||||
### `subagent/provider-added` — emit
|
||||
|
||||
@@ -811,7 +811,7 @@ A provider became resolvable in the registry.
|
||||
|
||||
Types: [SubagentProvider](../core-data-structures/subagent.md)
|
||||
|
||||
Source: [`packages/subagent/subagent/src/index.ts:114`](../../packages/subagent/subagent/src/index.ts)
|
||||
Source: [`packages/subagent/subagent/src/index.ts:115`](../../packages/subagent/subagent/src/index.ts)
|
||||
|
||||
### `subagent/provider-removed` — emit
|
||||
|
||||
@@ -826,7 +826,7 @@ A provider left the registry. Accepted runs remain holder-owned.
|
||||
'subagent/provider-removed'(name: string): void
|
||||
```
|
||||
|
||||
Source: [`packages/subagent/subagent/src/index.ts:120`](../../packages/subagent/subagent/src/index.ts)
|
||||
Source: [`packages/subagent/subagent/src/index.ts:121`](../../packages/subagent/subagent/src/index.ts)
|
||||
|
||||
### `subagent/start` — emit
|
||||
|
||||
@@ -848,7 +848,7 @@ A provider established a ready child. For in-process providers, `ctx.agents.get(
|
||||
|
||||
Types: [Scoped](../core-data-structures/scope.md) · [SubagentService](../core-data-structures/subagent.md)
|
||||
|
||||
Source: [`packages/subagent/subagent/src/index.ts:131`](../../packages/subagent/subagent/src/index.ts)
|
||||
Source: [`packages/subagent/subagent/src/index.ts:132`](../../packages/subagent/subagent/src/index.ts)
|
||||
|
||||
## `system-prompt/*`
|
||||
|
||||
|
||||
@@ -2033,7 +2033,7 @@ async start(name: string, request: SubagentStartRequest): Promise<SubagentRun>
|
||||
|
||||
Types: [ActivationState](../core-data-structures/subagent.md) · [ContentBlock](../core-data-structures/core.md) · [ContinuableStart](../core-data-structures/subagent.md) · [ContinuableStartSpec](../core-data-structures/subagent.md) · [MessageId](../core-data-structures/core.md) · [SessionId](../core-data-structures/core.md) · [SubagentAuthority](../core-data-structures/subagent.md) · [SubagentFollowupOptions](../core-data-structures/subagent.md) · [SubagentProvider](../core-data-structures/subagent.md) · [SubagentRun](../core-data-structures/subagent.md) · [SubagentStartRequest](../core-data-structures/subagent.md)
|
||||
|
||||
Source: [`packages/subagent/subagent/src/index.ts:173`](../../packages/subagent/subagent/src/index.ts)
|
||||
Source: [`packages/subagent/subagent/src/index.ts:174`](../../packages/subagent/subagent/src/index.ts)
|
||||
|
||||
## `ctx.subprocess` — `SubprocessService` (abstract seam)
|
||||
|
||||
|
||||
@@ -12,8 +12,8 @@ This matrix shows which packages dispatch each harness-owned event and which pac
|
||||
| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:280`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) |
|
||||
| `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:289`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) |
|
||||
| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:463`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `apiproxy`, [`goal-session`](../packages/goal/goal-session), [`session-telemetry`](../packages/telemetry/session-telemetry), [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/dequeue` | `emit` | [`packages/core/agent/src/types.ts:327`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/discard` | `emit` | [`packages/core/agent/src/types.ts:339`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/dequeue` | `emit` | [`packages/core/agent/src/types.ts:327`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`subagent`](../packages/subagent/subagent), [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/discard` | `emit` | [`packages/core/agent/src/types.ts:339`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`subagent`](../packages/subagent/subagent), [`tui`](../packages/ui/tui) |
|
||||
| `agent/inbox/enqueue` | `emit` | [`packages/core/agent/src/types.ts:308`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`goal-session`](../packages/goal/goal-session) |
|
||||
| `agent/inbox/update` | `emit` | [`packages/core/agent/src/types.ts:317`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `apiproxy` |
|
||||
| `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:376`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard), [`tui`](../packages/ui/tui) |
|
||||
@@ -41,10 +41,10 @@ This matrix shows which packages dispatch each harness-owned event and which pac
|
||||
| `settings/document-updated` | `emit` | [`packages/settings/settings/src/index.ts:150`](../packages/settings/settings/src/index.ts) | [`settings`](../packages/settings/settings) (`events.dispatch`) | `apiproxy` |
|
||||
| `settings/updated` | `emit` | [`packages/settings/settings/src/index.ts:137`](../packages/settings/settings/src/index.ts) | [`settings`](../packages/settings/settings) (`events.dispatch`) | [`settings`](../packages/settings/settings) |
|
||||
| `skills/change` | `emit` | [`packages/skill/skill/src/index.ts:188`](../packages/skill/skill/src/index.ts) | [`skill`](../packages/skill/skill) (`events.dispatch`) | [`tui`](../packages/ui/tui) |
|
||||
| `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:140`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc), [`subagent`](../packages/subagent/subagent) |
|
||||
| `subagent/provider-added` | `emit` | [`packages/subagent/subagent/src/index.ts:114`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`emit`) | [`subagent`](../packages/subagent/subagent), [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
| `subagent/provider-removed` | `emit` | [`packages/subagent/subagent/src/index.ts:120`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`subagent`](../packages/subagent/subagent), [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
| `subagent/start` | `emit` | [`packages/subagent/subagent/src/index.ts:131`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`subagent`](../packages/subagent/subagent) |
|
||||
| `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:141`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc), [`subagent`](../packages/subagent/subagent) |
|
||||
| `subagent/provider-added` | `emit` | [`packages/subagent/subagent/src/index.ts:115`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`emit`) | [`subagent`](../packages/subagent/subagent), [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
| `subagent/provider-removed` | `emit` | [`packages/subagent/subagent/src/index.ts:121`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`subagent`](../packages/subagent/subagent), [`tool-subagent`](../packages/subagent/tool-subagent) |
|
||||
| `subagent/start` | `emit` | [`packages/subagent/subagent/src/index.ts:132`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`subagent`](../packages/subagent/subagent) |
|
||||
| `system-prompt/assemble` | `waterfall` | [`packages/core/system-prompt/src/index.ts:29`](../packages/core/system-prompt/src/index.ts) | [`system-prompt`](../packages/core/system-prompt) (`waterfall`) | [`agent`](../packages/core/agent), [`system-prompt`](../packages/core/system-prompt) |
|
||||
| `system-prompt/change` | `emit` | [`packages/core/system-prompt/src/index.ts:35`](../packages/core/system-prompt/src/index.ts) | [`system-prompt`](../packages/core/system-prompt) (`emit`) | - |
|
||||
| `telemetry/record` | `waterfall` | [`packages/telemetry/session-telemetry/src/index.ts:41`](../packages/telemetry/session-telemetry/src/index.ts) | [`session-telemetry`](../packages/telemetry/session-telemetry) (`waterfall`) | - |
|
||||
|
||||
@@ -168,6 +168,12 @@ interface Activation {
|
||||
* a new Activation. Every converging releaser shares this one teardown.
|
||||
*/
|
||||
disposal: Promise<void> | undefined
|
||||
/**
|
||||
* Accepted waking message ids this manager has not yet seen leave the inbox.
|
||||
* `Agent.status` is still `idle` in the window between `followup()` and the
|
||||
* microtask that admits it, so settlement must not treat that gap as quiet.
|
||||
*/
|
||||
readonly accepted: Set<MessageId>
|
||||
/** Renewed whenever a settlement watcher must re-observe quiescence. */
|
||||
poke: PromiseWithResolvers<void>
|
||||
}
|
||||
@@ -353,6 +359,11 @@ export class SubagentContinuationManager {
|
||||
return activation.disposal.then(() => undefined, () => undefined)
|
||||
}
|
||||
await this.authorizeLive(authority, activation)
|
||||
// The caller signal owns admission until acceptance, so re-check it
|
||||
// here: the outer check cannot cover an abort that landed while
|
||||
// authorization yielded, and enqueueing afterwards would return a
|
||||
// message id for a delivery the caller already cancelled.
|
||||
options.signal.throwIfAborted()
|
||||
return this.submit(activation, content, options.source, authority)
|
||||
})
|
||||
/* v8 ignore start -- only the lost-cutoff arm above returns undefined, so only that
|
||||
@@ -413,10 +424,15 @@ export class SubagentContinuationManager {
|
||||
|
||||
/**
|
||||
* Derive residency from Agent quiescence and the owned-child set. `running`
|
||||
* covers an active admission, an open turn, or waking inbox work.
|
||||
* covers an active admission, an open turn, or accepted waking inbox work.
|
||||
*
|
||||
* `Agent.status` alone is insufficient: it stays `idle` between an accepted
|
||||
* waking send and the microtask that admits it, so a synchronous inbox
|
||||
* observer would see `settled` while a turn is already queued. `accepted`
|
||||
* holds the ids this manager admitted but has not yet seen drained.
|
||||
*/
|
||||
private stateOf(activation: Activation): ActivationState {
|
||||
if (activation.handle.agent.status === 'running') return 'running'
|
||||
if (activation.handle.agent.status === 'running' || activation.accepted.size > 0) return 'running'
|
||||
if (activation.ownedChildren.size > 0) return 'waiting'
|
||||
return 'settled'
|
||||
}
|
||||
@@ -528,6 +544,7 @@ export class SubagentContinuationManager {
|
||||
ownedChildren: new Set(),
|
||||
observer,
|
||||
disposal: undefined,
|
||||
accepted: new Set(),
|
||||
poke: Promise.withResolvers<void>(),
|
||||
}
|
||||
// After transfer, any failure must dispose the created handle, remove the
|
||||
@@ -550,6 +567,22 @@ export class SubagentContinuationManager {
|
||||
await activation.disposal.catch(() => undefined)
|
||||
throw error
|
||||
}
|
||||
// Every accepted id leaves the inbox exactly once, through dequeue or
|
||||
// discard. Clearing it there is what lets `stateOf()` distinguish a truly
|
||||
// quiet Agent from one whose accepted turn has not been admitted yet.
|
||||
// Registered through the child's own scoped context, so scope filtering
|
||||
// already restricts both listeners to this exact agent.
|
||||
handle.agent.ctx.on('agent/inbox/dequeue', (_agent, item) => {
|
||||
/* v8 ignore next -- a dequeue of an id this manager never admitted needs
|
||||
* another sender on the same child, which no current path allows. */
|
||||
if (activation.accepted.delete(item.message.id)) this.wake(activation)
|
||||
})
|
||||
handle.agent.ctx.on('agent/inbox/discard', (_agent, items) => {
|
||||
// Deleting every id in the batch is unconditional; waking once afterwards
|
||||
// costs nothing and avoids branching on which ids this manager admitted.
|
||||
for (const item of items) activation.accepted.delete(item.message.id)
|
||||
this.wake(activation)
|
||||
})
|
||||
// Resident: publish the start edge before any turn can run, so observers
|
||||
// see this epoch before its first request.
|
||||
observer.start()
|
||||
@@ -604,7 +637,15 @@ export class SubagentContinuationManager {
|
||||
// establish it before the message can enter the child's inbox.
|
||||
if (authority.kind === 'parent') this.acquireOwnership(authority.agent, activation.childId)
|
||||
const message = createUserMessage({ content, source })
|
||||
activation.handle.agent.followup(message)
|
||||
// `Agent.followup()` publishes `agent/inbox/enqueue` synchronously, so its
|
||||
// observers must see this Activation as busy before the call begins.
|
||||
activation.accepted.add(message.id)
|
||||
try {
|
||||
activation.handle.agent.followup(message)
|
||||
} catch (error: unknown) {
|
||||
activation.accepted.delete(message.id)
|
||||
throw error
|
||||
}
|
||||
// Accepted waking work keeps this Activation live until whenIdle() observes
|
||||
// the complete waking suffix.
|
||||
this.wake(activation)
|
||||
@@ -730,8 +771,17 @@ export class SubagentContinuationManager {
|
||||
'ACTIVATION_TEARDOWN_FAILED',
|
||||
)
|
||||
}
|
||||
// Quiesce before the checkpoint: a turn still running would keep
|
||||
// appending events the flush cannot cover, and a slow flush would let
|
||||
// model and tool work continue for the whole shutdown.
|
||||
activation.handle.agent.cancel({ kind: 'parent' })
|
||||
await activation.handle.agent.whenIdle()
|
||||
const durability = await this.checkpoint(activation)
|
||||
failure ??= durability
|
||||
// Publish the terminal edge while the child is STILL registered:
|
||||
// consumers resolve `ctx.agents.get(info.id)` in `subagent/end` to run
|
||||
// in the child's own cwd and scope, which handle disposal removes.
|
||||
activation.observer.settle(activation.handle.agent, failure)
|
||||
} finally {
|
||||
this.activations.delete(childId)
|
||||
try {
|
||||
@@ -746,7 +796,6 @@ export class SubagentContinuationManager {
|
||||
// Release ownership even on failure: a retained failed child would
|
||||
// pin its ancestors in `waiting` forever.
|
||||
this.releaseOwnership(childId)
|
||||
activation.observer.settle(activation.handle.agent, failure)
|
||||
}
|
||||
}
|
||||
if (failure !== undefined) throw failure
|
||||
|
||||
@@ -36,6 +36,7 @@ import type { Scoped } from '@deepseek-ai/dsh-scope'
|
||||
import { assertObjectJsonSchema } from '@deepseek-ai/dsh-tools'
|
||||
import type { ContentBlock, MessageId } from '@deepseek-ai/dsh-llm'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { findLastMessageTurnEnd } from '@deepseek-ai/dsh-session'
|
||||
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type {
|
||||
ContinuableCreateRequest,
|
||||
@@ -377,7 +378,7 @@ export class SubagentService extends Service {
|
||||
const output = failure === undefined ? lastAssistantOutput(child) : undefined
|
||||
this.emitLifecycle('subagent/end', {
|
||||
...identity,
|
||||
stopReason: failure === undefined ? 'completed' : 'error',
|
||||
stopReason: failure === undefined ? childStopReason(child) : 'error',
|
||||
...output === undefined ? {} : { lastAssistantMessage: output },
|
||||
}, parent)
|
||||
},
|
||||
@@ -459,6 +460,37 @@ export class SubagentService extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Why this child's last ordinary turn ended, for the terminal lifecycle edge.
|
||||
* The child's own `turn/end` is authoritative: teardown succeeding says nothing
|
||||
* about whether the model errored, hit its token ceiling, or was cancelled, so
|
||||
* deriving the reason from disposal would report failed work as completed.
|
||||
* @param child - the settling child agent whose log is read.
|
||||
* @returns its terminal stop reason; `completed` when no ordinary turn closed.
|
||||
*/
|
||||
function childStopReason(child: Agent): SubagentResult['stopReason'] {
|
||||
const reason = findLastMessageTurnEnd(child.session.events)?.data.reason
|
||||
// No ordinary turn closed, so nothing failed either.
|
||||
if (reason === undefined) return 'completed'
|
||||
switch (reason.kind) {
|
||||
case 'max-tokens':
|
||||
return 'max-tokens'
|
||||
case 'aborted':
|
||||
case 'interrupted':
|
||||
case 'disposed':
|
||||
return 'aborted'
|
||||
case 'error':
|
||||
return 'error'
|
||||
case 'completed':
|
||||
return 'completed'
|
||||
/* v8 ignore next 3 -- `TurnEndReason` is merge-extensible, so this arm needs a
|
||||
* backend that adds a variant; treating an unnameable reason as success would
|
||||
* report failed work as completed. */
|
||||
default:
|
||||
return 'error'
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The child's last assistant message content, for one Activation's terminal
|
||||
* lifecycle edge. Absent when no assistant message reached the log.
|
||||
|
||||
@@ -643,6 +643,152 @@ describe('continuable durability and teardown', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('continuable review regressions', () => {
|
||||
it('reports the child\'s own terminal reason, not teardown success', async () => {
|
||||
// The child hits its token ceiling; teardown still succeeds.
|
||||
const { ctx, parent } = await setupWith(new MockAdapter([
|
||||
[{ type: 'block-start', index: 0, blockType: 'text' },
|
||||
{ type: 'text-delta', index: 0, text: 'partial' },
|
||||
{ type: 'block-end', index: 0, block: { type: 'text', text: 'partial' } },
|
||||
{ type: 'finish', reason: { kind: 'max-tokens' } }],
|
||||
]))
|
||||
const ends: SubagentRunEndInfo[] = []
|
||||
ctx.on('subagent/end', (info) => { ends.push(info) })
|
||||
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
|
||||
await vi.waitFor(() => { expect(ends).toHaveLength(1) })
|
||||
// Deriving this from disposal success would report the failure as completed.
|
||||
expect(ends[0]!.stopReason).toBe('max-tokens')
|
||||
})
|
||||
|
||||
it('rejects a live delivery whose caller signal aborted before admission', async () => {
|
||||
const releaseFirst = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([{ chunks: textResponse('working'), gate: releaseFirst.promise }])
|
||||
const { ctx, parent } = await setupWith(adapter)
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
|
||||
const child = ctx.agents.get(started.childId)!
|
||||
const before = child.session.events.length
|
||||
|
||||
const controller = new AbortController()
|
||||
controller.abort('caller gave up')
|
||||
await expect(followup(ctx, { kind: 'user' }, started.childId, message('cancelled'), controller.signal))
|
||||
.rejects.toThrow()
|
||||
|
||||
// Nothing was enqueued, so no later turn can carry it.
|
||||
releaseFirst.resolve(undefined)
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
const loaded = await ctx.sessionPersistence.load(started.childId)
|
||||
expect(hasUserText(loaded.events, 'cancelled')).toBe(false)
|
||||
expect(before).toBeGreaterThan(0)
|
||||
})
|
||||
|
||||
it('publishes the terminal edge while the child agent is still resolvable', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('answer')])
|
||||
const resolvable: boolean[] = []
|
||||
// Consumers resolve the child in `subagent/end` to run in its own cwd.
|
||||
ctx.on('subagent/end', (info) => {
|
||||
resolvable.push(ctx.agents.get(info.id) !== undefined)
|
||||
})
|
||||
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
|
||||
await vi.waitFor(() => { expect(resolvable).toHaveLength(1) })
|
||||
expect(resolvable[0]).toBe(true)
|
||||
})
|
||||
|
||||
it('cancels a running turn before the final durability checkpoint', async () => {
|
||||
const hold = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([{ chunks: textResponse('slow'), gate: hold.promise }])
|
||||
const { ctx, parent } = await setupWith(adapter)
|
||||
const order: string[] = []
|
||||
ctx.on('session/flush', (session) => {
|
||||
if (session.header.parentSession !== undefined) order.push('flush')
|
||||
})
|
||||
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
const child = await vi.waitFor(() => {
|
||||
const found = ctx.agents.get(started.childId)
|
||||
expect(found).toBeDefined()
|
||||
return found!
|
||||
})
|
||||
child.ctx.on('agent/cancel-requested', () => { order.push('cancel') })
|
||||
|
||||
const drained = ctx.subagents.drainContinuable()
|
||||
hold.resolve(undefined)
|
||||
await drained
|
||||
|
||||
// Flushing a still-running turn cannot cover the events cancellation adds.
|
||||
expect(order.indexOf('cancel')).toBeGreaterThanOrEqual(0)
|
||||
expect(order.indexOf('cancel')).toBeLessThan(order.lastIndexOf('flush'))
|
||||
})
|
||||
|
||||
it('releases an accepted message that is discarded instead of run', async () => {
|
||||
const hold = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([{ chunks: textResponse('working'), gate: hold.promise }])
|
||||
const { ctx, parent } = await setupWith(adapter)
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
|
||||
// Queue a turn, then cancel so it is discarded rather than dequeued. The
|
||||
// Activation must still reach settlement instead of waiting on that id.
|
||||
await followup(ctx, { kind: 'user' }, started.childId, message('discarded'))
|
||||
|
||||
const drained = ctx.subagents.drainContinuable()
|
||||
hold.resolve(undefined)
|
||||
await drained
|
||||
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
const loaded = await ctx.sessionPersistence.load(started.childId)
|
||||
expect(hasUserText(loaded.events, 'discarded')).toBe(false)
|
||||
})
|
||||
|
||||
it('reports completed when no ordinary turn closed', async () => {
|
||||
const { ctx, parent } = await setup([])
|
||||
const ends: SubagentRunEndInfo[] = []
|
||||
ctx.on('subagent/end', (info) => { ends.push(info) })
|
||||
// Block admission so the child's only turn never opens.
|
||||
ctx.on('agent/prompt-submit', async (subject, _content, _source, _signal, next) => {
|
||||
if (subject === parent) return next()
|
||||
return { kind: 'block', reason: 'blocked by policy' }
|
||||
})
|
||||
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
|
||||
await vi.waitFor(() => { expect(ends).toHaveLength(1) })
|
||||
expect(ends[0]!.stopReason).toBe('completed')
|
||||
})
|
||||
|
||||
it('never reports settled while an accepted message is still in the inbox', async () => {
|
||||
const releaseFirst = Promise.withResolvers<undefined>()
|
||||
const adapter = new GatedAdapter([
|
||||
{ chunks: textResponse('first'), gate: releaseFirst.promise },
|
||||
{ chunks: textResponse('second') },
|
||||
])
|
||||
const { ctx, parent } = await setupWith(adapter)
|
||||
const states: (string | undefined)[] = []
|
||||
// A synchronous inbox observer runs before the admitting microtask, the
|
||||
// exact window where `Agent.status` is still idle.
|
||||
ctx.on('agent/inbox/enqueue', (agent) => {
|
||||
if (agent.session.header.parentSession !== undefined) {
|
||||
states.push(ctx.subagents.activationState(agent.id))
|
||||
}
|
||||
})
|
||||
|
||||
const started = await ctx.subagents.startContinuable(startSpec(parent))
|
||||
await vi.waitFor(() => { expect(adapter.requests).toHaveLength(1) })
|
||||
await followup(ctx, { kind: 'user' }, started.childId, message('queued'))
|
||||
|
||||
expect(states.length).toBeGreaterThan(0)
|
||||
expect(states).not.toContain('settled')
|
||||
releaseFirst.resolve(undefined)
|
||||
await waitNoActivation(ctx, started.childId)
|
||||
})
|
||||
})
|
||||
|
||||
describe('continuable lifecycle observation', () => {
|
||||
it('emits one paired start/end per residency epoch', async () => {
|
||||
const { ctx, parent } = await setup([textResponse('first'), textResponse('second')])
|
||||
|
||||
Reference in New Issue
Block a user