refactor(agent-loop): clarify pending message flow

This commit is contained in:
_Kerman
2026-07-24 17:55:08 +08:00
parent 1d9662a020
commit b56628ced7
3 changed files with 20 additions and 23 deletions
@@ -906,8 +906,8 @@ export const EVENT_API: readonly EventApiEntry[] = [
name: 'agent/inbox/enqueue', name: 'agent/inbox/enqueue',
mode: 'emit', mode: 'emit',
signature: '\'agent/inbox/enqueue\'(this: Scoped<Agent>, agent: Agent, message: AgentMessage): void', signature: '\'agent/inbox/enqueue\'(this: Scoped<Agent>, agent: Agent, message: AgentMessage): void',
jsDoc: '/**\n * A frozen item entered the queued or steering inbox.\n * @param agent - the owning agent.\n * @param message - accepted content, source, and correlation identity.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */', jsDoc: '/**\n * An item entered the queued or steering inbox.\n * @param agent - the owning agent.\n * @param message - accepted content, source, and correlation identity.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */',
summary: 'A frozen item entered the queued or steering inbox.', summary: 'An item entered the queued or steering inbox.',
}, },
{ {
name: 'agent/prompt-submit', name: 'agent/prompt-submit',
+16 -17
View File
@@ -13,7 +13,6 @@ import { createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope' import type { Scope } from '@deepseek-ai/dsh-scope'
import type { import type {
AgentMessage, AgentMessage,
AgentMessageId as AgentMessageIdType,
CancelOptions, CancelOptions,
AgentInterruptReason, AgentInterruptReason,
AgentOptions, AgentOptions,
@@ -92,7 +91,7 @@ export class ReactLoopAgent extends Agent {
send( send(
content: ContentBlock[], content: ContentBlock[],
options: SendOptions = { target: 'next-turn', wakeup: true, source: { kind: 'user' } }, options: SendOptions = { target: 'next-turn', wakeup: true, source: { kind: 'user' } },
): AgentMessageIdType { ): AgentMessageId {
const id = AgentMessageId(randomUUID()) const id = AgentMessageId(randomUUID())
const { target, wakeup, source } = options const { target, wakeup, source } = options
if (target === 'next-step' && !wakeup) { if (target === 'next-step' && !wakeup) {
@@ -135,10 +134,10 @@ export class ReactLoopAgent extends Agent {
if (cause.kind !== 'disposed') emitAgentEvent(this.loopCtx, this, 'agent/cancel-requested', cause) if (cause.kind !== 'disposed') emitAgentEvent(this.loopCtx, this, 'agent/cancel-requested', cause)
} }
if (!options.keepInbox) { if (!options.keepInbox) {
const discarded = [ const discarded: AgentMessage[] = [...this.queued]
...this.queued, for (const message of this.outbox) {
...this.outbox.filter((item): item is PendingMessage => 'id' in item), if ('id' in message) discarded.push(message)
] }
// Clear before abort observers run: replacement work belongs to the next turn. // Clear before abort observers run: replacement work belongs to the next turn.
this.queued.length = 0 this.queued.length = 0
this.outbox.length = 0 this.outbox.length = 0
@@ -429,18 +428,18 @@ export class ReactLoopAgent extends Agent {
/** Commit the outbox and report whether it contained steering. */ /** Commit the outbox and report whether it contained steering. */
private drainOutbox(turn: number): boolean { private drainOutbox(turn: number): boolean {
let steered = false let steered = false
for (const item of this.outbox.splice(0)) { for (const message of this.outbox.splice(0)) {
if (!('id' in item)) { if ('id' in message) {
this.session.append('user/message', item, { surfaceOp: 'append' }) steered = true
continue emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)
this.session.append(
'steering/message',
{ turn, content: message.content, source: message.source },
{ surfaceOp: 'append' },
)
} else {
this.session.append('user/message', message, { surfaceOp: 'append' })
} }
steered = true
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item)
this.session.append(
'steering/message',
{ turn, content: item.content, source: item.source },
{ surfaceOp: 'append' },
)
} }
return steered return steered
} }
+2 -4
View File
@@ -158,7 +158,7 @@ export abstract class Agent {
/** /**
* The unified delivery primitive over the (`target` × `wakeup`) matrix. * The unified delivery primitive over the (`target` × `wakeup`) matrix.
* Detaches, validates, and freezes one lossless-JSON item, then routes it: * It routes the caller's typed content and source as follows:
* *
* - `next-turn` (default) queues an item that becomes the sole ordinary * - `next-turn` (default) queues an item that becomes the sole ordinary
* message of its own FIFO-ordered turn; `wakeup` (default `true`) wakes a * message of its own FIFO-ordered turn; `wakeup` (default `true`) wakes a
@@ -169,8 +169,6 @@ export abstract class Agent {
* without running the model: an open turn stages it for the next safe log * without running the model: an open turn stages it for the next safe log
* position, while an idle injection appends it immediately without opening * position, while an idle injection appends it immediately without opening
* a turn. * a turn.
*
* Invalid input throws synchronously before any notification, enqueue, or append.
* @param content - the model-facing content blocks to deliver. * @param content - the model-facing content blocks to deliver.
* @param options - target queue, wakeup decision, and source. * @param options - target queue, wakeup decision, and source.
* @returns the accepted message's {@link AgentMessageId}, stable across its `agent/inbox/*` events. * @returns the accepted message's {@link AgentMessageId}, stable across its `agent/inbox/*` events.
@@ -289,7 +287,7 @@ declare module 'cordis' {
*/ */
'agent/status'(this: Scoped<Agent>, agent: Agent, status: AgentStatus): void 'agent/status'(this: Scoped<Agent>, agent: Agent, status: AgentStatus): void
/** /**
* A frozen item entered the queued or steering inbox. * An item entered the queued or steering inbox.
* @param agent - the owning agent. * @param agent - the owning agent.
* @param message - accepted content, source, and correlation identity. * @param message - accepted content, source, and correlation identity.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.