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.

.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).
 

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)
       }
     }

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

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'
 

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<void> {
+  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)
+  })
+})

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<void> | 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)

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,