Merge branch 'codex/simp-agent-entry-state' into codex/simp-unify-agent-session-id
# Conflicts: # docs/cordis-catalog/events.md # docs/event-producer-consumer.md # packages/core/agent-loop/src/agent.ts # packages/core/agent-loop/tests/contract-regressions.spec.ts # packages/core/agent-loop/tests/loop.spec.ts # website/zh-CN/api/harness/events.md
This commit is contained in:
@@ -47,7 +47,7 @@ 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`)
|
||||
|
||||
|
||||
@@ -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() },
|
||||
})
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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 ContinuationDecision } from '@deepseek-ai/dsh-agent'
|
||||
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { prepareReactLoopAgent } from '../src/agent.ts'
|
||||
import * as Invariants from '@deepseek-ai/dsh-invariants'
|
||||
@@ -250,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(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!: ReactLoopAgent
|
||||
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', () => {
|
||||
|
||||
@@ -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(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' }]
|
||||
},
|
||||
}))
|
||||
@@ -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(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 () => {
|
||||
|
||||
Reference in New Issue
Block a user