refactor(agent): expose resolved input acceptance
This commit is contained in:
Tianyi Cui
@@ -52,7 +52,7 @@ Configured agents start automatically. A model call requires both `provider` and
|
||||
|
||||
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.
|
||||
|
||||
`ReactLoopAgent` maps the public `send()`/`queue()`/`steer()`/`inject()` intents onto native-private `#acceptInput`. 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.
|
||||
`ReactLoopAgent.acceptInput()` implements the public fully resolved acceptance path. The `send()`/`queue()`/`steer()`/`inject()` helpers resolve every optional field before delegating to it; direct callers provide mandatory content, source, contexts, metadata, target, and wakeup facts through `ResolvedAgentInput`. `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()` or equivalent `acceptInput()` routing 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()` and non-waking next-step acceptance require an empty context tuple, bypass both FIFOs, and append 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`)
|
||||
|
||||
|
||||
@@ -17,11 +17,12 @@ import type {
|
||||
CancelOptions,
|
||||
HookContext,
|
||||
InjectOptions,
|
||||
ResolvedAgentInput,
|
||||
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 JsonValue, type Session, type SessionId } from '@deepseek-ai/dsh-session'
|
||||
import type { ContentBlock } 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, agentMessage, type InboxMessage } from './inbox.ts'
|
||||
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
|
||||
@@ -41,17 +42,6 @@ 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 ResolvedAgentInput = {
|
||||
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. */
|
||||
@@ -241,8 +231,8 @@ export class ReactLoopAgent implements Agent {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
}
|
||||
|
||||
/** Accept one fully resolved agent input through the concrete driver's private routing matrix. */
|
||||
#acceptInput(input: ResolvedAgentInput): AgentMessageId {
|
||||
/** Accept one fully resolved agent input through the concrete driver's routing matrix. */
|
||||
acceptInput(input: ResolvedAgentInput): AgentMessageId {
|
||||
this.assertNotDisposed()
|
||||
const id = AgentMessageId(randomUUID())
|
||||
const { target, wakeup } = input
|
||||
@@ -262,7 +252,7 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
|
||||
send(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
return this.#acceptInput({
|
||||
return this.acceptInput({
|
||||
content,
|
||||
target: 'next-turn',
|
||||
wakeup: true,
|
||||
@@ -273,7 +263,7 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
|
||||
queue(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
return this.#acceptInput({
|
||||
return this.acceptInput({
|
||||
content,
|
||||
target: 'next-turn',
|
||||
wakeup: false,
|
||||
@@ -284,7 +274,7 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
|
||||
steer(content: ContentBlock[], options?: SendOptions): AgentMessageId {
|
||||
return this.#acceptInput({
|
||||
return this.acceptInput({
|
||||
content,
|
||||
target: 'next-step',
|
||||
wakeup: true,
|
||||
@@ -295,7 +285,7 @@ export class ReactLoopAgent implements Agent {
|
||||
}
|
||||
|
||||
inject(content: ContentBlock[], options?: InjectOptions): AgentMessageId {
|
||||
return this.#acceptInput({
|
||||
return this.acceptInput({
|
||||
content,
|
||||
target: 'next-step',
|
||||
wakeup: false,
|
||||
|
||||
@@ -83,6 +83,40 @@ describe('Agent', () => {
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('acceptInput exposes the fully resolved delivery path without applying helper defaults', async () => {
|
||||
const adapter = new MockAdapter([textResponse('accepted')])
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
const enqueued = Promise.withResolvers<{ id: string; source: unknown; wakeup: boolean }>()
|
||||
ctx.on('agent/inbox/enqueue', (subject, message) => {
|
||||
if (subject === agent) enqueued.resolve(message)
|
||||
})
|
||||
|
||||
const id = agent.acceptInput({
|
||||
content: [{ type: 'text', text: 'advanced input' }],
|
||||
source: { kind: 'plugin', plugin: 'advanced-caller' },
|
||||
contexts: [],
|
||||
meta: { caller: 'advanced' },
|
||||
target: 'next-turn',
|
||||
wakeup: true,
|
||||
})
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(await enqueued.promise).toMatchObject({
|
||||
id,
|
||||
source: { kind: 'plugin', plugin: 'advanced-caller' },
|
||||
wakeup: true,
|
||||
})
|
||||
expect(agent.session.events.find(event => event.type === 'user/message'))
|
||||
.toMatchObject({
|
||||
data: {
|
||||
source: { kind: 'plugin', plugin: 'advanced-caller' },
|
||||
meta: { caller: 'advanced' },
|
||||
},
|
||||
})
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('send() throws after disposal', async () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
@@ -54,12 +54,13 @@ Turn and step boundaries and the model token stream are durable `session/event`
|
||||
|
||||
### Agent interface (`types.ts`)
|
||||
|
||||
`Agent` is a structural interface. Public delivery methods name caller intent; the concrete driver keeps queue targeting and wakeup routing private ([decision](../../../.agents/notes/implemented/architecture/2026-07-24-intent-named-agent-delivery.md)). `send()`, `queue()`, and `steer()` return an opaque `AgentMessageId` carried by that FIFO item's `agent/inbox/enqueue`/`dequeue`/`discard` events. Each snapshots content, resolved source, attached contexts, and model-hidden metadata as one detached, deeply frozen lossless-JSON record before notification and enqueue; invalid data throws synchronously. Omitting `options.source` attests direct human input as `{ kind: 'user' }`, so every non-human producer labels its content.
|
||||
`Agent` is a structural interface. `send()`, `queue()`, `steer()`, and `inject()` name common caller intents; `acceptInput(ResolvedAgentInput)` exposes the same acceptance path when a caller already has exact routing facts ([decision](../../../.agents/notes/implemented/architecture/2026-07-24-intent-named-agent-delivery.md)). Every `ResolvedAgentInput` field is mandatory, and its discriminated union excludes attached contexts from non-waking next-step injection. FIFO acceptance returns an opaque `AgentMessageId` carried by that item's `agent/inbox/enqueue`/`dequeue`/`discard` events. The driver snapshots content, resolved source, attached contexts, and model-hidden metadata as one detached, deeply frozen lossless-JSON record before notification and enqueue; invalid data throws synchronously. The helpers apply defaults: omitting `options.source` on `send()`, `queue()`, or `steer()` attests direct human input as `{ kind: 'user' }`, so every non-human producer labels its content.
|
||||
|
||||
- `agent.send(content, options?)` — queue one independent FIFO message as its own turn and wake the driver. After admission, separate contexts become injected `user/message` events, while prompt-prefix contexts are baked before the effective request in the same `user/message`; a block or replacement of the default additional-context decision can discard them. The [one-send-one-turn Agent Note](../../../.agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md) owns the turn rationale.
|
||||
- `agent.queue(content, options?)` — queue the same ordinary message without waking an idle driver. A lone queued item leaves `whenIdle()` resolved and rides along before the next waking message.
|
||||
- `agent.steer(content, options?)` — while running, queue steering for the next checkpoint without dispatching `agent/prompt-submit`; while idle, create a waking ordinary turn. Attached contexts remain in the same frozen record; separate contexts append immediately after the steering event, while prompt-prefix contexts are baked into that event. Both survive late-steering conversion to queued input and disappear with their message on cancellation or terminal discard. Policy can still stop before another step; after turn close and its checkpoint, remaining steering becomes later queued input unless terminal turn policy, cancellation, or disposal discards it.
|
||||
- `agent.inject(content, options?)` — accept detached in-session context without running the model; the next request sees its `user/message` (default plugin source) with `content` rendered verbatim as a user-role message. `InjectOptions` deliberately has no attached contexts. `options.meta` persists opaque JSON state without rendering it. While a turn is open the injection joins that turn, deferring FIFO while the current tool batch executes and draining before turn close if execution is interrupted; while idle it is wrapped in a one-shot `injection` turn and durability checkpoint ([the turn-enclosure invariant](../../../.agents/notes/implemented/architecture/2026-06-15-turn-enclosure-invariant.md)). Injection bypasses the FIFOs and emits no `agent/inbox/*` event.
|
||||
- `agent.acceptInput(input)` — accept a fully specified route without helper defaults. `next-turn` targets the ordinary FIFO; `next-step` with wakeup targets steering and falls back to a waking ordinary turn while idle; `next-step` without wakeup is injection and requires `contexts: []`. Callers provide `meta: undefined` explicitly when they have no metadata.
|
||||
- `agent.cancel(cause?, options?)` — cancel the active turn and, unless `options.keepInbox`, ALL pending work: an omitted cause means `{ kind: 'user' }`; TypeScript restricts callers to the `user | parent` union, and an active holder copies its discriminant into a detached frozen signal reason before aborting. An effective call emits `agent/cancel-requested` with the cause before clearing queued and steering work; dropped items are reported on `agent/inbox/discard`, and observers may synchronize state but cannot veto cancellation. `keepInbox: true` aborts the turn but preserves queued and steering items (no discard, and un-started work is not dropped). The same-process typed seam adds no runtime validation or compatibility fallback for untyped callers. Repeated active-turn cancellation is first-wins for the signal, and idle cancellation is a safe no-op with no notification. ACP maps to `user`, while in-process parent propagation maps to `parent`. The cause is runtime-only; durable `turn/end` stays coarse `aborted`.
|
||||
- `agent.whenIdle()` — resolve once the agent reaches quiescence after settling out of `running` (idle → immediately; disposed → awaits the loop exit). A non-owner's quiescence-observation hook: it observes the work settling WITHOUT tearing the agent down. Teardown is separate — a lifecycle owner stops and unregisters via `AgentHandle.dispose()`, which awaits the loop exit directly.
|
||||
- `agent.session`, `agent.status`, `agent.options`, `agent.id`
|
||||
@@ -78,7 +79,7 @@ Turn and step boundaries and the model token stream are durable `session/event`
|
||||
|
||||
#### What the model sees
|
||||
|
||||
`send`, `steer`, and `inject` feed the owning session. `agent/prompt-submit`, `agent/session-prefix`, and other declared events let plugins block a prompt or add request material; this interface contributes no fixed prose itself.
|
||||
The four helpers and `acceptInput` feed the owning session. `agent/prompt-submit`, `agent/session-prefix`, and other declared events let plugins block a prompt or add request material; this interface contributes no fixed prose itself.
|
||||
|
||||
#### Token effect
|
||||
|
||||
|
||||
@@ -68,16 +68,17 @@ export function AgentMessageId(id: string): AgentMessageId {
|
||||
|
||||
/**
|
||||
* One accepted FIFO message, carried by the `agent/inbox/*` live events. `id`
|
||||
* is the value `send`, `queue`, or `steer` returned to the caller, stable across
|
||||
* this message's enqueue, dequeue, and discard events. Source defaults are
|
||||
* already applied, so these are the exact values the item was accepted with.
|
||||
* is the value returned by the accepting helper or {@link Agent.acceptInput},
|
||||
* stable across this message's enqueue, dequeue, and discard events. Source
|
||||
* defaults, when applicable, are already applied, so these are the exact values
|
||||
* the item was accepted with.
|
||||
* `steering` is true for an item drained between steps; otherwise it is claimed
|
||||
* at a turn boundary. `SendOptions.meta` is intentionally omitted: it is durable
|
||||
* model-hidden state that lands on the eventual `user/message`/
|
||||
* `steering/message`, not live-event routing data.
|
||||
*/
|
||||
export interface AgentMessage {
|
||||
/** The id returned by the accepting `send`, `queue`, or `steer` call. */
|
||||
/** The id returned by the accepting helper or {@link Agent.acceptInput}. */
|
||||
id: AgentMessageId
|
||||
content: ContentBlock[]
|
||||
source: MessageSource
|
||||
@@ -102,7 +103,7 @@ export interface CancelOptions {
|
||||
* An agent's lifecycle state, emitted on every transition as `agent/status`:
|
||||
* `idle` (parked, waiting for queued work), `running` (the driver is draining
|
||||
* work and may be closing or checkpointing a turn), `disposed` (terminal — no
|
||||
* transition leaves it, and `send`/`queue`/`steer`/`inject` throw).
|
||||
* transition leaves it, and every delivery method throws).
|
||||
*/
|
||||
export type AgentStatus = 'idle' | 'running' | 'disposed'
|
||||
|
||||
@@ -120,6 +121,22 @@ export interface HookContext {
|
||||
meta?: JsonValue
|
||||
}
|
||||
|
||||
/**
|
||||
* Fully specified input for {@link Agent.acceptInput}. Unlike the intent-named
|
||||
* helpers, this form applies no defaults: callers provide content, source,
|
||||
* contexts, metadata (including explicit `undefined`), target, and wakeup.
|
||||
* The union excludes attached contexts from non-waking next-step injection.
|
||||
*/
|
||||
export type ResolvedAgentInput = {
|
||||
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: [] }
|
||||
)
|
||||
|
||||
/**
|
||||
* Prompt interception result. `allow.content` replaces the prompt. Each
|
||||
* `additionalContexts` entry follows its declared placement: separate context
|
||||
@@ -223,6 +240,19 @@ export interface Agent {
|
||||
*/
|
||||
inject(content: ContentBlock[], options?: InjectOptions): AgentMessageId
|
||||
|
||||
/**
|
||||
* Accept one fully specified input through the same snapshot and routing path
|
||||
* as the four intent-named helpers. `next-turn` targets the ordinary FIFO;
|
||||
* `next-step`/wakeup targets steering (falling back to an ordinary waking turn
|
||||
* while idle); and `next-step` without wakeup injects durable context without
|
||||
* running the model. Every field is mandatory and no source or routing default
|
||||
* is applied. Invalid input throws synchronously before notification, enqueue,
|
||||
* or append.
|
||||
* @param input - the resolved content, attribution, context, metadata, and routing facts.
|
||||
* @returns the accepted input's {@link AgentMessageId}, carried by FIFO lifecycle events when applicable.
|
||||
*/
|
||||
acceptInput(input: ResolvedAgentInput): AgentMessageId
|
||||
|
||||
/**
|
||||
* Clear queued and steering work — unless `keepInbox` — and abort the active
|
||||
* turn. An effective call first emits `agent/cancel-requested` with the
|
||||
@@ -275,8 +305,9 @@ declare module 'cordis' {
|
||||
* A detached, frozen item entered the agent's inbox (queued or steering
|
||||
* FIFO). Source defaults are already applied, so `message` holds the exact
|
||||
* accepted values. This is the enqueue-time live signal; the durable record
|
||||
* is the eventual `user/message`/`steering/message`. Injection
|
||||
* through `agent.inject()` bypasses the FIFOs and does not emit this.
|
||||
* is the eventual `user/message`/`steering/message`. Injection through
|
||||
* `agent.inject()` or equivalent `acceptInput()` routing bypasses the FIFOs
|
||||
* and does not emit this.
|
||||
* @param agent - the agent whose inbox received the item.
|
||||
* @param message - the accepted message (its returned `id`, content, source, contexts, steering, and wakeup facts).
|
||||
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
|
||||
|
||||
@@ -15,6 +15,7 @@ import type {
|
||||
ContinuationStop,
|
||||
CreateAgentOptions,
|
||||
InjectOptions,
|
||||
ResolvedAgentInput,
|
||||
ResumeAgentOptions,
|
||||
SendOptions,
|
||||
} from '@deepseek-ai/dsh-agent'
|
||||
@@ -31,6 +32,7 @@ function stubAgent(rawId: string, overrides: Partial<Agent> = {}): Agent {
|
||||
queue: () => AgentMessageId('stub'),
|
||||
steer: () => AgentMessageId('stub'),
|
||||
inject: () => AgentMessageId('stub'),
|
||||
acceptInput: () => AgentMessageId('stub'),
|
||||
cancel() {},
|
||||
whenIdle() { return Promise.resolve() },
|
||||
...overrides,
|
||||
@@ -38,10 +40,20 @@ function stubAgent(rawId: string, overrides: Partial<Agent> = {}): Agent {
|
||||
}
|
||||
|
||||
describe('AgentRegistry', () => {
|
||||
it('keeps concrete delivery routing out of public options', () => {
|
||||
it('keeps helper options semantic and makes advanced input fully specified', () => {
|
||||
type OptionalInputKey = {
|
||||
[Key in keyof ResolvedAgentInput]-?: Record<never, never> extends Pick<ResolvedAgentInput, Key>
|
||||
? Key
|
||||
: never
|
||||
}[keyof ResolvedAgentInput]
|
||||
|
||||
expectTypeOf<'target' extends keyof SendOptions ? true : false>().toEqualTypeOf<false>()
|
||||
expectTypeOf<'wakeup' extends keyof SendOptions ? true : false>().toEqualTypeOf<false>()
|
||||
expectTypeOf<'contexts' extends keyof InjectOptions ? true : false>().toEqualTypeOf<false>()
|
||||
expectTypeOf<Parameters<Agent['acceptInput']>[0]>().toEqualTypeOf<ResolvedAgentInput>()
|
||||
expectTypeOf<OptionalInputKey>().toEqualTypeOf<never>()
|
||||
expectTypeOf<Extract<ResolvedAgentInput, { target: 'next-step'; wakeup: false }>['contexts']>()
|
||||
.toEqualTypeOf<[]>()
|
||||
})
|
||||
|
||||
it('allows terminal stop policy to cooperate asynchronously with turn cancellation', () => {
|
||||
|
||||
Reference in New Issue
Block a user