From 2a51ef85fb2fc318ccae1fb99242ccd533ac56d2 Mon Sep 17 00:00:00 2001 From: _Kerman Date: Sun, 26 Jul 2026 20:50:02 +0800 Subject: [PATCH] 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. --- packages/core/agent-loop/src/index.ts | 13 ++++- .../agent-loop/tests/scope-lifecycle.spec.ts | 39 +++++++++++++++ packages/core/tools/src/index.ts | 20 +++++++- packages/core/tools/tests/tools.spec.ts | 48 +++++++++++++++++++ packages/goal/goal-session/src/index.ts | 8 ++++ .../goal-session/tests/goal-session.spec.ts | 36 ++++++++++++++ 6 files changed, 161 insertions(+), 3 deletions(-) diff --git a/packages/core/agent-loop/src/index.ts b/packages/core/agent-loop/src/index.ts index 46eb18fddd..28a482b3bb 100644 --- a/packages/core/agent-loop/src/index.ts +++ b/packages/core/agent-loop/src/index.ts @@ -374,7 +374,18 @@ export class AgentLoop extends Service implements AgentFactory { if (machine === undefined) await machineReady.promise if (machine !== undefined) { 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() } } finally { diff --git a/packages/core/agent-loop/tests/scope-lifecycle.spec.ts b/packages/core/agent-loop/tests/scope-lifecycle.spec.ts index 9f69f5ff6d..b63d0fda27 100644 --- a/packages/core/agent-loop/tests/scope-lifecycle.spec.ts +++ b/packages/core/agent-loop/tests/scope-lifecycle.spec.ts @@ -1042,4 +1042,43 @@ describe('agent scope lifecycle', () => { 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() + }) + }) diff --git a/packages/core/tools/src/index.ts b/packages/core/tools/src/index.ts index c316550768..067c854cb9 100644 --- a/packages/core/tools/src/index.ts +++ b/packages/core/tools/src/index.ts @@ -657,6 +657,12 @@ export class ToolRegistry extends Service { private concludingExecutions = new WeakSet() /** Enclosing transport tokens marked terminal by a successful nested call. */ private concludingParents = new Set() + /** + * 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() /** Original caller cancellation, kept outside the wrapper-mutable execution object. */ private cancellationStates = new WeakMap() /** 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 finalizeContent = definition?.finalizeContent?.bind(definition) const concludingExecutions = this.concludingExecutions - const concludingParents = this.concludingParents + const pendingParentConclusions = this.pendingParentConclusions const base = { token, callId, @@ -992,7 +998,9 @@ export class ToolRegistry extends Service { }, concludeTurn(): void { 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 { @@ -1206,6 +1214,14 @@ export class ToolRegistry extends Service { } catch (error: unknown) { 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.concludingParents.delete(exec.token) return finalResult diff --git a/packages/core/tools/tests/tools.spec.ts b/packages/core/tools/tests/tools.spec.ts index 2ade59a510..c704cfbada 100644 --- a/packages/core/tools/tests/tools.spec.ts +++ b/packages/core/tools/tests/tools.spec.ts @@ -553,6 +553,54 @@ describe('ToolRegistry', () => { 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 => { + 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 () => { const ctx = await setup() ctx.tools.register({ diff --git a/packages/goal/goal-session/src/index.ts b/packages/goal/goal-session/src/index.ts index 002fd90a24..f6f0404fdc 100644 --- a/packages/goal/goal-session/src/index.ts +++ b/packages/goal/goal-session/src/index.ts @@ -192,6 +192,14 @@ export function apply(ctx: Context): void { if (!attempt.stale) applyOutcome(state, goal, outcome) } 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) diff --git a/packages/goal/goal-session/tests/goal-session.spec.ts b/packages/goal/goal-session/tests/goal-session.spec.ts index a7b3f59373..9af38b878c 100644 --- a/packages/goal/goal-session/tests/goal-session.spec.ts +++ b/packages/goal/goal-session/tests/goal-session.spec.ts @@ -430,6 +430,42 @@ describe('same-session goal driving', () => { 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 () => { const test = await harness([]) test.ctx.on('session/flush', () => Promise.reject(new Error('clear checkpoint failed')))