fix: close three seams the message-machine refactor left open
agent-loop lifecycle: dispose drains machine.done to true quiescence. cancel()'s own running-to-idle transition can legitimately re-enter through an automation listener (goal-session's idle drive runs synchronously to its first await) and replace done with a fresh admission after the single capture; teardown now re-cancels and re-awaits until the slot stabilizes, so the scope never unwinds under a live run. tools: a nested concludeTurn() stages on its own execution and promotes to the enclosing composite only on the call's authoritative successful verdict. A post-execute policy that converts the nested success into an error no longer lets a recovering composite stop the turn on a failed terminal operation (the Code Mode structured-output shape). goal-session: the driver owns its round durability barrier again. The loop's persistence is eager write-behind with no turn-end flush, so the old post-turn agent/error signal for flush failures never fires; a settled round now sets needsCheckpoint and re-enters drive, flushing before the next reservation and disarming on failure instead of queueing an autonomous round on state that was never persisted.
This commit is contained in:
@@ -374,7 +374,18 @@ export class AgentLoop extends Service implements AgentFactory {
|
|||||||
if (machine === undefined) await machineReady.promise
|
if (machine === undefined) await machineReady.promise
|
||||||
if (machine !== undefined) {
|
if (machine !== undefined) {
|
||||||
machine.cancel({ kind: 'disposed' })
|
machine.cancel({ kind: 'disposed' })
|
||||||
await Promise.allSettled([machine.done])
|
// Drain to TRUE quiescence: cancel's own synchronous event chain
|
||||||
|
// (running→idle) can legitimately re-enter through an automation
|
||||||
|
// listener (goal-session's idle drive) and replace `done` with a
|
||||||
|
// fresh admission before this await captures it. The replacement
|
||||||
|
// work is cancelled and drained in turn until the slot stabilizes.
|
||||||
|
let done = machine.done
|
||||||
|
while (true) {
|
||||||
|
await Promise.allSettled([done])
|
||||||
|
if (machine.done === done) break
|
||||||
|
done = machine.done
|
||||||
|
machine.cancel({ kind: 'disposed' })
|
||||||
|
}
|
||||||
await machine.scope.dispose()
|
await machine.scope.dispose()
|
||||||
}
|
}
|
||||||
} finally {
|
} finally {
|
||||||
|
|||||||
@@ -1042,4 +1042,43 @@ describe('agent scope lifecycle', () => {
|
|||||||
await ctx.fiber.dispose()
|
await ctx.fiber.dispose()
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('drains a run re-entered by cancel\'s own idle transition before removing the scope', async () => {
|
||||||
|
// Automation shaped like goal-session: the running→idle transition that
|
||||||
|
// disposal's cancel produces immediately queues a follow-up prompt. The
|
||||||
|
// teardown must drain that replacement run to true quiescence instead of
|
||||||
|
// awaiting only the first captured done and unwinding under a live run.
|
||||||
|
const adapter = new MockAdapter([textResponse('one'), textResponse('never awaited')])
|
||||||
|
const ctx = await harness(adapter)
|
||||||
|
const handle = await ctx.agents.create({
|
||||||
|
sessionId: SessionId('drain-reentered-run'),
|
||||||
|
agentOptions: { provider: 'mock', model: 'mock' },
|
||||||
|
})
|
||||||
|
const agent = handle.agent
|
||||||
|
let reentered = false
|
||||||
|
ctx.on('agent/status', (subject, status) => {
|
||||||
|
if (subject !== agent || status !== 'idle' || reentered) return
|
||||||
|
reentered = true
|
||||||
|
agent.followup({ content: [{ type: 'text', text: 'reentrant' }], source: { kind: 'user' } })
|
||||||
|
})
|
||||||
|
|
||||||
|
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
|
||||||
|
await waitForIdle(ctx, agent)
|
||||||
|
expect(reentered).toBe(true)
|
||||||
|
|
||||||
|
// Idle again: the reentrant admission was already claimed and settled (its
|
||||||
|
// prompt was blocked by nothing, so it ran) — arm a SECOND reentry that
|
||||||
|
// fires from the disposal cancel's idle transition itself.
|
||||||
|
reentered = false
|
||||||
|
await handle.dispose()
|
||||||
|
|
||||||
|
// The reentrant run either never started or was drained: the registries
|
||||||
|
// are empty and nothing still drives the detached session.
|
||||||
|
expect(ctx.agents.get(agent.id)).toBeUndefined()
|
||||||
|
expect(ctx.sessions.get(agent.id)).toBeUndefined()
|
||||||
|
const eventsAfter = agent.session.events.length
|
||||||
|
await new Promise(resolve => setTimeout(resolve, 30))
|
||||||
|
expect(agent.session.events.length).toBe(eventsAfter)
|
||||||
|
await ctx.fiber.dispose()
|
||||||
|
})
|
||||||
|
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -657,6 +657,12 @@ export class ToolRegistry extends Service {
|
|||||||
private concludingExecutions = new WeakSet<ToolExecution>()
|
private concludingExecutions = new WeakSet<ToolExecution>()
|
||||||
/** Enclosing transport tokens marked terminal by a successful nested call. */
|
/** Enclosing transport tokens marked terminal by a successful nested call. */
|
||||||
private concludingParents = new Set<ToolExecutionToken>()
|
private concludingParents = new Set<ToolExecutionToken>()
|
||||||
|
/**
|
||||||
|
* Nested conclusions staged until their call's final verdict: a post-execute
|
||||||
|
* policy may still convert the nested success into an error, and a failed
|
||||||
|
* terminal operation must not stop the turn through its composite.
|
||||||
|
*/
|
||||||
|
private pendingParentConclusions = new WeakMap<ToolRunContext, ToolExecutionToken>()
|
||||||
/** Original caller cancellation, kept outside the wrapper-mutable execution object. */
|
/** Original caller cancellation, kept outside the wrapper-mutable execution object. */
|
||||||
private cancellationStates = new WeakMap<ToolRunContext, ToolCancellationState>()
|
private cancellationStates = new WeakMap<ToolRunContext, ToolCancellationState>()
|
||||||
/** Definition-owned final content transform snapshotted before policy begins. */
|
/** Definition-owned final content transform snapshotted before policy begins. */
|
||||||
@@ -979,7 +985,7 @@ export class ToolRegistry extends Service {
|
|||||||
const definition = this.get(name, agent)
|
const definition = this.get(name, agent)
|
||||||
const finalizeContent = definition?.finalizeContent?.bind(definition)
|
const finalizeContent = definition?.finalizeContent?.bind(definition)
|
||||||
const concludingExecutions = this.concludingExecutions
|
const concludingExecutions = this.concludingExecutions
|
||||||
const concludingParents = this.concludingParents
|
const pendingParentConclusions = this.pendingParentConclusions
|
||||||
const base = {
|
const base = {
|
||||||
token,
|
token,
|
||||||
callId,
|
callId,
|
||||||
@@ -992,7 +998,9 @@ export class ToolRegistry extends Service {
|
|||||||
},
|
},
|
||||||
concludeTurn(): void {
|
concludeTurn(): void {
|
||||||
if (parent === undefined) concludingExecutions.add(this as unknown as ToolExecution)
|
if (parent === undefined) concludingExecutions.add(this as unknown as ToolExecution)
|
||||||
else concludingParents.add(parent)
|
// Staged, not propagated: only this nested call's authoritative
|
||||||
|
// successful result promotes the marker onto its parent.
|
||||||
|
else pendingParentConclusions.set(this as unknown as ToolRunContext, parent)
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
try {
|
try {
|
||||||
@@ -1206,6 +1214,14 @@ export class ToolRegistry extends Service {
|
|||||||
} catch (error: unknown) {
|
} catch (error: unknown) {
|
||||||
finalResult = this.materializeFinalResult(toolErrorResult(error))
|
finalResult = this.materializeFinalResult(toolErrorResult(error))
|
||||||
}
|
}
|
||||||
|
// Promote a staged nested conclusion only on the call's authoritative
|
||||||
|
// successful verdict — a policy-converted failure must not let the
|
||||||
|
// composite stop the turn on a failed terminal operation.
|
||||||
|
const stagedParent = this.pendingParentConclusions.get(exec)
|
||||||
|
if (stagedParent !== undefined) {
|
||||||
|
this.pendingParentConclusions.delete(exec)
|
||||||
|
if (!finalResult.isError) this.concludingParents.add(stagedParent)
|
||||||
|
}
|
||||||
this.notifyResult(exec, finalResult)
|
this.notifyResult(exec, finalResult)
|
||||||
this.concludingParents.delete(exec.token)
|
this.concludingParents.delete(exec.token)
|
||||||
return finalResult
|
return finalResult
|
||||||
|
|||||||
@@ -553,6 +553,54 @@ describe('ToolRegistry', () => {
|
|||||||
expect(nested.isError ? undefined : nested.value).toBe('')
|
expect(nested.isError ? undefined : nested.value).toBe('')
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('propagates a nested concludeTurn only when the nested verdict stays successful', async () => {
|
||||||
|
const ctx = await setup()
|
||||||
|
ctx.tools.register({
|
||||||
|
...echoTool,
|
||||||
|
name: 'terminal-nested',
|
||||||
|
async execute(_args, exec) {
|
||||||
|
exec.concludeTurn()
|
||||||
|
return 'staged'
|
||||||
|
},
|
||||||
|
})
|
||||||
|
// A composite that swallows its nested failure and returns success — the
|
||||||
|
// Code Mode shape the staged propagation exists for.
|
||||||
|
let call = 0
|
||||||
|
ctx.tools.register({
|
||||||
|
...echoTool,
|
||||||
|
name: 'composite',
|
||||||
|
async execute(_args, exec) {
|
||||||
|
call += 1
|
||||||
|
const nested = await ctx.tools.execute({
|
||||||
|
signal: exec.signal, callId: CallId(`nested-${call}`), name: 'terminal-nested', arguments: {}, parent: exec.token,
|
||||||
|
})
|
||||||
|
return nested.isError ? 'nested failed, composite recovered' : 'nested succeeded'
|
||||||
|
},
|
||||||
|
})
|
||||||
|
|
||||||
|
// A policy converts the nested success into an error: the staged
|
||||||
|
// conclusion must NOT reach the composite, or the loop would stop the
|
||||||
|
// turn on a failed terminal operation.
|
||||||
|
const veto = ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => {
|
||||||
|
if (exec.name !== 'terminal-nested') return next()
|
||||||
|
return { kind: 'block', feedback: [{ type: 'text', text: 'nested success rejected' }] }
|
||||||
|
})
|
||||||
|
const recovered = await ctx.tools.execute({
|
||||||
|
signal: testToolSignal, callId: CallId('composite-vetoed'), name: 'composite', arguments: {},
|
||||||
|
})
|
||||||
|
expect(recovered.isError).toBe(false)
|
||||||
|
expect(recovered.concludesTurn).toBeUndefined()
|
||||||
|
veto()
|
||||||
|
|
||||||
|
// The same nested call succeeding promotes the marker: the composite's
|
||||||
|
// own successful result now carries concludesTurn.
|
||||||
|
const concluded = await ctx.tools.execute({
|
||||||
|
signal: testToolSignal, callId: CallId('composite-ok'), name: 'composite', arguments: {},
|
||||||
|
})
|
||||||
|
expect(concluded.isError).toBe(false)
|
||||||
|
expect(concluded.concludesTurn).toBe(true)
|
||||||
|
})
|
||||||
|
|
||||||
it('returns isError results for unknown tools and throwing tools', async () => {
|
it('returns isError results for unknown tools and throwing tools', async () => {
|
||||||
const ctx = await setup()
|
const ctx = await setup()
|
||||||
ctx.tools.register({
|
ctx.tools.register({
|
||||||
|
|||||||
@@ -192,6 +192,14 @@ export function apply(ctx: Context): void {
|
|||||||
if (!attempt.stale) applyOutcome(state, goal, outcome)
|
if (!attempt.stale) applyOutcome(state, goal, outcome)
|
||||||
}
|
}
|
||||||
if (!readyToDrive(state)) return
|
if (!readyToDrive(state)) return
|
||||||
|
// The loop's persistence is eager write-behind with no turn-end flush,
|
||||||
|
// so this driver owns the round's durability barrier: checkpoint the
|
||||||
|
// settled round before reserving another (re-entering drive through
|
||||||
|
// the flush path above), disarming on failure instead of queueing an
|
||||||
|
// autonomous round on state that was never persisted.
|
||||||
|
state.needsCheckpoint = true
|
||||||
|
state.requested = true
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
const goal = currentGoal(state)
|
const goal = currentGoal(state)
|
||||||
|
|||||||
@@ -430,6 +430,42 @@ describe('same-session goal driving', () => {
|
|||||||
expect(test.adapter.requests).toHaveLength(0)
|
expect(test.adapter.requests).toHaveLength(0)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('disarms instead of reserving another round when the round checkpoint fails', async () => {
|
||||||
|
const test = await harness([textResponse('round one ran')])
|
||||||
|
// The loop persists eagerly with no turn-end flush, so the driver owns
|
||||||
|
// the round durability barrier. Let goal creation's checkpoint pass, then
|
||||||
|
// fail the flush that settles round one: no second round may be reserved
|
||||||
|
// on state that was never persisted.
|
||||||
|
let flushes = 0
|
||||||
|
test.ctx.on('session/flush', () => {
|
||||||
|
flushes += 1
|
||||||
|
// Flush 1 is goal creation's checkpoint; flush 2 settles round one.
|
||||||
|
return flushes >= 2 ? Promise.reject(new Error('round checkpoint failed')) : undefined
|
||||||
|
})
|
||||||
|
test.ctx.goals.create(test.agent, { objective: 'no autonomous rounds without durability', maxGoalRounds: 5 })
|
||||||
|
|
||||||
|
const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
|
||||||
|
|
||||||
|
expect(goal).toMatchObject({ phase: 'active', roundsStarted: 1 })
|
||||||
|
expect(test.adapter.requests).toHaveLength(1)
|
||||||
|
})
|
||||||
|
|
||||||
|
it('reserves the next round only after the settled round checkpoint succeeds', async () => {
|
||||||
|
const test = await harness([textResponse('round one'), textResponse('round two')])
|
||||||
|
const flushes: number[] = []
|
||||||
|
test.ctx.on('session/flush', () => { flushes.push(test.adapter.requests.length) })
|
||||||
|
test.ctx.goals.create(test.agent, { objective: 'checkpoint between rounds', maxGoalRounds: 2 })
|
||||||
|
|
||||||
|
const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
|
||||||
|
|
||||||
|
expect(goal?.blockedReason?.code).toBe('round-limit')
|
||||||
|
expect(goal?.roundsStarted).toBe(2)
|
||||||
|
expect(test.adapter.requests).toHaveLength(2)
|
||||||
|
// A flush was observed after round one settled and before round two
|
||||||
|
// dispatched (recorded request count 1 at flush time).
|
||||||
|
expect(flushes).toContain(1)
|
||||||
|
})
|
||||||
|
|
||||||
it('contains a checkpoint failure after a clear notification leaves no current goal', async () => {
|
it('contains a checkpoint failure after a clear notification leaves no current goal', async () => {
|
||||||
const test = await harness([])
|
const test = await harness([])
|
||||||
test.ctx.on('session/flush', () => Promise.reject(new Error('clear checkpoint failed')))
|
test.ctx.on('session/flush', () => Promise.reject(new Error('clear checkpoint failed')))
|
||||||
|
|||||||
Reference in New Issue
Block a user