fix(agent-loop): address review — quiet-item parking, meta, discard balance
Resolve six review findings on the unified-send change: - quiet (wakeup:false) queued items no longer un-park the driver; the inbox distinguishes hasWakingQueued (drives the loop, idle/quiescence) from hasQueued (anything to dequeue), so a lone quiet item parks at idle and rides the next waking send. whenIdle/cancel settle off the waking signal, so cancelling a parked quiet item no longer hangs whenIdle. - SendOptions.meta on queued/steering sends now reaches the durable user/message and steering/message (was dropped except on injection). - a terminal agent/turn-stop that drops pending steering emits agent/inbox/discard so the enqueue-dequeue-or-discard ledger balances. - the loop-authored continuation reason is snapshotted and frozen like a public send. - gen-cordis-api collects exported classes (body-stripped) so the now- abstract-class Agent and its transitive shapes reappear in the API catalog. Adds regression tests for each and re-records the affected snapshot.
This commit is contained in:
Turtle
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
2026-07-22-unified-send-and-coalesced-user-messages.md: d88cd60f5c09f7961d7a59dbbd0703abdd26cc1a
|
||||
2026-07-22-unified-send-and-coalesced-user-messages.zh.md: d16ed9f763a350e61abc00d3971e4a7711e8f610
|
||||
2026-07-22-unified-send-and-coalesced-user-messages.md: e969012c8172bbba78f04943be46b7dc866ccb7f
|
||||
2026-07-22-unified-send-and-coalesced-user-messages.zh.md: 35cc183a7c7bfd8a649557e50e6e7d379f8f8193
|
||||
|
||||
@@ -36,6 +36,8 @@ Separately, `context/message` and `user/message` had converged: the surface proj
|
||||
|
||||
The delivery surface is now one primitive plus three self-documenting presets, and the (`target` × `wakeup`) matrix makes previously-unreachable combinations explicit. One durable message type serves prompts, injected context, and goal rounds, so the surface projection and every "human prompt?" check simplify to a `source` test. The cost: `Agent` became an abstract class, so object-literal test fakes must supply `followup` and cannot spread a class-typed value without re-casting (prototype methods are non-enumerable); the goal fold's channel split moved from event type to `source.round`; and every consumer that filtered `context/message` now filters `user/message` by source. The turn-enclosure and reconstruction invariants are unchanged — an idle injection still wraps a one-shot turn, now emitting `user/message` instead of `context/message`.
|
||||
|
||||
`wakeup` is the "should the model run" signal, so the inbox distinguishes `hasWakingQueued` (drives the loop and idle/quiescence decisions) from `hasQueued` (anything to dequeue): a lone `next-turn`/no-wakeup item stays parked at idle and rides along the next waking send, and `whenIdle`/`cancel` settle quiescence off the waking signal. `SendOptions.meta` on a queued or steering send is carried onto the durable `user/message`/`steering/message`, matching injection. Every FIFO exit publishes exactly one lifecycle event: a terminal `agent/turn-stop` that drops pending steering now emits `agent/inbox/discard` for it, and a loop-authored continuation reason is snapshotted and frozen like a public send. `gen-cordis-api` collects exported classes (body-stripped) so the now-class `Agent` and its transitive shapes still appear in the model-facing API catalog.
|
||||
|
||||
## Related
|
||||
|
||||
- [one-send-one-turn](../simplification/2026-07-17-one-send-one-turn.md) — the one-claimed-message-per-turn rule this builds on.
|
||||
|
||||
@@ -36,6 +36,8 @@ agent 的对外驱动接口逐渐长出三个近乎平行的动词——`send`
|
||||
|
||||
投递接口现在是一个原语加三个自解释的预设，(`target` × `wakeup`) 矩阵把此前无法表达的组合显式化。一种持久消息类型同时服务提示词、注入的上下文和 goal 轮次，因此对外接口的投影和每一处“是否人类提示词？”检查都简化为一次 `source` 判断。代价是：`Agent` 变成了抽象类，因此对象字面量形式的测试替身必须提供 `followup`，且无法在不重新做类型转换的情况下展开一个类类型的值（原型方法不可枚举）；goal 折叠的通道区分从事件类型改到了 `source.round`；此前过滤 `context/message` 的每个消费方现在改为按来源过滤 `user/message`。轮次封闭与重建的不变量保持不变——空闲状态下的一次注入仍然封装成一个一次性轮次，只是现在发出 `user/message` 而非 `context/message`。
|
||||
|
||||
`wakeup` 是“模型是否应当运行”的信号，因此 inbox 区分 `hasWakingQueued`（驱动 loop 以及空闲/静默判定）与 `hasQueued`（是否有任何可 dequeue 的项）：一个孤立的 `next-turn`/no-wakeup 队列项会停泊在空闲状态，并随下一次唤醒 send 一同带出，而 `whenIdle`/`cancel` 依据唤醒信号来结算静默。排队 send 或 steering send 上的 `SendOptions.meta` 会被带到持久的 `user/message`/`steering/message` 上，与注入保持一致。每一次 FIFO 退出都恰好发布一个生命周期事件：一次会丢弃待处理 steering 项的终止性 `agent/turn-stop` 现在会为它发出 `agent/inbox/discard`，而由 loop 生成的继续原因会像一次对外 send 那样被快照并冻结。`gen-cordis-api` 收集导出的类（剥除方法体），因此如今已是类的 `Agent` 及其传递涉及的形状仍会出现在面向模型的 API 目录中。
|
||||
|
||||
## 相关
|
||||
|
||||
- [one-send-one-turn](../simplification/2026-07-17-one-send-one-turn.md)——本决策所依托的“每轮次只认领一条消息”规则。
|
||||
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -1151,6 +1151,14 @@ export const EVENT_API: readonly EventApiEntry[] = [
|
||||
|
||||
/** Shapes of every exported type the SERVICE_API signatures reference (transitively), sorted by name. */
|
||||
export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
{
|
||||
name: 'Agent',
|
||||
declaration: 'export abstract class Agent {\n abstract readonly id: SessionId;\n abstract readonly options: AgentOptions;\n abstract readonly session: Session;\n abstract readonly status: AgentStatus;\n abstract readonly ctx: Context;\n abstract send(content: ContentBlock[], options?: SendOptions): AgentMessageId;\n abstract cancel(cause?: AgentCancelCause, options?: CancelOptions): void;\n abstract whenIdle(): Promise<void>;\n followup(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId;\n steer(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId;\n inject(content: ContentBlock[], options?: AliasSendOptions): AgentMessageId;\n}',
|
||||
},
|
||||
{
|
||||
name: 'AgentCancelCause',
|
||||
declaration: 'export type AgentCancelCause = {\n readonly kind: \'user\';\n} | {\n readonly kind: \'parent\';\n};',
|
||||
},
|
||||
{
|
||||
name: 'AgentFactory',
|
||||
declaration: 'export interface AgentFactory {\n createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle>;\n resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle>;\n}',
|
||||
@@ -1159,10 +1167,22 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'AgentHandle',
|
||||
declaration: 'export interface AgentHandle {\n agent: Agent;\n dispose(): Promise<void>;\n}',
|
||||
},
|
||||
{
|
||||
name: 'AgentMessageId',
|
||||
declaration: 'export type AgentMessageId = Branded<\'AgentMessageId\'>;',
|
||||
},
|
||||
{
|
||||
name: 'AgentOptions',
|
||||
declaration: 'export interface AgentOptions {\n provider?: string;\n model?: string;\n}',
|
||||
},
|
||||
{
|
||||
name: 'AgentStatus',
|
||||
declaration: 'export type AgentStatus = \'idle\' | \'running\' | \'disposed\';',
|
||||
},
|
||||
{
|
||||
name: 'AliasSendOptions',
|
||||
declaration: 'export type AliasSendOptions = Omit<SendOptions, \'target\' | \'wakeup\'>;',
|
||||
},
|
||||
{
|
||||
name: 'ApprovalOutcome',
|
||||
declaration: 'export type ApprovalOutcome = \'allowed-once\' | \'rejected\' | \'cancelled\' | \'unavailable\';',
|
||||
@@ -1255,6 +1275,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'CallId',
|
||||
declaration: 'export type CallId = Branded<\'CallId\'>;',
|
||||
},
|
||||
{
|
||||
name: 'CancelOptions',
|
||||
declaration: 'export interface CancelOptions {\n keepInbox?: boolean;\n}',
|
||||
},
|
||||
{
|
||||
name: 'CodeBindingErrorClass',
|
||||
declaration: 'export interface CodeBindingErrorClass {\n name: string;\n memberNameProperty: string;\n}',
|
||||
@@ -1495,6 +1519,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'JsonValue',
|
||||
declaration: 'export type JsonValue = null | boolean | number | string | JsonValue[] | {\n [key: string]: JsonValue;\n};',
|
||||
},
|
||||
{
|
||||
name: 'LlmAdapter',
|
||||
declaration: 'export abstract class LlmAdapter {\n providerInfo(provider: string): LlmProviderInfo;\n listModels(_provider: string): Promise<readonly LlmModelInfo[]>;\n resolveModelContext(_provider: string, _model: string): Promise<LlmModelContext | undefined>;\n abstract stream(options: GenerateOptions): AsyncIterable<StreamChunk>;\n}',
|
||||
},
|
||||
{
|
||||
name: 'LlmCallConfig',
|
||||
declaration: 'export interface LlmCallConfig {\n provider: string;\n model: string;\n temperature?: number;\n maxTokens?: number;\n stop?: string[];\n}',
|
||||
@@ -1695,6 +1723,18 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'ScopeKey',
|
||||
declaration: 'export type ScopeKey = object;',
|
||||
},
|
||||
{
|
||||
name: 'SendOptions',
|
||||
declaration: 'export interface SendOptions {\n target?: SendTarget;\n wakeup?: boolean;\n source?: MessageSource;\n contexts?: HookContext[];\n meta?: JsonValue;\n}',
|
||||
},
|
||||
{
|
||||
name: 'SendTarget',
|
||||
declaration: 'export type SendTarget = \'next-turn\' | \'next-step\';',
|
||||
},
|
||||
{
|
||||
name: 'Session',
|
||||
declaration: 'export class Session {\n private log: SessionEvent[];\n private readonly surfaceManager;\n get surface(): SessionSurface {\n return this.surfaceManager;\n }\n readonly header: SessionHeader;\n get id(): SessionId {\n return this.header.id;\n }\n constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader);\n private eventsSnapshot: readonly SessionEvent[] | undefined;\n get events(): readonly SessionEvent[] {\n this.eventsSnapshot ??= Object.freeze([...this.log]);\n return this.eventsSnapshot;\n }\n get seq(): number {\n return this.log.length;\n }\n append<T extends SessionEventType>(type: T, data: SessionEventMap[T], ...opts: T extends SurfaceEventType ? [\n opts: SurfaceIntent\n ] : [\n ]): SessionEvent<T>;\n private headerFold: EpochHeader | undefined;\n private headerFoldSeq;\n requestHeader(): EpochHeader | undefined;\n private derived: Message[];\n private derivedNodes;\n private derivedGeneration;\n deriveMessages(): Message[];\n deriveEventMessage(event: SessionEvent): Message | null;\n}',
|
||||
},
|
||||
{
|
||||
name: 'SessionEvent',
|
||||
declaration: 'export type SessionEvent<T extends SessionEventType = SessionEventType> = {\n [K in SessionEventType]: {\n type: K;\n seq: number;\n time: number;\n data: SessionEventMap[K];\n } & (K extends SurfaceEventType ? {\n sourceEventSeqs?: number[];\n surfaceOp?: SurfaceOp;\n } : object);\n}[T];',
|
||||
@@ -1767,6 +1807,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'SessionReferenceInput',
|
||||
declaration: 'export interface SessionReferenceInput {\n sessionId: SessionId;\n label?: string;\n}',
|
||||
},
|
||||
{
|
||||
name: 'SessionSurface',
|
||||
declaration: 'export interface SessionSurface {\n readonly nodes: readonly number[];\n readonly replaceGeneration: number;\n}',
|
||||
},
|
||||
{
|
||||
name: 'SessionSurfaceSnapshot',
|
||||
declaration: 'export interface SessionSurfaceSnapshot {\n session: SessionHeader;\n capturedThroughSeq: number | null;\n events: SurfaceEvent[];\n}',
|
||||
@@ -1899,6 +1943,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
name: 'SurfaceEventType',
|
||||
declaration: 'export type SurfaceEventType = \'user/message\' | \'assistant/message\' | \'tool/result\' | \'steering/message\';',
|
||||
},
|
||||
{
|
||||
name: 'SurfaceIntent',
|
||||
declaration: 'export interface SurfaceIntent {\n surfaceOp: SurfaceOp;\n sourceEventSeqs?: number[];\n}',
|
||||
},
|
||||
{
|
||||
name: 'SurfaceOp',
|
||||
declaration: 'export type SurfaceOp = \'append\' | {\n op: \'replace\';\n start: number;\n end: number;\n};',
|
||||
|
||||
@@ -201,7 +201,10 @@ export class ReactLoopAgent extends Agent {
|
||||
id: AgentMessageId, content: ContentBlock[], source: MessageSource, wakeup: boolean, options?: SendOptions,
|
||||
): InboxMessage {
|
||||
const contexts = options?.contexts ?? []
|
||||
const accepted = snapshotJsonValue({ id, content, source, contexts, wakeup })
|
||||
const accepted = snapshotJsonValue({
|
||||
id, content, source, contexts, wakeup,
|
||||
...options?.meta !== undefined ? { meta: options.meta } : {},
|
||||
})
|
||||
if (accepted === undefined) {
|
||||
throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
|
||||
}
|
||||
@@ -344,6 +347,11 @@ export class ReactLoopAgent extends Agent {
|
||||
agentEvents(this.loopCtx, this).emit('agent/cancel-requested', resolvedCause)
|
||||
}
|
||||
if (!keepInbox) {
|
||||
// Whether the parked driver was already scheduled to run: a waking item
|
||||
// woke `waitForQueued`, so the loop WILL resume and settle idle waiters
|
||||
// itself through the pre-run-cancel path (possibly after a replacement
|
||||
// prompt). Only a lone quiet item leaves the loop truly parked.
|
||||
const willResume = this.#inbox.hasWakingQueued
|
||||
// Snapshot before clearing so the discard notification carries the exact
|
||||
// dropped items; a replacement synchronously enqueued by an
|
||||
// `agent/cancel-requested` observer belongs to the next turn, not here.
|
||||
@@ -354,6 +362,15 @@ export class ReactLoopAgent extends Agent {
|
||||
const items = discarded.map(({ message, steering }) => agentMessage(message, steering))
|
||||
agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
|
||||
}
|
||||
// Clearing a parked quiet (`wakeup:false`) item reaches quiescence with no
|
||||
// status transition and without waking the parked driver, so settle any
|
||||
// `whenIdle` waiter here. When a waking item was present the loop resumes
|
||||
// and settles itself; while `running` (including the post-turn flush
|
||||
// window) the driver still owns the eventual idle transition. So settle
|
||||
// only for a parked, non-running agent whose sole cleared work was quiet.
|
||||
if (cancellation === undefined && !willResume && this._status !== 'running') {
|
||||
this.settleIdleWaiters()
|
||||
}
|
||||
}
|
||||
cancellation?.request(resolvedCause)
|
||||
}
|
||||
@@ -365,7 +382,9 @@ export class ReactLoopAgent extends Agent {
|
||||
*/
|
||||
whenIdle(): Promise<void> {
|
||||
if (this._status === 'disposed') return this.done
|
||||
if (this._status !== 'running' && !this.#inbox.hasQueued) return Promise.resolve()
|
||||
// A lone quiet (`wakeup:false`) queued item leaves the agent quiescent — the
|
||||
// driver stays parked — so gate on hasWakingQueued, not hasQueued.
|
||||
if (this._status !== 'running' && !this.#inbox.hasWakingQueued) return Promise.resolve()
|
||||
// Agent-owned waiters survive concurrent fiber disposal.
|
||||
return new Promise<void>((resolve) => {
|
||||
this.idleWaiters.push(() => {
|
||||
|
||||
@@ -7,6 +7,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. */
|
||||
@@ -17,6 +18,8 @@ export interface InboxMessage {
|
||||
contexts: HookContext[]
|
||||
/** Whether the item is marked to wake the driver or force a continuation. */
|
||||
wakeup: boolean
|
||||
/** Opaque durable JSON state retained on the durable message but hidden from the model. */
|
||||
meta?: JsonValue
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -39,11 +42,22 @@ export class Inbox {
|
||||
private steeringMessages: InboxMessage[] = []
|
||||
private wakeup: (() => void) | undefined
|
||||
|
||||
/** True while queued messages are pending — read by the idle wait's fast path and the loop's turn-start checks. */
|
||||
/** True while any queued message is pending — read by cancellation's discard snapshot and the turn-start dequeue guard. */
|
||||
get hasQueued(): boolean {
|
||||
return this.queuedMessages.length > 0
|
||||
}
|
||||
|
||||
/**
|
||||
* True while a queued message wants to wake the driver — the "should the loop
|
||||
* run" signal read by the idle wait's fast path, the loop's idle-publish
|
||||
* check, and `whenIdle`. A `wakeup:false` (quiet) item alone leaves this
|
||||
* false, so the driver stays parked until a waking send (or a waking item
|
||||
* ahead of it in FIFO order) drives the loop; the quiet item then rides along.
|
||||
*/
|
||||
get hasWakingQueued(): boolean {
|
||||
return this.queuedMessages.some(message => message.wakeup)
|
||||
}
|
||||
|
||||
/** True while steering messages are pending — read by cancellation and the loop's stop-override check. */
|
||||
get hasSteering(): boolean {
|
||||
return this.steeringMessages.length > 0
|
||||
@@ -116,7 +130,7 @@ export class Inbox {
|
||||
* loop can exit).
|
||||
*/
|
||||
waitForQueued(cancel: Promise<void>): Promise<void> {
|
||||
if (this.hasQueued) return Promise.resolve()
|
||||
if (this.hasWakingQueued) return Promise.resolve()
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
this.wakeup = resolve
|
||||
void cancel.then(resolve)
|
||||
|
||||
@@ -202,9 +202,11 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
|
||||
while (!handle.isDisposed()) {
|
||||
// An idle listener can enqueue and cancel replacement work before the next
|
||||
// wait is installed. Consume that empty marker before parking the driver.
|
||||
// A quiet (`wakeup:false`) item alone must not un-park the loop, so gate on
|
||||
// hasWakingQueued, not hasQueued.
|
||||
if (handle.isPreRunCancelled()) {
|
||||
handle.clearPreRunCancel()
|
||||
if (!handle.inbox.hasQueued) {
|
||||
if (!handle.inbox.hasWakingQueued) {
|
||||
handle.settleIdle()
|
||||
handle.setStatus('idle')
|
||||
continue
|
||||
@@ -218,7 +220,7 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
|
||||
// a replacement prompt still runs before the eventual idle transition.
|
||||
if (handle.isPreRunCancelled()) {
|
||||
handle.clearPreRunCancel()
|
||||
if (!handle.inbox.hasQueued) {
|
||||
if (!handle.inbox.hasWakingQueued) {
|
||||
// Settle before publishing idle: the already-idle path has no status
|
||||
// transition, while an idle listener can register waiters for new work.
|
||||
handle.settleIdle()
|
||||
@@ -235,10 +237,11 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
|
||||
}
|
||||
|
||||
// A synchronous `running` listener can cancel before `runTurn`; balance the
|
||||
// status only when no replacement prompt was queued by that listener.
|
||||
// status only when no waking replacement prompt was queued by that listener
|
||||
// (a lone quiet item parks at idle rather than driving a turn).
|
||||
if (cancellation.signal.aborted) {
|
||||
handle.clearTurnCancellation(cancellation)
|
||||
if (!handle.inbox.hasQueued) {
|
||||
if (!handle.inbox.hasWakingQueued) {
|
||||
handle.setStatus('idle')
|
||||
continue
|
||||
}
|
||||
@@ -266,7 +269,9 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
|
||||
if (!terminalStopped) handle.inbox.enqueue(message)
|
||||
}
|
||||
|
||||
if (!handle.inbox.hasQueued) handle.setStatus('idle')
|
||||
// Park at idle unless a waking item still wants the model to run; a lone
|
||||
// quiet (`wakeup:false`) item stays queued but does not keep the loop busy.
|
||||
if (!handle.inbox.hasWakingQueued) handle.setStatus('idle')
|
||||
}
|
||||
}
|
||||
|
||||
@@ -282,7 +287,10 @@ async function runTurn(
|
||||
for (const message of messages) {
|
||||
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' })
|
||||
session.append('steering/message', {
|
||||
turn, ...prepared.data,
|
||||
...message.meta === undefined ? {} : { meta: message.meta },
|
||||
}, { surfaceOp: 'append' })
|
||||
for (const context of prepared.separateContexts) {
|
||||
session.append('user/message', {
|
||||
content: context.content,
|
||||
@@ -364,7 +372,10 @@ async function runTurn(
|
||||
// `allow.content` REPLACES the prompt bytes (a rewrite); absent keeps them.
|
||||
const content = promptDecision.content ?? message.content
|
||||
const prepared = preparePromptMessage(content, message.source, promptDecision.additionalContexts ?? [])
|
||||
session.append('user/message', prepared.data, { surfaceOp: 'append' })
|
||||
session.append('user/message', {
|
||||
...prepared.data,
|
||||
...message.meta === undefined ? {} : { meta: message.meta },
|
||||
}, { surfaceOp: 'append' })
|
||||
// Separate contexts still enter THIS turn through inject(). Prefix
|
||||
// contexts are already baked into the user/message with their durable
|
||||
// display envelope, so appending them again would duplicate model input.
|
||||
@@ -543,10 +554,15 @@ 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 = {
|
||||
id: AgentMessageId(randomUUID()), content: decision.reason.content,
|
||||
source: decision.reason.source, contexts: [], wakeup: true,
|
||||
}
|
||||
// Detach and freeze the listener-owned reason like a public steer, so an
|
||||
// enqueue listener or the producer cannot mutate the durable/model-visible
|
||||
// steering message before it drains.
|
||||
const item: InboxMessage = deepFreeze({
|
||||
id: AgentMessageId(randomUUID()),
|
||||
content: structuredClone(decision.reason.content),
|
||||
source: structuredClone(decision.reason.source),
|
||||
contexts: [], wakeup: true,
|
||||
})
|
||||
handle.inbox.steer(item)
|
||||
events.emit('agent/inbox/enqueue', agentMessage(item, true))
|
||||
}
|
||||
@@ -572,7 +588,13 @@ async function runTurn(
|
||||
if (terminalStop) {
|
||||
terminalStopped = true
|
||||
// Terminal stop discards steering but preserves ordinary queued prompts.
|
||||
handle.inbox.drainSteering()
|
||||
// Publish a discard for every dropped steering item so the enqueue ⇒
|
||||
// dequeue-or-discard ledger stays balanced (the outstanding-count
|
||||
// invariant and correlation consumers must not be left with dangling ids).
|
||||
const dropped = handle.inbox.drainSteering()
|
||||
if (dropped.length > 0) {
|
||||
events.emit('agent/inbox/discard', dropped.map(item => agentMessage(item, true)))
|
||||
}
|
||||
shouldContinue = false
|
||||
}
|
||||
|
||||
|
||||
@@ -117,6 +117,38 @@ describe('Agent.cancel()', () => {
|
||||
expect(userTexts(agent)).toEqual(['preserved', 'wake it'])
|
||||
})
|
||||
|
||||
it('a lone quiet (wakeup:false) send 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 })
|
||||
await agent.whenIdle()
|
||||
expect(agent.status).toBe('idle')
|
||||
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
|
||||
|
||||
// A later waking send drives the loop, and the quiet item rides along first.
|
||||
send(agent, 'wake')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(userTexts(agent)).toEqual(['quiet', 'wake'])
|
||||
})
|
||||
|
||||
it('cancelling a parked quiet item settles a pending whenIdle() without a later send', async () => {
|
||||
const adapter = new MockAdapter([textResponse('reply')])
|
||||
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 })
|
||||
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).
|
||||
agent.cancel({ kind: 'user' })
|
||||
await idle
|
||||
expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false)
|
||||
})
|
||||
|
||||
it('pre-step cancel drops the about-to-start turn (no turn is opened)', async () => {
|
||||
const adapter = new MockAdapter([textResponse('should not run')])
|
||||
const ctx = await harness(adapter)
|
||||
|
||||
@@ -86,4 +86,35 @@ describe('inbox FIFO-conservation invariant', () => {
|
||||
expect(warn.mock.calls.flat().some(arg => String(arg).includes('agent/inbox'))).toBe(false)
|
||||
expect(warn.mock.calls.flat().some(arg => String(arg).includes('INVARIANT'))).toBe(false)
|
||||
})
|
||||
|
||||
it('stays balanced when a terminal stop discards pending steering', async () => {
|
||||
const adapter = new MockAdapter([textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
const discards: number[] = []
|
||||
ctx.on('agent/inbox/discard', (subject, messages) => { if (subject === agent) discards.push(messages.length) })
|
||||
|
||||
// A continuation reason enqueues a steering item; a terminal stop then drops
|
||||
// it. The drop must emit a discard so the enqueue ⇒ dequeue-or-discard
|
||||
// ledger stays balanced (no dangling outstanding id).
|
||||
ctx.on('agent/turn-continuation', async (subject, _turn, _default, _signal, next) => {
|
||||
if (subject !== agent) return next()
|
||||
return { action: 'continue' as const, reason: { content: [{ type: 'text', text: 'keep going' }], source: { kind: 'plugin', plugin: 'loop' } } }
|
||||
})
|
||||
let stopped = false
|
||||
ctx.on('agent/turn-stop', (subject) => {
|
||||
if (subject !== agent || stopped) return undefined
|
||||
stopped = true
|
||||
return { action: 'stop' as const }
|
||||
})
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(discards).toEqual([1]) // the dropped steering item was reported
|
||||
expect(warn.mock.calls.flat().some(arg => String(arg).includes('agent/inbox'))).toBe(false)
|
||||
expect(warn.mock.calls.flat().some(arg => String(arg).includes('INVARIANT'))).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -526,6 +526,28 @@ describe('agent loop', () => {
|
||||
expect(agent.session.events.some(event => event.type === 'user/message' && event.data.source.kind === 'plugin')).toBe(false)
|
||||
})
|
||||
|
||||
it('preserves SendOptions.meta on the durable user/message and steering/message', async () => {
|
||||
const adapter = new MockAdapter([toolCallResponse('c1', 'noop', {}), textResponse('done')])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineContentToolFixture({
|
||||
name: 'noop', description: '', parameters: {},
|
||||
async execute() {
|
||||
// Running steer carries its own meta onto the durable steering/message.
|
||||
agent.steer([{ type: 'text', text: 's' }], { source: { kind: 'plugin', plugin: 'p' }, meta: { steer: 1 } })
|
||||
return []
|
||||
},
|
||||
}))
|
||||
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 } })
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const user = agent.session.events.find(e => e.type === 'user/message')
|
||||
expect(user?.type === 'user/message' && user.data.meta).toEqual({ prompt: 1 })
|
||||
const steering = agent.session.events.find(e => e.type === 'steering/message')
|
||||
expect(steering?.type === 'steering/message' && steering.data.meta).toEqual({ steer: 1 })
|
||||
})
|
||||
|
||||
it('agent/turn-continuation can force-continue (/loop pattern) and force-stop', async () => {
|
||||
// force-continue: model never calls tools, but a plugin forces 3 steps
|
||||
const adapter = new MockAdapter([
|
||||
|
||||
@@ -30,8 +30,35 @@ function quote(value: string): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Collect exported interface and type shapes; omit names declared in multiple
|
||||
* packages rather than risk serving the wrong package's shape.
|
||||
* Reduce an exported class to its type shape: drop method/constructor bodies
|
||||
* and property initializers so the catalog serves member signatures, not
|
||||
* implementation. An abstract class (e.g. `Agent`) is a public type consumers
|
||||
* program against, so it belongs in the type closure alongside interfaces.
|
||||
*/
|
||||
function classShape(node: ts.ClassDeclaration): ts.ClassDeclaration {
|
||||
const members = node.members.map((member): ts.ClassElement => {
|
||||
if (ts.isMethodDeclaration(member)) {
|
||||
return ts.factory.updateMethodDeclaration(
|
||||
member, member.modifiers, member.asteriskToken, member.name, member.questionToken,
|
||||
member.typeParameters, member.parameters, member.type, undefined)
|
||||
}
|
||||
if (ts.isConstructorDeclaration(member)) {
|
||||
return ts.factory.updateConstructorDeclaration(member, member.modifiers, member.parameters, undefined)
|
||||
}
|
||||
if (ts.isPropertyDeclaration(member)) {
|
||||
return ts.factory.updatePropertyDeclaration(
|
||||
member, member.modifiers, member.name, member.questionToken ?? member.exclamationToken, member.type, undefined)
|
||||
}
|
||||
return member
|
||||
})
|
||||
return ts.factory.updateClassDeclaration(
|
||||
node, node.modifiers, node.name, node.typeParameters, node.heritageClauses, members)
|
||||
}
|
||||
|
||||
/**
|
||||
* Collect exported interface, type-alias, and (body-stripped) class shapes;
|
||||
* omit names declared in multiple packages rather than risk serving the wrong
|
||||
* package's shape.
|
||||
*/
|
||||
function collectTypeDecls(scanRoot: string = root): Map<string, string> {
|
||||
const printer = ts.createPrinter({ removeComments: true })
|
||||
@@ -41,14 +68,16 @@ function collectTypeDecls(scanRoot: string = root): Map<string, string> {
|
||||
const abs = resolve(scanRoot, rel)
|
||||
const sf = ts.createSourceFile(abs, readFileSync(abs, 'utf8'), ts.ScriptTarget.Latest, true)
|
||||
for (const stmt of sf.statements) {
|
||||
if (!ts.isInterfaceDeclaration(stmt) && !ts.isTypeAliasDeclaration(stmt)) continue
|
||||
const named = ts.isInterfaceDeclaration(stmt) || ts.isTypeAliasDeclaration(stmt) || ts.isClassDeclaration(stmt)
|
||||
if (!named || stmt.name === undefined) continue
|
||||
if (!(stmt.modifiers?.some(m => m.kind === ts.SyntaxKind.ExportKeyword) ?? false)) continue
|
||||
const name = stmt.name.text
|
||||
if (decls.has(name)) {
|
||||
ambiguous.add(name)
|
||||
continue
|
||||
}
|
||||
const printed = printer.printNode(ts.EmitHint.Unspecified, stmt, sf).replace(/\r/g, '')
|
||||
const emit = ts.isClassDeclaration(stmt) ? classShape(stmt) : stmt
|
||||
const printed = printer.printNode(ts.EmitHint.Unspecified, emit, sf).replace(/\r/g, '')
|
||||
decls.set(name, printed.length > MAX_DECL_CHARS
|
||||
? `${printed.slice(0, MAX_DECL_CHARS)} /* …truncated — full shape in source */`
|
||||
: printed)
|
||||
|
||||
Reference in New Issue
Block a user