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

Merge latest origin/master into parallel-tool-call

Browse Source 
# Conflicts:
#	docs/architecture.md
#	examples/acp-agent/tests/snapshots/bash-spill/session.jsonl
#	examples/acp-agent/tests/snapshots/escalation-approved/session.jsonl
#	examples/acp-agent/tests/snapshots/escalation-rejected/session.jsonl
#	examples/acp-agent/tests/snapshots/hook-cc-pretool-ask/session.jsonl
#	packages/core/agent-loop/src/loop.ts
This commit is contained in:
Tianyi Cui
2026-07-18 15:09:39 +08:00
parent 21ec178841 7a29676c01
commit 50f0291b93
198 changed files with 6086 additions and 4036 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -46,15 +46,17 @@ Configured agents start automatically. A model call requires both `provider` and
- `ReactLoopAgent` — the concrete `Agent` implementation. Its inbox is a JavaScript native-private field, and one prepared session can be claimed by only one concrete driver. Everything observable happens through session events and the `agent/*` event taxonomy.
`Inbox`, `runLoop`, and the instance-bound publication/start controls are package-internal. The package root does not export them, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than constructing or starting the driver internals. `ReactLoopAgent.send()` and running `steer()` materialize content plus resolved source once as detached, deeply frozen lossless JSON, then share that accepted record between `agent/queued` and the inbox; malformed data throws before either boundary.
`Inbox`, `runLoop`, and the instance-bound publication/start controls are package-internal. The package root does not export them, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than constructing or starting the driver internals. `ReactLoopAgent.send()`, running `steer()`, and open-turn `inject()` materialize content plus resolved source once as detached, deeply frozen lossless JSON; malformed data throws before enqueue or append. An injection that arrives while the current step executes assistant tool calls stays in a FIFO until the batch settles; successful batches place it after the complete result batch, and interrupted batches drain it before the turn closes.
### Loop lifecycle (`loop.ts`)
The driver owns one agent for its lifetime. It records turn, step, request, stream, and tool boundaries in the session log; live extension events coordinate policy around those durable facts. The [architecture turn flow](../../../docs/architecture.md#turn-flow) and generated [event catalog](../../../docs/cordis-catalog/events.md) are the authoritative sequence and signatures.
Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
Plugin failure ends the current turn, not the loop. Cancellation clears pending work and aborts the current step without leaking to the next prompt. Terminal continuation stops remain authoritative through turn close and durability flush.
Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and context remain model-ordered. Abort stops new calls, drains started results, discards their context, and follows the normal abort path.
Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, then drains accepted batch context before the turn closes through the normal abort path.
### What belongs to plugins

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

@@ -8,7 +8,7 @@
import type { Context } from 'cordis'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type { AgentId, AgentOptions, AgentStatus, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { AgentId, AgentOptions, AgentStatus, HookContext, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
@@ -152,6 +152,10 @@ export class ReactLoopAgent implements Agent {
* this set before the lifecycle unregisters the agent or detaches its session.
*/
private pendingIdleFlushes = new Set<Promise<void>>()
/** Whether the current step is executing an assistant tool-call batch. */
private toolBatchActive = false
/** Open-turn injections waiting for the active assistant tool-call batch to close. */
private deferredInjections: HookContext[] = []
constructor(
private loopCtx: Context,
@@ -194,12 +198,11 @@ export class ReactLoopAgent implements Agent {
}
/**
* Accept one public send/steer payload as the exact detached record shared by
* the live notification and inbox. Lossless-JSON materialization reads every
* nested field once; deep freeze prevents an observer from rewriting queued
* work before the loop drains it.
* Accept one public message payload as a detached record. Lossless-JSON
* materialization reads every nested field once; deep freeze prevents later
* caller mutation before an inbox or deferred-injection queue drains it.
*/
private acceptInboxMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
private acceptMessage(content: ContentBlock[], options?: SendOptions): InboxMessage {
const source = this.resolveSource(options)
const accepted = snapshotJsonValue({ content, source })
if (accepted === undefined) {
@@ -208,6 +211,15 @@ export class ReactLoopAgent implements Agent {
return deepFreeze(accepted)
}
/** Detach one context before it can outlive its caller in the active-batch FIFO. */
private acceptContext(context: HookContext): HookContext {
const accepted = snapshotJsonValue(context)
if (accepted === undefined) {
throw new TypeError('agent context must be losslessly JSON-serializable')
}
return deepFreeze(accepted)
}
/** Reject a driving operation once teardown has synchronously closed the agent. */
private assertNotDisposed(): void {
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
@@ -215,7 +227,7 @@ export class ReactLoopAgent implements Agent {
send(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
const accepted = this.acceptInboxMessage(content, options)
const accepted = this.acceptMessage(content, options)
this.#inbox.enqueue(accepted)
const info = { source: accepted.source, steering: false } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -224,7 +236,7 @@ export class ReactLoopAgent implements Agent {
steer(content: ContentBlock[], options?: SendOptions): void {
this.assertNotDisposed()
if (this._status !== 'running') { this.send(content, options); return }
const accepted = this.acceptInboxMessage(content, options)
const accepted = this.acceptMessage(content, options)
this.#inbox.steer(accepted)
const info = { source: accepted.source, steering: true } as const
agentEvents(this.loopCtx, this).emit('agent/queued', accepted.content, info)
@@ -240,10 +252,15 @@ export class ReactLoopAgent implements Agent {
...options?.meta !== undefined ? { meta: options.meta } : {},
}
if (isTurnOpen(this.session)) {
// A turn is open in the LOG (decided from the log, not agent status —
// status can be `running` with no turn open): the context/message is
// turn-enclosed by that turn, so append it directly.
this.session.append('context/message', context, { surfaceOp: 'append' })
const accepted = this.acceptContext(context)
// Provider protocols require every assistant tool-call batch to be
// followed only by its tool results. Historical interrupted batches do
// not own new context; only the currently executing batch may defer it.
if (this.toolBatchActive) {
this.deferredInjections.push(accepted)
return
}
this.session.append('context/message', accepted, { surfaceOp: 'append' })
return
}
// No turn open: wrap the injection in a one-shot turn so every event stays
@@ -283,6 +300,34 @@ export class ReactLoopAgent implements Agent {
}
}
/** Append deferred open-turn injections after the loop closes a tool-result batch. */
private drainDeferredInjections(): void {
const pending = this.deferredInjections.splice(0)
for (const accepted of pending) {
this.session.append('context/message', accepted, { surfaceOp: 'append' })
}
}
/**
* Run one tool-call batch and drain its deferred context before settlement.
* The loop-owned acceptor remains valid after public disposal begins because
* the interrupted turn stays open until this batch settles.
*/
private async withToolBatch<T>(
run: (acceptContext: (context: HookContext) => void) => Promise<T>,
): Promise<T> {
this.toolBatchActive = true
const acceptContext = (context: HookContext): void => {
this.deferredInjections.push(this.acceptContext(context))
}
try {
return await run(acceptContext)
} finally {
this.toolBatchActive = false
this.drainDeferredInjections()
}
}
cancel(reason?: string): void {
// Arm only for current work; an idle marker would cancel the next prompt.
if (this._status === 'running' || this.currentAbort !== undefined || this.#inbox.hasQueued || this.#inbox.hasSteering) {
@@ -348,6 +393,7 @@ export class ReactLoopAgent implements Agent {
isCancelled: () => this.cancelRequested,
cancelReason: () => this.cancelReason,
clearCancel: () => { this.cancelRequested = false },
withToolBatch: run => this.withToolBatch(run),
// Pre-step cancellation re-parks without emitting a status transition.
settleIdle: () => { this.settleIdleWaiters() },
})

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

@@ -10,7 +10,7 @@ import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, Messag
import { isDeepStrictEqual } from 'node:util'
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, PromptDecision } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
import { canonicalHeader } from '@deepseek-ai/dsh-session'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
@@ -89,6 +89,8 @@ export interface LoopHandle {
clearCancel(): void
/** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
settleIdle(): void
/** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
readonly withToolBatch: <T>(run: (acceptContext: (context: HookContext) => void) => Promise<T>) => Promise<T>
}
/**
@@ -334,8 +336,7 @@ async function runTurn(
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
try {
stepOutcome = await runStep(
ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages,
transmission, abort.signal, handle.maxParallelToolCalls)
ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
} finally {
@@ -472,6 +473,7 @@ async function runStep(
ctx: Context,
events: AgentEventDispatch,
agent: ReactLoopAgent,
handle: LoopHandle,
turn: number,
step: number,
assembly: PromptAssembly,
@@ -479,7 +481,6 @@ async function runStep(
boundaryMessages: Message[],
transmission: TransmissionLog,
signal: AbortSignal,
maxParallelToolCalls: number,
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
const { session, options } = agent
@@ -541,7 +542,9 @@ async function runStep(
const assembled = assembler.message()
const assembledContent = structuredClone(assembled.content)
let message: Message = withoutToolCalls(assembled)
message = withoutToolCalls(await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message)))
message = withoutToolCalls(await processStepResult(
events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
))
// Preserve usage even when max-token truncation produced no content.
recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
return { hadToolCalls: false, finish: assembler.finish }
@@ -551,28 +554,55 @@ async function runStep(
const assembled = assembler.message()
const assembledContent = structuredClone(assembled.content)
let message: Message = assembled
message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
message = await processStepResult(
events, session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs,
)
const toolCalls = message.content.filter(block => block.type === 'tool-call')
// Empty messages exist only to carry usage; the helper also omits empty chunk provenance.
// Every successful call records its completion anchor, including explicit
// empty chunk provenance for a contentless, usage-less provider response.
recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
// Dispatch may overlap; policy, results, and context remain model-ordered.
const pendingContext = toolCalls.length > 0
? await executeToolCalls(ctx, agent, turn, step, toolCalls, signal, maxParallelToolCalls)
: []
// Dispatch may overlap; policy, durable results, and result context stay model-ordered.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
return handle.withToolBatch(async (acceptContext) => {
await executeToolCalls(
ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
)
return { hadToolCalls: true, finish: assembler.finish }
})
}
// Context follows the complete result batch to preserve call/result adjacency.
for (const context of pendingContext) {
agent.inject(context.content, {
source: context.source,
...context.envelope !== undefined ? { envelope: context.envelope } : {},
...context.meta !== undefined ? { meta: context.meta } : {},
})
/** Preserve successful-call accounting without retaining output that result processing rejected. */
async function processStepResult(
events: AgentEventDispatch,
session: Session,
turn: number,
step: number,
config: LlmCallConfig,
assembledContent: ContentBlock[],
message: Message,
assembler: BlockAssembler,
chunkSeqs: number[],
): Promise<Message> {
try {
return await events.waterfall(
'agent/step-result', turn, step, message, () => Promise.resolve(message),
)
} catch (error: unknown) {
recordAssistantMessage(
session,
turn,
step,
config,
assembledContent,
{ ...message, content: [] },
assembler,
chunkSeqs,
false,
)
throw error
}
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
}
/** Record one content-or-usage assistant message with replay-safe provenance. */
@@ -585,8 +615,8 @@ function recordAssistantMessage(
message: Message,
assembler: BlockAssembler,
chunkSeqs: number[],
preserveReplayState = true,
): void {
if (message.content.length === 0 && assembler.usage === undefined) return
session.append(
'assistant/message',
{
@@ -596,11 +626,11 @@ function recordAssistantMessage(
provenance: assistantProvenance(
config,
assembler.replayState,
isDeepStrictEqual(message.content, assembledContent),
preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
),
...assembler.usage === undefined ? {} : { usage: assembler.usage },
},
{ surfaceOp: 'append', ...chunkSeqs.length > 0 ? { sourceEventSeqs: chunkSeqs } : {} },
{ surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
)
}

22
packages/core/agent-loop/src/request-log.ts
View File

@@ -1,11 +1,12 @@
/**
* Per-loop-instance request-header bookkeeping for reconstructability. The
* comparison baseline is the header folded from the session log, so a fresh
* loop instance needs no special resume or fork state.
* comparison baseline is folded from the session log; a fresh instance anchors
* it with an initial/resume snapshot and later logs full changed snapshots.
*
* @module dsh-agent-loop/request-log
*/
import { diffHeader, headerEquals, applyHeaderDelta } from '@deepseek-ai/dsh-session'
import { headerEquals } from '@deepseek-ai/dsh-session'
import type { EpochHeader, Session } from '@deepseek-ai/dsh-session'
import type { Message } from '@deepseek-ai/dsh-llm'
@@ -32,10 +33,8 @@ export function createTransmissionLog(): TransmissionLog {
}
/**
* Append whatever header event makes the log reproduce this request's header.
* The first request from an instance always records a full `initial` or `resume`
* snapshot. Later requests record nothing when unchanged, a round-tripping
* delta when expressible, or a full `fallback` snapshot otherwise.
* Append the full header snapshot owed by this request: initial/resume for the
* instance's first request, nothing when unchanged, or change otherwise.
*
* @param session - the session whose log explains the request.
* @param state - this loop instance's bookkeeping (mutated on first log).
@@ -52,12 +51,5 @@ export function recordRequestHeader(session: Session, state: TransmissionLog, he
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const baseline = session.requestHeader()!
if (headerEquals(baseline, header)) return
const delta = diffHeader(baseline, header)
/* v8 ignore next -- headerEquals false ⟹ diffHeader defined: both compare the same four parts */
if (delta === undefined) return
if (headerEquals(applyHeaderDelta(baseline, delta), header)) {
session.append('request/header-delta', delta)
} else {
session.append('request/header', { header, reason: 'fallback' })
}
session.append('request/header', { header, reason: 'change' })
}

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

@@ -1,8 +1,8 @@
/**
* Schedules one assistant step's tool calls. Exclusive calls form barriers;
* parallel calls use a bounded rolling pool and are reclassified before start.
* Dispatch may overlap, while policy, results, and context remain model-ordered.
* Abort stops replenishment and drains started calls.
* Dispatch may overlap, while policy, results, and result context remain
* model-ordered. Abort stops replenishment and drains started calls.
*
* Each started call records `tool/call`; `tool/result` commits in model order,
* preserving derived history when audit events interleave with earlier results.
@@ -31,8 +31,8 @@ interface Slot {
/**
* Schedule one assistant step's tool calls by their live concurrency mode.
* Started calls receive ordered results; abort drains them, discards their
* buffered context, and rethrows so the turn owns final error handling.
* Started calls receive ordered results. Abort drains them and rethrows after
* accepting their context into the batch FIFO owned by the caller.
*
* @param ctx - loop context that owns the tool registry.
* @param agent - agent and session receiving the call lifecycle.
@@ -41,7 +41,7 @@ interface Slot {
* @param toolCalls - assistant calls in model order.
* @param signal - abort signal shared by the step.
* @param maxParallel - validated in-flight cap.
* @returns buffered contexts in model call order.
* @param acceptContext - accepts committed result context into the active batch.
*/
export async function executeToolCalls(
ctx: Context,
@@ -51,7 +51,8 @@ export async function executeToolCalls(
toolCalls: ToolCallBlock[],
signal: AbortSignal,
maxParallel: number,
): Promise<HookContext[]> {
acceptContext: (context: HookContext) => void,
): Promise<void> {
const { session } = agent
// Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
@@ -66,7 +67,6 @@ export async function executeToolCalls(
},
}))
const pendingContext: HookContext[] = []
let next = 0
while (next < planned.length) {
// Commit before classifying again so registry changes affect unstarted calls.
@@ -74,9 +74,8 @@ export async function executeToolCalls(
const first = planned[next]!
const mode = ctx.tools.executionMode(first.exec).kind
const group = mode === 'parallel' ? planned.slice(next) : [first]
next += await runGroup(ctx, session, turn, step, group, mode, signal, maxParallel, pendingContext)
next += await runGroup(ctx, session, turn, step, group, mode, signal, maxParallel, acceptContext)
}
return pendingContext
}
/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
@@ -92,8 +91,8 @@ function parseArguments(raw: string): unknown {
* Run one exclusive barrier or parallel pool. Later calls are reclassified
* before start; an exclusive reclassification waits for the current pool to
* drain and remains for the caller's next barrier. Results and contexts commit
* in model order. Abort stops starts, drains and commits started calls, discards
* their contexts, and throws.
* in model order. Abort stops starts, drains and commits started calls, accepts
* their contexts into the owning batch, and throws.
*/
async function runGroup(
ctx: Context,
@@ -104,7 +103,7 @@ async function runGroup(
mode: ToolExecutionMode['kind'],
signal: AbortSignal,
maxParallel: number,
pendingContext: HookContext[],
acceptContext: (context: HookContext) => void,
): Promise<number> {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
@@ -127,7 +126,7 @@ async function runGroup(
: ctx.tools[TOOL_REGISTRY_SCHEDULER].finish(slot.exec, slot.result)
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
appendToolResult(session, turn, step, call!.block, result, callSeqs[committed]!)
pendingContext.push(...result.additionalContexts ?? [])
for (const context of result.additionalContexts ?? []) acceptContext(context)
committed++
}
}
@@ -189,7 +188,7 @@ async function runGroup(
}
if (aborted) {
// Started calls are committed; their context is discarded with the aborted step.
// Started calls and accepted context settle before the turn records the abort.
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
throw new Error(String(signal.reason ?? 'aborted'))
}

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

@@ -3,12 +3,12 @@ import { Context } from 'cordis'
import LlmService, { CallId, ContentBlock, MessageSource, 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, { defineTool } from '@deepseek-ai/dsh-tools'
import ToolRegistry, { defineTool, type PostToolDecision } from '@deepseek-ai/dsh-tools'
import AgentRegistry, { AgentId, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS, ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
import { prepareReactLoopAgent } from '../src/agent.ts'
import * as Invariants from '@deepseek-ai/dsh-invariants'
import { MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
import { maxTokensResponse, MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
/** Regression tests for agent-loop boundary, identity, and lifecycle contracts. */
@@ -132,6 +132,73 @@ describe('session log records what agent/step-result actually produced', () => {
})
})
describe('successful provider completion survives agent/step-result failure', () => {
async function expectContentlessCompletionAnchor(
response: StreamChunk[],
id: string,
providerText: string,
): Promise<void> {
const adapter = new MockAdapter([response])
const ctx = await harness(adapter)
await ctx.plugin(Invariants)
const agent = ctx.agentLoop.create(AgentId(id), { provider: 'mock', model: 'mock' })
const failure = new Error(`${id} result processing failed`)
const reported: Error[] = []
ctx.on('agent/step-result', async () => {
throw failure
})
ctx.on('agent/error', (subject, _turn, _step, error) => {
if (subject === agent) reported.push(error)
})
send(agent, 'go')
await waitForIdle(ctx, agent)
const events = [...agent.session.events]
const chunks = events.filter(event => event.type === 'assistant/chunk')
const completions = events.filter(event => event.type === 'assistant/message')
expect(completions).toHaveLength(1)
expect(completions[0]?.type === 'assistant/message' && completions[0].data).toEqual({
turn: 1,
step: 1,
content: [],
provenance: { provider: 'mock', model: 'mock' },
usage: { inputTokens: 10, outputTokens: providerText.length },
})
expect(completions[0]?.sourceEventSeqs).toEqual(chunks.map(event => event.seq))
expect(agent.session.deriveMessages()).toEqual([
{ role: 'user', content: [{ type: 'text', text: 'go' }] },
])
expect(reported).toHaveLength(1)
expect(reported[0]).toBe(failure)
const turnEnd = events.findLast(event => event.type === 'turn/end')
expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({
kind: 'error',
step: 1,
message: failure.message,
})
}
it('records one content-less anchor when ordinary stop result processing rejects', async () => {
const providerText = 'ordinary provider output'
await expectContentlessCompletionAnchor(
textResponse(providerText),
'a-step-result-stop-failure',
providerText,
)
})
it('records one content-less anchor when max-token result processing rejects', async () => {
const providerText = 'truncated provider output'
await expectContentlessCompletionAnchor(
maxTokensResponse(providerText),
'a-step-result-max-token-failure',
providerText,
)
})
})
describe('abort during tool execution ends the turn', () => {
it('aborting the in-flight step inside a tool prevents both remaining tools and the next model step', async () => {
const adapter = new MockAdapter([
@@ -182,6 +249,190 @@ describe('abort during tool execution ends the turn', () => {
expect(adapter.requests).toHaveLength(1) // no follow-up model call
expect(reasons).toEqual([{ kind: 'aborted', reason: 'user interrupt' }])
})
it('records context accepted before a tool-step abort in the same turn', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'aborter', {})])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-abort-injection'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
parameters: {},
async execute() {
agent.inject([{ type: 'text', text: 'accepted before abort' }], { source: { kind: 'plugin', plugin: 'test' } })
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
return [{ type: 'text', text: 'done' }]
},
}))
ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: 'accepted result context after abort' }],
source: { kind: 'plugin', plugin: 'test' },
}],
}))
send(agent, 'go')
await waitForIdle(ctx, agent)
const events = [...agent.session.events]
expect(events
.filter(event => event.type === 'tool/result' || event.type === 'context/message'
|| event.type === 'step/end' || event.type === 'turn/end')
.map(event => event.type))
.toEqual(['tool/result', 'context/message', 'context/message', 'step/end', 'turn/end'])
expect(events
.filter(event => event.type === 'context/message')
.map(event => event.data.content))
.toEqual([
[{ type: 'text', text: 'accepted before abort' }],
[{ type: 'text', text: 'accepted result context after abort' }],
])
})
it('records post-tool context when a later call aborts the batch', async () => {
const adapter = new MockAdapter([[
{ type: 'block-start', index: 0, blockType: 'tool-call' },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('c1'), name: 'first', arguments: '{}' } },
{ type: 'block-start', index: 1, blockType: 'tool-call' },
{ type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('c2'), name: 'aborter', arguments: '{}' } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
] satisfies StreamChunk[]])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-later-abort-context'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'first',
description: '',
parameters: {},
async execute() {
return [{ type: 'text', text: 'first done' }]
},
}))
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
parameters: {},
async execute() {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
return [{ type: 'text', text: 'aborted' }]
},
}))
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => {
if (exec.callId !== CallId('c1')) return next()
return {
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: 'accepted after first result' }],
source: { kind: 'plugin', plugin: 'test' },
}],
}
})
send(agent, 'go')
await waitForIdle(ctx, agent)
const events = [...agent.session.events]
expect(events
.filter(event => event.type === 'tool/result' || event.type === 'context/message'
|| event.type === 'step/end' || event.type === 'turn/end')
.map(event => event.type))
.toEqual(['tool/result', 'tool/result', 'context/message', 'step/end', 'turn/end'])
expect(events.find(event => event.type === 'context/message')?.data.content)
.toEqual([{ type: 'text', text: 'accepted after first result' }])
})
it('drains deferred context before disposal reaches quiescence', async () => {
const adapter = new MockAdapter([toolCallResponse('c1', 'waiter', {})])
const ctx = await harness(adapter)
const started = Promise.withResolvers<undefined>()
let agent!: ReactLoopAgent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(AgentId('a-dispose-injection'), { provider: 'mock', model: 'mock' })
}, { inject: ['agentLoop'] }))
ctx.tools.register(defineTool({
name: 'waiter',
description: '',
parameters: {},
async execute(_args, exec) {
agent.inject([{ 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')
await new Promise<void>((resolve) => {
if (signal.aborted) resolve()
else signal.addEventListener('abort', () => { resolve() }, { once: true })
})
return [{ type: 'text', text: 'done' }]
},
}))
ctx.on('tools/post-execute', async (): Promise<PostToolDecision> => ({
kind: 'accept',
additionalContexts: [{
content: [{ type: 'text', text: 'accepted result context during disposal' }],
source: { kind: 'plugin', plugin: 'test' },
}],
}))
send(agent, 'go')
await started.promise
await fiber.dispose()
expect(agent.session.events
.filter(event => event.type === 'context/message')
.map(event => event.data.content))
.toEqual([
[{ type: 'text', text: 'accepted before disposal' }],
[{ type: 'text', text: 'accepted result context during disposal' }],
])
expect(agent.session.events.find(event => event.type === 'turn/end')?.data.reason)
.toEqual({ kind: 'disposed' })
})
it('limits injection deferral to the current tool batch', async () => {
const adapter = new MockAdapter([
[
{ type: 'block-start', index: 0, blockType: 'tool-call' },
{ type: 'block-end', index: 0, block: { type: 'tool-call', id: CallId('c1'), name: 'aborter', arguments: '{}' } },
{ type: 'block-start', index: 1, blockType: 'tool-call' },
{ type: 'block-end', index: 1, block: { type: 'tool-call', id: CallId('c2'), name: 'second', arguments: '{}' } },
{ type: 'finish', reason: { kind: 'tool-calls' } },
] satisfies StreamChunk[],
textResponse('later turn'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a-historical-tool-pair'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'aborter',
description: '',
parameters: {},
async execute() {
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('user interrupt')
return [{ type: 'text', text: 'done' }]
},
}))
ctx.tools.register(defineTool({
name: 'second',
description: '',
parameters: {},
async execute() {
return [{ type: 'text', text: 'must not run' }]
},
}))
send(agent, 'leave an unmatched historical call')
await waitForIdle(ctx, agent)
ctx.on('agent/pre-step', (subject, turn) => {
if (subject === agent && turn === 2) {
agent.inject([{ type: 'text', text: 'new turn context' }], { source: { kind: 'plugin', plugin: 'test' } })
}
})
send(agent, 'start a text-only turn')
await waitForIdle(ctx, agent)
expect(agent.session.events.find(event => event.type === 'context/message')?.data.content)
.toEqual([{ type: 'text', text: 'new turn context' }])
expect(JSON.stringify(adapter.requests[1]?.messages)).toContain('new turn context')
})
})
describe('steering from late extension points is never stranded', () => {
@@ -1118,9 +1369,10 @@ describe('tool result call identity', () => {
})
})
describe('surface: assistant/message omits sourceEventSeqs when no chunks streamed', () => {
it('a step-result listener injecting content over an empty stream appends with surfaceOp but no sourceEventSeqs', async () => {
// Injected result content with no chunks must omit empty sourceEventSeqs.
describe('surface: assistant/message records exact empty provenance when no chunks streamed', () => {
it('a step-result listener injecting content over an empty stream records sourceEventSeqs []', async () => {
// The explicit empty source set distinguishes a known empty provider
// stream from legacy events whose provenance was not recorded.
const adapter = new MockAdapter([[]])
const ctx = await harness(adapter)
await ctx.plugin(Invariants)
@@ -1137,7 +1389,7 @@ describe('surface: assistant/message omits sourceEventSeqs when no chunks stream
const recorded = agent.session.events.find(e => e.type === 'assistant/message')!
expect(recorded.type).toBe('assistant/message')
expect(recorded.surfaceOp).toBe('append')
expect(recorded.sourceEventSeqs).toBeUndefined()
expect(recorded.sourceEventSeqs).toEqual([])
// The injected content reaches derived history.
expect(JSON.stringify(agent.session.deriveMessages())).toContain('injected')
})

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

@@ -375,8 +375,8 @@ describe('agent/session-prefix', () => {
expect(request.messages[0]).toEqual(reminder)
}
// The anchoring snapshot is the prefix's durable record — and the ONLY
// header event: reuse means no request/header-delta ever.
const headerEvents = events(agent).filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
// header event: reuse means no changed snapshot ever.
const headerEvents = events(agent).filter(e => e.type === 'request/header')
expect(headerEvents).toHaveLength(1)
expect(headerEvents[0]?.type === 'request/header' && headerEvents[0].data.header.messagePrefix).toEqual([reminder])
// Never session history: the derivation starts at the real user prompt.
@@ -492,7 +492,7 @@ describe('agent/session-prefix', () => {
// cached prefix is a deep-frozen clone, so step 2's request is unchanged.
held.content = [{ type: 'text', text: 'v2' }]
expect(adapter.requests[1]!.messages[0]).toEqual({ role: 'user', content: [{ type: 'text', text: 'v1' }] })
expect(events(agent).filter(e => e.type === 'request/header-delta')).toHaveLength(0)
expect(events(agent).filter(e => e.type === 'request/header')).toHaveLength(1)
})
})

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

@@ -410,22 +410,30 @@ describe('agent loop', () => {
expect(requestText).not.toContain('<context source=')
})
it('inject() while running appends into the open turn (no extra synthetic turn)', async () => {
it('defers inject() during tool execution until after the tool result', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'noticer', {}, 'calling'),
textResponse('done'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
// A tool that injects mid-execution: at this point the agent is running, so
// inject must append the context/message into the ALREADY-open turn rather
// than wrap it in its own one-shot turn.
let visibleDuringTool = false
const meta = { kind: 'deferred-test', version: 1 }
ctx.tools.register(defineTool({
name: 'noticer',
description: 'injects a notice',
parameters: {},
async execute() {
agent.inject([{ type: 'text', text: 'mid-turn notice' }], { source: { kind: 'plugin', plugin: 'x' } })
await Promise.resolve()
const first = { type: 'text' as const, text: 'mid-turn notice' }
agent.inject([first], {
source: { kind: 'plugin', plugin: 'x' },
envelope: 'raw',
meta,
})
first.text = 'mutated after inject'
agent.inject([{ type: 'text', text: 'second notice' }], { source: { kind: 'plugin', plugin: 'x' } })
visibleDuringTool = agent.session.events.some(e => e.type === 'context/message')
return [{ type: 'text', text: 'ok' }]
},
}))
@@ -433,13 +441,67 @@ describe('agent loop', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
// Exactly ONE turn ran (no synthetic injection turn), and the mid-turn
// context/message sits inside it.
expect(visibleDuringTool).toBe(false)
// The injection stays in the open turn, but its user-role context cannot
// split the assistant tool call from the provider's tool-result message.
const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
expect(turnStarts).toHaveLength(1)
const ts0 = turnStarts[0]!
expect(ts0.type === 'turn/start' && ts0.data.trigger.kind).toBe('message')
expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
const result = agent.session.events.find(e => e.type === 'tool/result')!
const contexts = agent.session.events.filter(e => e.type === 'context/message')
expect(contexts).toHaveLength(2)
expect(result.seq).toBeLessThan(contexts[0]!.seq)
expect(contexts[0]?.type === 'context/message' && contexts[0].data).toMatchObject({
envelope: 'raw',
meta,
})
expect(contexts.flatMap(event => event.type === 'context/message' ? event.data.content : []))
.toEqual([
{ type: 'text', text: 'mid-turn notice' },
{ type: 'text', text: 'second notice' },
])
const secondRequest = adapter.requests[1]!.messages
const resultIndex = secondRequest.findIndex(message =>
message.content.some(block => block.type === 'tool-result'))
const contextIndexes = secondRequest.flatMap((message, index) =>
message.content.some(block => block.type === 'text'
&& (block.text.includes('mid-turn notice') || block.text.includes('second notice')))
? [index]
: [])
expect(resultIndex).toBeGreaterThanOrEqual(0)
expect(contextIndexes).toHaveLength(2)
expect(contextIndexes.every(index => index > resultIndex)).toBe(true)
})
it('rejects non-JSON context before it enters the active tool-batch FIFO', async () => {
const adapter = new MockAdapter([
toolCallResponse('c1', 'invalid-injector', {}, 'calling'),
textResponse('done'),
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('invalid-context'), { provider: 'mock', model: 'mock' })
ctx.tools.register(defineTool({
name: 'invalid-injector',
description: 'attempts an invalid context injection',
parameters: {},
async execute() {
expect(() => {
agent.inject([{ type: 'text', text: 'invalid' }], {
source: { kind: 'plugin', plugin: 'test' },
meta: { bigint: 1n } as never,
})
}).toThrow('agent context must be losslessly JSON-serializable')
return [{ type: 'text', text: 'rejected invalid context' }]
},
}))
send(agent, 'go')
await waitForIdle(ctx, agent)
expect(agent.session.events.some(event => event.type === 'context/message')).toBe(false)
})
it('agent/turn-continuation can force-continue (/loop pattern) and force-stop', async () => {
@@ -743,10 +805,9 @@ describe('agent loop', () => {
})
})
it('appends no assistant/message for a max-tokens step with empty content and no usage', async () => {
// A max-tokens step truncated to a dropped tool call AND with no usage chunk has nothing to
// record: empty content and no accounting → no assistant/message (the empty-content host
// exists only to carry usage).
it('appends an empty completion anchor for a max-tokens step with no usage', async () => {
// The truncated tool call is dropped from durable content, while the
// successful provider call still needs an exact replay anchor.
const callId = CallId('c1')
const adapter = new MockAdapter([[
{ type: 'block-start', index: 0, blockType: 'tool-call' },
@@ -770,14 +831,20 @@ describe('agent loop', () => {
await waitForIdle(ctx, agent)
expect(reasons).toEqual([{ kind: 'max-tokens' }])
expect(agent.session.events.some(e => e.type === 'assistant/message')).toBe(false)
const assistant = agent.session.events.find(e => e.type === 'assistant/message')!
expect(assistant.type === 'assistant/message' && assistant.data).toEqual({
turn: 1,
step: 1,
content: [],
provenance: { provider: 'mock', model: 'mock' },
})
expect(assistant.sourceEventSeqs?.length).toBeGreaterThan(0)
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
})
it('appends no assistant/message for a normal stop finish with empty content and no usage', async () => {
// A clean `stop` finish that streamed nothing assembled (no blocks) and
// carried no usage chunk has nothing to record: the content-or-usage guard
// on the normal step path suppresses a pure trace-only empty assistant/message.
it('appends an empty completion anchor for a normal stop with no usage', async () => {
// A clean content-less call stays absent from derived messages but remains
// a durable successful-call boundary for replay consumers.
const adapter = new MockAdapter([[{ type: 'finish', reason: { kind: 'stop' } }]])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
@@ -789,7 +856,14 @@ describe('agent loop', () => {
await waitForIdle(ctx, agent)
expect(reasons).toEqual([{ kind: 'completed' }])
expect(agent.session.events.some(e => e.type === 'assistant/message')).toBe(false)
const assistant = agent.session.events.find(e => e.type === 'assistant/message')!
expect(assistant.type === 'assistant/message' && assistant.data).toEqual({
turn: 1,
step: 1,
content: [],
provenance: { provider: 'mock', model: 'mock' },
})
expect(assistant.sourceEventSeqs?.length).toBe(1)
expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }])
})

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

@@ -1,9 +1,8 @@
/**
* recordRequestHeader unit tests: exactly one of four things per request —
* recordRequestHeader unit tests: exactly one of three things per request —
* an 'initial' snapshot (log has no header yet), a 'resume' snapshot (fresh
* loop instance over a log that has one), nothing (header unchanged), a
* round-tripping delta, or a 'fallback' snapshot when the delta encoding
* cannot express the change (pure tool reordering).
* loop instance over a log that has one), nothing (header unchanged), or a
* full 'change' snapshot.
*/
import { describe, expect, it } from 'vitest'
@@ -23,7 +22,7 @@ function openSession(id: string): Session {
}
function headerEvents(session: Session): SessionEvent[] {
return session.events.filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
return session.events.filter(e => e.type === 'request/header')
}
describe('recordRequestHeader', () => {
@@ -53,8 +52,8 @@ describe('recordRequestHeader', () => {
expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('resume')
})
it('logs a round-tripping delta for a mid-run change, and the fold reproduces the header', () => {
const session = openSession('rl-delta')
it("logs a full 'change' snapshot for a mid-run change, and the fold reproduces the header", () => {
const session = openSession('rl-change')
const state = createTransmissionLog()
const first = canonicalHeader({ config: { provider: 'mock', model: 'm' }, system: 'a\nb', tools: [tool('t')] })
recordRequestHeader(session, state, first)
@@ -63,12 +62,12 @@ describe('recordRequestHeader', () => {
recordRequestHeader(session, state, second)
const events = headerEvents(session)
expect(events).toHaveLength(2)
expect(events[1]?.type).toBe('request/header-delta')
expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('change')
expect(session.requestHeader()).toEqual(second)
})
it("records a change the delta cannot express (pure reordering) as a 'fallback' snapshot", () => {
const session = openSession('rl-fallback')
it("records a pure tool reordering as a 'change' snapshot", () => {
const session = openSession('rl-reorder')
const state = createTransmissionLog()
const first = canonicalHeader({ config: { provider: 'mock', model: 'm' }, tools: [tool('a'), tool('b')] })
recordRequestHeader(session, state, first)
@@ -77,9 +76,7 @@ describe('recordRequestHeader', () => {
recordRequestHeader(session, state, reordered)
const events = headerEvents(session)
expect(events).toHaveLength(2)
expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('fallback')
// The fold still lands on the exact header — deltas are an encoding
// optimization, never a correctness dependency.
expect(events[1]?.type === 'request/header' && events[1].data.reason).toBe('change')
expect(session.requestHeader()).toEqual(reordered)
})
})

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

@@ -1,9 +1,8 @@
/**
* Loop-level reconstructability: every request the loop sends is a pure function of the
* session log — messages are the derivation at the step/start boundary, the header is the fold
* of request/header* events — and every request is an append-extension of its predecessor
* unless a logged event (compaction replace, header change) explains the difference. Mock-adapter
* requests are the observable, and the final offline rebuild states the full contract end to end.
* session log — messages derive at the step/start boundary and the header is the latest
* request/header snapshot. Each request extends its predecessor unless a logged compaction
* replacement or header change explains the difference.
*/
import { describe, expect, it } from 'vitest'
@@ -85,7 +84,7 @@ describe('request stability across the loop', () => {
expect(Object.isFrozen(request.messages)).toBe(true)
}
// One anchoring header snapshot; no further header events (nothing changed).
const headerEvents = agent.session.events.filter(e => e.type === 'request/header' || e.type === 'request/header-delta')
const headerEvents = agent.session.events.filter(e => e.type === 'request/header')
expect(headerEvents).toHaveLength(1)
expect(headerEvents[0]?.type === 'request/header' && headerEvents[0].data.reason).toBe('initial')
})
@@ -122,8 +121,8 @@ describe('request stability across the loop', () => {
content: [{ type: 'text', text: '[summary of turn 1]' }],
source: { kind: 'plugin', plugin: 'test-compact' },
}, {
surfaceOp: { op: 'replace', start: nodes[0]!.seq, end: nodes[1]!.seq },
sourceEventSeqs: [nodes[0]!.seq, nodes[1]!.seq],
surfaceOp: { op: 'replace', start: nodes[0]!, end: nodes[1]! },
sourceEventSeqs: [nodes[0]!, nodes[1]!],
})
})
@@ -137,7 +136,7 @@ describe('request stability across the loop', () => {
expect(agent.session.events.filter(e => e.type === 'request/header')).toHaveLength(1)
})
it('a real system-prompt change is a logged header delta; a stable prompt logs nothing', async () => {
it('a real system-prompt change is a full changed-header snapshot; a stable prompt logs nothing', async () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two'), textResponse('three')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(AgentId('a1'), { provider: 'mock', model: 'mock' })
@@ -147,14 +146,15 @@ describe('request stability across the loop', () => {
send(agent, 'second')
await waitForIdle(ctx, agent)
// Identical assembly re-rendered per step is NOT a change.
expect(agent.session.events.filter(e => e.type === 'request/header-delta')).toHaveLength(0)
expect(agent.session.events.filter(e => e.type === 'request/header')).toHaveLength(1)
ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'new guidance' })
send(agent, 'third')
await waitForIdle(ctx, agent)
const deltas = agent.session.events.filter(e => e.type === 'request/header-delta')
expect(deltas).toHaveLength(1)
const snapshots = agent.session.events.filter(e => e.type === 'request/header')
expect(snapshots).toHaveLength(2)
expect(snapshots[1]?.data.reason).toBe('change')
expect(adapter.requests[2]!.system).toContain('new guidance')
// History is preserved across the change — only the header moved.
expect(adapter.requests[2]!.messages.length).toBeGreaterThan(adapter.requests[1]!.messages.length)
@@ -260,9 +260,9 @@ describe('request stability across the loop', () => {
send(agent, 'second')
await waitForIdle(ctx, agent)
// No delta was logged (nothing really changed), and the session's own
// No changed snapshot was logged (nothing really changed), and the session's own
// fold is immutable state.
expect(agent.session.events.filter(e => e.type === 'request/header-delta')).toHaveLength(0)
expect(agent.session.events.filter(e => e.type === 'request/header')).toHaveLength(1)
expect(Object.isFrozen(agent.session.requestHeader())).toBe(true)
expect(adapter.requests[1]!.temperature).toBeUndefined()
})
@@ -296,7 +296,7 @@ describe('request stability across the loop', () => {
const rebuilt = new Session(SessionId(`rebuild-${index}`), structuredClone(events.slice(0, stepStart.seq)))
expect(structuredClone(request.messages)).toEqual(rebuilt.deriveMessages())
// Header: the fold of request/header* events up to this step's dispatch
// Header: the latest request/header snapshot up to this step's dispatch
// (its header event sits between step/start and the first chunk).
const firstChunk = events.find(e => e.type === 'assistant/chunk' && e.seq > stepStart.seq)!
const header = foldRequestHeader(events.slice(0, firstChunk.seq))!

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

@@ -502,7 +502,7 @@ describe('tool-call scheduler: abort handling', () => {
.toEqual([CallId('c1')])
})
it('stops replenishing after abort, commits started results, and drops buffered additional contexts', async () => {
it('stops replenishing after abort, commits started results, and drains accepted additional contexts', async () => {
const adapter = new MockAdapter([
multiCall([1, 2, 3, 4].map(n => ({ id: `c${n}`, name: 'p', args: { id: String(n) } }))),
textResponse('should never be requested'),
@@ -528,7 +528,12 @@ describe('tool-call scheduler: abort handling', () => {
.toEqual([CallId('c1'), CallId('c2')])
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
.toEqual([CallId('c1'), CallId('c2')])
expect(events(agent).filter(e => e.type === 'context/message')).toEqual([])
const settled = events(agent).filter(e => e.type === 'tool/result' || e.type === 'context/message')
expect(settled.map(e => e.type))
.toEqual(['tool/result', 'tool/result', 'context/message', 'context/message'])
expect(settled.filter(e => e.type === 'context/message')
.map(e => (e.data.content[0] as { text: string }).text))
.toEqual(['ctx-c1', 'ctx-c2'])
})
it('does not run an exclusive barrier after a parallel group aborts', async () => {