feat(agent): rename InboxItemInfo to AgentMessage with an id; send returns it
Add a branded AgentMessageId assigned to each accepted send message and returned from send/followup/steer/inject (was void). Rename the inbox event payload InboxItemInfo to AgentMessage, carrying that id so a caller can correlate a queued item with its enqueue/dequeue/discard events.
This commit is contained in:
@@ -6,15 +6,16 @@
|
||||
* @module dsh-agent-loop/agent
|
||||
*/
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Context } from 'cordis'
|
||||
import { agentEvents } from '@deepseek-ai/dsh-agent'
|
||||
import { agentEvents, AgentMessageId } from '@deepseek-ai/dsh-agent'
|
||||
import { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type { AgentCancelCause, AgentOptions, AgentStatus, CancelOptions, HookContext, SendOptions } from '@deepseek-ai/dsh-agent'
|
||||
import { deepFreeze, errorChain } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import { snapshotJsonValue, type Session, type SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { DISPOSED_INTERRUPT_REASON, TurnCancellation } from './cancellation.ts'
|
||||
import { Inbox, inboxInfo, type InboxMessage } from './inbox.ts'
|
||||
import { Inbox, agentMessage, type InboxMessage } from './inbox.ts'
|
||||
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
|
||||
|
||||
/** Sessions already claimed by a concrete driver construction. */
|
||||
@@ -196,9 +197,11 @@ export class ReactLoopAgent extends Agent {
|
||||
* materialization reads every nested field once; deep freeze prevents later
|
||||
* caller mutation before an inbox or deferred-injection queue drains it.
|
||||
*/
|
||||
private acceptMessage(content: ContentBlock[], source: MessageSource, wakeup: boolean, options?: SendOptions): InboxMessage {
|
||||
private acceptMessage(
|
||||
id: AgentMessageId, content: ContentBlock[], source: MessageSource, wakeup: boolean, options?: SendOptions,
|
||||
): InboxMessage {
|
||||
const contexts = options?.contexts ?? []
|
||||
const accepted = snapshotJsonValue({ content, source, contexts, wakeup })
|
||||
const accepted = snapshotJsonValue({ id, content, source, contexts, wakeup })
|
||||
if (accepted === undefined) {
|
||||
throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
|
||||
}
|
||||
@@ -219,23 +222,25 @@ export class ReactLoopAgent extends Agent {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
}
|
||||
|
||||
send(content: ContentBlock[], options?: SendOptions): void {
|
||||
send(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
this.assertNotDisposed()
|
||||
const id = AgentMessageId(randomUUID())
|
||||
const target = options?.target ?? 'next-turn'
|
||||
const wakeup = options?.wakeup ?? true
|
||||
// next-step/no-wakeup is injection: durable context without running the model.
|
||||
if (target === 'next-step' && !wakeup) { this.injectContext(content, options); return }
|
||||
if (target === 'next-step' && !wakeup) { this.injectContext(content, options); return id }
|
||||
// next-step/wakeup is steering into the running turn; idle falls back to a
|
||||
// woken follow-up turn (there is no active turn to attach to).
|
||||
const steering = target === 'next-step' && this._status === 'running'
|
||||
const source = options?.source ?? { kind: 'user' }
|
||||
const accepted = this.acceptMessage(content, source, wakeup, options)
|
||||
const accepted = this.acceptMessage(id, content, source, wakeup, options)
|
||||
if (steering) {
|
||||
this.#inbox.steer(accepted)
|
||||
} else {
|
||||
this.#inbox.enqueue(accepted, wakeup)
|
||||
}
|
||||
agentEvents(this.loopCtx, this).emit('agent/inbox/enqueue', inboxInfo(accepted, steering))
|
||||
agentEvents(this.loopCtx, this).emit('agent/inbox/enqueue', agentMessage(accepted, steering))
|
||||
return id
|
||||
}
|
||||
|
||||
/** The `next-step`/no-wakeup injection path: durable context, no FIFO, no run. */
|
||||
@@ -346,7 +351,7 @@ export class ReactLoopAgent extends Agent {
|
||||
// Clear work already present before abort observers run.
|
||||
this.#inbox.clear()
|
||||
if (discarded.length > 0) {
|
||||
const items = discarded.map(({ message, steering }) => inboxInfo(message, steering))
|
||||
const items = discarded.map(({ message, steering }) => agentMessage(message, steering))
|
||||
agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -7,10 +7,11 @@
|
||||
*/
|
||||
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { HookContext, InboxItemInfo } from '@deepseek-ai/dsh-agent'
|
||||
import type { AgentMessage, AgentMessageId, HookContext } from '@deepseek-ai/dsh-agent'
|
||||
|
||||
/** One message waiting in an agent's inbox. */
|
||||
/** One message waiting in an agent's inbox; `id` is the value `send` returned. */
|
||||
export interface InboxMessage {
|
||||
id: AgentMessageId
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
contexts: HookContext[]
|
||||
@@ -22,10 +23,10 @@ export interface InboxMessage {
|
||||
* Build the `agent/inbox/*` event payload for one inbox item.
|
||||
* @param message - the accepted inbox record.
|
||||
* @param steering - whether the item is in the steering FIFO (`next-step`).
|
||||
* @returns the live-event facts for enqueue/dequeue/discard.
|
||||
* @returns the live-event message for enqueue/dequeue/discard.
|
||||
*/
|
||||
export function inboxInfo(message: InboxMessage, steering: boolean): InboxItemInfo {
|
||||
return { content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup }
|
||||
export function agentMessage(message: InboxMessage, steering: boolean): AgentMessage {
|
||||
return { id: message.id, content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup }
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@@ -5,11 +5,12 @@
|
||||
* @module dsh-agent-loop/loop
|
||||
*/
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Context } from 'cordis'
|
||||
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { isDeepStrictEqual } from 'node:util'
|
||||
import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm'
|
||||
import { agentEvents, agentInterruptReasonOf, assembleContextFor } from '@deepseek-ai/dsh-agent'
|
||||
import { agentEvents, agentInterruptReasonOf, assembleContextFor, AgentMessageId } from '@deepseek-ai/dsh-agent'
|
||||
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
|
||||
import { canonicalHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { PromptMessageData, Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
@@ -19,7 +20,7 @@ import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import { executeToolCalls } from './tool-calls.ts'
|
||||
import { inboxInfo, type Inbox, type InboxMessage } from './inbox.ts'
|
||||
import { agentMessage, type Inbox, type InboxMessage } from './inbox.ts'
|
||||
import type { TurnCancellation } from './cancellation.ts'
|
||||
|
||||
/** Normalize thrown values while preserving an existing error code. */
|
||||
@@ -279,7 +280,7 @@ async function runTurn(
|
||||
const drainSteering = (): boolean => {
|
||||
const messages = handle.inbox.drainSteering()
|
||||
for (const message of messages) {
|
||||
events.emit('agent/inbox/dequeue', inboxInfo(message, true))
|
||||
events.emit('agent/inbox/dequeue', agentMessage(message, true))
|
||||
const prepared = preparePromptMessage(message.content, message.source, message.contexts)
|
||||
session.append('steering/message', { turn, ...prepared.data }, { surfaceOp: 'append' })
|
||||
for (const context of prepared.separateContexts) {
|
||||
@@ -297,7 +298,7 @@ async function runTurn(
|
||||
const message = handle.inbox.dequeueQueued()
|
||||
/* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
|
||||
if (!message) throw new Error('runTurn invariant violated: no queued message at turn start')
|
||||
events.emit('agent/inbox/dequeue', inboxInfo(message, false))
|
||||
events.emit('agent/inbox/dequeue', agentMessage(message, false))
|
||||
const trigger: TurnTrigger = { kind: 'message', source: message.source }
|
||||
|
||||
let reason: TurnEndReason = { kind: 'completed' }
|
||||
@@ -542,9 +543,12 @@ async function runTurn(
|
||||
// enqueue event a public steer would, so the inbox ledger stays balanced
|
||||
// (every FIFO entry has a matching enqueue before its dequeue/discard).
|
||||
if (decision.action === 'continue' && decision.reason) {
|
||||
const item: InboxMessage = { content: decision.reason.content, source: decision.reason.source, contexts: [], wakeup: true }
|
||||
const item: InboxMessage = {
|
||||
id: AgentMessageId(randomUUID()), content: decision.reason.content,
|
||||
source: decision.reason.source, contexts: [], wakeup: true,
|
||||
}
|
||||
handle.inbox.steer(item)
|
||||
events.emit('agent/inbox/enqueue', inboxInfo(item, true))
|
||||
events.emit('agent/inbox/enqueue', agentMessage(item, true))
|
||||
}
|
||||
let shouldContinue = decision.action === 'continue'
|
||||
|
||||
|
||||
Reference in New Issue
Block a user