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

Merge branch 'codex/simp-ui-identity-residue' into codex/simp-hide-concrete-agent-loop

Browse Source 
# Conflicts:
#	docs/event-producer-consumer.md
#	packages/core/agent-loop/README.md
#	packages/core/agent-loop/tests/contract-regressions.spec.ts
This commit is contained in:
Tianyi Cui
2026-07-18 14:52:34 +08:00
parent a42c77f883 7d34e33b9e
commit 2db38ecf05
21 changed files with 434 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -45,7 +45,7 @@ Configured agents start automatically. A model call requires both `provider` and
### Internal concrete driver
The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start 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. The concrete `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. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
The concrete `Agent` class, its `Inbox`, `runLoop`, and instance-bound publication/start 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. The concrete `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. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy.
### Loop lifecycle (`loop.ts`)

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 { AgentOptions, AgentStatus, InjectOptions, SendOptions } from '@deepseek-ai/dsh-agent'
import type { 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'
@@ -149,6 +149,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,
@@ -189,12 +193,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) {
@@ -203,6 +206,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`)
@@ -210,7 +222,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)
@@ -219,7 +231,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)
@@ -235,10 +247,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
@@ -278,6 +295,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) {
@@ -342,6 +387,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() },
})

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

@@ -86,6 +86,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>
}
/**
@@ -331,7 +333,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)
ctx, events, agent, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
} finally {
@@ -468,6 +470,7 @@ async function runStep(
ctx: Context,
events: AgentEventDispatch,
agent: ReactLoopAgent,
handle: LoopHandle,
turn: number,
step: number,
assembly: PromptAssembly,
@@ -558,56 +561,49 @@ async function runStep(
// Tool execution stays sequential; recheck abort around each normalized result.
const toolCalls = message.content.filter(block => block.type === 'tool-call')
// Buffer context until all results are appended to preserve call/result adjacency.
const pendingContext: HookContext[] = []
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
let parsedArguments: unknown
try {
parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
} catch {
parsedArguments = call.arguments
if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
return handle.withToolBatch(async (acceptContext) => {
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
let parsedArguments: unknown
try {
parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
} catch {
parsedArguments = call.arguments
}
// TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
// see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
arguments: parsedArguments,
agent,
signal,
})
session.append('tool/result', {
turn, step,
// Correlation comes from the immutable execution input; the result does
// not duplicate this authoritative transcript identity.
callId: call.id,
content: result.content,
isError: result.isError,
...result.error ? { error: result.error } : {},
// Persist tool-owned presentation data for replay.
...result.meta !== undefined ? { meta: result.meta } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
// Accept into the batch FIFO immediately; entries remain deferred until
// every recorded result settles and survive abort or disposal afterward.
for (const context of result.additionalContexts ?? []) acceptContext(context)
// The signal may flip while the tool is awaited.
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
/* v8 ignore stop */
}
// TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
// see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
arguments: parsedArguments,
agent,
signal,
})
session.append('tool/result', {
turn, step,
// Correlation comes from the immutable execution input; the result does
// not duplicate this authoritative transcript identity.
callId: call.id,
content: result.content,
isError: result.isError,
...result.error ? { error: result.error } : {},
// Persist tool-owned presentation data for replay.
...result.meta !== undefined ? { meta: result.meta } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
pendingContext.push(...result.additionalContexts ?? [])
// The signal may flip while the tool is awaited.
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
/* v8 ignore stop */
}
// Append buffered context after the complete result batch.
for (const context of pendingContext) {
agent.inject(context.content, {
source: context.source,
...context.envelope !== undefined ? { envelope: context.envelope } : {},
...context.meta !== undefined ? { meta: context.meta } : {},
})
}
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
return { hadToolCalls: true, finish: assembler.finish }
})
}
/** Preserve successful-call accounting without retaining output that result processing rejected. */

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

@@ -3,9 +3,8 @@ 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, { type Agent, type ContinuationDecision } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { prepareReactLoopAgent } from '../src/agent.ts'
import * as Invariants from '@deepseek-ai/dsh-invariants'
@@ -254,6 +253,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(SessionId('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(SessionId('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!: Agent
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
agent = inner.agentLoop.create(SessionId('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(SessionId('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', () => {

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

@@ -414,22 +414,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(SessionId('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' }]
},
}))
@@ -437,13 +445,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(SessionId('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 () => {