refactor(agent): name delivery methods by intent
This commit is contained in:
Tianyi Cui
@@ -50,9 +50,9 @@ Configured agents start automatically. A model call requires both `provider` and
|
||||
|
||||
### Internal concrete driver
|
||||
|
||||
The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
|
||||
The concrete `ReactLoopAgent` adapter, its `Inbox`, `runLoop`, and instance-bound publication/start controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
|
||||
|
||||
The unified `send()` primitive materializes content, resolved source, and attached contexts once as a detached, deeply frozen lossless-JSON record, then routes it by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO (waking the driver unless `wakeup: false`); if claimed, it is the sole ordinary message in its turn, and its contexts are the prompt waterfall's default additional contexts that materialize only after admission. Absent or `separate` placement appends an independent injected `user/message`; `prompt-prefix` placement bakes the context, the stable `## My request:` delimiter, and effective request into one `user/message`, whose model-hidden envelope retains display content and context descriptors. The waterfall's returned allow is authoritative, so a listener wrapping `next()` preserves downstream `content` and `additionalContexts` unless it intentionally replaces them. A successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, a prompt block, or a pre-start failure may drop its contexts with the message. A running `next-step`/wakeup `steer()` enters the same record shape in the steering FIFO without dispatching `agent/prompt-submit`; its next checkpoint applies the same separate-or-prefix placement to `steering/message`, while policy can still stop before another step. Steering left after turn close and its checkpoint becomes later queued input with contexts intact unless terminal turn policy, cancellation, or disposal discards it. `next-step`/no-wakeup `inject()` bypasses the FIFOs and appends durable context directly: an open-turn injection uses the same accepted-value boundary but defers in a FIFO while the current step executes assistant tool calls (successful batches place it after all results, interrupted batches drain it before turn close), and an idle injection wraps a one-shot `injection` turn. Every FIFO enqueue publishes `agent/inbox/enqueue`; the driver's claims publish `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`. Malformed data throws before enqueue or append.
|
||||
`ReactLoopAgent` maps the public `send()`/`queue()`/`steer()`/`inject()` intents onto native-private `#acceptDelivery`. Each public method resolves every optional field before the private mechanism receives mandatory content, source, contexts, metadata, target, and wakeup facts; no configurable delivery primitive crosses the package seam. `send()` and `queue()` join the ordinary FIFO, respectively waking or leaving an idle driver parked. If claimed, an ordinary item is the sole message in its turn, and its contexts are the prompt waterfall's default additional contexts that materialize only after admission. Absent or `separate` placement appends an independent injected `user/message`; `prompt-prefix` placement bakes the context, the stable `## My request:` delimiter, and effective request into one `user/message`, whose model-hidden envelope retains display content and context descriptors. The waterfall's returned allow is authoritative, so a listener wrapping `next()` preserves downstream `content` and `additionalContexts` unless it intentionally replaces them. A successor waits for the preceding ordinary turn's checkpoint to settle, while cancellation, disposal, a prompt block, or a pre-start failure may drop its contexts with the message. Running `steer()` enters the same record shape in the steering FIFO without dispatching `agent/prompt-submit`; its next checkpoint applies the same separate-or-prefix placement to `steering/message`, while policy can still stop before another step. Steering left after turn close and its checkpoint becomes later queued input with contexts intact unless terminal turn policy, cancellation, or disposal discards it. `inject()` accepts no attached contexts, bypasses both FIFOs, and appends durable context directly: an open-turn injection defers in a FIFO while the current step executes assistant tool calls (successful batches place it after all results, interrupted batches drain it before turn close), and an idle injection wraps a one-shot `injection` turn. Every FIFO enqueue publishes `agent/inbox/enqueue`; the driver's claims publish `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`. Malformed data throws before enqueue or append.
|
||||
|
||||
### Loop lifecycle (`loop.ts`)
|
||||
|
||||
|
||||
@@ -9,11 +9,19 @@
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Context } from 'cordis'
|
||||
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 type {
|
||||
Agent,
|
||||
AgentCancelCause,
|
||||
AgentOptions,
|
||||
AgentStatus,
|
||||
CancelOptions,
|
||||
HookContext,
|
||||
InjectOptions,
|
||||
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 { snapshotJsonValue, type JsonValue, type Session, type SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { DISPOSED_INTERRUPT_REASON, TurnCancellation } from './cancellation.ts'
|
||||
import { Inbox, agentMessage, type InboxMessage } from './inbox.ts'
|
||||
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
|
||||
@@ -33,6 +41,17 @@ const bindContext = Symbol('dsh.agent-loop.bind-context')
|
||||
/** Module-private publication marker. */
|
||||
const publishAgent = Symbol('dsh.agent-loop.publish-agent')
|
||||
|
||||
/** Fully resolved input accepted only by the concrete driver's private delivery mechanism. */
|
||||
type ResolvedDelivery = {
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
meta: JsonValue | undefined
|
||||
} & (
|
||||
| { target: 'next-turn'; wakeup: boolean; contexts: HookContext[] }
|
||||
| { target: 'next-step'; wakeup: true; contexts: HookContext[] }
|
||||
| { target: 'next-step'; wakeup: false; contexts: [] }
|
||||
)
|
||||
|
||||
/** Factory-owned controls that can operate only on the agent created with them. */
|
||||
export interface PreparedReactLoopAgent {
|
||||
/** The unpublished concrete agent. */
|
||||
@@ -101,7 +120,7 @@ export function bindReactLoopAgentContext(agent: ReactLoopAgent, ctx: Context):
|
||||
* the loop driver. Everything observable happens through session events and
|
||||
* the agent/* event taxonomy — plugins never need this class.
|
||||
*/
|
||||
export class ReactLoopAgent extends Agent {
|
||||
export class ReactLoopAgent implements Agent {
|
||||
/** Queued + steering FIFOs; native-private so callers cannot bypass the public driving verbs. */
|
||||
readonly #inbox = new Inbox()
|
||||
|
||||
@@ -162,7 +181,6 @@ export class ReactLoopAgent extends Agent {
|
||||
public readonly session: Session,
|
||||
maxParallelToolCalls: number,
|
||||
) {
|
||||
super()
|
||||
this.maxParallelToolCalls = maxParallelToolCalls
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
this.disposed = promise
|
||||
@@ -197,13 +215,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(
|
||||
id: AgentMessageId, content: ContentBlock[], source: MessageSource, wakeup: boolean, options?: SendOptions,
|
||||
): InboxMessage {
|
||||
const contexts = options?.contexts ?? []
|
||||
private snapshotMessage(id: AgentMessageId, delivery: ResolvedDelivery): InboxMessage {
|
||||
const { content, source, contexts, wakeup, meta } = delivery
|
||||
const accepted = snapshotJsonValue({
|
||||
id, content, source, contexts, wakeup,
|
||||
...options?.meta !== undefined ? { meta: options.meta } : {},
|
||||
...meta !== undefined ? { meta } : {},
|
||||
})
|
||||
if (accepted === undefined) {
|
||||
throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
|
||||
@@ -225,18 +241,17 @@ export class ReactLoopAgent extends Agent {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
}
|
||||
|
||||
send(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
/** Accept one fully resolved intent through the concrete driver's private routing matrix. */
|
||||
#acceptDelivery(delivery: ResolvedDelivery): AgentMessageId {
|
||||
this.assertNotDisposed()
|
||||
const id = AgentMessageId(randomUUID())
|
||||
const target = options?.target ?? 'next-turn'
|
||||
const wakeup = options?.wakeup ?? true
|
||||
const { target, wakeup } = delivery
|
||||
// next-step/no-wakeup is injection: durable context without running the model.
|
||||
if (target === 'next-step' && !wakeup) { this.injectContext(content, options); return id }
|
||||
if (target === 'next-step' && !wakeup) { this.injectContext(delivery); 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).
|
||||
// waking ordinary 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(id, content, source, wakeup, options)
|
||||
const accepted = this.snapshotMessage(id, delivery)
|
||||
if (steering) {
|
||||
this.#inbox.steer(accepted)
|
||||
} else {
|
||||
@@ -246,13 +261,57 @@ export class ReactLoopAgent extends Agent {
|
||||
return id
|
||||
}
|
||||
|
||||
send(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
return this.#acceptDelivery({
|
||||
content,
|
||||
target: 'next-turn',
|
||||
wakeup: true,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
contexts: options?.contexts ?? [],
|
||||
meta: options?.meta,
|
||||
})
|
||||
}
|
||||
|
||||
queue(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
return this.#acceptDelivery({
|
||||
content,
|
||||
target: 'next-turn',
|
||||
wakeup: false,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
contexts: options?.contexts ?? [],
|
||||
meta: options?.meta,
|
||||
})
|
||||
}
|
||||
|
||||
steer(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
return this.#acceptDelivery({
|
||||
content,
|
||||
target: 'next-step',
|
||||
wakeup: true,
|
||||
source: options?.source ?? { kind: 'user' },
|
||||
contexts: options?.contexts ?? [],
|
||||
meta: options?.meta,
|
||||
})
|
||||
}
|
||||
|
||||
inject(content: ContentBlock[], options?: InjectOptions): AgentMessageId {
|
||||
return this.#acceptDelivery({
|
||||
content,
|
||||
target: 'next-step',
|
||||
wakeup: false,
|
||||
source: options?.source ?? { kind: 'plugin', plugin: '' },
|
||||
contexts: [],
|
||||
meta: options?.meta,
|
||||
})
|
||||
}
|
||||
|
||||
/** The `next-step`/no-wakeup injection path: durable context, no FIFO, no run. */
|
||||
private injectContext(content: ContentBlock[], options?: SendOptions): void {
|
||||
const source = options?.source ?? { kind: 'plugin', plugin: '' }
|
||||
private injectContext(delivery: Extract<ResolvedDelivery, { target: 'next-step'; wakeup: false }>): void {
|
||||
const { content, source, meta } = delivery
|
||||
const context = {
|
||||
content,
|
||||
source,
|
||||
...options?.meta !== undefined ? { meta: options.meta } : {},
|
||||
...meta !== undefined ? { meta } : {},
|
||||
}
|
||||
if (isTurnOpen(this.session)) {
|
||||
const accepted = this.acceptContext(context)
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* Per-agent message inbox: queued and steering FIFOs. Purely an in-memory
|
||||
* mechanism of the loop driver — the public surface is `Agent.send()` and its
|
||||
* fixed-preset aliases.
|
||||
* mechanism of the loop driver — callers use `Agent`'s intent-named delivery
|
||||
* methods instead.
|
||||
*
|
||||
* @module dsh-agent-loop/inbox
|
||||
*/
|
||||
@@ -10,7 +10,7 @@ import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-session'
|
||||
import type { AgentMessage, AgentMessageId, HookContext } from '@deepseek-ai/dsh-agent'
|
||||
|
||||
/** One message waiting in an agent's inbox; `id` is the value `send` returned. */
|
||||
/** One message waiting in an agent's inbox; `id` is the value its accepting delivery method returned. */
|
||||
export interface InboxMessage {
|
||||
id: AgentMessageId
|
||||
content: ContentBlock[]
|
||||
@@ -35,7 +35,7 @@ export function agentMessage(message: InboxMessage, steering: boolean): AgentMes
|
||||
/**
|
||||
* Per-agent inbox: a queued FIFO (dequeued once per turn start) and a steering FIFO
|
||||
* (drained between steps of a running turn). Purely an in-memory mechanism of
|
||||
* the loop — the public surface is `Agent.send()` and its aliases.
|
||||
* the loop — the public surface is `Agent`'s intent-named delivery methods.
|
||||
*/
|
||||
export class Inbox {
|
||||
private queuedMessages: InboxMessage[] = []
|
||||
@@ -78,7 +78,7 @@ export class Inbox {
|
||||
/**
|
||||
* Add a message to the steering FIFO. Deliberately no wakeup: steering is
|
||||
* drained between steps of a running turn, never by the idle wait —
|
||||
* `Agent.steer()` on an idle agent falls back to a woken follow-up instead.
|
||||
* `Agent.steer()` on an idle agent falls back to a waking ordinary turn instead.
|
||||
* @param message - the message to inject between steps of the running turn.
|
||||
*/
|
||||
steer(message: InboxMessage): void {
|
||||
|
||||
@@ -106,7 +106,7 @@ describe('Agent.cancel()', () => {
|
||||
ctx.on('agent/inbox/discard', (subject, items) => { if (subject === agent) discards.push(items) })
|
||||
|
||||
// Queue a turn WITHOUT waking the driver, so it sits in the inbox.
|
||||
agent.send([{ type: 'text', text: 'preserved' }], { target: 'next-turn', wakeup: false })
|
||||
agent.queue([{ type: 'text', text: 'preserved' }])
|
||||
// keepInbox cancel: no active turn, work preserved, no discard event.
|
||||
agent.cancel({ kind: 'user' }, { keepInbox: true })
|
||||
expect(discards).toEqual([])
|
||||
@@ -117,14 +117,14 @@ describe('Agent.cancel()', () => {
|
||||
expect(userTexts(agent)).toEqual(['preserved', 'wake it'])
|
||||
})
|
||||
|
||||
it('a lone quiet (wakeup:false) send leaves the agent parked at idle', async () => {
|
||||
it('a lone queued message leaves the agent parked at idle', async () => {
|
||||
const adapter = new MockAdapter([textResponse('reply')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
// A quiet item alone must NOT wake the driver: no turn runs and whenIdle
|
||||
// resolves (the agent is quiescent), leaving the item queued.
|
||||
agent.send([{ type: 'text', text: 'quiet' }], { target: 'next-turn', wakeup: false })
|
||||
agent.queue([{ type: 'text', text: 'quiet' }])
|
||||
await agent.whenIdle()
|
||||
expect(agent.status).toBe('idle')
|
||||
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
|
||||
@@ -140,7 +140,7 @@ describe('Agent.cancel()', () => {
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'quiet' }], { target: 'next-turn', wakeup: false })
|
||||
agent.queue([{ type: 'text', text: 'quiet' }])
|
||||
const idle = agent.whenIdle()
|
||||
// Cancel reaches quiescence with no status transition and no waking send;
|
||||
// whenIdle must still resolve (previously it hung until the next send).
|
||||
|
||||
@@ -539,7 +539,7 @@ describe('agent loop', () => {
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }], { target: 'next-turn', wakeup: true, meta: { prompt: 1 } })
|
||||
agent.send([{ type: 'text', text: 'go' }], { meta: { prompt: 1 } })
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const user = agent.session.events.find(e => e.type === 'user/message')
|
||||
|
||||
Reference in New Issue
Block a user