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

refactor(agent): minimize inbox message contract

Browse Source
This commit is contained in:
_Kerman
2026-07-24 17:00:42 +08:00
parent b3c1abac67
commit 90e69a3123
16 changed files with 58 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -12,7 +12,6 @@ import { Agent, AgentMessageId, agentCarrier, agentInterruptReasonOf, assembleCo
import { createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope'
import type {
AgentMessage,
AgentMessageId as AgentMessageIdType,
CancelOptions,
AgentInterruptReason,
@@ -48,20 +47,6 @@ interface OutboxItem extends PromptMessageData {
steering?: PendingMessage
}
/** Build one live inbox event payload from a pending message. */
function inboxMessage(message: PendingMessage, steering: boolean): AgentMessage {
return {
id: message.id,
content: message.content,
source: message.source,
steering,
wakeup: message.wakeup,
}
}
/** Stable runtime-only reason used when lifecycle teardown interrupts a turn. */
export const DISPOSED_INTERRUPT_REASON = Object.freeze({ kind: 'disposed' } as const)
/** Normalize thrown values while preserving an existing error code. */
function toError(error: unknown): Error & { code?: string } {
return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
@@ -69,12 +54,7 @@ function toError(error: unknown): Error & { code?: string } {
/** Rebuild the live {@link LlmError} for serializable provider facts; `cause` keeps the foreign original. */
function llmError(facts: LlmFailure, cause?: Error): LlmError {
return new LlmError(facts.message, facts.code, {
...facts.status === undefined ? {} : { status: facts.status },
...facts.providerRetryAfterMs === undefined ? {} : { providerRetryAfterMs: facts.providerRetryAfterMs },
...facts.requestId === undefined ? {} : { requestId: facts.requestId },
...cause === undefined ? {} : { cause },
})
return new LlmError(facts.message, facts.code, { ...facts, cause })
}
function withoutToolCalls(message: Message): Message {
@@ -154,7 +134,7 @@ export class ReactLoopAgent extends Agent {
} else {
this.queued.push(message)
}
emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', inboxMessage(message, steering))
emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message)
if (!steering && wakeup) this.kick()
return id
}
@@ -174,8 +154,8 @@ export class ReactLoopAgent extends Agent {
}
if (!options.keepInbox) {
const discarded = [
...this.queued.map(message => inboxMessage(message, false)),
...this.outbox.flatMap(item => item.steering === undefined ? [] : [inboxMessage(item.steering, true)]),
...this.queued,
...this.outbox.map(item => item.steering).filter(steering => steering !== undefined),
]
// Clear before abort observers run: replacement work belongs to the next turn.
this.queued.length = 0
@@ -212,7 +192,7 @@ export class ReactLoopAgent extends Agent {
const message = this.queued.shift()
if (message === undefined) return
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', inboxMessage(message, false))
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)
const admission = new AbortController()
this.abort = admission
this.done = this.loopCtx.agents.withInitiator(this, async () => {
@@ -472,7 +452,7 @@ export class ReactLoopAgent extends Agent {
continue
}
steered = true
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', inboxMessage(message, true))
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)
this.session.append('steering/message', { turn, ...data }, { surfaceOp: 'append' })
}
return steered

4
packages/core/agent-loop/src/index.ts
View File

@@ -24,7 +24,7 @@ import type { Session, SessionHeader } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
import { DISPOSED_INTERRUPT_REASON, ReactLoopAgent } from './agent.ts'
import { ReactLoopAgent } from './agent.ts'
import { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
/** Fiber states that cannot own or serve a new lifecycle. */
@@ -362,7 +362,7 @@ export class AgentLoop extends Service implements AgentFactory {
// sent after this point is the sender's bug — the registries are about
// to drop the agent, so nothing should still hold it.
if (machine !== undefined) {
machine.cancel(DISPOSED_INTERRUPT_REASON)
machine.cancel({ kind: 'disposed' })
await Promise.allSettled([machine.done])
await machine.scope.dispose()
}

27
packages/core/agent/src/types.ts
View File

@@ -2,23 +2,6 @@
* Public agent types and live-runtime events. Durable transcript facts and
* turn/step boundaries remain `@deepseek-ai/dsh-session` events.
*
* The agent is a naive message machine over the session log: prompts queue
* (one turn each), steering/context ride the outbox (taken whole at every
* step boundary), and the log re-derives the request history each step — so
* "edit history between steps" needs no dedicated seam. The extension surface
* is deliberately small:
*
* - `agent/prompt-submit` (waterfall): veto/rewrite a claimed prompt.
* - `agent/request` (waterfall): replace the call config per request.
* - `agent/step` (serial): awaited before every request is built — inject
* context, steer, or edit the log here; the request derives after it.
* - `agent/stopping` (serial): the turn is about to close — steer to object.
* - a tool result carrying `concludesTurn` ends the turn at its step (data,
* not a hook): the terminal-tool pattern.
* - `agent/idle` (emit): one per turn close, carrying why it ended. Error
* recovery is a consumer loop: observe an error idle, fix (edit the log,
* wait out a rate limit), then `agent.retry()`.
*
* @module @deepseek-ai/dsh-agent/types
*/
@@ -100,19 +83,13 @@ export function AgentMessageId(id: string): AgentMessageId {
* One accepted {@link Agent.send} message, carried by the `agent/inbox/*` live
* events. `id` is the value `send` 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. `steering`
* is true for a `next-step` item drained between steps; a `next-turn` item is
* claimed at a turn boundary.
* applied, so these are the exact values the item was accepted with.
*/
export interface AgentMessage {
/** The id `send` returned for this message. */
id: AgentMessageId
content: ContentBlock[]
source: MessageSource
/** Whether the item joined the steering FIFO (`next-step`) rather than the queued FIFO. */
steering: boolean
/** Whether the item is marked to wake the driver or force a continuation. */
wakeup: boolean
}
/** Options for {@link Agent.cancel}. */
@@ -319,7 +296,7 @@ declare module 'cordis' {
/**
* A frozen item entered the queued or steering inbox.
* @param agent - the owning agent.
* @param message - accepted routing data and correlation identity.
* @param message - accepted content, source, and correlation identity.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/

16
packages/core/agent/tests/invariant.spec.ts
View File

@@ -58,24 +58,24 @@ describe('agent status invariants', () => {
})
describe('agent inbox invariants', () => {
const info = (steering: boolean) => ({ id: AgentMessageId('m'), content: [], source: { kind: 'user' as const }, steering, wakeup: true })
const info = () => ({ id: AgentMessageId('m'), content: [], source: { kind: 'user' as const } })
it('accepts a dequeue and a discard covered by prior enqueues', async () => {
const ctx = await setup()
const agent = mockAgent('i1')
const at = scopeTarget(agent, agent)
expect(() => {
ctx.emit(at, 'agent/inbox/enqueue', agent, info(false))
ctx.emit(at, 'agent/inbox/enqueue', agent, info(true))
ctx.emit(at, 'agent/inbox/dequeue', agent, info(false))
ctx.emit(at, 'agent/inbox/discard', agent, [info(true)])
ctx.emit(at, 'agent/inbox/enqueue', agent, info())
ctx.emit(at, 'agent/inbox/enqueue', agent, info())
ctx.emit(at, 'agent/inbox/dequeue', agent, info())
ctx.emit(at, 'agent/inbox/discard', agent, [info()])
}).not.toThrow()
})
it('rejects a dequeue with no outstanding item', async () => {
const ctx = await setup()
const agent = mockAgent('i2')
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/inbox/dequeue', agent, info(false)) })
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/inbox/dequeue', agent, info()) })
.toThrow(/without a matching prior enqueue/)
})
@@ -83,8 +83,8 @@ describe('agent inbox invariants', () => {
const ctx = await setup()
const agent = mockAgent('i3')
const at = scopeTarget(agent, agent)
ctx.emit(at, 'agent/inbox/enqueue', agent, info(false))
expect(() => { ctx.emit(at, 'agent/inbox/discard', agent, [info(false), info(true)]) })
ctx.emit(at, 'agent/inbox/enqueue', agent, info())
expect(() => { ctx.emit(at, 'agent/inbox/discard', agent, [info(), info()]) })
.toThrow(/dropped 2 items but only 1 were outstanding/)
})

4
packages/core/scope/tests/invariant.spec.ts
View File

@@ -42,8 +42,8 @@ describe('scoped-dispatch invariants', () => {
'agent/created': [agent],
'agent/disposed': [agent],
'agent/status': [agent, 'idle'],
'agent/inbox/enqueue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' }, steering: false, wakeup: true }],
'agent/inbox/dequeue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' }, steering: false, wakeup: true }],
'agent/inbox/enqueue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' } }],
'agent/inbox/dequeue': [agent, { id: AgentMessageId('m'), content: [], source: { kind: 'user' } }],
'agent/inbox/discard': [agent, []],
'agent/cancel-requested': [agent, { kind: 'user' }],
'agent/session-start': [agent, 'startup'],

24
packages/ui/tui/src/index.ts
View File

@@ -1592,15 +1592,10 @@ export function createTuiChat(
let toolsExpanded = false
let streaming: StreamingAssistantComponent | undefined
let runningStatus: RunningStatus | undefined
// 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.
// TUI steering submissions that the loop has not yet drained, shown as a
// badge on the status line. Each entry is the submitted message's serialized
// source, so an unrelated steering/message cannot consume its slot. Leaving
// `running` clears entries discarded by cancellation.
const pendingSteering: string[] = []
let disposed = false
let shuttingDown: Promise<void> | undefined
@@ -2505,7 +2500,10 @@ export function createTuiChat(
if (disposed) {
appendNotice(`Agent "${agent.id}" is disposed.`, 'error')
} else if (agent.status === 'running') {
agent.steer(content, { source: { kind: 'user' } })
const source = { kind: 'user' } as const
agent.steer(content, { source })
pendingSteering.push(JSON.stringify(source))
refreshStatus()
} else {
agent.followup(content, { source: { kind: 'user' } })
}
@@ -2751,11 +2749,6 @@ export function createTuiChat(
renderEvent(event, { addHistory: false, renderChunks: true })
requestRender()
})
const disposeQueued = ctx.on('agent/inbox/enqueue', (subject, info) => {
if (subject !== agent || !info.steering) return
pendingSteering.push(JSON.stringify(info.source))
refreshStatus()
})
const disposeStatus = ctx.on('agent/status', (subject, status) => {
if (subject !== agent) return
// Leaving 'running' ends the turn's status line; clear any badge so the
@@ -2784,7 +2777,6 @@ export function createTuiChat(
disposeCommandChanges()
stopBannerReveal()
disposeSessionEvents()
disposeQueued()
disposeStatus()
disposeError()
disposeAgent()

21
packages/ui/tui/tests/tui.spec.ts
View File

@@ -574,8 +574,9 @@ describe('pi-tui chat lifecycle and transcript', () => {
expect(result.terminal.output).toContain('— Enter sends steering, Esc cancels')
expect(result.terminal.output).not.toContain('queued')
const queueSteering = (text: string): void => {
result.ctx.emit('agent/inbox/enqueue', result.agent, { id: AgentMessageId('stub'), content: [{ type: 'text', text }], source: { kind: 'user' }, steering: true, wakeup: true })
const submitSteering = (text: string): void => {
result.terminal.send(text)
result.terminal.send('\r')
}
const drainSteering = (text: string): void => {
result.session.append('steering/message', { turn: 1, content: [{ type: 'text', text }], source: { kind: 'user' } }, { surfaceOp: 'append' })
@@ -584,20 +585,19 @@ describe('pi-tui chat lifecycle and transcript', () => {
// A steering queue for a different agent never touches this status line.
const other = { ...result.agent, id: SessionId('other') } as unknown as Agent
result.terminal.output = ''
result.ctx.emit('agent/inbox/enqueue', other, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'elsewhere' }], source: { kind: 'user' }, steering: true, wakeup: true })
result.ctx.emit('agent/inbox/enqueue', other, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'elsewhere' }], source: { kind: 'user' } })
await tick()
expect(result.terminal.output).not.toContain('queued')
// Two steering messages queue while the turn runs.
queueSteering('first')
submitSteering('first')
result.terminal.output = ''
queueSteering('second')
submitSteering('second')
await tick()
expect(result.terminal.output).toContain('2 queued · Enter sends steering, Esc cancels')
// A non-steering queue (an idle-style send) leaves the badge untouched.
// Draining one submitted message decrements the badge.
result.terminal.output = ''
result.ctx.emit('agent/inbox/enqueue', result.agent, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'sent' }], source: { kind: 'user' }, steering: false, wakeup: true })
drainSteering('first')
await tick()
expect(result.terminal.output).toContain('1 queued')
@@ -613,7 +613,7 @@ describe('pi-tui chat lifecycle and transcript', () => {
// A drain with no matching queued entry is ignored rather than underflowing.
result.terminal.output = ''
drainSteering('continuation')
queueSteering('after')
submitSteering('after')
await tick()
expect(result.terminal.output).toContain('1 queued')
@@ -649,9 +649,8 @@ describe('pi-tui chat lifecycle and transcript', () => {
it('derives the fine-grained turn phase from session lifecycle events', async () => {
// A live event before the turn runs has no status controller to move.
const idle = await setup()
// A steering queue arriving while idle has no status line to badge, so the
// refresh is a no-op beyond requesting a render.
idle.ctx.emit('agent/inbox/enqueue', idle.agent, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'early' }], source: { kind: 'user' }, steering: true, wakeup: true })
// Inbox notifications do not affect the status phase while idle.
idle.ctx.emit('agent/inbox/enqueue', idle.agent, { id: AgentMessageId('stub'), content: [{ type: 'text', text: 'early' }], source: { kind: 'user' } })
idle.session.append('tool/call', { turn: 1, step: 0, callId: 'pre' as never, name: 'bash', arguments: '{}' })
await tick()
expect(idle.terminal.output).not.toContain('Executing tools')