Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge origin/master into worktree/llm-reasoning-effort

Browse Source
This commit is contained in:
Yichen Jiang
2026-07-25 07:59:21 +08:00
parent 8372340f9c f7b36bd36d
commit 1c66759235
856 changed files with 23972 additions and 1907 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
packages/core/agent-loop/src/agent.ts
View File

@@ -6,15 +6,25 @@
* @module dsh-agent-loop/agent
*/
import { randomUUID } from 'node:crypto'
import type { Context } from 'cordis'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type { AgentCancelCause, AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { agentEvents, AgentMessageId } from '@deepseek-ai/dsh-agent'
import type {
Agent,
AgentCancelCause,
AgentOptions,
AgentStatus,
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 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, type InboxMessage } from './inbox.ts'
import { Inbox, agentMessage, type InboxMessage } from './inbox.ts'
import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
/** Sessions already claimed by a concrete driver construction. */
@@ -190,19 +200,17 @@ export class ReactLoopAgent implements Agent {
for (const resolve of waiters) resolve()
}
private resolveSource(options?: SendOptions): MessageSource {
return options?.source ?? { kind: 'user' }
}
/**
* Accept one public message payload as a detached record. Lossless-JSON
* materialization reads every nested field once; deep freeze prevents later
* caller mutation before an inbox or deferred-injection queue drains it.
*/
private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
const source = this.resolveSource(options)
const contexts = options?.contexts ?? []
const accepted = snapshotJsonValue({ content, source, contexts })
private snapshotMessage(id: AgentMessageId, input: ResolvedAgentInput): InboxMessage {
const { content, source, contexts, wakeup, meta } = input
const accepted = snapshotJsonValue({
id, content, source, contexts, wakeup,
...meta !== undefined ? { meta } : {},
})
if (accepted === undefined) {
throw new TypeError('agent message content, source, and contexts must be losslessly JSON-serializable')
}
@@ -223,33 +231,81 @@ export class ReactLoopAgent implements Agent {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
}
send(content: ContentBlock[], options?: SendOptions): void {
/** Accept one fully resolved agent input through the concrete driver's routing matrix. */
send(input: ResolvedAgentInput): AgentMessageId {
this.assertNotDisposed()
const accepted = this.acceptMessage(content, options)
this.#inbox.enqueue(accepted)
const info = { source: accepted.source, contexts: accepted.contexts, steering: false } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
const id = AgentMessageId(randomUUID())
const { target, wakeup } = input
// next-step/no-wakeup is injection: durable context without running the model.
if (target === 'next-step' && !wakeup) { this.injectContext(input); return id }
// next-step/wakeup is steering into the running turn; idle falls back to a
// waking ordinary turn (there is no active turn to attach to).
const steering = target === 'next-step' && this._status === 'running'
const accepted = this.snapshotMessage(id, input)
if (steering) {
this.#inbox.steer(accepted)
} else {
this.#inbox.enqueue(accepted, wakeup)
}
agentEvents(this.loopCtx, this).emit('agent/inbox/enqueue', agentMessage(accepted, steering))
return id
}
steer(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
if (this._status !== 'running') { this.send(content, options); return }
const accepted = this.acceptMessage(content, options)
this.#inbox.steer(accepted)
const info = { source: accepted.source, contexts: accepted.contexts, steering: true } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
followup(content: ContentBlock[], options?: SendOptions): AgentMessageId {
return this.send({
content,
target: 'next-turn',
wakeup: true,
source: options?.source ?? { kind: 'user' },
contexts: options?.contexts ?? [],
meta: options?.meta,
})
}
inject(content: ContentBlock[], options?: InjectOptions): void {
this.assertNotDisposed()
const source = this.resolveSource(options)
const context = {
queue(content: ContentBlock[], options?: SendOptions): AgentMessageId {
return this.send({
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.send({
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.send({
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(input: Extract<ResolvedAgentInput, { target: 'next-step'; wakeup: false }>): void {
const { content, source, meta } = input
// Detach and validate the payload before any append, so malformed input
// cannot open a one-shot turn or otherwise mutate the session.
const accepted = this.acceptContext({
content,
source,
...options?.meta !== undefined ? { meta: options.meta } : {},
}
...meta !== undefined ? { meta } : {},
})
if (isTurnOpen(this.session)) {
const accepted = this.acceptContext(context)
// Provider protocols require every assistant tool-call batch to be
// followed only by its tool results. Historical interrupted batches do
// not own new context; only the currently executing batch may defer it.
@@ -257,27 +313,29 @@ export class ReactLoopAgent implements Agent {
this.deferredInjections.push(accepted)
return
}
this.session.append('context/message', accepted, { surfaceOp: 'append' })
this.session.append('user/message', accepted, { surfaceOp: 'append' })
return
}
// No turn open: wrap the injection in a one-shot turn so every event stays
// turn-enclosed (the durability/replay boundary is the turn).
// turn-enclosed (the durability/replay boundary is the turn). The payload is
// validated above, but `Session.append` can still reject a turn/start
// pre-commit (append re-entrancy from a session/event listener, or an
// internal-dispatch veto), so the finally owes a turn/end only when
// turn/start actually committed.
const turn = lastTurnNumber(this.session) + 1
// Once turn/start enters the log, a turn/end is owed even if the message
// append fails acceptance or pre-commit validation. The finally re-checks
// the log and closes only a turn that actually opened; post-commit observers
// are contained by Session and cannot create a false append failure.
try {
this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
this.session.append('context/message', context, { surfaceOp: 'append' })
this.session.append('user/message', accepted, { surfaceOp: 'append' })
} finally {
// Close the turn if turn/start made it into the log. A pre-commit veto
// must escape rather than being mistaken for a committed turn/end.
if (isTurnOpen(this.session)) {
this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
}
// Decide the durability checkpoint from the log: an accepted one-shot
// turn must be flushed even when its message append was the failing step.
// Checkpoint only an accepted one-shot turn: a turn/start rejected
// pre-commit recorded nothing, so it owes no flush (and a spurious flush
// would emit a phantom-turn agent/error). The payload is validated up
// front, so a committed turn/start is always followed by its user/message.
const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
// Keep inject() synchronous: report checkpoint failures live instead of
// rejecting the caller, and track the task so disposal still drains it.
@@ -301,7 +359,7 @@ export class ReactLoopAgent implements Agent {
private drainDeferredInjections(): void {
const pending = this.deferredInjections.splice(0)
for (const accepted of pending) {
this.session.append('context/message', accepted, { surfaceOp: 'append' })
this.session.append('user/message', accepted, { surfaceOp: 'append' })
}
}
@@ -325,10 +383,14 @@ export class ReactLoopAgent implements Agent {
}
}
cancel(cause?: AgentCancelCause): void {
cancel(cause?: AgentCancelCause, options?: CancelOptions): void {
const resolvedCause = cause ?? { kind: 'user' }
const keepInbox = options?.keepInbox ?? false
const cancellation = this.turnCancellation
const preRun = cancellation === undefined && (this.#inbox.hasQueued || this.#inbox.hasSteering)
// keepInbox preserves pending work, so un-started items must not arm the
// pre-run cancel path that would otherwise drop the next queued turn.
const preRun = !keepInbox && cancellation === undefined
&& (this.#inbox.hasQueued || this.#inbox.hasSteering)
if (cancellation !== undefined || preRun) {
if (preRun) this.preRunCancelled = true
// Coordination consumers must update their own state before this call
@@ -336,9 +398,24 @@ export class ReactLoopAgent implements Agent {
// contained by the fused dispatcher and cannot veto cancellation.
agentEvents(this.loopCtx, this).emit('agent/cancel-requested', resolvedCause)
}
// Clear work already present before abort observers run. A replacement
// synchronously enqueued by an observer belongs to the next turn.
this.#inbox.clear()
if (!keepInbox) {
// 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.
const discarded = this.#inbox.pending()
// Clear work already present before abort observers run.
this.#inbox.clear()
if (discarded.length > 0) {
const items = discarded.map(({ message, steering }) => agentMessage(message, steering))
agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
}
// No idle-waiter settle here: a `whenIdle` waiter exists only while the
// agent is `running` or a waking item is queued, and neither is left
// quiescent by clearing the inbox — a lone quiet item takes `whenIdle`'s
// fast path (no waiter), a waking item keeps the woken driver running,
// and a running agent owns its own idle transition (including the
// post-turn flush window).
}
cancellation?.request(resolvedCause)
}
@@ -349,7 +426,9 @@ export class ReactLoopAgent implements 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(() => {
@@ -407,8 +486,21 @@ export class ReactLoopAgent implements Agent {
*/
private [stopDriver](): Promise<void> | void {
if (this._status !== 'disposed') {
// Snapshot any still-pending inbox items, then CLEAR and mark disposed
// BEFORE emitting the discard — mirroring cancel()'s snapshot→clear→emit
// order so a re-entrant followup()/cancel() from a discard listener throws
// `disposed` (or finds an empty inbox) instead of leaking or double-
// discarding an id. `followup()` emits enqueue unconditionally, so the discard
// is unconditional too (even on an unpublished rollback) to keep every
// enqueued id matched.
const discarded = this.#inbox.pending()
this.#inbox.clear()
this._status = 'disposed'
this.resolveDisposed()
if (discarded.length > 0) {
const items = discarded.map(({ message, steering }) => agentMessage(message, steering))
agentEvents(this.loopCtx, this).emit('agent/inbox/discard', items)
}
// Release whenIdle waiters BEFORE the (guarded) event emit — they are
// internal state that must settle even if a listener throws below. Each
// waiter chains `done`, so it resolves only once the loop actually exits.

71
packages/core/agent-loop/src/inbox.ts
View File

@@ -1,54 +1,91 @@
/**
* Per-agent message inbox: queued and steering FIFOs. Purely an in-memory
* mechanism of the loop driver — the public surface is `Agent.send()` and
* `Agent.steer()`.
* mechanism of the loop driver — callers use `Agent`'s intent-named delivery
* methods instead.
*
* @module dsh-agent-loop/inbox
*/
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
import type { HookContext } from '@deepseek-ai/dsh-agent'
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. */
/** One message waiting in an agent's inbox; `id` is the value its accepting delivery method returned. */
export interface InboxMessage {
id: AgentMessageId
content: ContentBlock[]
source: MessageSource
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
}
/**
* 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 message for enqueue/dequeue/discard.
*/
export function agentMessage(message: InboxMessage, steering: boolean): AgentMessage {
// Frozen: the fused emitter passes this exact object to every listener in
// turn, so one listener must not be able to mutate a field (`id`, `steering`,
// `content`, …) a later listener then observes. `message` is already a frozen
// inbox record, so its nested fields need no re-clone.
return Object.freeze({
id: message.id, 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
* the loop — the public surface is `Agent.send()` / `Agent.steer()`.
* the loop — the public surface is `Agent`'s intent-named delivery methods.
*/
export class Inbox {
private queuedMessages: InboxMessage[] = []
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 follow-up (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
}
/**
* Add a message to the queued FIFO and wake a parked {@link waitForQueued}.
* Add a message to the queued FIFO, waking a parked {@link waitForQueued}
* unless the item opted out. A non-waking item still runs once any woken
* item or later wakeup drives the parked loop.
* @param message - the message to queue for the next turn start.
* @param wake - whether to wake a parked idle wait (default true).
*/
enqueue(message: InboxMessage): void {
enqueue(message: InboxMessage, wake = true): void {
this.queuedMessages.push(message)
this.wakeup?.()
if (wake) this.wakeup?.()
}
/**
* 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 `send()` 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 {
@@ -71,6 +108,18 @@ export class Inbox {
return this.steeringMessages.splice(0)
}
/**
* Snapshot the pending items (queued then steering, FIFO order) without
* removing them — the discard notification's payload source.
* @returns the pending items paired with whether each is steering.
*/
pending(): { message: InboxMessage; steering: boolean }[] {
return [
...this.queuedMessages.map(message => ({ message, steering: false })),
...this.steeringMessages.map(message => ({ message, steering: true })),
]
}
/**
* Discard all pending messages (queued + steering) without delivering them —
* used by `cancel()`, which drops un-started work rather than draining it into
@@ -88,7 +137,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)

72
packages/core/agent-loop/src/loop.ts
View File

@@ -5,11 +5,12 @@
* @module dsh-agent-loop/loop
*/
import { randomUUID } from 'node:crypto'
import type { Context } from 'cordis'
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm'
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 { agentEvents, agentInterruptReasonOf, assembleContextFor, AgentMessageId } 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'
@@ -19,7 +20,7 @@ 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 } from './inbox.ts'
import { agentMessage, type Inbox, type InboxMessage } from './inbox.ts'
import type { TurnCancellation } from './cancellation.ts'
/** Normalize thrown values while preserving an existing error code. */
@@ -201,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
@@ -217,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()
@@ -234,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
}
@@ -260,12 +264,22 @@ export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
handle.clearTurnCancellation(cancellation)
}
// Late steering becomes queued input unless terminal policy stopped the turn.
for (const message of handle.inbox.drainSteering()) {
if (!terminalStopped) handle.inbox.enqueue(message)
// Late steering (arriving after runTurn returns, e.g. during the post-turn
// flush) becomes queued input — unless terminal policy stopped the turn, in
// which case it is dropped and must publish a discard so its enqueue is
// still matched (the invariant only catches a NEGATIVE count, not a leak).
const lateSteering = handle.inbox.drainSteering()
if (terminalStopped) {
if (lateSteering.length > 0) {
events.emit('agent/inbox/discard', lateSteering.map(message => agentMessage(message, true)))
}
} else {
for (const message of lateSteering) 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')
}
}
@@ -279,10 +293,14 @@ async function runTurn(
const drainSteering = (): boolean => {
const messages = handle.inbox.drainSteering()
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('context/message', {
session.append('user/message', {
content: context.content,
source: context.source,
...context.meta === undefined ? {} : { meta: context.meta },
@@ -296,6 +314,7 @@ async function runTurn(
const message = handle.inbox.dequeueQueued()
/* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
if (!message) throw new Error('runTurn invariant violated: no queued message at turn start')
events.emit('agent/inbox/dequeue', agentMessage(message, false))
const trigger: TurnTrigger = { kind: 'message', source: message.source }
let reason: TurnEndReason = { kind: 'completed' }
@@ -361,7 +380,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.
@@ -536,9 +558,21 @@ 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: [] })
// 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))
}
let shouldContinue = decision.action === 'continue'
@@ -562,7 +596,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
}