From 2bc994900b01d9562c38d73b5036457e592a197a Mon Sep 17 00:00:00 2001 From: _Kerman Date: Sun, 26 Jul 2026 18:30:26 +0800 Subject: [PATCH] fix(agent-loop): order admission publication and gate idle on committed turns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three ordering fixes on the admission and settlement boundaries. kick() installs the abort owner, marks the interval busy (running is emitted before any listener can observe the claim), and installs the pending done BEFORE publishing agent/inbox/dequeue: a dequeue listener that cancels or disposes now finds live cancellation and quiescence ownership instead of the previous activity's settled state, and claimed prompt admission — including asynchronous prompt-submit hooks — sits inside the running interval where cancel routing can reach it. The admission-rejected path yields one microtask before continueOrIdle so the idle transition cannot fire inside send()'s synchronous extent. agent/idle now names only committed turns: a run that aborts or fails before its turn/start commits exits without the notification, since there is no durable turn/end for settlement consumers to act against. The event's JSDoc states the narrowed contract. --- packages/core/agent-loop/src/agent.ts | 30 ++++++++++++++++++++++++--- packages/core/agent/src/types.ts | 6 ++++-- 2 files changed, 31 insertions(+), 5 deletions(-) diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 7176fdeb0d..05193b7180 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -201,9 +201,18 @@ export class ReactLoopAgent implements Agent { // eslint-disable-next-line @typescript-eslint/no-non-null-assertion const { message } = this.queued.shift()! - emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message) const admission = new AbortController() this.abort = admission + // Claimed admission is part of the running interval: it is cancellable + // activity, so observers (and their cancel routing) must see it. + if (!this.busy) { + this.busy = true + emitAgentEvent(this.loopCtx, this, 'agent/status', 'running') + } + // The admission body runs synchronously up to the prompt-submit + // waterfall's first await, so the waterfall snapshots its listeners + // before a disposal initiated by the running-status emit above can + // unregister a vetoing plugin. this.done = this.loopCtx.agents.withInitiator(this, async () => { const signal = admission.signal const trigger: TurnTrigger = { kind: 'message', source: message.source } @@ -235,11 +244,19 @@ export class ReactLoopAgent implements Agent { // still owns the slot here and releasing it unconditionally is exact. this.abort = undefined if (admitted === undefined) { + // A synchronously aborted admission would otherwise publish idle + // inside send()'s own synchronous extent, before any post-send + // subscriber could observe the transition. + await Promise.resolve() this.continueOrIdle() return } await this.run(trigger, admitted) }) + // Published only after the abort owner and pending done are installed: a + // dequeue listener that cancels or disposes must find live cancellation + // and quiescence ownership, not the previous activity's settled state. + emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message) } /** @@ -260,6 +277,7 @@ export class ReactLoopAgent implements Agent { const signal = controller.signal const turn = this.lastTurn + 1 let step = 0 + let opened = false let reason: TurnEndReason = { kind: 'completed' } let idle: IdleReason = { kind: 'completed' } let retry = false @@ -272,6 +290,7 @@ export class ReactLoopAgent implements Agent { // Committed: publish the turn to the machine's own bookkeeping and let // the admitted input enter the log it now belongs to. this.turnOpen = true + opened = true this.lastTurn = turn for (const input of admitted) { this.session.append('user/message', input, { surfaceOp: 'append' }) @@ -368,7 +387,10 @@ export class ReactLoopAgent implements Agent { if (retry) { await this.run({ kind: 'retry' }) } else { - emitAgentEvent(this.loopCtx, this, 'agent/idle', turn, idle) + // agent/idle names only committed turns: a run aborted or rejected + // before turn/start has no durable turn/end for consumers to settle + // against, so it exits without the notification. + if (opened) emitAgentEvent(this.loopCtx, this, 'agent/idle', turn, idle) this.continueOrIdle() } } @@ -592,7 +614,9 @@ export class ReactLoopAgent implements Agent { if (this.abort !== undefined) return if (this.queued.some(item => item.wakeup)) { this.kick() - } else if (this.busy) { + } else { + // Every caller sits inside an admission or run whose install marked the + // interval busy, so the flag is still set here. this.busy = false emitAgentEvent(this.loopCtx, this, 'agent/status', 'idle') } diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 511e6e3510..a20151686d 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -379,8 +379,10 @@ declare module 'cordis' { /** * One drain chain reached its terminal turn: that turn's `turn/end` is * already committed. Automatically recovered failed turns do not emit this - * notification. `reason` says why; model-request recovery is exhausted when - * an error reaches it. + * notification, and neither does a run that aborts or fails before its + * `turn/start` commits — there is no durable turn to settle against. + * `reason` says why; model-request recovery is exhausted when an error + * reaches it. * @param agent - the agent whose turn closed. * @param turn - the terminal turn number. * @param reason - why the terminal turn ended, with live error facts when it failed.