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

Merge remote-tracking branch 'origin/master' into codex/status-bar-token-metrics

Browse Source 
# Conflicts:
#	docs/cordis-catalog/events.md
#	docs/cordis-catalog/services.md
#	docs/core-data-structures/llm-streaming.i18n.yaml
#	docs/event-producer-consumer.md
#	packages/core/agent-loop/README.i18n.yaml
#	packages/core/agent-loop/tests/request-reconstruction.spec.ts
#	packages/core/agent/README.i18n.yaml
#	packages/llm/llm/README.i18n.yaml
This commit is contained in:
Hypatia May
2026-07-28 19:54:58 +08:00
parent fc6eb4e7a3 466d13ed9c
commit 2f1ced77ca
363 changed files with 5880 additions and 3017 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/core/agent-loop/README.i18n.yaml
View File

@@ -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 packages/core/agent-loop/README.md
README.md: b686e796c33f24d98560c212af6de0180d67993b
README.zh.md: f6596fa3721e6f39844f73856554e6082f27a53d
README.md: d25779a7104bc0896e0063db236515ce543b6b5a
README.zh.md: ba9332b64c062d71b721007ac66acb484b02cddd

2
packages/core/agent-loop/README.md
View File

@@ -55,7 +55,7 @@ Configured agents start automatically. A model call requires both `provider` and
The concrete `ReactLoopAgent`, its queued input, outbox, and run controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue` with the resolved queued-or-steering placement; taking it publishes `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue` with the resolved queued-or-steering placement; taking it publishes `agent/inbox/dequeue` with the same placement, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`.
### Loop lifecycle (`agent.ts`)

2
packages/core/agent-loop/README.zh.md
View File

@@ -55,7 +55,7 @@ interface Config {
实体 `ReactLoopAgent`、其排队输入、outbox 与运行控制均为包内部实现。包根只导出插件／服务／配置契约，包导出映射不提供 `./src/*` 逃逸路径；生命周期拥有方通过 `ctx.agents` 创建 agent，而不是点名、构造或启动驱动器内部组件。一个准备完成的会话只能由一个实体驱动器认领；所有可观测行为都通过会话事件和 `agent/*` 事件分类体系发生。
统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。每次 inbox 入队都会发布 `agent/inbox/enqueue`，并携带解析出的 queued 或 steering 路由归类；取走它会发布 `agent/inbox/dequeue`；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。
统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。每次 inbox 入队都会发布 `agent/inbox/enqueue`，并携带解析出的 queued 或 steering 路由归类；取走它会发布 `agent/inbox/dequeue`，并携带相同的路由归类；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。
### 循环生命周期（`agent.ts`）

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

@@ -7,13 +7,11 @@
* @module dsh-agent-loop/agent
*/
import { randomUUID } from 'node:crypto'
import type { Context } from 'cordis'
import { AgentMessageId, agentCarrier, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
import { agentCarrier, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
import { createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope'
import type {
AgentMessage,
Agent,
CancelOptions,
AgentInterruptReason,
@@ -27,11 +25,21 @@ import type {
SendOptions,
} from '@deepseek-ai/dsh-agent'
import {
BlockAssembler, LlmError, assertNever, deepFreeze, errorChain, isHarnessError, llmFailureOf, llmRetryPolicyOf, markAgentLoopRequest,
BlockAssembler,
LlmError,
assertNever,
createAssistantMessage,
deepFreeze,
errorChain,
freezeMessage,
isHarnessError,
llmFailureOf,
llmRetryPolicyOf,
markAgentLoopRequest,
} from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, LlmCallConfig, LlmFailure, Message, PreparedLlmCall, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm'
import { canonicalHeader, headerEquals } from '@deepseek-ai/dsh-session'
import type { Session, SessionId, TurnEndReason, TurnTrigger, UserMessageData } from '@deepseek-ai/dsh-session'
import type { AssistantMessage, Session, SessionId, TurnEndReason, TurnTrigger, UserMessage } from '@deepseek-ai/dsh-session'
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
import { executeToolCalls } from './tool-calls.ts'
@@ -47,9 +55,9 @@ type StepOutcome =
*/
export class ReactLoopAgent implements Agent {
/** Prompts awaiting individual turns. */
private queued: { message: AgentMessage; wakeup: boolean }[] = []
private queued: { message: UserMessage; wakeup: boolean }[] = []
/** Input taken into the session log at step boundaries. */
private outbox: (UserMessageData | AgentMessage)[] = []
private outbox: { message: UserMessage; steering: boolean }[] = []
/** Whether observers see a running interval; consecutive turns share it. */
private busy = false
@@ -93,30 +101,22 @@ export class ReactLoopAgent implements Agent {
/** Accept and route one unified send item. */
send(
input: UserMessageData,
message: UserMessage,
options: SendOptions,
): AgentMessageId {
const { content, source } = deepFreeze(structuredClone(input))
): void {
const { target, wakeup } = options
const id = AgentMessageId(randomUUID())
if (target === 'next-step' && !wakeup) {
if (this.acceptsNextStep) {
this.outbox.push({ content, source })
return id
this.outbox.push({ message, steering: false })
return
}
this.session.append('user/message', { content, source }, { surfaceOp: 'append' })
return id
this.session.append('user/message', message, { surfaceOp: 'append' })
return
}
const placement: InboxPlacement = target === 'next-step' && this.acceptsNextStep ? 'steering' : 'queued'
const message: AgentMessage = {
id,
content,
source,
}
deepFreeze(message)
if (placement === 'steering') {
this.outbox.push(message)
this.outbox.push({ message, steering: true })
} else {
this.queued.push({ message, wakeup })
}
@@ -125,28 +125,27 @@ export class ReactLoopAgent implements Agent {
// can cancel or dispose.
if (placement === 'queued' && wakeup) this.scheduleKick()
emitAgentEvent(this.loopCtx, this, 'agent/inbox/enqueue', message, placement)
return id
}
/** Queue one ordinary prompt turn and wake the driver. */
followup(input: UserMessageData): AgentMessageId {
return this.send(input, {
followup(input: UserMessage): void {
this.send(input, {
target: 'next-turn',
wakeup: true,
})
}
/** Steer the open turn, falling back to a waking prompt while idle. */
steer(input: UserMessageData): AgentMessageId {
return this.send(input, {
steer(input: UserMessage): void {
this.send(input, {
target: 'next-step',
wakeup: true,
})
}
/** Append model-facing context without waking the driver. */
inject(input: UserMessageData): AgentMessageId {
return this.send(input, {
inject(input: UserMessage): void {
this.send(input, {
target: 'next-step',
wakeup: false,
})
@@ -171,8 +170,8 @@ export class ReactLoopAgent implements Agent {
}
if (!options.keepInbox) {
const discarded = this.queued.map(item => item.message)
for (const message of this.outbox) {
if ('id' in message) discarded.push(message)
for (const item of this.outbox) {
if (item.steering) discarded.push(item.message)
}
// Clear before abort observers run: replacement work belongs to the next turn.
this.queued.length = 0
@@ -244,19 +243,21 @@ export class ReactLoopAgent implements Agent {
const trigger: TurnTrigger = { kind: 'message', source: message.source }
// Admitted input stays on the stack until its turn/start commits: the
// turn owns it only once the turn exists in the log.
let admitted: UserMessageData[] | undefined
let admitted: UserMessage[] | undefined
try {
signal.throwIfAborted()
const decision = await this.loopCtx.waterfall(
agentCarrier(this), 'agent/prompt-submit', this, message.content, message.source, signal,
agentCarrier(this), 'agent/prompt-submit', this, message, signal,
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
)
signal.throwIfAborted()
if (decision.kind === 'allow') {
admitted = [{ content: decision.content ?? message.content, source: message.source }]
admitted = [decision.content === undefined
? message
: freezeMessage({ ...message, content: decision.content })]
for (const context of decision.additionalContexts ?? []) {
admitted.push({ content: context.content, source: context.source })
admitted.push(freezeMessage(context))
}
}
} catch (error: unknown) {
@@ -292,7 +293,7 @@ export class ReactLoopAgent implements Agent {
// Published only after the abort owner and pending done are installed: a
// dequeue listener that cancels or disposes must find live cancellation
// and quiescence ownership, not the previous activity's settled state.
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message, 'queued')
}
/**
@@ -301,7 +302,7 @@ export class ReactLoopAgent implements Agent {
*/
private async run(
trigger: TurnTrigger,
admitted: UserMessageData[] = [],
admitted: UserMessage[] = [],
inheritedOutboxLength = 0,
priorFailures: readonly LlmFailure[] = Object.freeze([]),
): Promise<void> {
@@ -353,7 +354,7 @@ export class ReactLoopAgent implements Agent {
// one, and the agent/turn-stopping drain below is skipped for the same
// reason.
if (outcome.concluded) break steps
if (outcome.continueTurn || this.outbox.some(item => 'id' in item)) continue
if (outcome.continueTurn || this.outbox.some(item => item.steering)) continue
break
case 'request-failed': {
// step() reports request failures only after step/start commits
@@ -526,22 +527,25 @@ export class ReactLoopAgent implements Agent {
}
// Truncated (max-tokens) output cannot owe tool calls.
const assembled = assembler.message()
const assembled = assembler.blocks()
const content = finish.kind === 'max-tokens'
? assembled.content.filter(block => block.type !== 'tool-call')
: assembled.content
? assembled.filter(block => block.type !== 'tool-call')
: assembled
const message: AssistantMessage = createAssistantMessage({
content,
source: {
provider: request.provider,
model: request.model,
...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {},
},
})
session.append(
'assistant/message',
{
turn,
step,
content,
provenance: {
provider: request.provider,
model: request.model,
...assembler.replayState !== undefined ? { replayState: assembler.replayState } : {},
},
message,
...assembler.usage === undefined ? {} : { usage: assembler.usage },
},
{ surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
@@ -552,7 +556,7 @@ export class ReactLoopAgent implements Agent {
if (toolCalls.length > 0) {
({ concluded } = await executeToolCalls(
this.loopCtx, turn, step, toolCalls, signal,
context => this.outbox.push({ content: context.content, source: context.source }),
context => this.outbox.push({ message: freezeMessage(context), steering: false }),
))
}
@@ -650,17 +654,17 @@ export class ReactLoopAgent implements Agent {
/** Commit the outbox and report whether it contained steering. */
private drainOutbox(turn: number, limit = this.outbox.length): boolean {
let steered = false
for (const message of this.outbox.splice(0, limit)) {
if ('id' in message) {
for (const item of this.outbox.splice(0, limit)) {
if (item.steering) {
steered = true
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message)
emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.message, 'steering')
this.session.append(
'steering/message',
{ turn, content: message.content, source: message.source },
{ turn, message: item.message },
{ surfaceOp: 'append' },
)
} else {
this.session.append('user/message', message, { surfaceOp: 'append' })
this.session.append('user/message', item.message, { surfaceOp: 'append' })
}
}
return steered
@@ -672,14 +676,14 @@ export class ReactLoopAgent implements Agent {
* accepted beside it cannot split from the request it accompanies.
*/
private flushRejectedAdmissionContexts(): void {
if (this.outbox.some(message => 'id' in message)) return
if (this.outbox.some(item => item.steering)) return
const contexts = this.outbox.splice(0)
for (let index = 0; index < contexts.length; index += 1) {
const context = contexts[index]
const item = contexts[index]
/* v8 ignore next 2 -- the steering precheck proves this batch is context-only */
if (context === undefined || 'id' in context) throw new Error('rejected-admission context batch changed')
if (item === undefined || item.steering) throw new Error('rejected-admission context batch changed')
try {
this.session.append('user/message', context, { surfaceOp: 'append' })
this.session.append('user/message', item.message, { surfaceOp: 'append' })
} catch (error: unknown) {
this.outbox.unshift(...contexts.slice(index))
throw error

17
packages/core/agent-loop/src/tool-calls.ts
View File

@@ -10,8 +10,8 @@
*/
import type { Context } from 'cordis'
import { assertNever, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
import type { Session, UserMessageData } from '@deepseek-ai/dsh-session'
import { assertNever, createToolResultMessage, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
import type { Session, UserMessage } from '@deepseek-ai/dsh-session'
import { TOOL_ABORTED_BEFORE_DISPATCH, TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
/** One tool call after argument parsing, ready to schedule. */
@@ -57,7 +57,7 @@ export async function executeToolCalls(
step: number,
toolCalls: ToolCallBlock[],
signal: AbortSignal,
acceptContext: (context: UserMessageData) => void,
acceptContext: (context: UserMessage) => void,
): Promise<{ concluded: boolean }> {
const agent = ctx.agents.requireInitiator()
const { session } = agent
@@ -119,7 +119,7 @@ async function runGroup(
group: PlannedCall[],
mode: ToolExecutionMode['kind'],
signal: AbortSignal,
acceptContext: (context: UserMessageData) => void,
acceptContext: (context: UserMessage) => void,
): Promise<GroupOutcome> {
const { session } = ctx.agents.requireInitiator()
const { maxParallelToolCalls } = ctx.agentLoop.config
@@ -246,13 +246,14 @@ function appendToolResult(
result: ToolExecutionResult,
callSeq: number,
): void {
session.append('tool/result', {
turn, step,
// Correlation stays with the loop's authoritative model-transcript call id;
// registry results deliberately do not duplicate it.
const message = createToolResultMessage({
callId: block.id,
content: result.content,
isError: result.isError,
})
session.append('tool/result', {
turn, step,
message,
...result.error?.info ? { error: result.error.info } : {},
// The tool's private presentation payload (e.g. a result-time diff),
// persisted so a UI bridge reproduces the card on replay.

6
packages/core/agent-loop/tests/agent-initiator.spec.ts
View File

@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
import { Context, type Fiber } from 'cordis'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import LlmService, { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
@@ -41,7 +41,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
}
function send(agent: Agent, text: string): void {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
/** Adapter that holds both drivers at the same awaited continuation. */
@@ -164,7 +164,7 @@ describe('AgentLoop initiator scope', () => {
if (context.agent === agent) capture(context.signal)
return next()
})
ctx.on('agent/prompt-submit', async (subject, _content, _source, signal, next) => {
ctx.on('agent/prompt-submit', async (subject, _message, signal, next) => {
if (subject === agent) {
expect(ctx.agents.requireInitiator()).toBe(agent)
admissionSignals.push(signal)

41
packages/core/agent-loop/tests/agent.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
@@ -21,10 +22,40 @@ async function harness(adapter: MockAdapter): Promise<Context> {
}
function send(agent: Agent, text: string): void {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
describe('Agent', () => {
it('does not echo caller-owned message identities from delivery methods', async () => {
const adapter = new MockAdapter([
textResponse('one'),
textResponse('two'),
textResponse('three'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const message = (text: string) => createUserMessage({
content: [{ type: 'text' as const, text }],
source: { kind: 'user' as const },
})
const call = (method: 'send' | 'inject' | 'followup' | 'steer', args: unknown[]): unknown => {
const implementation: unknown = Reflect.get(agent, method)
if (typeof implementation !== 'function') throw new Error(`missing Agent.${method}`)
return Reflect.apply(implementation, agent, args)
}
expect(call('send', [message('quiet'), {
target: 'next-turn',
wakeup: false,
}])).toBeUndefined()
expect(call('inject', [message('context')])).toBeUndefined()
expect(call('followup', [message('followup')])).toBeUndefined()
expect(call('steer', [message('steering')])).toBeUndefined()
await agent.whenIdle()
expect(adapter.requests).toHaveLength(3)
})
it('idle inject() appends context without opening a turn or requesting a flush', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
@@ -32,7 +63,7 @@ describe('Agent', () => {
let flushes = 0
ctx.on('session/flush', () => { flushes += 1 })
agent.inject({ content: [{ type: 'text', text: 'context' }], source: { kind: 'plugin', plugin: 'p' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'context' }], source: { kind: 'plugin', plugin: 'p' } }))
expect(agent.session.events.map(event => event.type)).toEqual(['user/message'])
expect(agent.status).toBe('idle')
@@ -45,7 +76,7 @@ describe('Agent', () => {
const ctx = await harness(new MockAdapter([textResponse('ok')]))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.inject({ content: [{ type: 'text', text: 'empty plugin source' }], source: { kind: 'plugin', plugin: '' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'empty plugin source' }], source: { kind: 'plugin', plugin: '' } }))
const injected = agent.session.events.at(-1)
expect(injected?.type === 'user/message' && injected.data.source)
@@ -57,7 +88,7 @@ describe('Agent', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
expect(() => {
agent.inject({ content: [{ type: 'text', text: 'x', bad: 1n } as never], source: { kind: 'plugin', plugin: 'p' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'x', bad: 1n } as never], source: { kind: 'plugin', plugin: 'p' } }))
}).toThrow(/non-JSON-serializable/)
expect(agent.session.events).toHaveLength(0)
})
@@ -67,7 +98,7 @@ describe('Agent', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.steer({ content: [{ type: 'text', text: 'steer idle' }], source: { kind: 'plugin', plugin: 'test' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'steer idle' }], source: { kind: 'plugin', plugin: 'test' } }))
await agent.whenIdle()
expect(agent.session.events.some(event => event.type === 'user/message')).toBe(true)

23
packages/core/agent-loop/tests/cancel.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
/**
* Tests for the queue-aware `Agent.cancel()` primitive. `cancel()` is the broad verb — it
* clears queued + steering work, aborts the active turn, and drops work not yet claimed by the
@@ -33,7 +34,7 @@ async function harness(adapter: MockAdapter) {
}
function send(agent: Agent, text: string) {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
/** Resolve on the agent's next idle transition (event-based, not status poll). */
@@ -63,7 +64,7 @@ describe('Agent.cancel()', () => {
ctx.on('agent/cancel-requested', (subject, cause) => {
if (subject !== agent) return
seen.push(`first:${cause.kind}`)
subject.followup({ content: [{ type: 'text', text: 'queued by cancel observer' }], source: { kind: 'user' } })
subject.followup(createUserMessage({ content: [{ type: 'text', text: 'queued by cancel observer' }], source: { kind: 'user' } }))
throw new Error('observer failed')
})
ctx.on('agent/cancel-requested', (subject, cause) => {
@@ -108,7 +109,7 @@ describe('Agent.cancel()', () => {
ctx.on('agent/cancel-requested', (subject, cause) => { if (subject === agent) cancelRequests.push(cause) })
// Queue a turn WITHOUT waking the driver, so it sits in the inbox.
agent.send({ content: [{ type: 'text', text: 'preserved' }], source: { kind: 'user' } }, { target: 'next-turn', wakeup: false })
agent.send(createUserMessage({ content: [{ type: 'text', text: 'preserved' }], source: { kind: 'user' } }), { target: 'next-turn', wakeup: false })
// keepInbox cancel: no active turn, work preserved, no discard event. With
// nothing to abort and nothing discarded, the call is a documented no-op,
// so it emits no cancel-requested either.
@@ -129,7 +130,7 @@ describe('Agent.cancel()', () => {
// 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({ content: [{ type: 'text', text: 'quiet' }], source: { kind: 'user' } }, { target: 'next-turn', wakeup: false })
agent.send(createUserMessage({ content: [{ type: 'text', text: 'quiet' }], source: { kind: 'user' } }), { 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)
@@ -145,7 +146,7 @@ describe('Agent.cancel()', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.send({ content: [{ type: 'text', text: 'quiet' }], source: { kind: 'user' } }, { target: 'next-turn', wakeup: false })
agent.send(createUserMessage({ content: [{ type: 'text', text: 'quiet' }], source: { kind: 'user' } }), { target: 'next-turn', wakeup: false })
const idle = agent.whenIdle()
agent.cancel({ kind: 'user' })
await idle
@@ -337,8 +338,10 @@ describe('Agent.cancel()', () => {
const result = agent.session.events.find(event => event.type === 'tool/result')
expect(call?.type === 'tool/call' ? call.data.callId : undefined).toBe('c1')
expect(result?.type === 'tool/result' ? result.data : undefined).toMatchObject({
callId: 'c1',
isError: true,
message: {
source: { kind: 'tool', callId: 'c1' },
content: [{ type: 'tool-result', toolCallId: 'c1', isError: true }],
},
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
})
@@ -578,7 +581,7 @@ describe('Agent.cancel()', () => {
expect(agent.status).toBe('running')
// Steer (joins the running turn's steering FIFO), then cancel: the steering
// must be dropped, NOT re-enqueued as a new queued turn.
agent.steer({ content: [{ type: 'text', text: 'steer text' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'steer text' }], source: { kind: 'user' } }))
agent.cancel({ kind: 'user' })
await waitForIdle(ctx, agent)
@@ -591,7 +594,7 @@ describe('Agent.cancel()', () => {
// The steering text was dropped — it never reached the log.
const flat = agent.session.events
.filter(e => e.type === 'steering/message')
.flatMap(e => e.type === 'steering/message' ? e.data.content : [])
.flatMap(e => e.type === 'steering/message' ? e.data.message.content : [])
.flatMap(b => b.type === 'text' ? [b.text] : [])
expect(flat).not.toContain('steer text')
})
@@ -693,7 +696,7 @@ describe('Agent.cancel()', () => {
switch (stage) {
case 'prompt-submit':
ctx.on('agent/prompt-submit', async (subject, _content, _source, signal, next) => {
ctx.on('agent/prompt-submit', async (subject, _message, signal, next) => {
if (subject === agent) await blockUntilAbort(signal)
return next()
})

15
packages/core/agent-loop/tests/config-session-id.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { mkdtemp, rm } from 'node:fs/promises'
@@ -98,7 +99,7 @@ describe('config-driven session id', () => {
first = ctx.agents.get(SessionId('config-exact-reload'))
}
expect(first).toBeDefined()
first!.followup({ content: [{ type: 'text', text: 'remember me' }], source: { kind: 'user' } })
first!.followup(createUserMessage({ content: [{ type: 'text', text: 'remember me' }], source: { kind: 'user' } }))
await waitForIdle(ctx, first!)
await firstLoop.dispose()
@@ -110,7 +111,7 @@ describe('config-driven session id', () => {
}
expect(second).toBeDefined()
expect(JSON.stringify(second!.session.deriveMessages())).toContain('remember me')
second!.followup({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } })
second!.followup(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } }))
await waitForIdle(ctx, second!)
await ctx.sessions.flush(second!.session)
const loaded = await ctx.sessionPersistence.load(SessionId('config-exact-reload'))
@@ -137,7 +138,7 @@ describe('config-driven session id', () => {
cleanupStarted.resolve(undefined)
await cleanupGate.promise
})
first.inject({ content: [{ type: 'text', text: 'persist before replacement' }], source: { kind: 'plugin', plugin: 'test' } })
first.inject(createUserMessage({ content: [{ type: 'text', text: 'persist before replacement' }], source: { kind: 'plugin', plugin: 'test' } }))
await ctx.sessions.flush(first.session)
expect(JSON.stringify((await ctx.sessionPersistence.inspect(sessionId)).events))
.toContain('persist before replacement')
@@ -181,7 +182,7 @@ describe('config-driven session id', () => {
cleanupStarted.resolve(undefined)
await cleanupGate.promise
})
first.inject({ content: [{ type: 'text', text: 'persist before cancellation' }], source: { kind: 'plugin', plugin: 'test' } })
first.inject(createUserMessage({ content: [{ type: 'text', text: 'persist before cancellation' }], source: { kind: 'plugin', plugin: 'test' } }))
await ctx.sessions.flush(first.session)
expect(JSON.stringify((await ctx.sessionPersistence.inspect(sessionId)).events))
.toContain('persist before cancellation')
@@ -345,7 +346,7 @@ describe('config-driven session id', () => {
expect(a1.id).toBe(a1.session.id)
expect(a1.session.id).toMatch(idPattern)
expect(ctx1.agents.get(SessionId('cfg'))).toBeUndefined()
a1.followup({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }))
await waitForIdle(ctx1, a1)
await ctx1.fiber.dispose()
@@ -364,7 +365,7 @@ describe('config-driven session id', () => {
expect(a2.id).toBe(a2.session.id)
expect(a2.session.id).toMatch(idPattern)
expect(a2.session.id).not.toBe(a1.session.id)
a2.followup({ content: [{ type: 'text', text: 'q2' }], source: { kind: 'user' } })
a2.followup(createUserMessage({ content: [{ type: 'text', text: 'q2' }], source: { kind: 'user' } }))
await waitForIdle(ctx2, a2)
await ctx2.fiber.dispose()
})
@@ -385,7 +386,7 @@ describe('config-driven session id', () => {
await ctx1.plugin(SessionPersistenceJsonl, { root })
ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')]))
const a1 = (await ctx1.agents.create({ sessionId: SessionId('sticky-1') })).agent
a1.followup({ content: [{ type: 'text', text: 'remember me' }], source: { kind: 'user' } })
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'remember me' }], source: { kind: 'user' } }))
await waitForIdle(ctx1, a1)
await ctx1.fiber.dispose()

48
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, CallId, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture, type PostToolDecision } from '@deepseek-ai/dsh-tools'
@@ -50,7 +50,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
}
function send(agent: Agent, text: string) {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
describe('assistant replay provenance', () => {
@@ -66,11 +66,11 @@ describe('assistant replay provenance', () => {
await waitForIdle(ctx, agent)
const recorded = agent.session.events.find(event => event.type === 'assistant/message')
expect(recorded?.type === 'assistant/message' && recorded.data.provenance).toEqual({
provider: 'mock', model: 'next-model', replayState,
expect(recorded?.type === 'assistant/message' && recorded.data.message.source).toEqual({
kind: 'model', provider: 'mock', model: 'next-model', replayState,
})
expect(agent.session.deriveMessages().at(-1)?.provenance).toEqual({
provider: 'mock', model: 'next-model', replayState,
expect(agent.session.deriveMessages().at(-1)?.source).toEqual({
kind: 'model', provider: 'mock', model: 'next-model', replayState,
})
})
})
@@ -85,17 +85,17 @@ describe('abort during tool execution ends the turn', () => {
description: '',
parameters: {},
async execute() {
agent.inject({ content: [{ type: 'text', text: 'accepted before abort' }], source: { kind: 'plugin', plugin: 'test' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'accepted before abort' }], source: { kind: 'plugin', plugin: 'test' } }))
agent.cancel({ kind: 'user' })
return [{ type: 'text', text: 'done' }]
},
}))
ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
kind: 'accept',
additionalContexts: [{
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: 'accepted result context after abort' }],
source: { kind: 'plugin', plugin: 'test' },
}],
})],
}))
send(agent, 'go')
@@ -148,10 +148,10 @@ describe('abort during tool execution ends the turn', () => {
if (exec.callId !== CallId('c1')) return next()
return {
kind: 'accept',
additionalContexts: [{
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: 'accepted after first result' }],
source: { kind: 'plugin', plugin: 'test' },
}],
})],
}
})
@@ -185,7 +185,7 @@ describe('abort during tool execution ends the turn', () => {
description: '',
parameters: {},
async execute(_args, exec) {
agent.inject({ content: [{ type: 'text', text: 'accepted before disposal' }], source: { kind: 'plugin', plugin: 'test' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'accepted before disposal' }], source: { kind: 'plugin', plugin: 'test' } }))
started.resolve(undefined)
const signal = exec.signal
if (!signal) throw new Error('tool execution signal is missing')
@@ -198,10 +198,10 @@ describe('abort during tool execution ends the turn', () => {
}))
ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
kind: 'accept',
additionalContexts: [{
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: 'accepted result context during disposal' }],
source: { kind: 'plugin', plugin: 'test' },
}],
})],
}))
send(agent, 'go')
@@ -254,7 +254,7 @@ describe('abort during tool execution ends the turn', () => {
await waitForIdle(ctx, agent)
ctx.on('agent/step', (subject, turn) => {
if (subject === agent && turn === 2) {
agent.inject({ content: [{ type: 'text', text: 'new turn context' }], source: { kind: 'plugin', plugin: 'test' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'new turn context' }], source: { kind: 'plugin', plugin: 'test' } }))
}
})
send(agent, 'start a text-only turn')
@@ -282,7 +282,7 @@ describe('steering from late extension points is never stranded', () => {
ctx.on('agent/turn-stopping', () => {
if (!steeredOnce) {
steeredOnce = true
agent.steer({ content: [{ type: 'text', text: 'one more thing' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'one more thing' }], source: { kind: 'user' } }))
}
})
@@ -307,7 +307,7 @@ describe('steering from late extension points is never stranded', () => {
ctx.on('session/event', (subject, event) => {
if (subject !== agent.session || event.type !== 'step/end' || steeredOnce) return
steeredOnce = true
agent.steer({ content: [{ type: 'text', text: 'goal reminder from step/end' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'goal reminder from step/end' }], source: { kind: 'user' } }))
})
send(agent, 'go')
@@ -339,7 +339,7 @@ describe('steering from late extension points is never stranded', () => {
if (event.type === 'turn/end' && !steeredOnce) {
steeredOnce = true
expect(agent.acceptsNextStep).toBe(false)
agent.steer({ content: [{ type: 'text', text: 'too late for this turn' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'too late for this turn' }], source: { kind: 'user' } }))
}
})
@@ -494,7 +494,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
description: '',
parameters: {},
async execute() {
agent.steer({ content: [{ type: 'text', text: 's' }], source: { kind: 'plugin', plugin: 'goal' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 's' }], source: { kind: 'plugin', plugin: 'goal' } }))
return []
},
}))
@@ -516,13 +516,13 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
{ kind: 'plugin', plugin: 'goal' },
])
expect(queuedShapes).toEqual([
['content', 'id', 'source'],
['content', 'id', 'source'],
['content', 'id', 'role', 'source'],
['content', 'id', 'role', 'source'],
])
expect(placements).toEqual(['queued', 'steering'])
// The drain appends the durable steering/message with the caller's source
// intact — the log, not a transient emit, is where consumers read it.
const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.source] : [])
const steeringSources = agent.session.events.flatMap(e => e.type === 'steering/message' ? [e.data.message.source] : [])
expect(steeringSources).toEqual([{ kind: 'plugin', plugin: 'goal' }])
})
@@ -554,7 +554,7 @@ describe('turn numbering continues across seeded sessions', () => {
const turns: number[] = []
ctx2.on('session/event', (_s, event) => { if (event.type === 'turn/start') turns.push(event.data.turn) })
forked.followup({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } })
forked.followup(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } }))
await new Promise<void>((resolve) => {
ctx2.on('agent/status', (subject, status) => {
if (subject === forked && status === 'idle') resolve()
@@ -1105,7 +1105,7 @@ describe('tool result call identity', () => {
const resultEvent = [...agent.session.events].find(e => e.type === 'tool/result')
expect(resultEvent?.type).toBe('tool/result')
if (resultEvent?.type === 'tool/result') {
expect(resultEvent.data.callId).toBe(CallId('c1'))
expect(resultEvent.data.message.source.callId).toBe(CallId('c1'))
}
// And deriveMessages pairs the tool-result with the assistant tool-call:

6
packages/core/agent-loop/tests/coverage-edges.spec.ts
View File

@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId, LlmError, StreamChunk, errorChain } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, CallId, LlmError, StreamChunk, errorChain } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
@@ -38,7 +38,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
}
function send(agent: Agent, text: string) {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
describe('tool JSON parse', () => {
@@ -249,7 +249,7 @@ describe('structured tool error propagation (the runtime-validation Agent Note,
await waitForIdle(ctx, agent)
const toolResult = agent.session.events.find(e => e.type === 'tool/result')
expect(toolResult?.type === 'tool/result' && toolResult.data.isError).toBe(true)
expect(toolResult?.type === 'tool/result' && toolResult.data.message.content[0].isError).toBe(true)
expect(toolResult?.type === 'tool/result' && toolResult.data.error)
.toEqual({ name: 'HarnessError', code: 'BOOM' })
})

119
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -1,17 +1,16 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, CallId } from '@deepseek-ai/dsh-llm'
import SessionStore, {
SessionId,
type SessionEvent,
type TurnEndReason,
type UserMessageData,
type UserMessage,
} from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, {
type Agent,
type AgentMessage,
type InboxPlacement,
type PromptDecision,
type SessionStartSource,
@@ -53,7 +52,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
}
function send(agent: Agent, text: string) {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
function events(agent: Agent): SessionEvent[] {
@@ -67,8 +66,8 @@ describe('agent/prompt-submit', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const seen: string[] = []
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next) => {
seen.push(content.map(b => (b.type === 'text' ? b.text : '')).join(''))
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next) => {
seen.push(message.content.map(b => (b.type === 'text' ? b.text : '')).join(''))
return next()
})
@@ -80,13 +79,13 @@ describe('agent/prompt-submit', () => {
expect(userMsg?.type === 'user/message' && userMsg.data.content).toEqual([{ type: 'text', text: 'hello' }])
})
it('snapshots and freezes input before publishing or awaiting admission', async () => {
it('publishes frozen input without replacing its identity', async () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('owned-input'), { provider: 'mock', model: 'mock' })
const entered = Promise.withResolvers<undefined>()
const decision = Promise.withResolvers<PromptDecision>()
const observed: AgentMessage[] = []
const observed: UserMessage[] = []
ctx.on('agent/inbox/enqueue', (subject, message) => {
if (subject !== agent) return
expect(Object.isFrozen(message)).toBe(true)
@@ -105,30 +104,32 @@ describe('agent/prompt-submit', () => {
entered.resolve(undefined)
return decision.promise
})
const input: UserMessageData = {
const input: UserMessage = createUserMessage({
content: [{ type: 'text', text: 'accepted text' }],
source: { kind: 'plugin', plugin: 'accepted source' },
}
})
const idle = waitForIdle(ctx, agent)
agent.followup(input)
await entered.promise
const block = input.content[0]
if (block?.type === 'text') block.text = 'caller mutation'
if (input.source.kind === 'plugin') input.source.plugin = 'caller mutation'
expect(() => {
if (block?.type === 'text') block.text = 'caller mutation'
}).toThrow(TypeError)
expect(() => {
if (input.source.kind === 'plugin') input.source.plugin = 'caller mutation'
}).toThrow(TypeError)
decision.resolve({ kind: 'allow' })
await idle
expect(observed).toHaveLength(1)
expect(observed[0]).toBe(input)
expect(observed[0]).toMatchObject({
content: [{ type: 'text', text: 'accepted text' }],
source: { kind: 'plugin', plugin: 'accepted source' },
})
const userMsg = events(agent).find(event => event.type === 'user/message')
expect(userMsg?.type === 'user/message' && userMsg.data).toEqual({
content: [{ type: 'text', text: 'accepted text' }],
source: { kind: 'plugin', plugin: 'accepted source' },
})
expect(userMsg?.type === 'user/message' && userMsg.data).toEqual(input)
})
it('allow with content REWRITES the prompt before it is recorded', async () => {
@@ -157,10 +158,10 @@ describe('agent/prompt-submit', () => {
ctx.on('agent/prompt-submit', async (): Promise<PromptDecision> =>
({
kind: 'allow',
additionalContexts: [{
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: '<system-reminder>extra ctx</system-reminder>' }],
source: { kind: 'plugin', plugin: 'test' },
}],
})],
}))
send(agent, 'go')
@@ -185,7 +186,9 @@ describe('agent/prompt-submit', () => {
({
kind: 'allow',
content: [{ type: 'text', text: 'REWRITTEN prompt' }],
additionalContexts: [{ content: [{ type: 'text', text: 'injected ctx' }], source: { kind: 'plugin', plugin: 'test' } }],
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: 'injected ctx' }], source: { kind: 'plugin', plugin: 'test' },
})],
}))
let preStepDerived: string | undefined
@@ -213,7 +216,7 @@ describe('agent/prompt-submit', () => {
const reasons: TurnEndReason[] = []
ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
agent.followup({ content: [{ type: 'text', text: 'do something' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'do something' }], source: { kind: 'user' } }))
await agent.whenIdle()
// the model was never called
@@ -248,11 +251,11 @@ describe('agent/prompt-submit', () => {
expect(agent.acceptsNextStep).toBe(true)
expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
agent.inject({
agent.inject(createUserMessage({
content: [{ type: 'text', text: 'attached context' }],
source: { kind: 'plugin', plugin: 'test' },
})
agent.steer({ content: [{ type: 'text', text: 'admission steering' }], source: { kind: 'user' } })
}))
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'admission steering' }], source: { kind: 'user' } }))
expect(events(agent).some(event => event.type === 'user/message')).toBe(false)
expect(placements).toEqual(['queued', 'steering'])
@@ -272,7 +275,7 @@ describe('agent/prompt-submit', () => {
.toEqual([{ type: 'text', text: 'admitted prompt' }])
expect(staged[2]?.type === 'user/message' && staged[2].data.content)
.toEqual([{ type: 'text', text: 'attached context' }])
expect(staged[3]?.type === 'steering/message' && staged[3].data.content)
expect(staged[3]?.type === 'steering/message' && staged[3].data.message.content)
.toEqual([{ type: 'text', text: 'admission steering' }])
const request = JSON.stringify(adapter.requests[0]?.messages)
expect(request).toContain('admitted prompt')
@@ -295,11 +298,11 @@ describe('agent/prompt-submit', () => {
send(agent, 'blocked prompt')
await entered.promise
expect(agent.acceptsNextStep).toBe(true)
agent.inject({
agent.inject(createUserMessage({
content: [{ type: 'text', text: 'staged context' }],
source: { kind: 'plugin', plugin: 'test' },
})
agent.steer({ content: [{ type: 'text', text: 'staged steering' }], source: { kind: 'user' } })
}))
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'staged steering' }], source: { kind: 'user' } }))
decision.resolve({ kind: 'block', reason: 'policy' })
await blockedIdle
@@ -330,22 +333,22 @@ describe('agent/prompt-submit', () => {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next) => {
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next) => {
const decision = await next()
return content.some(block => block.type === 'text' && block.text === 'blocked prompt')
return message.content.some(block => block.type === 'text' && block.text === 'blocked prompt')
? { kind: 'block', reason: 'policy' }
: decision
})
ctx.on('agent/prompt-submit', async (subject, content, _source, _signal, next) => {
if (content.some(block => block.type === 'text' && block.text === 'blocked prompt')) {
subject.inject({
ctx.on('agent/prompt-submit', async (subject, message, _signal, next) => {
if (message.content.some(block => block.type === 'text' && block.text === 'blocked prompt')) {
subject.inject(createUserMessage({
content: [{ type: 'text', text: 'earlier state change' }],
source: { kind: 'plugin', plugin: 'test' },
})
subject.steer({
}))
subject.steer(createUserMessage({
content: [{ type: 'text', text: 'earlier steering' }],
source: { kind: 'user' },
})
}))
}
return next()
})
@@ -365,7 +368,7 @@ describe('agent/prompt-submit', () => {
])
expect(staged[1]?.type === 'user/message' && staged[1].data.content)
.toEqual([{ type: 'text', text: 'earlier state change' }])
expect(staged[2]?.type === 'steering/message' && staged[2].data.content)
expect(staged[2]?.type === 'steering/message' && staged[2].data.message.content)
.toEqual([{ type: 'text', text: 'earlier steering' }])
expect(staged[3]?.type === 'user/message' && staged[3].data.content)
.toEqual([{ type: 'text', text: 'later prompt' }])
@@ -385,10 +388,10 @@ describe('agent/prompt-submit', () => {
const idle = waitForIdle(ctx, agent)
send(agent, 'blocked prompt')
await entered.promise
agent.inject({
agent.inject(createUserMessage({
content: [{ type: 'text', text: 'independent context' }],
source: { kind: 'plugin', plugin: 'test' },
})
}))
decision.resolve({ kind: 'block', reason: 'policy' })
await idle
@@ -417,12 +420,12 @@ describe('agent/prompt-submit', () => {
return decision.promise
})
agent.followup({ content: [{ type: 'text', text: 'blocked prompt' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'blocked prompt' }], source: { kind: 'user' } }))
await entered.promise
agent.inject({
agent.inject(createUserMessage({
content: [{ type: 'text', text: 'retained context' }],
source: { kind: 'plugin', plugin: 'test' },
})
}))
decision.resolve({ kind: 'block', reason: 'policy' })
await agent.whenIdle()
@@ -442,8 +445,8 @@ describe('agent/prompt-submit', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next): Promise<PromptDecision> => {
const text = content.map(b => (b.type === 'text' ? b.text : '')).join('')
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next): Promise<PromptDecision> => {
const text = message.content.map(b => (b.type === 'text' ? b.text : '')).join('')
return text === 'secret' ? { kind: 'block', reason: 'policy: no secrets' } : next()
})
@@ -525,7 +528,7 @@ describe('agent/session-start', () => {
const ctx = await harness(adapter)
ctx.on('agent/session-start', (agent) => {
agent.inject({ content: [{ type: 'text', text: 'session preamble' }], source: { kind: 'plugin', plugin: 'test' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'session preamble' }], source: { kind: 'plugin', plugin: 'test' } }))
})
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -579,10 +582,10 @@ describe('tool additionalContexts buffering across a step', () => {
ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> =>
({
kind: 'accept',
additionalContexts: [{
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: `ctx-${exec.callId}` }],
source: { kind: 'plugin', plugin: 'p' },
}],
})],
}))
send(agent, 'go')
@@ -611,8 +614,12 @@ describe('tool additionalContexts buffering across a step', () => {
ctx.tools.register(defineContentToolFixture({
name: 'composite', description: 'composite', parameters: {},
async execute(_args, exec) {
exec.deferContext({ content: [{ type: 'text', text: 'nested-a' }], source: { kind: 'plugin', plugin: 'a' } })
exec.deferContext({ content: [{ type: 'text', text: 'nested-b' }], source: { kind: 'plugin', plugin: 'b' } })
exec.deferContext(createUserMessage({
content: [{ type: 'text', text: 'nested-a' }], source: { kind: 'plugin', plugin: 'a' },
}))
exec.deferContext(createUserMessage({
content: [{ type: 'text', text: 'nested-b' }], source: { kind: 'plugin', plugin: 'b' },
}))
return [{ type: 'text', text: 'outer result' }]
},
}))
@@ -654,9 +661,9 @@ describe('tools/pre-execute gate (native-plugin permission pattern, end-to-end t
expect(ran).toBe(false)
const result = events(agent).find(e => e.type === 'tool/result')
expect(result?.type === 'tool/result' && result.data.isError).toBe(true)
expect(result?.type === 'tool/result' && result.data.message.content[0].isError).toBe(true)
expect(result?.type === 'tool/result'
&& result.data.content.some(b => b.type === 'text' && b.text.includes('blocked dangerous tool'))).toBe(true)
&& result.data.message.content[0].content.some(b => b.type === 'text' && b.text.includes('blocked dangerous tool'))).toBe(true)
})
})
@@ -669,11 +676,11 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
apply(ctx: Context) {
// 1. SessionStart: seed a standing instruction.
ctx.on('agent/session-start', (agent, source) => {
agent.inject({ content: [{ type: 'text', text: `policy active (started: ${source})` }], source: { kind: 'plugin', plugin: 'native-guard' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: `policy active (started: ${source})` }], source: { kind: 'plugin', plugin: 'native-guard' } }))
})
// 2. PromptSubmit: block a forbidden prompt, annotate the rest.
ctx.on('agent/prompt-submit', async (_agent, content, _source, _signal, next): Promise<PromptDecision> => {
const text = content.map(b => (b.type === 'text' ? b.text : '')).join('')
ctx.on('agent/prompt-submit', async (_agent, message, _signal, next): Promise<PromptDecision> => {
const text = message.content.map(b => (b.type === 'text' ? b.text : '')).join('')
if (text.includes('rm -rf')) return { kind: 'block', reason: 'destructive prompt blocked' }
return next()
})
@@ -686,7 +693,9 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
ctx.on('tools/post-execute', async (_exec, _result, next): Promise<PostToolDecision> => {
const decision = await next()
if (decision.kind === 'accept') {
return { kind: 'accept', additionalContexts: [{ content: [{ type: 'text', text: 'audited' }], source: { kind: 'plugin', plugin: 'native-guard' } }] }
return { kind: 'accept', additionalContexts: [createUserMessage({
content: [{ type: 'text', text: 'audited' }], source: { kind: 'plugin', plugin: 'native-guard' },
})] }
}
return decision
})
@@ -713,7 +722,7 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
// prompt allowed → user-sourced user/message recorded
expect(log.some(e => e.type === 'user/message' && e.data.source.kind === 'user')).toBe(true)
// tool ran (echo allowed) and post-execute attached "audited" context
expect(log.some(e => e.type === 'tool/result' && !e.data.isError)).toBe(true)
expect(log.some(e => e.type === 'tool/result' && !e.data.message.content[0].isError)).toBe(true)
expect(log.some(e => e.type === 'user/message' && e.data.source.kind === 'plugin'
&& e.data.content.some(b => b.type === 'text' && b.text === 'audited'))).toBe(true)
// NO hook/* events — a native plugin needs none

14
packages/core/agent-loop/tests/invariant.spec.ts
View File

@@ -3,7 +3,7 @@ import { Context } from 'cordis'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import InvariantService from '@deepseek-ai/dsh-invariants'
import * as AgentLoopInvariant from '@deepseek-ai/dsh-agent-loop/invariant'
import { markAgentLoopRequest, type GenerateOptions } from '@deepseek-ai/dsh-llm'
import { createUserMessage, markAgentLoopRequest, type GenerateOptions } from '@deepseek-ai/dsh-llm'
async function setup(): Promise<Context> {
const ctx = new Context()
@@ -26,7 +26,9 @@ async function requestSetup() {
const ctx = await setup()
const session = ctx.sessions.create(SessionId('req-check'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
const boundary = session.deriveMessages()
session.append('step/start', { turn: 1, step: 1 })
session.append('request/header', { header: { config: { provider: 'mock', model: 'm' } }, reason: 'initial' })
@@ -42,7 +44,9 @@ describe('request-reconstruction invariant', () => {
it('uses the step boundary rather than content appended afterward', async () => {
const { ctx, session, boundary } = await requestSetup()
session.append('user/message', { content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' } }, { surfaceOp: 'append' })
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' },
}), { surfaceOp: 'append' })
const options = loopRequest({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
expect(() => { dispatch(ctx, options) }).not.toThrow()
})
@@ -119,7 +123,9 @@ describe('request-reconstruction invariant', () => {
await ctx.plugin(AgentLoopInvariant)
const session = ctx.sessions.create(SessionId('prepend-check'))
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' },
}), { surfaceOp: 'append' })
session.append('step/start', { turn: 1, step: 1 })
session.append('request/header', { header: { config: { provider: 'mock', model: 'm' } }, reason: 'initial' })
const divergent = loopRequest({

124
packages/core/agent-loop/tests/loop.spec.ts
View File

@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
@@ -42,7 +42,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
}
function send(agent: Agent, text: string) {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
describe('agent loop', () => {
@@ -284,7 +284,7 @@ describe('agent loop', () => {
parameters: {},
async execute() {
// steer while the turn is running (during tool execution)
agent.steer({ content: [{ type: 'text', text: 'change of plans' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'change of plans' }], source: { kind: 'user' } }))
return [{ type: 'text', text: 'tool done' }]
},
}))
@@ -312,8 +312,8 @@ describe('agent loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const idle = waitForIdle(ctx, agent)
agent.steer({ content: [{ type: 'text', text: 'first idle steer' }], source: { kind: 'user' } })
agent.steer({ content: [{ type: 'text', text: 'second idle steer' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'first idle steer' }], source: { kind: 'user' } }))
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'second idle steer' }], source: { kind: 'user' } }))
await idle
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
@@ -338,7 +338,7 @@ describe('agent loop', () => {
ctx.on('agent/step', (subject) => {
if (subject !== agent || !fail) return
fail = false
subject.steer({ content: [{ type: 'text', text: 'pending steering' }], source: { kind: 'user' } })
subject.steer(createUserMessage({ content: [{ type: 'text', text: 'pending steering' }], source: { kind: 'user' } }))
throw new Error('step failed')
})
@@ -363,13 +363,17 @@ describe('agent loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.inject({ content: [{ type: 'text', text: 'file changed: a.ts' }], source: { kind: 'plugin', plugin: 'watcher' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'file changed: a.ts' }], source: { kind: 'plugin', plugin: 'watcher' } }))
expect(agent.status).toBe('idle')
expect(adapter.requests).toHaveLength(0)
expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(0)
expect(agent.session.events.at(-1)).toMatchObject({
type: 'user/message',
data: { source: { kind: 'plugin', plugin: 'watcher' } },
data: {
role: 'user',
content: [{ type: 'text', text: 'file changed: a.ts' }],
source: { kind: 'plugin', plugin: 'watcher' },
},
})
send(agent, 'go')
@@ -385,7 +389,7 @@ describe('agent loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('raw-context'), { provider: 'mock', model: 'mock' })
const text = '<system-reminder>Additional instructions from: pkg/AGENTS.md</system-reminder>'
agent.inject({ content: [{ type: 'text', text }], source: { kind: 'plugin', plugin: 'workspace-context' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'plugin', plugin: 'workspace-context' } }))
send(agent, 'go')
await waitForIdle(ctx, agent)
@@ -412,9 +416,9 @@ describe('agent loop', () => {
async execute() {
await Promise.resolve()
const first = { type: 'text' as const, text: 'mid-turn notice' }
agent.inject({ content: [first], source: { kind: 'plugin', plugin: 'x' } })
agent.inject(createUserMessage({ content: [first], source: { kind: 'plugin', plugin: 'x' } }))
first.text = 'mutated after inject'
agent.inject({ content: [{ type: 'text', text: 'second notice' }], source: { kind: 'plugin', plugin: 'x' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'second notice' }], source: { kind: 'plugin', plugin: 'x' } }))
visibleDuringTool = agent.session.events.some(e => e.type === 'user/message' && e.data.source.kind === 'plugin')
return [{ type: 'text', text: 'ok' }]
},
@@ -467,7 +471,7 @@ describe('agent loop', () => {
parameters: {},
async execute() {
expect(() => {
agent.inject({ content: [{ type: 'text', text: 'invalid' }], source: { kind: 'plugin', plugin: 'test', bigint: 1n } as never })
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'invalid' }], source: { kind: 'plugin', plugin: 'test', bigint: 1n } as never }))
}).toThrow('agent context must be losslessly JSON-serializable')
return [{ type: 'text', text: 'rejected invalid context' }]
},
@@ -492,7 +496,7 @@ describe('agent loop', () => {
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
ctx.on('agent/turn-stopping', (subject) => {
if (steps < 3) {
subject.steer({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'plugin', plugin: 'loop-test' } })
subject.steer(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'plugin', plugin: 'loop-test' } }))
}
})
@@ -537,7 +541,7 @@ describe('agent loop', () => {
parameters: {},
async execute(_args, exec) {
// Steering lands while the concluding tool is still executing.
agent.steer({ content: [{ type: 'text', text: 'late steering' }], source: { kind: 'user' } })
agent.steer(createUserMessage({ content: [{ type: 'text', text: 'late steering' }], source: { kind: 'user' } }))
exec.concludeTurn()
return [{ type: 'text', text: 'final' }]
},
@@ -625,10 +629,10 @@ describe('agent loop', () => {
ctx.on('agent/step', (subject) => {
if (subject === agent && !injected) {
injected = true
subject.session.append('user/message', {
subject.session.append('user/message', createUserMessage({
content: [{ type: 'text', text: 'INJECTED-IN-PRE-STEP' }],
source: { kind: 'plugin', plugin: 'test' },
}, { surfaceOp: 'append' })
}), { surfaceOp: 'append' })
}
})
@@ -740,7 +744,7 @@ describe('agent loop', () => {
// (step 2 is a plain stop with no tool calls → stops).
ctx.on('agent/turn-stopping', (subject) => {
if (steps < 2) {
subject.steer({ content: [{ type: 'text', text: 'continue after truncation' }], source: { kind: 'plugin', plugin: 'max-tokens-test' } })
subject.steer(createUserMessage({ content: [{ type: 'text', text: 'continue after truncation' }], source: { kind: 'plugin', plugin: 'max-tokens-test' } }))
}
})
@@ -753,9 +757,24 @@ describe('agent loop', () => {
expect(steps).toBe(2)
expect(adapter.requests).toHaveLength(2)
expect(adapter.requests[1]!.messages).toEqual([
{ role: 'user', content: [{ type: 'text', text: 'go' }] },
{ role: 'assistant', content: [{ type: 'text', text: 'first half' }], provenance: { provider: 'mock', model: 'mock' } },
{ role: 'user', content: [{ type: 'text', text: 'continue after truncation' }] },
{
id: expect.any(String) as unknown,
role: 'user',
content: [{ type: 'text', text: 'go' }],
source: { kind: 'user' },
},
{
id: expect.any(String) as unknown,
role: 'assistant',
content: [{ type: 'text', text: 'first half' }],
source: { kind: 'model', provider: 'mock', model: 'mock' },
},
{
id: expect.any(String) as unknown,
role: 'user',
content: [{ type: 'text', text: 'continue after truncation' }],
source: { kind: 'plugin', plugin: 'max-tokens-test' },
},
])
expect(reasons).toEqual([{ kind: 'max-tokens' }])
})
@@ -808,13 +827,26 @@ describe('agent loop', () => {
expect(executions).toBe(0)
expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false)
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
expect(agent.session.deriveMessages()).toEqual([{
id: expect.any(String) as unknown,
role: 'user',
content: [{ type: 'text', text: 'go' }],
source: { kind: 'user' },
}])
expect(reasons).toEqual([{ kind: 'max-tokens' }])
// Empty content still needs an assistant/message to carry usage; derivation
// skips that host so it does not create a spurious assistant turn.
const assistantMessage = agent.session.events.find(e => e.type === 'assistant/message')
expect(assistantMessage?.type === 'assistant/message' && assistantMessage.data).toEqual({
turn: 1, step: 1, content: [], provenance: { provider: 'mock', model: 'mock' }, usage: { inputTokens: 10, outputTokens: 5 },
turn: 1,
step: 1,
message: {
id: expect.any(String) as unknown,
role: 'assistant',
content: [],
source: { kind: 'model', provider: 'mock', model: 'mock' },
},
usage: { inputTokens: 10, outputTokens: 5 },
})
})
@@ -848,11 +880,20 @@ describe('agent loop', () => {
expect(assistant.type === 'assistant/message' && assistant.data).toEqual({
turn: 1,
step: 1,
content: [],
provenance: { provider: 'mock', model: 'mock' },
message: {
id: expect.any(String) as unknown,
role: 'assistant',
content: [],
source: { kind: 'model', provider: 'mock', model: 'mock' },
},
})
expect(assistant.sourceEventSeqs?.length).toBeGreaterThan(0)
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
expect(agent.session.deriveMessages()).toEqual([{
id: expect.any(String) as unknown,
role: 'user',
content: [{ type: 'text', text: 'go' }],
source: { kind: 'user' },
}])
})
it('appends an empty completion anchor for a normal stop with no usage', async () => {
@@ -873,11 +914,20 @@ describe('agent loop', () => {
expect(assistant.type === 'assistant/message' && assistant.data).toEqual({
turn: 1,
step: 1,
content: [],
provenance: { provider: 'mock', model: 'mock' },
message: {
id: expect.any(String) as unknown,
role: 'assistant',
content: [],
source: { kind: 'model', provider: 'mock', model: 'mock' },
},
})
expect(assistant.sourceEventSeqs?.length).toBe(1)
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
expect(agent.session.deriveMessages()).toEqual([{
id: expect.any(String) as unknown,
role: 'user',
content: [{ type: 'text', text: 'go' }],
source: { kind: 'user' },
}])
})
it('keeps safe max-tokens assistant content while dropping truncated tool calls', async () => {
@@ -898,8 +948,18 @@ describe('agent loop', () => {
expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false)
expect(agent.session.deriveMessages()).toEqual([
{ role: 'user', content: [{ type: 'text', text: 'go' }] },
{ role: 'assistant', content: [{ type: 'text', text: 'partial text' }], provenance: { provider: 'mock', model: 'mock' } },
{
id: expect.any(String) as unknown,
role: 'user',
content: [{ type: 'text', text: 'go' }],
source: { kind: 'user' },
},
{
id: expect.any(String) as unknown,
role: 'assistant',
content: [{ type: 'text', text: 'partial text' }],
source: { kind: 'model', provider: 'mock', model: 'mock' },
},
])
})
@@ -966,9 +1026,9 @@ describe('agent loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const idle = waitForIdle(ctx, agent)
agent.followup({ content: [{ type: 'text', text: 'user message' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'user message' }], source: { kind: 'user' } }))
await Promise.resolve()
agent.followup({ content: [{ type: 'text', text: 'plugin message' }], source: { kind: 'plugin', plugin: 'test' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'plugin message' }], source: { kind: 'plugin', plugin: 'test' } }))
await idle
const triggers = agent.session.events

8
packages/core/agent-loop/tests/properties.spec.ts
View File

@@ -12,8 +12,8 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService from '@deepseek-ai/dsh-llm'
import { createUserMessage, LlmAdapter } from '@deepseek-ai/dsh-llm'
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
import { LlmAdapter } 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'
@@ -115,7 +115,7 @@ describe('agent loop scheduling properties', () => {
const { seen: trace } = recordStatus(ctx, agent)
const idle = nextIdle(ctx, agent)
// Send all in one synchronous tick: they queue before the loop wakes.
for (const text of texts) agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
for (const text of texts) agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
await idle
// No message lost: every send appears as a user/message, in order.
@@ -142,7 +142,7 @@ describe('agent loop scheduling properties', () => {
const agent = ctx.agentLoop.create(SessionId('a'), { provider: 'mock', model: 'mock' })
for (const text of texts) {
const idle = nextIdle(ctx, agent)
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
await idle
}
// Each send was drained at a separate turn start: N turns, 1..N.
@@ -171,7 +171,7 @@ describe('agent loop scheduling properties', () => {
for (const step of steps) {
const idle = nextIdle(ctx, agent)
lastIdle = idle
agent.followup({ content: [{ type: 'text', text: step.text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: step.text }], source: { kind: 'user' } }))
if (step.settle) await idle
}
await lastIdle

5
packages/core/agent-loop/tests/request-cache.e2e.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService from '@deepseek-ai/dsh-llm'
@@ -73,10 +74,10 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('log-derived request cache hits (
const agent = ctx.agentLoop.create(SessionId('cache-e2e'), { provider: 'deepseek', model: 'deepseek-v4-flash' })
// Turn 1: forces a tool call → at least two steps (two model requests).
agent.followup({ content: [{ type: 'text', text: 'Look up the key "deploy-color" with the lookup tool and tell me the value.' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'Look up the key "deploy-color" with the lookup tool and tell me the value.' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
// Turn 2: a follow-up over the same (longer) prefix.
agent.followup({ content: [{ type: 'text', text: 'Thanks. Repeat that value one more time.' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'Thanks. Repeat that value one more time.' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
const usages = [...agent.session.events]

10
packages/core/agent-loop/tests/request-error.spec.ts
View File

@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import LlmService, { LlmError } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, LlmError } from '@deepseek-ai/dsh-llm'
import type { LlmFailure, ResolvedRetryPolicy } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
@@ -40,7 +40,7 @@ describe('agent/request-error', () => {
recoveries += 1
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await agent.whenIdle()
expect(recoveries).toBe(0)
@@ -82,7 +82,7 @@ describe('agent/request-error', () => {
return { kind: 'retry' }
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await agent.whenIdle()
expect(seen.map(item => ({
@@ -126,7 +126,7 @@ describe('agent/request-error', () => {
return { kind: 'retry' }
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)
@@ -148,7 +148,7 @@ describe('agent/request-error', () => {
throw new Error('recovery failed')
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await agent.whenIdle()
expect(adapter.requests).toHaveLength(1)

15
packages/core/agent-loop/tests/request-reconstruction.spec.ts
View File

@@ -7,7 +7,7 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import LlmService, { LlmAdapter, LlmError, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import LlmService, { createUserMessage, LlmAdapter, LlmError, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
import type {
GenerateOptions, LlmModelReasoningInfo, LlmResolvedModelInfo, StreamChunk,
} from '@deepseek-ai/dsh-llm'
@@ -43,7 +43,7 @@ function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
}
function send(agent: Agent, text: string) {
agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text }], source: { kind: 'user' } }))
}
/** Assert `previous` is a strict value-prefix of `current`. */
@@ -460,10 +460,10 @@ describe('request stability across the loop', () => {
preStep()
const session = agent.session
const nodes = session.surface.nodes
session.append('user/message', {
session.append('user/message', createUserMessage({
content: [{ type: 'text', text: '[summary of turn 1]' }],
source: { kind: 'plugin', plugin: 'test-compact' },
}, {
}), {
surfaceOp: { op: 'replace', start: nodes[0]!, end: nodes[1]! },
sourceEventSeqs: [nodes[0]!, nodes[1]!],
})
@@ -512,7 +512,7 @@ describe('request stability across the loop', () => {
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
if (!injected) {
injected = true
agent.inject({ content: [{ type: 'text', text: '[late context]' }], source: { kind: 'plugin', plugin: 'test' } })
agent.inject(createUserMessage({ content: [{ type: 'text', text: '[late context]' }], source: { kind: 'plugin', plugin: 'test' } }))
}
return next()
})
@@ -543,7 +543,10 @@ describe('request stability across the loop', () => {
ctx.on('llm/stream', (options, next) => {
// The historical failure mode this design kills: a listener rewriting
// request content in place. The freeze turns it into a loud error.
options.messages.push({ role: 'user', content: [{ type: 'text', text: 'sneaky' }] })
options.messages.push(createUserMessage({
content: [{ type: 'text', text: 'sneaky' }],
source: { kind: 'plugin', plugin: 'test' },
}))
return next()
})

13
packages/core/agent-loop/tests/resume.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { mkdtemp, rm } from 'node:fs/promises'
@@ -146,7 +147,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
const adapter1 = new MockAdapter([textResponse('a')])
const { ctx: ctx1, root } = await persistentHarness(adapter1)
const a1 = (await ctx1.agents.create({ sessionId: SessionId('nocwd-sess') })).agent
a1.followup({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }))
await waitForIdle(ctx1, a1)
await ctx1.fiber.dispose()
@@ -174,7 +175,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
ctx1.on('agent/session-start', (_agent, source) => void sources1.push(source))
const a1 = (await ctx1.agents.create({ sessionId: SessionId('start-sess') })).agent
expect(sources1).toEqual(['startup'])
a1.followup({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }))
await waitForIdle(ctx1, a1)
await ctx1.fiber.dispose()
@@ -475,9 +476,9 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
const adapter1 = new MockAdapter([textResponse('answer')])
const { ctx: ctx1, root } = await persistentHarness(adapter1)
const a1 = (await ctx1.agents.create({ sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent
a1.followup({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } })
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }))
await waitForIdle(ctx1, a1)
a1.inject({ content: [{ type: 'text', text: 'background task 42 finished' }], source: { kind: 'plugin', plugin: 'tool-bash' } })
a1.inject(createUserMessage({ content: [{ type: 'text', text: 'background task 42 finished' }], source: { kind: 'plugin', plugin: 'tool-bash' } }))
await a1.whenIdle()
await ctx1.fiber.dispose()
@@ -503,7 +504,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
const adapter1 = new MockAdapter([textResponse('first answer')])
const { ctx: ctx1, root } = await persistentHarness(adapter1)
const a1 = (await ctx1.agents.create({ sessionId: SessionId('sess-resume'), meta: { cwd: '/w' } })).agent
a1.followup({ content: [{ type: 'text', text: 'first question' }], source: { kind: 'user' } })
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'first question' }], source: { kind: 'user' } }))
await waitForIdle(ctx1, a1)
const events1 = [...a1.session.events]
const seqs1 = events1.map(e => e.seq)
@@ -530,7 +531,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
expect(a2.session.deriveMessages()).toEqual(replay.deriveMessages())
// …and a new turn continues numbering (turn 2) with contiguous seqs.
a2.followup({ content: [{ type: 'text', text: 'second question' }], source: { kind: 'user' } })
a2.followup(createUserMessage({ content: [{ type: 'text', text: 'second question' }], source: { kind: 'user' } }))
await waitForIdle(ctx2, a2)
const allSeqs = a2.session.events.map(e => e.seq)
expect(allSeqs).toEqual(allSeqs.map((_, i) => i)) // 0..N contiguous, no duplicates

11
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { describe, expect, it } from 'vitest'
import { Context, symbols, type EffectMeta, type Fiber } from 'cordis'
import LlmService from '@deepseek-ai/dsh-llm'
@@ -203,11 +204,11 @@ describe('agent scope lifecycle', () => {
if (event.type === 'user/message') heard.push('a-sees:user-message')
})
b.followup({ content: text('for b'), source: { kind: 'user' } })
b.followup(createUserMessage({ content: text('for b'), source: { kind: 'user' } }))
await waitForIdle(ctx, b)
expect(heard).toEqual([]) // nothing of b's leaked into a's scope
a.followup({ content: text('for a'), source: { kind: 'user' } })
a.followup(createUserMessage({ content: text('for a'), source: { kind: 'user' } }))
await waitForIdle(ctx, a)
expect(heard).toContain('a-sees:a:running')
expect(heard).toContain('a-sees:user-message')
@@ -934,7 +935,7 @@ describe('agent scope lifecycle', () => {
if (event.type === 'turn/start') { off(); resolve() }
})
})
agent.followup({ content: text('work'), source: { kind: 'user' } })
agent.followup(createUserMessage({ content: text('work'), source: { kind: 'user' } }))
await turnOpen
await owner.dispose()
expect(order).toEqual([
@@ -1058,10 +1059,10 @@ describe('agent scope lifecycle', () => {
ctx.on('agent/status', (subject, status) => {
if (subject !== agent || status !== 'idle' || reentered) return
reentered = true
agent.followup({ content: [{ type: 'text', text: 'reentrant' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'reentrant' }], source: { kind: 'user' } }))
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
expect(reentered).toBe(true)

94
packages/core/agent-loop/tests/tool-calls.spec.ts
View File

@@ -5,7 +5,7 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
import { createUserMessage, CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import LlmService from '@deepseek-ai/dsh-llm'
@@ -105,7 +105,7 @@ describe('tool-call scheduler: grouping and barriers', () => {
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 3)
expect(gated.started).toEqual(['1', '2', '3'])
gated.release('1'); gated.release('2'); gated.release('3')
@@ -133,7 +133,7 @@ describe('tool-call scheduler: grouping and barriers', () => {
async execute(args) { order.push(`w-${args.id}`); return [{ type: 'text', text: 'w' }] },
}))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
expect(order).toEqual(['r-start-A1', 'r-end-A1', 'w-A2', 'r-start-A3', 'r-end-A3'])
@@ -169,7 +169,7 @@ describe('tool-call scheduler: grouping and barriers', () => {
}))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => replacement.started.length === 1)
await new Promise(r => setTimeout(r, 5))
expect(replacement.started).toEqual(['1'])
@@ -200,11 +200,11 @@ describe('tool-call scheduler: grouping and barriers', () => {
})
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => initial.started.length === 2)
initial.release('1')
await until(() => events(agent).some(event =>
event.type === 'tool/result' && event.data.callId === CallId('c1')))
event.type === 'tool/result' && event.data.message.source.callId === CallId('c1')))
await new Promise(r => setTimeout(r, 5))
expect(replacement.started).toEqual([])
initial.release('2')
@@ -226,7 +226,7 @@ describe('tool-call scheduler: model-order results despite out-of-order settleme
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 2)
gated.release('2')
await new Promise(r => setTimeout(r, 5))
@@ -236,7 +236,7 @@ describe('tool-call scheduler: model-order results despite out-of-order settleme
await waitForIdle(ctx, agent)
const results = events(agent).filter(e => e.type === 'tool/result')
expect(results.map(e => e.data.callId)).toEqual([CallId('c1'), CallId('c2')])
expect(results.map(e => e.data.message.source.callId)).toEqual([CallId('c1'), CallId('c2')])
})
it('derived history pairs calls in model order regardless of tool/call log interleaving', async () => {
@@ -248,7 +248,7 @@ describe('tool-call scheduler: model-order results despite out-of-order settleme
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 2)
gated.release('2'); gated.release('1')
await waitForIdle(ctx, agent)
@@ -295,7 +295,7 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 2)
await new Promise(r => setTimeout(r, 5))
expect(gated.started).toEqual(['1', '2'])
@@ -304,14 +304,16 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
expect(gated.started).toEqual(['1', '2', '3'])
expect(events(agent)
.filter(e => e.type === 'tool/call' || e.type === 'tool/result')
.map(e => `${e.type}:${String(e.data.callId)}`)
.map(e => e.type === 'tool/call'
? `${e.type}:${String(e.data.callId)}`
: `${e.type}:${String(e.data.message.source.callId)}`)
.slice(0, 4))
.toEqual(['tool/call:c1', 'tool/call:c2', 'tool/result:c1', 'tool/call:c3'])
gated.release('2'); gated.release('3')
await until(() => gated.started.length === 4)
gated.release('4')
await waitForIdle(ctx, agent)
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.message.source.callId))
.toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
})
@@ -324,7 +326,7 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 1)
await new Promise(r => setTimeout(r, 5))
expect(gated.started).toEqual(['1'])
@@ -350,7 +352,7 @@ describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () =>
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 1)
await new Promise(r => setTimeout(r, 5))
expect(gated.started).toEqual(['1'])
@@ -377,7 +379,7 @@ describe('tool-call scheduler: ordered middleware and additional contexts', () =
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => { post.push(String(exec.callId)); return next() })
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 3)
gated.release('3'); gated.release('2'); gated.release('1')
await waitForIdle(ctx, agent)
@@ -395,10 +397,12 @@ describe('tool-call scheduler: ordered middleware and additional contexts', () =
const gated = gatedParallelTool('p')
ctx.tools.register(gated.tool)
ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> =>
({ kind: 'accept', additionalContexts: [{ content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } }] }))
({ kind: 'accept', additionalContexts: [createUserMessage({
content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' },
})] }))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 2)
gated.release('2'); gated.release('1')
await waitForIdle(ctx, agent)
@@ -436,7 +440,7 @@ describe('tool-call scheduler: ordered middleware and additional contexts', () =
})
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 1)
gated.release('1')
await waitForIdle(ctx, agent)
@@ -444,9 +448,9 @@ describe('tool-call scheduler: ordered middleware and additional contexts', () =
expect(gated.started).toEqual(['1'])
expect(post).toEqual(['c1', 'c2'])
const results = events(agent).filter(e => e.type === 'tool/result')
expect(results.map(e => e.data.callId)).toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
expect((results[1]!.data.content[0] as { text: string }).text).toContain('blocked by policy')
expect((results[2]!.data.content[0] as { text: string }).text).toContain('pre exploded')
expect(results.map(e => e.data.message.source.callId)).toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
expect((results[1]!.data.message.content[0].content[0] as { text: string }).text).toContain('blocked by policy')
expect((results[2]!.data.message.content[0].content[0] as { text: string }).text).toContain('pre exploded')
})
})
@@ -466,15 +470,15 @@ describe('tool-call scheduler: abort handling', () => {
}
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
expect(gated.started).toEqual([])
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2')])
expect(events(agent).filter(e => e.type === 'tool/result').map(e => ({
callId: e.data.callId,
isError: e.data.isError,
callId: e.data.message.source.callId,
isError: e.data.message.content[0].isError,
error: e.data.error,
}))).toEqual([
{ callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
@@ -498,15 +502,15 @@ describe('tool-call scheduler: abort handling', () => {
return next()
})
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
expect(gated.started).toEqual([])
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2')])
expect(events(agent).filter(e => e.type === 'tool/result').map(e => ({
callId: e.data.callId,
isError: e.data.isError,
callId: e.data.message.source.callId,
isError: e.data.message.content[0].isError,
error: e.data.error,
}))).toEqual([
{ callId: CallId('c1'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } },
@@ -524,11 +528,13 @@ describe('tool-call scheduler: abort handling', () => {
ctx.tools.register(gated.tool)
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => ({
...await next(),
additionalContexts: [{ content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } }],
additionalContexts: [createUserMessage({
content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' },
})],
}))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 2)
agent.cancel({ kind: 'user' })
gated.release('1')
@@ -538,12 +544,24 @@ describe('tool-call scheduler: abort handling', () => {
expect(gated.started).toEqual(['1', '2'])
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.message.source.callId))
.toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
expect(events(agent).filter(e => e.type === 'tool/result').slice(-2).map(e => e.data))
expect(events(agent).filter(e => e.type === 'tool/result').slice(-2).map(e => ({
callId: e.data.message.source.callId,
isError: e.data.message.content[0].isError,
error: e.data.error,
})))
.toEqual([
expect.objectContaining({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } }),
expect.objectContaining({ callId: CallId('c4'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } }),
{
callId: CallId('c3'),
isError: true,
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
},
{
callId: CallId('c4'),
isError: true,
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
},
])
const settled = events(agent).filter(e => e.type === 'tool/result'
|| (e.type === 'user/message' && e.data.source.kind === 'plugin'))
@@ -575,7 +593,7 @@ describe('tool-call scheduler: abort handling', () => {
}))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await until(() => gated.started.length === 2)
agent.cancel({ kind: 'user' })
gated.release('1')
@@ -586,6 +604,12 @@ describe('tool-call scheduler: abort handling', () => {
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
expect(events(agent).filter(e => e.type === 'tool/result').at(-1)?.data)
.toMatchObject({ callId: CallId('c3'), isError: true, error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH } })
.toMatchObject({
message: {
source: { kind: 'tool', callId: CallId('c3') },
content: [{ isError: true }],
},
error: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
})
})
})

5
packages/core/agent-loop/tests/tool-order.spec.ts
View File

@@ -1,3 +1,4 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
/**
* Loop-level tool-order determinism: the request/header event — and therefore the frozen
* request the adapter receives — carries the assembly's canonical tool order (system-prompt's
@@ -58,7 +59,7 @@ async function runTurn(registrationOrder: string[], toolOrder?: SystemPromptConf
const ctx = await harness(adapter, toolOrder)
for (const name of registrationOrder) registerNamed(ctx, name)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
return { ctx, agent, adapter }
}
@@ -102,7 +103,7 @@ describe('loop-level canonical tool order', () => {
if (error instanceof Error) errors.push(error)
})
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }))
await waitForIdle(ctx, agent)
expect(adapter.requests).toHaveLength(0)
expect(errors.map(e => e.message)).toEqual(['toolOrder lists unregistered tool "ghost"; known tools: alpha'])