From 95e75ba3e05636741f1a8508a8fa3d0f49f2a70f Mon Sep 17 00:00:00 2001 From: Turtle Date: Thu, 23 Jul 2026 19:37:21 +0800 Subject: [PATCH] fix(agent-loop): balance inbox invariant on continuation-reason steer The FIFO-conservation invariant fired on the loop-authored continuation reason path: a continue-with-reason decision entered the steering FIFO without an agent/inbox/enqueue, so its later dequeue/discard had no matching enqueue. Emit the enqueue for that steer too, add a regression test that mounts the invariant over a continue-with-reason turn and a cancel, and hoist the duplicated inboxInfo helper into inbox.ts. Found by fresh-eye review. --- ...nified-send-and-coalesced-user-messages.md | 2 +- packages/core/agent-loop/src/agent.ts | 13 +-- packages/core/agent-loop/src/inbox.ts | 12 ++- packages/core/agent-loop/src/loop.ts | 17 ++-- .../agent-loop/tests/inbox-invariant.spec.ts | 89 +++++++++++++++++++ packages/ui/tui/src/index.ts | 23 ++--- packages/ui/tui/tests/tui.spec.ts | 5 +- 7 files changed, 128 insertions(+), 33 deletions(-) create mode 100644 packages/core/agent-loop/tests/inbox-invariant.spec.ts diff --git a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md index 0dfa875d0b..3bf721f0fd 100644 --- a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md +++ b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md @@ -18,7 +18,7 @@ Separately, `context/message` and `user/message` had converged: the surface proj **Goal replay disambiguates by round, not type.** A goal state change is a round-zero goal-sourced `user/message` carrying `goal/change` metadata; a positive round is an admitted continuation prompt. `decodeGoalEvent` now takes a `user/message` and still fails loud on goal metadata under a non-goal source or a goal source lacking metadata. -**Three inbox events replace agent/queued.** `agent/inbox/enqueue` (an item entered a FIFO; carries `target`/`wakeup` on `InboxItemInfo`), `agent/inbox/dequeue` (the driver claimed one), and `agent/inbox/discard` (`cancel()` dropped pending items). Injection never touches a FIFO and emits none of these. The `dsh-agent` invariant companion asserts FIFO conservation: a per-agent outstanding count that dequeue and discard can never drive negative. +**Three inbox events replace agent/queued.** `agent/inbox/enqueue` (an item entered a FIFO; carries `target`/`wakeup` on `InboxItemInfo`), `agent/inbox/dequeue` (the driver claimed one), and `agent/inbox/discard` (`cancel()` dropped pending items). Injection never touches a FIFO and emits none of these. Every FIFO entry publishes an enqueue, including the loop-authored continuation-reason steer (`agent/turn-continuation` returning `{ action: 'continue', reason }`), so the ledger stays balanced with its later dequeue or discard. The `dsh-agent` invariant companion asserts FIFO conservation: a per-agent outstanding count that dequeue and discard can never drive negative. **cancel gains keepInbox.** `cancel(cause?, { keepInbox? })`; when true it aborts the active turn but preserves queued and steering items (no discard event, and un-started work is not dropped). diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index d118fb3063..670945a8fa 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -9,12 +9,12 @@ import type { Context } from 'cordis' import { agentEvents } from '@deepseek-ai/dsh-agent' import { Agent } from '@deepseek-ai/dsh-agent' -import type { AgentCancelCause, AgentOptions, AgentStatus, CancelOptions, HookContext, InboxItemInfo, SendOptions } 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, type InboxMessage } from './inbox.ts' +import { Inbox, inboxInfo, type InboxMessage } from './inbox.ts' import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts' /** Sessions already claimed by a concrete driver construction. */ @@ -205,11 +205,6 @@ export class ReactLoopAgent extends Agent { return deepFreeze(accepted) } - /** Build the `agent/inbox/*` payload for one accepted item. */ - private inboxInfo(message: InboxMessage, steering: boolean): InboxItemInfo { - return { content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup } - } - /** Detach one context before it can outlive its caller in the active-batch FIFO. */ private acceptContext(context: HookContext): HookContext { const accepted = snapshotJsonValue(context) @@ -240,7 +235,7 @@ export class ReactLoopAgent extends Agent { } else { this.#inbox.enqueue(accepted, wakeup) } - agentEvents(this.loopCtx, this).emit('agent/inbox/enqueue', this.inboxInfo(accepted, steering)) + agentEvents(this.loopCtx, this).emit('agent/inbox/enqueue', inboxInfo(accepted, steering)) } /** The `next-step`/no-wakeup injection path: durable context, no FIFO, no run. */ @@ -351,7 +346,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 }) => this.inboxInfo(message, steering)) + const items = discarded.map(({ message, steering }) => inboxInfo(message, steering)) agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items) } } diff --git a/packages/core/agent-loop/src/inbox.ts b/packages/core/agent-loop/src/inbox.ts index a5ddd87b1f..e6bbbcfb7f 100644 --- a/packages/core/agent-loop/src/inbox.ts +++ b/packages/core/agent-loop/src/inbox.ts @@ -7,7 +7,7 @@ */ import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm' -import type { HookContext } from '@deepseek-ai/dsh-agent' +import type { HookContext, InboxItemInfo } from '@deepseek-ai/dsh-agent' /** One message waiting in an agent's inbox. */ export interface InboxMessage { @@ -18,6 +18,16 @@ export interface InboxMessage { wakeup: boolean } +/** + * 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. + */ +export function inboxInfo(message: InboxMessage, steering: boolean): InboxItemInfo { + return { content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup } +} + /** * 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 diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index c13fa5a74f..11733b17e8 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -10,7 +10,7 @@ import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFai 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 type { AgentEventDispatch, ContinuationDecision, HookContext, InboxItemInfo, PromptDecision, RequestError, RequestErrorDecision } 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' import { createTransmissionLog, recordRequestHeader } from './request-log.ts' @@ -19,14 +19,9 @@ 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 type { Inbox, InboxMessage } from './inbox.ts' +import { inboxInfo, type Inbox, type InboxMessage } from './inbox.ts' import type { TurnCancellation } from './cancellation.ts' -/** Build the `agent/inbox/dequeue` payload for one claimed inbox item. */ -function inboxInfo(message: InboxMessage, steering: boolean): InboxItemInfo { - return { content: message.content, source: message.source, contexts: message.contexts, steering, wakeup: message.wakeup } -} - /** Normalize thrown values while preserving an existing error code. */ function toError(error: unknown): RequestError { return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error }) @@ -543,9 +538,13 @@ async function runTurn( break } - // A continuation reason becomes next-step steering. + // A continuation reason becomes next-step steering. Publish the same + // 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) { - handle.inbox.steer({ content: decision.reason.content, source: decision.reason.source, contexts: [], wakeup: true }) + const item: InboxMessage = { content: decision.reason.content, source: decision.reason.source, contexts: [], wakeup: true } + handle.inbox.steer(item) + events.emit('agent/inbox/enqueue', inboxInfo(item, true)) } let shouldContinue = decision.action === 'continue' diff --git a/packages/core/agent-loop/tests/inbox-invariant.spec.ts b/packages/core/agent-loop/tests/inbox-invariant.spec.ts new file mode 100644 index 0000000000..ac84c3d157 --- /dev/null +++ b/packages/core/agent-loop/tests/inbox-invariant.spec.ts @@ -0,0 +1,89 @@ +/** + * Regression: the dsh-agent FIFO-conservation invariant must stay balanced on + * the loop-authored continuation-reason steering path. A continue-with-reason + * decision enters the steering FIFO and later drains (or is discarded by + * cancel); both must be matched by an enqueue event so the invariant's + * outstanding count never goes negative. + * @module dsh-agent-loop/tests/inbox-invariant + */ + +import { describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import LlmService from '@deepseek-ai/dsh-llm' +import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' +import SystemPrompt from '@deepseek-ai/dsh-system-prompt' +import ToolRegistry from '@deepseek-ai/dsh-tools' +import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent' +import * as AgentInvariant from '@deepseek-ai/dsh-agent/invariant' +import InvariantService from '@deepseek-ai/dsh-invariants' +import AgentLoop from '@deepseek-ai/dsh-agent-loop' +import { MockAdapter, textResponse } from './mock-adapter.ts' + +async function harness(adapter: MockAdapter) { + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(InvariantService) + await ctx.plugin(AgentInvariant) + await ctx.plugin(AgentLoop, { agents: [] }) + ctx.llm.registerAdapter(['mock'], adapter) + return ctx +} + +function waitForIdle(ctx: Context, agent: Agent): Promise { + return new Promise((resolve) => { + const dispose = ctx.on('agent/status', (subject, status) => { + if (subject === agent && status === 'idle') { dispose(); resolve() } + }) + }) +} + +describe('inbox FIFO-conservation invariant', () => { + it('stays balanced when a continuation reason enters and drains the steering FIFO', async () => { + const adapter = new MockAdapter([textResponse('step 1'), textResponse('step 2')]) + 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' }) + + let forced = false + ctx.on('agent/turn-continuation', async (_agent, _turn, _default, _signal, next) => { + if (forced) return next() + forced = true + return { action: 'continue' as const, reason: { content: [{ type: 'text', text: 'keep going' }], source: { kind: 'plugin', plugin: 'loop' } } } + }) + + agent.send([{ type: 'text', text: 'go' }]) + await waitForIdle(ctx, agent) + + expect(adapter.requests).toHaveLength(2) + // The continuation reason drained as a steering/message on the second step. + expect(agent.session.events.some(e => e.type === 'steering/message')).toBe(true) + // No invariant violation was logged. + 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 cancel discards a pending continuation reason', async () => { + const adapter = new MockAdapter([textResponse('only step')]) + 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' }) + + // Force a continuation reason, then cancel from the same checkpoint so the + // reason sits in the steering FIFO when the inbox is discarded. + ctx.on('agent/turn-continuation', async (subject, _turn, _default, _signal, next) => { + if (subject !== agent) return next() + queueMicrotask(() => { agent.cancel({ kind: 'user' }) }) + return { action: 'continue' as const, reason: { content: [{ type: 'text', text: 'keep going' }], source: { kind: 'plugin', plugin: 'loop' } } } + }) + + agent.send([{ type: 'text', text: 'go' }]) + await waitForIdle(ctx, agent) + + 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) + }) +}) diff --git a/packages/ui/tui/src/index.ts b/packages/ui/tui/src/index.ts index d473c4ac97..d493c26a9c 100644 --- a/packages/ui/tui/src/index.ts +++ b/packages/ui/tui/src/index.ts @@ -1487,14 +1487,15 @@ export function createTuiChat( let toolsExpanded = false let streaming: StreamingAssistantComponent | undefined let runningStatus: RunningStatus | undefined - // Steering messages queued during the running turn (`agent/inbox/enqueue`) - // that the loop has not yet drained, shown as a badge on the status line. Each - // entry is the queued message's serialized source: a drain (`steering/message`) - // removes one MATCHING entry, so loop-authored steering — continuation reasons - // enter the inbox without an `agent/inbox/enqueue` event — cannot consume a - // pending user message's slot. Cleared on leaving `running`, which also absorbs a - // cancellation that discards the queue without logging drains; the status - // line exists only while running, so idle carries no badge to keep current. + // Steering messages queued during the running turn (`agent/inbox/enqueue` + // with `info.steering`) that the loop has not yet drained, shown as a badge on + // the status line. Each entry is the queued message's serialized source: a + // drain (`steering/message`) removes one MATCHING entry, so a loop-authored + // continuation reason (which enqueues and drains under its own source) pushes + // and pops its own slot and cannot consume a pending user message's slot. + // Cleared on leaving `running`, which also absorbs a cancellation that + // discards the queue without logging drains; the status line exists only + // while running, so idle carries no badge to keep current. const pendingSteering: string[] = [] let disposed = false let shuttingDown: Promise | undefined @@ -2558,9 +2559,9 @@ export function createTuiChat( advanceTurnPhase(event) if (event.type === 'steering/message') { // A queued steering message reached the model as it drained; drop its - // entry from the badge. Matching by source keeps loop-authored steering - // (e.g. continuation reasons), which logs here without a matching - // `agent/inbox/enqueue` increment, from consuming a pending user slot. + // entry from the badge. Matching by source keeps a loop-authored + // continuation reason popping its own enqueued slot rather than a pending + // user message's slot. const drained = pendingSteering.indexOf(JSON.stringify(event.data.source)) if (drained >= 0) { pendingSteering.splice(drained, 1) diff --git a/packages/ui/tui/tests/tui.spec.ts b/packages/ui/tui/tests/tui.spec.ts index 667db4abb3..b45b7a826f 100644 --- a/packages/ui/tui/tests/tui.spec.ts +++ b/packages/ui/tui/tests/tui.spec.ts @@ -594,8 +594,9 @@ describe('pi-tui chat lifecycle and transcript', () => { await tick() expect(result.terminal.output).toContain('1 queued') - // A loop-authored steering event (plugin source, no matching agent/inbox/enqueue) - // cannot consume a pending user slot, even when it drains first. + // A steering/message whose source matches no pending badge entry (here a + // plugin source with no tracked enqueue) pops nothing, so it cannot consume + // a pending user slot even when it drains first. result.terminal.output = '' result.session.append('steering/message', { turn: 1,