Merge branch 'codex/simp-unify-agent-session-id' into codex/simp-ui-identity-residue
# Conflicts: # docs/event-producer-consumer.md
This commit is contained in:
Tianyi Cui
@@ -31,6 +31,7 @@ The config-driven `ctx.agentLoop.create()` path keeps its agent owned by the loo
|
||||
|
||||
```ts
|
||||
interface Config {
|
||||
maxParallelToolCalls?: number // default 10; 1 is serial
|
||||
agents: Array<{
|
||||
id: string // required
|
||||
provider?: string
|
||||
@@ -41,7 +42,7 @@ interface Config {
|
||||
}
|
||||
```
|
||||
|
||||
Configured agents start automatically. A model call requires both `provider` and `model`; `agent/request` may supply a missing pair before dispatch. `cwd` applies only to fresh sessions, while `resumeSessionId` retains persisted metadata. Configured agents use the deployment persona, and programmatic setup can shadow it per agent. This plugin supplies the per-agent `provider`, `model`, and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
|
||||
Configured agents start automatically. A model call requires both `provider` and `model`; `agent/request` may supply a missing pair before dispatch. `maxParallelToolCalls` bounds every agent's rolling pool for parallel-safe calls and defaults to `10`. `cwd` applies only to fresh sessions, while `resumeSessionId` retains persisted metadata. Configured agents use the deployment persona, and programmatic setup can shadow it per agent. This plugin supplies the per-agent `provider`, `model`, and `cwd` prompt variables; harness identity and deployment persona belong to `dsh-system-prompt`.
|
||||
|
||||
### Exported concrete class
|
||||
|
||||
@@ -57,6 +58,8 @@ Every provider call that reaches a successful finish appends exactly one `assist
|
||||
|
||||
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 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
|
||||
|
||||
Everything that goes beyond "call the model, run the tools, repeat" belongs to plugins listening on the event taxonomy:
|
||||
@@ -83,7 +86,7 @@ Everything that goes beyond "call the model, run the tools, repeat" belongs to p
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Tool calls within a step execute sequentially** — parallel execution waits on concurrency-safety metadata in the tool contract (see `dsh-tools`).
|
||||
- **Classification is unary** — calls whose safety depends on comparing siblings or resources must remain exclusive ([rationale](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md)).
|
||||
- **Config labels are fresh by default** — omitting `sessionId` creates a fresh `${id}-session-<uuid>` on every startup; exact resume-or-create behavior requires an explicit stable `sessionId`, while `resumeSessionId` requires existing persisted history.
|
||||
- **Config agents have no per-agent persona field or setup hook** — they use the deployment persona; scoped persona/tool composition is available only through the programmatic `ctx.agents.create()` / `resume()` factory options.
|
||||
- **No built-in turn budget** — the default continuation is `continue` whenever a step had tool calls or steering; bounding a runaway turn requires an `agent/turn-continuation` force-stop plugin.
|
||||
|
||||
@@ -54,15 +54,20 @@ export interface PreparedReactLoopAgent {
|
||||
* @param id - the concrete agent identity.
|
||||
* @param options - loop options for the agent.
|
||||
* @param session - the prepared session the agent will own.
|
||||
* @param maxParallelToolCalls - resolved in-flight cap for this agent.
|
||||
* @returns the agent and closures bound only to that exact instance.
|
||||
*/
|
||||
export function prepareReactLoopAgent(
|
||||
ctx: Context, id: SessionId, options: AgentOptions, session: Session,
|
||||
ctx: Context,
|
||||
id: SessionId,
|
||||
options: AgentOptions,
|
||||
session: Session,
|
||||
maxParallelToolCalls: number,
|
||||
): PreparedReactLoopAgent {
|
||||
if (claimedDriverSessions.has(session)) {
|
||||
throw new Error(`session "${session.id}" already has a concrete agent driver`)
|
||||
}
|
||||
const agent = new ReactLoopAgent(ctx, id, options, session)
|
||||
const agent = new ReactLoopAgent(ctx, id, options, session, maxParallelToolCalls)
|
||||
claimedDriverSessions.add(session)
|
||||
const dispose = () => agent[stopDriver]()
|
||||
return {
|
||||
@@ -143,6 +148,8 @@ export class ReactLoopAgent implements Agent {
|
||||
* the `disposed` transition fires and leave the promise hanging.
|
||||
*/
|
||||
private idleWaiters: (() => void)[] = []
|
||||
/** Maximum parallel-safe calls allowed in one step. */
|
||||
private readonly maxParallelToolCalls: number
|
||||
/**
|
||||
* Durability checkpoints started by idle {@link inject} calls. `inject()` is
|
||||
* synchronous, so it cannot await them itself; the driver disposer drains
|
||||
@@ -159,7 +166,9 @@ export class ReactLoopAgent implements Agent {
|
||||
public readonly id: SessionId,
|
||||
public readonly options: AgentOptions,
|
||||
public readonly session: Session,
|
||||
maxParallelToolCalls: number,
|
||||
) {
|
||||
this.maxParallelToolCalls = maxParallelToolCalls
|
||||
const { promise, resolve } = Promise.withResolvers<void>()
|
||||
this.disposed = promise
|
||||
this.resolveDisposed = resolve
|
||||
@@ -380,6 +389,7 @@ export class ReactLoopAgent implements Agent {
|
||||
this.driverStarted = true
|
||||
this.done = runLoop(this.loopCtx, this, {
|
||||
inbox: this.#inbox,
|
||||
maxParallelToolCalls: this.maxParallelToolCalls,
|
||||
setStatus: (status) => { this.setStatus(status) },
|
||||
setAbort: controller => void (this.currentAbort = controller),
|
||||
disposed: this.disposed,
|
||||
|
||||
6
packages/core/agent-loop/src/constants.ts
Normal file
6
packages/core/agent-loop/src/constants.ts
Normal file
@@ -0,0 +1,6 @@
|
||||
/** Shared agent-loop scheduler defaults.
|
||||
* @module dsh-agent-loop/constants
|
||||
*/
|
||||
|
||||
/** Default maximum in-flight parallel-safe calls per agent step. */
|
||||
export const DEFAULT_MAX_PARALLEL_TOOL_CALLS = 10
|
||||
@@ -31,6 +31,7 @@ import {
|
||||
ReactLoopAgent,
|
||||
} from './agent.ts'
|
||||
import type { PreparedReactLoopAgent } from './agent.ts'
|
||||
import { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
|
||||
|
||||
export { ReactLoopAgent } from './agent.ts'
|
||||
|
||||
@@ -97,6 +98,15 @@ function signalAbortError(id: SessionId, signal: AbortSignal): Error {
|
||||
return new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
|
||||
}
|
||||
|
||||
/** Resolve the deployment-wide scheduler cap at the owning config boundary. */
|
||||
function resolveMaxParallelToolCalls(value: number | undefined): number {
|
||||
const maxParallelToolCalls = value ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
|
||||
if (!Number.isInteger(maxParallelToolCalls) || maxParallelToolCalls < 1) {
|
||||
throw new Error('maxParallelToolCalls must be a positive integer')
|
||||
}
|
||||
return maxParallelToolCalls
|
||||
}
|
||||
|
||||
/**
|
||||
* Caller-owned create/resume transaction through rollback-covered publication
|
||||
* and quiescent teardown. Resources remain private until the final registry
|
||||
@@ -187,13 +197,13 @@ class AgentCreationTransaction {
|
||||
}
|
||||
|
||||
/** Construct the driver and scope, then install their complete ordered lifecycle. */
|
||||
prepare(options: AgentOptions, session: Session): ReactLoopAgent {
|
||||
prepare(options: AgentOptions, session: Session, maxParallelToolCalls: number): ReactLoopAgent {
|
||||
this.assertActive()
|
||||
const gate = Promise.withResolvers<void>()
|
||||
this.preparing = gate.promise
|
||||
try {
|
||||
this.session = session
|
||||
const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session)
|
||||
const driver = prepareReactLoopAgent(this.loopCtx, this.id, options, session, maxParallelToolCalls)
|
||||
this.driver = driver
|
||||
const agent = driver.agent
|
||||
const scope = createScope(this.loopCtx, agent)
|
||||
@@ -354,8 +364,15 @@ declare module 'cordis' {
|
||||
}
|
||||
}
|
||||
|
||||
/** Plugin configuration for declarative startup agents. */
|
||||
export { DEFAULT_MAX_PARALLEL_TOOL_CALLS }
|
||||
|
||||
/** Agent-loop plugin configuration. */
|
||||
export interface Config {
|
||||
/**
|
||||
* Maximum parallel-safe calls in flight per agent step. `1` is serial;
|
||||
* omission defaults to {@link DEFAULT_MAX_PARALLEL_TOOL_CALLS}.
|
||||
*/
|
||||
maxParallelToolCalls?: number
|
||||
/** Agents created or resumed at plugin startup. */
|
||||
agents: (AgentOptions & {
|
||||
/** Stable config label used in logs and as the fresh combined-id prefix. */
|
||||
@@ -393,6 +410,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
|
||||
/** Runtime schema for declarative agents. */
|
||||
static Config = z.object({
|
||||
maxParallelToolCalls: z.number().step(1).min(1).default(DEFAULT_MAX_PARALLEL_TOOL_CALLS),
|
||||
agents: z.array(z.object({
|
||||
id: z.string().required(),
|
||||
sessionId: z.string().min(1),
|
||||
@@ -404,12 +422,15 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
}) as unknown as z<Config>
|
||||
|
||||
private readonly ownership: FactoryOwnership
|
||||
/** Resolved concurrency cap for every driver created by this factory. */
|
||||
private readonly maxParallelToolCalls: number
|
||||
/** Plain holder prevents Cordis from re-tracing the factory's dependency context through a caller shadow. */
|
||||
private readonly runtime: { ctx: Context }
|
||||
|
||||
constructor(ctx: Context, public config: Config) {
|
||||
super(ctx, 'agentLoop')
|
||||
validateConfiguredAgents(config.agents)
|
||||
this.maxParallelToolCalls = resolveMaxParallelToolCalls(config.maxParallelToolCalls)
|
||||
this.ownership = new FactoryOwnership(ctx.fiber)
|
||||
this.runtime = { ctx }
|
||||
ctx.effect(() => () => this.ownership.dispose(), 'agentLoop.transactions()')
|
||||
@@ -524,7 +545,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
const transaction = new AgentCreationTransaction(loopCtx, this.ctx, this.ownership, id)
|
||||
try {
|
||||
const session = loopCtx.sessions.prepare(id, { meta })
|
||||
const agent = transaction.prepare(options, session)
|
||||
const agent = transaction.prepare(options, session, this.maxParallelToolCalls)
|
||||
transaction.publish('startup')
|
||||
return agent
|
||||
} catch (error: unknown) {
|
||||
@@ -542,6 +563,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* @returns the published handle.
|
||||
*/
|
||||
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
|
||||
const agentOptions = options.agentOptions ?? {}
|
||||
const transaction = new AgentCreationTransaction(
|
||||
this.runtime.ctx,
|
||||
ownerCtx,
|
||||
@@ -554,7 +576,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
...options.seed === undefined ? {} : { seed: options.seed },
|
||||
...options.meta === undefined ? {} : { meta: options.meta },
|
||||
})
|
||||
const agent = transaction.prepare(options.agentOptions ?? {}, session)
|
||||
const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
|
||||
await transaction.waitFor(options.setup?.(agent.ctx))
|
||||
transaction.assertActive()
|
||||
return transaction.publish('startup')
|
||||
@@ -586,6 +608,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
persistence: SessionPersistence,
|
||||
options: ResumeAgentOptions,
|
||||
): Promise<AgentHandle> {
|
||||
const agentOptions = options.agentOptions ?? {}
|
||||
const transaction = new AgentCreationTransaction(
|
||||
this.runtime.ctx,
|
||||
ownerCtx,
|
||||
@@ -605,7 +628,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
...loaded.meta.seedLength === undefined ? {} : { seedLength: loaded.meta.seedLength },
|
||||
},
|
||||
})
|
||||
const agent = transaction.prepare(options.agentOptions ?? {}, session)
|
||||
const agent = transaction.prepare(agentOptions, session, this.maxParallelToolCalls)
|
||||
await transaction.waitFor(options.setup?.(agent.ctx))
|
||||
transaction.assertActive()
|
||||
return transaction.publish('resume')
|
||||
|
||||
@@ -18,6 +18,7 @@ import type { TransmissionLog } from './request-log.ts'
|
||||
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
|
||||
import type {} from '@deepseek-ai/dsh-tools'
|
||||
import { executeToolCalls } from './tool-calls.ts'
|
||||
import type { ReactLoopAgent } from './agent.ts'
|
||||
import type { Inbox } from './inbox.ts'
|
||||
|
||||
@@ -73,6 +74,8 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
|
||||
export interface LoopHandle {
|
||||
/** Native-private agent inbox handed to the driver only at internal startup. */
|
||||
readonly inbox: Inbox
|
||||
/** Maximum parallel-safe calls allowed in one step. */
|
||||
readonly maxParallelToolCalls: number
|
||||
setStatus(status: 'idle' | 'running'): void
|
||||
setAbort(controller: AbortController | undefined): void
|
||||
/** Resolves when the agent is disposed — unblocks the idle wait. */
|
||||
@@ -559,49 +562,13 @@ async function runStep(
|
||||
// empty chunk provenance for a contentless, usage-less provider response.
|
||||
recordAssistantMessage(session, turn, step, header.config, assembledContent, message, assembler, chunkSeqs)
|
||||
|
||||
// Tool execution stays sequential; recheck abort around each normalized result.
|
||||
// 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) => {
|
||||
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 */
|
||||
}
|
||||
await executeToolCalls(
|
||||
ctx, agent, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
|
||||
)
|
||||
return { hadToolCalls: true, finish: assembler.finish }
|
||||
})
|
||||
}
|
||||
|
||||
229
packages/core/agent-loop/src/tool-calls.ts
Normal file
229
packages/core/agent-loop/src/tool-calls.ts
Normal file
@@ -0,0 +1,229 @@
|
||||
/**
|
||||
* 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 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.
|
||||
* @module dsh-agent-loop/tool-calls
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import { assertNever, type ToolCallBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { HookContext } from '@deepseek-ai/dsh-agent'
|
||||
import type { Session } from '@deepseek-ai/dsh-session'
|
||||
import { TOOL_REGISTRY_SCHEDULER, type ToolExecutionInput, type ToolExecutionMode, type ToolExecutionResult, type ToolRunContext } from '@deepseek-ai/dsh-tools'
|
||||
import type { ReactLoopAgent } from './agent.ts'
|
||||
|
||||
/** One tool call after argument parsing, ready to schedule. */
|
||||
interface PlannedCall {
|
||||
block: ToolCallBlock
|
||||
exec: ToolExecutionInput
|
||||
}
|
||||
|
||||
/** Settled dispatch awaiting model-order finalization. */
|
||||
interface Slot {
|
||||
exec: ToolRunContext
|
||||
result: ToolExecutionResult
|
||||
needsPost: boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule one assistant step's tool calls by their live concurrency mode.
|
||||
* 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.
|
||||
* @param turn - current turn number.
|
||||
* @param step - current step number.
|
||||
* @param toolCalls - assistant calls in model order.
|
||||
* @param signal - abort signal shared by the step.
|
||||
* @param maxParallel - validated in-flight cap.
|
||||
* @param acceptContext - accepts committed result context into the active batch.
|
||||
*/
|
||||
export async function executeToolCalls(
|
||||
ctx: Context,
|
||||
agent: ReactLoopAgent,
|
||||
turn: number,
|
||||
step: number,
|
||||
toolCalls: ToolCallBlock[],
|
||||
signal: AbortSignal,
|
||||
maxParallel: number,
|
||||
acceptContext: (context: HookContext) => void,
|
||||
): Promise<void> {
|
||||
const { session } = agent
|
||||
|
||||
// Inputs are distinct because tools/execute wrappers may replace `exec.signal`.
|
||||
const planned: PlannedCall[] = toolCalls.map(block => ({
|
||||
block,
|
||||
exec: {
|
||||
callId: block.id,
|
||||
name: block.name,
|
||||
arguments: parseArguments(block.arguments),
|
||||
agent,
|
||||
signal,
|
||||
},
|
||||
}))
|
||||
|
||||
let next = 0
|
||||
while (next < planned.length) {
|
||||
// Commit before classifying again so registry changes affect unstarted calls.
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
|
||||
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, acceptContext)
|
||||
}
|
||||
}
|
||||
|
||||
/** Parse model arguments, preserving invalid JSON as text and mapping empty input to `{}`. */
|
||||
function parseArguments(raw: string): unknown {
|
||||
try {
|
||||
return raw ? JSON.parse(raw) : {}
|
||||
} catch {
|
||||
return raw
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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, accepts
|
||||
* their contexts into the owning batch, and throws.
|
||||
*/
|
||||
async function runGroup(
|
||||
ctx: Context,
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
group: PlannedCall[],
|
||||
mode: ToolExecutionMode['kind'],
|
||||
signal: AbortSignal,
|
||||
maxParallel: number,
|
||||
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'))
|
||||
const slots: (Slot | undefined)[] = group.map(() => undefined)
|
||||
// Started slots retain their tool/call seq for result provenance.
|
||||
const callSeqs: number[] = group.map(() => -1)
|
||||
let nextToStart = 0
|
||||
let committed = 0
|
||||
let started = 0
|
||||
let aborted: boolean = signal.aborted
|
||||
|
||||
// `committed` advances only across contiguous model-order slots.
|
||||
const commitReady = async (): Promise<void> => {
|
||||
while (committed < group.length) {
|
||||
const slot = slots[committed]
|
||||
if (slot === undefined) break
|
||||
const call = group[committed]
|
||||
const result = slot.needsPost
|
||||
? await ctx.tools[TOOL_REGISTRY_SCHEDULER].finalize(slot.exec, slot.result)
|
||||
: 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]!)
|
||||
for (const context of result.additionalContexts ?? []) acceptContext(context)
|
||||
committed++
|
||||
}
|
||||
}
|
||||
|
||||
const inFlight = new Map<number, Promise<number>>()
|
||||
|
||||
const startCall = async (index: number): Promise<void> => {
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded index
|
||||
const call = group[index]!
|
||||
callSeqs[index] = appendToolCall(session, turn, step, call.block)
|
||||
started++
|
||||
const prepared = await ctx.tools[TOOL_REGISTRY_SCHEDULER].prepare(call.exec)
|
||||
switch (prepared.kind) {
|
||||
case 'dispatch': {
|
||||
const promise = ctx.tools[TOOL_REGISTRY_SCHEDULER].dispatch(prepared.exec).then((outcome) => {
|
||||
slots[index] = { exec: prepared.exec, result: outcome.result, needsPost: outcome.kind === 'post-result' }
|
||||
return index
|
||||
})
|
||||
inFlight.set(index, promise)
|
||||
break
|
||||
}
|
||||
case 'post-result':
|
||||
slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: true }
|
||||
break
|
||||
case 'final-result':
|
||||
slots[index] = { exec: prepared.exec, result: prepared.result, needsPost: false }
|
||||
break
|
||||
/* v8 ignore next -- closed-union exhaustiveness guard */
|
||||
default:
|
||||
assertNever(prepared, 'tool-call scheduler prepare result')
|
||||
}
|
||||
}
|
||||
|
||||
const fillPool = async (): Promise<void> => {
|
||||
while (!aborted && nextToStart < group.length && inFlight.size < maxParallel) {
|
||||
// Re-read later modes after ordered commits so registry changes can create a barrier.
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- bounded by the loop condition
|
||||
const nextCall = group[nextToStart]!
|
||||
if (nextToStart > 0 && mode === 'parallel'
|
||||
&& ctx.tools.executionMode(nextCall.exec).kind !== 'parallel') break
|
||||
await startCall(nextToStart)
|
||||
nextToStart++
|
||||
await commitReady()
|
||||
// Abort may arrive while pre-execute awaits.
|
||||
if (signal.aborted) aborted = true
|
||||
}
|
||||
}
|
||||
|
||||
// Ordered pre-execute may await; only dispatch/body overlaps.
|
||||
// TODO: Drain every started call before rethrowing a scheduler error; tool
|
||||
// bodies must not outlive the failed turn.
|
||||
await fillPool()
|
||||
while (inFlight.size > 0) {
|
||||
const settledIndex = await Promise.race(inFlight.values())
|
||||
inFlight.delete(settledIndex)
|
||||
await commitReady()
|
||||
// Abort may arrive while a tool or ordered commit awaits.
|
||||
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
|
||||
if (signal.aborted) aborted = true
|
||||
await fillPool()
|
||||
}
|
||||
|
||||
if (aborted) {
|
||||
// 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'))
|
||||
}
|
||||
/* v8 ignore next -- unreachable: a non-aborted group commits every started call */
|
||||
if (committed !== started) throw new Error('tool-call scheduler: uncommitted settled calls')
|
||||
return started
|
||||
}
|
||||
|
||||
/** Append a started call and return its provenance sequence. */
|
||||
function appendToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): number {
|
||||
const event = session.append('tool/call', { turn, step, callId: block.id, name: block.name, arguments: block.arguments })
|
||||
return event.seq
|
||||
}
|
||||
|
||||
/** Append a model-ordered result linked to its call event. */
|
||||
function appendToolResult(
|
||||
session: Session,
|
||||
turn: number,
|
||||
step: number,
|
||||
block: ToolCallBlock,
|
||||
result: ToolExecutionResult,
|
||||
callSeq: number,
|
||||
): void {
|
||||
session.append('tool/result', {
|
||||
turn, step,
|
||||
// Correlation stays with the loop's authoritative model-transcript call id;
|
||||
// registry results deliberately do not duplicate it.
|
||||
callId: block.id,
|
||||
content: result.content,
|
||||
isError: result.isError,
|
||||
...result.error ? { error: result.error } : {},
|
||||
// The tool's private presentation payload (e.g. a result-time diff),
|
||||
// persisted so a UI bridge reproduces the card on replay.
|
||||
...result.meta !== undefined ? { meta: result.meta } : {},
|
||||
}, { surfaceOp: 'append', sourceEventSeqs: [callSeq] })
|
||||
}
|
||||
@@ -5,7 +5,7 @@ import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import AgentLoop, { DEFAULT_MAX_PARALLEL_TOOL_CALLS, ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { bindReactLoopAgentContext, prepareReactLoopAgent } from '../src/agent.ts'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
@@ -52,10 +52,14 @@ describe('ReactLoopAgent', () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('exclusive-driver'))
|
||||
const prepared = prepareReactLoopAgent(ctx, SessionId('first-driver'), { provider: 'mock', model: 'mock' }, session)
|
||||
const prepared = prepareReactLoopAgent(
|
||||
ctx, SessionId('first-driver'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
|
||||
)
|
||||
|
||||
expect(() => prepared.agent.ctx).toThrow('context is not bound')
|
||||
expect(() => prepareReactLoopAgent(ctx, SessionId('second-driver'), { provider: 'mock', model: 'mock' }, session))
|
||||
expect(() => prepareReactLoopAgent(
|
||||
ctx, SessionId('second-driver'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
|
||||
))
|
||||
.toThrow('already has a concrete agent driver')
|
||||
|
||||
await prepared.dispose()
|
||||
@@ -253,7 +257,9 @@ describe('ReactLoopAgent', () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('test'))
|
||||
const prepared = prepareReactLoopAgent(ctx, SessionId('bare'), { provider: 'mock', model: 'mock' }, session)
|
||||
const prepared = prepareReactLoopAgent(
|
||||
ctx, SessionId('bare'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
|
||||
)
|
||||
const { agent } = prepared
|
||||
|
||||
prepared.markPublished()
|
||||
@@ -271,7 +277,9 @@ describe('ReactLoopAgent', () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const session = ctx.sessions.create(SessionId('pre-start-dispose'))
|
||||
const prepared = prepareReactLoopAgent(ctx, SessionId('pre-start-dispose'), { provider: 'mock', model: 'mock' }, session)
|
||||
const prepared = prepareReactLoopAgent(
|
||||
ctx, SessionId('pre-start-dispose'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
|
||||
)
|
||||
|
||||
await prepared.dispose()
|
||||
expect(prepared.agent.status).toBe('disposed')
|
||||
@@ -368,7 +376,9 @@ describe('ReactLoopAgent', () => {
|
||||
const adapter = new MockAdapter(['hang'])
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const session = ctx.sessions.create(SessionId('bare'))
|
||||
const prepared = prepareReactLoopAgent(ctx, SessionId('bare'), { provider: 'mock', model: 'mock' }, session)
|
||||
const prepared = prepareReactLoopAgent(
|
||||
ctx, SessionId('bare'), { provider: 'mock', model: 'mock' }, session, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
|
||||
)
|
||||
const { agent } = prepared
|
||||
prepared.markPublished()
|
||||
const dispose = prepared.startDriver()
|
||||
|
||||
@@ -5,7 +5,7 @@ import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
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 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 { maxTokensResponse, MockAdapter, textResponse, toolCallResponse } from './mock-adapter.ts'
|
||||
@@ -822,7 +822,9 @@ describe('turn numbering continues across seeded sessions', () => {
|
||||
ctx2.llm.registerAdapter(['mock'], second)
|
||||
|
||||
const seeded = ctx2.sessions.create(SessionId('forked'), { seed: [...agent.session.events] })
|
||||
const prepared = prepareReactLoopAgent(ctx2, SessionId('forked-agent'), { provider: 'mock', model: 'mock' }, seeded)
|
||||
const prepared = prepareReactLoopAgent(
|
||||
ctx2, SessionId('forked-agent'), { provider: 'mock', model: 'mock' }, seeded, DEFAULT_MAX_PARALLEL_TOOL_CALLS,
|
||||
)
|
||||
const forked = prepared.agent
|
||||
prepared.markPublished()
|
||||
ctx2.effect(() => prepared.startDriver())
|
||||
|
||||
571
packages/core/agent-loop/tests/tool-calls.spec.ts
Normal file
571
packages/core/agent-loop/tests/tool-calls.spec.ts
Normal file
@@ -0,0 +1,571 @@
|
||||
/**
|
||||
* Exercises scheduler ordering and cancellation with deterministic gated tools.
|
||||
* ACP goldens own transcript-facing coverage.
|
||||
*/
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import ToolRegistry, { defineTool, type PostToolDecision, type PreToolDecision } from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
async function harness(adapter: MockAdapter, maxParallelToolCalls?: number) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, {
|
||||
agents: [],
|
||||
...maxParallelToolCalls === undefined ? {} : { maxParallelToolCalls },
|
||||
})
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
return ctx
|
||||
}
|
||||
|
||||
function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'idle') { dispose(); resolve() }
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
function events(agent: ReactLoopAgent): SessionEvent[] {
|
||||
return [...agent.session.events]
|
||||
}
|
||||
|
||||
/** Build one assistant response containing the supplied tool calls. */
|
||||
function multiCall(calls: { id: string; name: string; args: object }[]): StreamChunk[] {
|
||||
const chunks: StreamChunk[] = []
|
||||
calls.forEach((call, index) => {
|
||||
chunks.push(
|
||||
{ type: 'block-start', index, blockType: 'tool-call' },
|
||||
{ type: 'block-end', index, block: { type: 'tool-call', id: CallId(call.id), name: call.name, arguments: JSON.stringify(call.args) } },
|
||||
)
|
||||
})
|
||||
chunks.push(
|
||||
{ type: 'usage', usage: { inputTokens: 5, outputTokens: 5 } },
|
||||
{ type: 'finish', reason: { kind: 'tool-calls' } },
|
||||
)
|
||||
return chunks
|
||||
}
|
||||
|
||||
/** A tool whose calls block until the test releases them by callId. */
|
||||
function gatedTool(name: string, parallel: boolean) {
|
||||
const gates = new Map<string, () => void>()
|
||||
const started: string[] = []
|
||||
const tool = defineTool({
|
||||
name,
|
||||
description: `gated ${name}`,
|
||||
parameters: { id: { type: 'string', required: true } },
|
||||
...parallel ? { isConcurrencySafe: () => true } : {},
|
||||
async execute(args) {
|
||||
started.push(args.id)
|
||||
await new Promise<void>((resolve) => { gates.set(args.id, resolve) })
|
||||
return [{ type: 'text', text: `done-${args.id}` }]
|
||||
},
|
||||
})
|
||||
return {
|
||||
tool,
|
||||
started,
|
||||
release(id: string) { gates.get(id)?.(); gates.delete(id) },
|
||||
pending() { return [...gates.keys()] },
|
||||
}
|
||||
}
|
||||
|
||||
function gatedParallelTool(name: string) {
|
||||
return gatedTool(name, true)
|
||||
}
|
||||
|
||||
function gatedExclusiveTool(name: string) {
|
||||
return gatedTool(name, false)
|
||||
}
|
||||
|
||||
/** Poll until `predicate` holds, letting microtasks/timers drain between checks. */
|
||||
async function until(predicate: () => boolean): Promise<void> {
|
||||
for (let i = 0; i < 1000 && !predicate(); i++) await new Promise(r => setTimeout(r, 0))
|
||||
if (!predicate()) throw new Error('until: condition never held')
|
||||
}
|
||||
|
||||
describe('tool-call scheduler: grouping and barriers', () => {
|
||||
it('runs parallel-safe siblings concurrently (all start before any completes)', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }, { id: 'c3', name: 'p', args: { id: '3' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 3)
|
||||
expect(gated.started).toEqual(['1', '2', '3'])
|
||||
gated.release('1'); gated.release('2'); gated.release('3')
|
||||
await waitForIdle(ctx, agent)
|
||||
})
|
||||
|
||||
it('an exclusive call between two parallel-safe calls forms a barrier (3 groups)', async () => {
|
||||
const order: string[] = []
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([
|
||||
{ id: 'c1', name: 'r', args: { id: 'A1' } },
|
||||
{ id: 'c2', name: 'w', args: { id: 'A2' } },
|
||||
{ id: 'c3', name: 'r', args: { id: 'A3' } },
|
||||
]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'r', description: 'read', parameters: { id: { type: 'string', required: true } },
|
||||
isConcurrencySafe: () => true,
|
||||
async execute(args) { order.push(`r-start-${args.id}`); order.push(`r-end-${args.id}`); return [{ type: 'text', text: 'r' }] },
|
||||
}))
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'w', description: 'write', parameters: { id: { type: 'string', required: true } },
|
||||
async execute(args) { order.push(`w-${args.id}`); return [{ type: 'text', text: 'w' }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(order).toEqual(['r-start-A1', 'r-end-A1', 'w-A2', 'r-start-A3', 'r-end-A3'])
|
||||
})
|
||||
|
||||
it('reclassifies pending calls after an exclusive barrier replaces their tool', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([
|
||||
{ id: 'c1', name: 'replace', args: { id: '0' } },
|
||||
{ id: 'c2', name: 'x', args: { id: '1' } },
|
||||
{ id: 'c3', name: 'x', args: { id: '2' } },
|
||||
]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const replacement = gatedExclusiveTool('x')
|
||||
const disposeSafe = ctx.tools.register(defineTool({
|
||||
name: 'x',
|
||||
description: 'initially safe',
|
||||
parameters: { id: { type: 'string', required: true } },
|
||||
isConcurrencySafe: () => true,
|
||||
async execute(args) { return [{ type: 'text', text: `old-${args.id}` }] },
|
||||
}))
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'replace',
|
||||
description: 'replace x',
|
||||
parameters: { id: { type: 'string', required: true } },
|
||||
async execute() {
|
||||
disposeSafe()
|
||||
ctx.tools.register(replacement.tool)
|
||||
return [{ type: 'text', text: 'replaced' }]
|
||||
},
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => replacement.started.length === 1)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(replacement.started).toEqual(['1'])
|
||||
replacement.release('1')
|
||||
await until(() => replacement.started.length === 2)
|
||||
expect(replacement.started).toEqual(['1', '2'])
|
||||
replacement.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
})
|
||||
|
||||
it('stops replenishing when a result observer makes the next call exclusive', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([
|
||||
{ id: 'c1', name: 'x', args: { id: '1' } },
|
||||
{ id: 'c2', name: 'x', args: { id: '2' } },
|
||||
{ id: 'c3', name: 'x', args: { id: '3' } },
|
||||
]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter, 2)
|
||||
const initial = gatedParallelTool('x')
|
||||
const replacement = gatedExclusiveTool('x')
|
||||
const disposeInitial = ctx.tools.register(initial.tool)
|
||||
ctx.on('tools/result', (exec) => {
|
||||
if (exec.callId !== CallId('c1')) return
|
||||
disposeInitial()
|
||||
ctx.tools.register(replacement.tool)
|
||||
})
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => initial.started.length === 2)
|
||||
initial.release('1')
|
||||
await until(() => events(agent).some(event =>
|
||||
event.type === 'tool/result' && event.data.callId === CallId('c1')))
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(replacement.started).toEqual([])
|
||||
initial.release('2')
|
||||
await until(() => replacement.started.length === 1)
|
||||
expect(replacement.started).toEqual(['3'])
|
||||
replacement.release('3')
|
||||
await waitForIdle(ctx, agent)
|
||||
})
|
||||
})
|
||||
|
||||
describe('tool-call scheduler: model-order results despite out-of-order settlement', () => {
|
||||
it('commits tool/result in model order even when a later call settles first', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
gated.release('2')
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
const beforeFirst = events(agent).filter(e => e.type === 'tool/result')
|
||||
expect(beforeFirst).toEqual([])
|
||||
gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const results = events(agent).filter(e => e.type === 'tool/result')
|
||||
expect(results.map(e => e.data.callId)).toEqual([CallId('c1'), CallId('c2')])
|
||||
})
|
||||
|
||||
it('derived history pairs calls in model order regardless of tool/call log interleaving', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
gated.release('2'); gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const messages = agent.session.deriveMessages()
|
||||
const toolResults = messages.flatMap(m => m.content.filter(b => b.type === 'tool-result'))
|
||||
expect(toolResults.map(b => b.toolCallId)).toEqual([CallId('c1'), CallId('c2')])
|
||||
})
|
||||
})
|
||||
|
||||
describe('tool-call scheduler: rolling pool honors maxParallelToolCalls', () => {
|
||||
it('rejects invalid global maxParallelToolCalls config at plugin load', async () => {
|
||||
await expect(harness(new MockAdapter([]), 0)).rejects.toThrow()
|
||||
await expect(harness(new MockAdapter([]), 1.5)).rejects.toThrow()
|
||||
})
|
||||
|
||||
it('defensively rejects invalid caps when direct construction bypasses the config schema', () => {
|
||||
expect(() => new AgentLoop(new Context(), { agents: [], maxParallelToolCalls: 0 }))
|
||||
.toThrow('maxParallelToolCalls must be a positive integer')
|
||||
expect(() => new AgentLoop(new Context(), { agents: [], maxParallelToolCalls: 1.5 }))
|
||||
.toThrow('maxParallelToolCalls must be a positive integer')
|
||||
})
|
||||
|
||||
it('defaults the cap when direct construction bypasses the config schema', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
|
||||
expect(() => new AgentLoop(ctx, { agents: [] })).not.toThrow()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('starts at most the cap, replenishing as calls settle', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([1, 2, 3, 4].map(n => ({ id: `c${n}`, name: 'p', args: { id: String(n) } }))),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter, 2)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(gated.started).toEqual(['1', '2'])
|
||||
gated.release('1')
|
||||
await until(() => gated.started.length === 3)
|
||||
expect(gated.started).toEqual(['1', '2', '3'])
|
||||
expect(events(agent)
|
||||
.filter(e => e.type === 'tool/call' || e.type === 'tool/result')
|
||||
.map(e => `${e.type}:${String(e.data.callId)}`)
|
||||
.slice(0, 4))
|
||||
.toEqual(['tool/call:c1', 'tool/call:c2', 'tool/result:c1', 'tool/call:c3'])
|
||||
gated.release('2'); gated.release('3')
|
||||
await until(() => gated.started.length === 4)
|
||||
gated.release('4')
|
||||
await waitForIdle(ctx, agent)
|
||||
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2'), CallId('c3'), CallId('c4')])
|
||||
})
|
||||
|
||||
it('maxParallelToolCalls: 1 is fully serial (no second start before the first settles)', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter, 1)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 1)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(gated.started).toEqual(['1'])
|
||||
gated.release('1')
|
||||
await until(() => gated.started.length === 2)
|
||||
gated.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
})
|
||||
|
||||
it('applies the configured cap to every factory-created agent', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: '' })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [], maxParallelToolCalls: 1 })
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 1)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(gated.started).toEqual(['1'])
|
||||
gated.release('1')
|
||||
await until(() => gated.started.length === 2)
|
||||
gated.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
describe('tool-call scheduler: ordered middleware and additional contexts', () => {
|
||||
it('tools/pre-execute and tools/post-execute observe model call order', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }, { id: 'c3', name: 'p', args: { id: '3' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const pre: string[] = []
|
||||
const post: string[] = []
|
||||
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => { pre.push(String(exec.callId)); return next() })
|
||||
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => { post.push(String(exec.callId)); return next() })
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 3)
|
||||
gated.release('3'); gated.release('2'); gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(pre).toEqual([CallId('c1'), CallId('c2'), CallId('c3')].map(String))
|
||||
expect(post).toEqual([CallId('c1'), CallId('c2'), CallId('c3')].map(String))
|
||||
})
|
||||
|
||||
it('injects additional contexts in model call order, not settlement order', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter, 2)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
ctx.on('tools/post-execute', async (exec, _result): Promise<PostToolDecision> =>
|
||||
({ kind: 'accept', additionalContexts: [{ content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } }] }))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
gated.release('2'); gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
const log = events(agent)
|
||||
const contextTexts = log.filter(e => e.type === 'context/message')
|
||||
.map(e => (e.data.content[0] as { text: string }).text)
|
||||
expect(contextTexts).toEqual(['ctx-c1', 'ctx-c2'])
|
||||
const lastResult = log.findLastIndex(e => e.type === 'tool/result')
|
||||
const firstContext = log.findIndex(e => e.type === 'context/message')
|
||||
expect(lastResult).toBeLessThan(firstContext)
|
||||
})
|
||||
|
||||
it('orders pre-execute denials and errors without dispatching them', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([
|
||||
{ id: 'c1', name: 'p', args: { id: '1' } },
|
||||
{ id: 'c2', name: 'p', args: { id: '2' } },
|
||||
{ id: 'c3', name: 'p', args: { id: '3' } },
|
||||
]),
|
||||
textResponse('done'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const post: string[] = []
|
||||
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
|
||||
if (exec.callId === CallId('c2')) return { kind: 'deny', reason: 'blocked by policy' }
|
||||
if (exec.callId === CallId('c3')) throw new Error('pre exploded')
|
||||
return next()
|
||||
})
|
||||
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => {
|
||||
post.push(String(exec.callId))
|
||||
return next()
|
||||
})
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 1)
|
||||
gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(gated.started).toEqual(['1'])
|
||||
expect(post).toEqual(['c1', 'c2'])
|
||||
const results = events(agent).filter(e => e.type === 'tool/result')
|
||||
expect(results.map(e => e.data.callId)).toEqual([CallId('c1'), CallId('c2'), CallId('c3')])
|
||||
expect((results[1]!.data.content[0] as { text: string }).text).toContain('blocked by policy')
|
||||
expect((results[2]!.data.content[0] as { text: string }).text).toContain('pre exploded')
|
||||
})
|
||||
})
|
||||
|
||||
describe('tool-call scheduler: abort handling', () => {
|
||||
it('starts no calls when the signal is already aborted before a parallel group', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('should never be requested'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
ctx.on('session/event', (session, event) => {
|
||||
if (session === agent.session && event.type === 'assistant/message') {
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('already aborted')
|
||||
}
|
||||
})
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(gated.started).toEqual([])
|
||||
expect(events(agent).filter(e => e.type === 'tool/call')).toEqual([])
|
||||
expect(events(agent).filter(e => e.type === 'tool/result')).toEqual([])
|
||||
})
|
||||
|
||||
it('stops starting siblings when abort fires during ordered pre-execute', async () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([{ id: 'c1', name: 'p', args: { id: '1' } }, { id: 'c2', name: 'p', args: { id: '2' } }]),
|
||||
textResponse('should never be requested'),
|
||||
])
|
||||
const ctx = await harness(adapter)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
ctx.on('tools/pre-execute', async (exec, next): Promise<PreToolDecision> => {
|
||||
if (exec.callId === CallId('c1')) {
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('pre cancelled')
|
||||
}
|
||||
return next()
|
||||
})
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 1)
|
||||
await new Promise(r => setTimeout(r, 5))
|
||||
expect(gated.started).toEqual(['1'])
|
||||
gated.release('1')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1')])
|
||||
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1')])
|
||||
})
|
||||
|
||||
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'),
|
||||
])
|
||||
const ctx = await harness(adapter, 2)
|
||||
const gated = gatedParallelTool('p')
|
||||
ctx.tools.register(gated.tool)
|
||||
ctx.on('tools/post-execute', async (exec, _result, next): Promise<PostToolDecision> => ({
|
||||
...await next(),
|
||||
additionalContexts: [{ content: [{ type: 'text', text: `ctx-${exec.callId}` }], source: { kind: 'plugin', plugin: 'p' } }],
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop now')
|
||||
gated.release('1')
|
||||
gated.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(gated.started).toEqual(['1', '2'])
|
||||
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2')])
|
||||
expect(events(agent).filter(e => e.type === 'tool/result').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2')])
|
||||
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 () => {
|
||||
const adapter = new MockAdapter([
|
||||
multiCall([
|
||||
{ id: 'c1', name: 'p', args: { id: '1' } },
|
||||
{ id: 'c2', name: 'p', args: { id: '2' } },
|
||||
{ id: 'c3', name: 'x', args: { id: '3' } },
|
||||
]),
|
||||
textResponse('should never be requested'),
|
||||
])
|
||||
const ctx = await harness(adapter, 2)
|
||||
const gated = gatedParallelTool('p')
|
||||
const exclusive: string[] = []
|
||||
ctx.tools.register(gated.tool)
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'x',
|
||||
description: 'exclusive',
|
||||
parameters: { id: { type: 'string', required: true } },
|
||||
async execute(args) { exclusive.push(args.id); return [{ type: 'text', text: 'x' }] },
|
||||
}))
|
||||
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
agent.send([{ type: 'text', text: 'go' }])
|
||||
await until(() => gated.started.length === 2)
|
||||
;(agent as unknown as { currentAbort?: AbortController }).currentAbort?.abort('stop before barrier')
|
||||
gated.release('1')
|
||||
gated.release('2')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(exclusive).toEqual([])
|
||||
expect(events(agent).filter(e => e.type === 'tool/call').map(e => e.data.callId))
|
||||
.toEqual([CallId('c1'), CallId('c2')])
|
||||
})
|
||||
})
|
||||
@@ -9,6 +9,7 @@ import { join } from 'node:path'
|
||||
import { afterEach, describe, expect, it } from 'vitest'
|
||||
import {
|
||||
annotateSurface,
|
||||
collectEventEnvelopeTypes,
|
||||
collectLogEvents,
|
||||
collectSurfaceEventTypes,
|
||||
render,
|
||||
@@ -56,6 +57,7 @@ describe('gen-persistence-catalog collectLogEvents', () => {
|
||||
scope: 'fix',
|
||||
doc: 'A thing was recorded.',
|
||||
payload: '{ turn: number }',
|
||||
declaration: '/** A thing was recorded. */\n\'fix/happened\': { turn: number }',
|
||||
source: 'packages/core/fix/src/types.ts:3',
|
||||
})
|
||||
})
|
||||
@@ -102,10 +104,13 @@ describe('gen-persistence-catalog collectLogEvents', () => {
|
||||
it('collapses a newline-separated multi-line payload to a valid one-line fragment', () => {
|
||||
const events = collectLogEvents(make({
|
||||
'packages/group/fix/src/types.ts': merge(
|
||||
' /** Wide payload. */\n \'fix/wide\': {\n alpha: string[]\n range: { start: number; end: number }\n count: number\n }',
|
||||
' /** Wide payload. */\n \'fix/wide\': {\n /** Alpha values. */\n alpha: string[]\n range: { start: number; end: number }\n count: number\n }',
|
||||
),
|
||||
}))
|
||||
expect(events[0]?.payload).toBe('{ alpha: string[]; range: { start: number; end: number }; count: number }')
|
||||
expect(events[0]?.declaration).toBe(
|
||||
'/** Wide payload. */\n\'fix/wide\': {\n /** Alpha values. */\n alpha: string[]\n range: { start: number; end: number }\n count: number\n}',
|
||||
)
|
||||
})
|
||||
|
||||
it('hard-errors on a member with no description prose', () => {
|
||||
@@ -158,6 +163,59 @@ describe('gen-persistence-catalog collectLogEvents', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('gen-persistence-catalog collectEventEnvelopeTypes', () => {
|
||||
const declarations = `/** Event keys. */
|
||||
export type SessionEventType = keyof SessionEventMap
|
||||
/** Surface-producing event keys. */
|
||||
export type SurfaceEventType = 'fix/message'
|
||||
/** Surface placement. */
|
||||
export type SurfaceOp = 'append'
|
||||
/** One persisted event. */
|
||||
export type SessionEvent<T extends SessionEventType = SessionEventType> = { type: T }
|
||||
`
|
||||
|
||||
it('extracts the envelope declarations with their complete JSDoc in canonical order', () => {
|
||||
const entries = collectEventEnvelopeTypes(make({
|
||||
'packages/core/fix/package.json': OWNER_MANIFEST,
|
||||
'packages/core/fix/src/types.ts': declarations,
|
||||
}))
|
||||
expect(entries.map(entry => entry.name)).toEqual([
|
||||
'SessionEventType',
|
||||
'SurfaceEventType',
|
||||
'SurfaceOp',
|
||||
'SessionEvent',
|
||||
])
|
||||
expect(entries[3]).toMatchObject({
|
||||
declaration: '/** One persisted event. */\nexport type SessionEvent<T extends SessionEventType = SessionEventType> = { type: T }',
|
||||
source: 'packages/core/fix/src/types.ts:8',
|
||||
})
|
||||
})
|
||||
|
||||
it('hard-errors when an envelope declaration is missing', () => {
|
||||
expect(() => collectEventEnvelopeTypes(make({
|
||||
'packages/core/fix/package.json': OWNER_MANIFEST,
|
||||
'packages/core/fix/src/types.ts': declarations.replace('/** Surface placement. */\nexport type SurfaceOp = \'append\'\n', ''),
|
||||
}))).toThrow(/missing event-envelope declaration\(s\): SurfaceOp/)
|
||||
})
|
||||
|
||||
it('hard-errors on duplicate, unexported, undocumented, or mistagged envelope declarations', () => {
|
||||
const violations = new RegExp([
|
||||
'4 JSDoc completeness violation\\(s\\)',
|
||||
'[\\s\\S]*not exported',
|
||||
'[\\s\\S]*@mode tag',
|
||||
'[\\s\\S]*SurfaceOp.*no description prose',
|
||||
'[\\s\\S]*SessionEvent.*already declared',
|
||||
].join(''))
|
||||
expect(() => collectEventEnvelopeTypes(make({
|
||||
'packages/core/fix/package.json': OWNER_MANIFEST,
|
||||
'packages/core/fix/src/types.ts': declarations
|
||||
.replace('/** Event keys. */\nexport type SessionEventType', '/** Event keys.\n * @mode emit\n */\ntype SessionEventType')
|
||||
.replace('/** Surface placement. */\n', '')
|
||||
+ '/** Duplicate event. */\nexport type SessionEvent = { type: never }\n',
|
||||
}))).toThrow(violations)
|
||||
})
|
||||
})
|
||||
|
||||
describe('gen-persistence-catalog collectSurfaceEventTypes', () => {
|
||||
it('parses the literal union', () => {
|
||||
const types = collectSurfaceEventTypes(make({
|
||||
@@ -192,9 +250,21 @@ describe('gen-persistence-catalog annotateSurface + render', () => {
|
||||
scope: name.split('/')[0] ?? name,
|
||||
payload: '{ turn: number }',
|
||||
doc: `Records ${name}.`,
|
||||
declaration: `/** Records ${name}. */\n'${name}': { turn: number }`,
|
||||
source: 'packages/core/fix/src/types.ts:3',
|
||||
})
|
||||
|
||||
const envelopeTypes = [
|
||||
'SessionEventType',
|
||||
'SurfaceEventType',
|
||||
'SurfaceOp',
|
||||
'SessionEvent',
|
||||
].map(name => ({
|
||||
name: name as 'SessionEventType' | 'SurfaceEventType' | 'SurfaceOp' | 'SessionEvent',
|
||||
declaration: `/** ${name}. */\nexport type ${name} = never`,
|
||||
source: 'packages/core/fix/src/types.ts:1',
|
||||
}))
|
||||
|
||||
it('badges union members surface and everything else log-only', () => {
|
||||
const annotated = annotateSurface([entry('fix/message'), entry('fix/marker')], ['fix/message'])
|
||||
expect(annotated.map(e => [e.name, e.surface])).toEqual([['fix/message', true], ['fix/marker', false]])
|
||||
@@ -205,11 +275,13 @@ describe('gen-persistence-catalog annotateSurface + render', () => {
|
||||
.toThrow(/'fix\/ghost' name no declared log event/)
|
||||
})
|
||||
|
||||
it('renders badges, payload fences, and the generated-file header', () => {
|
||||
const out = render(annotateSurface([entry('fix/message'), entry('fix/marker')], ['fix/message']))
|
||||
it('renders badges, declaration fences, and the generated-file header', () => {
|
||||
const out = render(annotateSurface([entry('fix/message'), entry('fix/marker')], ['fix/message']), envelopeTypes)
|
||||
expect(out).toContain('Generated by scripts/gen-persistence-catalog.ts')
|
||||
expect(out).toContain('# Session Persistence Event Catalog')
|
||||
expect(out).toContain('```ts persistence-catalog\n/** SessionEventType. */\nexport type SessionEventType = never')
|
||||
expect(out).toContain('#### `fix/message` — surface')
|
||||
expect(out).toContain('#### `fix/marker` — log-only')
|
||||
expect(out).toContain('```ts persistence-catalog\n\'fix/marker\': { turn: number }\n```')
|
||||
expect(out).toContain('```ts persistence-catalog\n/** Records fix/marker. */\n\'fix/marker\': { turn: number }\n```')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -21,6 +21,7 @@ tools:
|
||||
- `ctx.tools.schemas(scope?: ScopeKey): ToolSchema[]` Schemas of everything the scope can see (without the `execute` functions). The shipped tools' schemas are catalogued in [docs/tool-catalog.md](../../../docs/tool-catalog.md), generated by booting each tool plugin and harvesting this method (see [the tool-schema-catalog RFC](../../../docs/rfc/implemented/process/2026-07-02-tool-schema-catalog.md)).
|
||||
- `ctx.tools.guard(guard: ToolGuard): () => void` Register a monotonic synchronous execution guard after `tools/pre-execute`: returning a reason denies the call, while `undefined` leaves it unchanged. A plain-context guard applies globally; an `agent.ctx` guard applies only to that agent. Later waterfall listeners cannot turn a guard denial back into permission. Disposed with the calling fiber.
|
||||
- `ctx.tools.execute(exec)` losslessly snapshots and freezes arguments, assigns an opaque token, runs the complete policy/dispatch/result pipeline, then independently snapshots the authoritative outcome before final observation. Invalid arguments use the same result path without reaching policy or the body; around wrappers may replace only `signal`.
|
||||
- `ctx.tools.executionMode(exec)` returns `parallel` only when the visible definition's `isConcurrencySafe(exec.arguments)` classifier returns exactly `true`; unknown, hidden, undeclared, invalid, or throwing classifications are exclusive.
|
||||
|
||||
### Injected services
|
||||
|
||||
@@ -32,7 +33,7 @@ The live registry pipeline has three transformable waterfalls followed by the ob
|
||||
|
||||
### Key types
|
||||
|
||||
- `ToolDefinition` — `ToolSchema` + `execute(args, exec)`, optional presentation callbacks, and cooperative `timeoutMs`.
|
||||
- `ToolDefinition` — `ToolSchema` + `execute(args, exec)`, optional presentation callbacks, cooperative `timeoutMs`, and optional per-call `isConcurrencySafe(args)` classification.
|
||||
- `ToolExecutionInput` — the caller-supplied call description: `{ callId, name, arguments, agent?, parent?, signal? }`; callers may pass an enclosing execution's opaque token as `parent` but never choose the new execution's own token.
|
||||
- `ToolExecutionToken` — a fresh branded `Symbol` assigned by the registry. It supports equality correlation only and never crosses a model, log, or worker boundary.
|
||||
- `ToolExecution` — the pipeline-owned call: immutable `{ token, callId, name, arguments, agent?, parent? }` identity plus optional operational `signal`, which an around wrapper may add, replace, remove, and restore. A nested call's `parent` is a `ToolExecutionToken`, not an execution object.
|
||||
@@ -87,6 +88,8 @@ See `defineTool`, `validateArgs`, `ToolArgsError`, `SchemaSpec`, `InferArgs`, an
|
||||
|
||||
Optional `timeoutMs` must be positive and finite; it is policy metadata, not model-visible schema.
|
||||
|
||||
Optional `isConcurrencySafe(args)` receives typed, softly validated arguments. Exact `true` permits concurrent dispatch/body execution; invalid input and all other outcomes remain exclusive. Opted-in bodies do not mutate parent-owned state, and shared-state races must commute or fail closed. The [parallel tool-call RFC](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md) owns the full safety contract.
|
||||
|
||||
### Structured-output schema subset
|
||||
|
||||
`StructuredOutputSchema` is the object-rooted raw JSON Schema subset used by subagents and workflows for machine-readable results. It accepts one scalar `type`, object `properties`/`required`/boolean `additionalProperties`, array `items`, and scalar `enum`/`const`. The annotations `description`, `title`, `default`, and `examples` are ignored but must remain JSON data. Type arrays, undeclared required keys, and unsupported keywords fail through `OutputSchemaError` rather than being ignored; `validateStructuredValue()` returns path-qualified violations without throwing.
|
||||
@@ -108,6 +111,10 @@ Under `code` or `both`, the registry exposes the reserved `run_code` transport a
|
||||
- **The dispatch bridge** (`run_code`'s execute): every binding call is JSON-normalized before dispatch (a value that does not survive — `BigInt`, circulars — rejects that one call, so the dispatched form and logged form are the same JSON value by construction), serialized through a per-run queue (even `Promise.all` executes underlying calls one at a time in submission order), given the outer execution's opaque token as `parent`, and run through the complete pre-execute → guards → execute → post-execute → result pipeline. A denial reaches the program as a binding rejection, and each sub-call is logged as a `tool/code-dispatch` session event with deterministic id `<parent>:code:<n>`; `deriveMessages()` does not surface that event. Token correlation lets commit-style observers defer an inner success until the final `run_code` result without exposing the live outer execution; ordinary tool side effects are not rolled back. Every sub-call `additionalContexts` entry is deferred through the outer `ToolRunContext` in dispatch order; the loop appends those contexts only after the parent `run_code` result, preserving adjacency and retaining each source/envelope/meta even when the program later fails.
|
||||
- **Settlement discipline**: the bridge owns a run-scoped abort that follows the outer signal in and fires when the run settles for any reason, so a budget expiry aborts an in-flight sub-tool instead of orphaning it; the bridge then drains its queue BEFORE returning, so every `tool/code-dispatch` lands inside the open turn. A failed run throws `CodeRunFailedError` (`code: 'CODE_RUN_FAILED'`, message = the failure kind + captured logs), which the pipeline converts to a structured `isError` the model self-corrects from.
|
||||
|
||||
### Parallel execution
|
||||
|
||||
The agent loop groups consecutive `parallel` calls into a bounded rolling pool and treats each `exclusive` call as an ordering barrier. Only dispatch/body overlaps; policy, durable results, and context retain model order. Code Mode bindings remain serial. The [parallel tool-call RFC](../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md) owns the shipped declarations and rationale.
|
||||
|
||||
## Model Experience
|
||||
|
||||
### Normal tool schemas
|
||||
@@ -145,7 +152,7 @@ The available tools:
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Native tool calls execute sequentially** — `ToolDefinition` carries no concurrency-safety metadata; adding it (and parallel execution in the loop) waits on the deferred tool-shapes review (`TODO(review)`).
|
||||
- **Concurrency policy is not an event seam** — `executionMode()` reads the resolved tool definition directly; plugins can only declare a classifier on definitions they own.
|
||||
- **`tools/pre-execute` deliberately cannot rewrite `exec.arguments`** — logged and rendered args would desync from what ran; the rewrite design is [a proposed RFC](../../../docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md).
|
||||
- **`defineTool`'s schema DSL is a deliberate subset** — string/number/boolean/object/array with string-only `enum`; `validateArgs` tolerates extra keys and never applies `default` (`XXX(unused-default)` flags removing that field); raw-registered JSON-Schema tools validate their own input.
|
||||
- **`timeoutMs` on a definition is declarative only** — the registry never enforces deadlines; enforcement requires the `@deepseek-ai/dsh-timeout-policy` wrapper.
|
||||
|
||||
@@ -117,9 +117,6 @@ declare module 'cordis' {
|
||||
}
|
||||
}
|
||||
|
||||
// TODO(concurrency): revisit these shapes when concurrency metadata becomes useful
|
||||
// (for example, a read-only hint that would permit safe parallel execution).
|
||||
|
||||
/** Tool output, optionally with lossless-JSON presentation metadata persisted for replay. */
|
||||
export type ToolExecuteReturn = ContentBlock[] | { content: ContentBlock[]; meta?: unknown }
|
||||
|
||||
@@ -134,6 +131,20 @@ export interface ToolDefinition extends ToolSchema {
|
||||
* cooperative implementation that can reach quiescence when the signal aborts.
|
||||
*/
|
||||
timeoutMs?: number
|
||||
/**
|
||||
* Pure synchronous classifier for overlap with sibling tool calls. Only
|
||||
* `true` opts in; omission, exceptions, non-`true` returns, and invalid
|
||||
* `defineTool` arguments are exclusive. This metadata is never model-visible.
|
||||
*
|
||||
* Opted-in executions must not mutate parent-owned state. Shared state must
|
||||
* tolerate concurrent dispatch; recorder races are permitted only when they
|
||||
* commute or fail closed. See the
|
||||
* [parallel-tool-call RFC](../../../../docs/rfc/implemented/feature/2026-07-10-parallel-tool-call-execution.md)
|
||||
* for the full contract.
|
||||
* @param args - parsed arguments; `defineTool` validates before calling.
|
||||
* @returns Whether this call may join a parallel group.
|
||||
*/
|
||||
isConcurrencySafe?(args: unknown): boolean
|
||||
/**
|
||||
* Optional: how to present the PENDING state of one call in a UI, derived from
|
||||
* the call's `args` (parsed arguments, `unknown` — the tool validates/narrows
|
||||
@@ -195,6 +206,14 @@ export interface ToolExecutionInput {
|
||||
signal?: AbortSignal
|
||||
}
|
||||
|
||||
/**
|
||||
* Scheduling mode for one pending call. `parallel` may overlap with siblings;
|
||||
* `exclusive` runs alone and forms an ordering barrier.
|
||||
*/
|
||||
export type ToolExecutionMode =
|
||||
| { kind: 'parallel' }
|
||||
| { kind: 'exclusive' }
|
||||
|
||||
/**
|
||||
* One pending tool call inside the registry pipeline. Parsed arguments cross
|
||||
* one lossless-JSON materialization boundary before policy and are deep-frozen;
|
||||
@@ -222,6 +241,47 @@ export interface ToolRunContext extends ToolExecution {
|
||||
deferContext(context: HookContext): void
|
||||
}
|
||||
|
||||
/**
|
||||
* Scheduler-only result after ordered pre-execute and guards. A `post-result`
|
||||
* still receives post-execute; a `final-result` bypasses it.
|
||||
* @internal
|
||||
*/
|
||||
export type ScheduledToolPreparation =
|
||||
| { kind: 'dispatch'; exec: ToolRunContext }
|
||||
| { kind: 'post-result'; exec: ToolRunContext; result: ToolExecutionResult }
|
||||
| { kind: 'final-result'; exec: ToolRunContext; result: ToolExecutionResult }
|
||||
|
||||
/**
|
||||
* Scheduler-only dispatch result. A `post-result` still receives post-execute;
|
||||
* a `final-result` already matches {@link ToolRegistry.execute} failure semantics.
|
||||
* @internal
|
||||
*/
|
||||
export type ScheduledToolDispatch =
|
||||
| { kind: 'post-result'; result: ToolExecutionResult }
|
||||
| { kind: 'final-result'; result: ToolExecutionResult }
|
||||
|
||||
/**
|
||||
* Symbol-keyed scheduler view that keeps pre/post policy ordered while
|
||||
* overlapping dispatch. Ordinary callers use {@link ToolRegistry.execute};
|
||||
* this is not a plugin seam.
|
||||
* @internal
|
||||
*/
|
||||
export interface ToolRegistryScheduler {
|
||||
/** Materialize input, run the ordered pre-execute/guard gate, and decide what stage follows. */
|
||||
prepare(exec: ToolExecutionInput): Promise<ScheduledToolPreparation>
|
||||
/** Run only the around-dispatch/body stage. */
|
||||
dispatch(exec: ToolRunContext): Promise<ScheduledToolDispatch>
|
||||
/** Run ordered post-execute finalization, then materialize and notify the final outcome. */
|
||||
finalize(exec: ToolRunContext, result: ToolExecutionResult): Promise<ToolExecutionResult>
|
||||
/** Materialize and notify a final outcome that must bypass post-execute. */
|
||||
finish(exec: ToolRunContext, result: ToolExecutionResult): ToolExecutionResult
|
||||
}
|
||||
|
||||
/**
|
||||
* Scheduler entry point omitted from the generated named service API.
|
||||
* @internal
|
||||
*/
|
||||
export const TOOL_REGISTRY_SCHEDULER: unique symbol = Symbol('@deepseek-ai/dsh-tools.scheduler')
|
||||
/** Structured error metadata for a failed tool call (alongside the model-facing text). */
|
||||
export interface ToolErrorInfo {
|
||||
name: string
|
||||
@@ -382,6 +442,16 @@ export class ToolRegistry extends Service {
|
||||
mode: z.union(['native', 'code', 'both'] as const).default('native'),
|
||||
})
|
||||
|
||||
/** Internal staged view consumed by `dsh-agent-loop`'s parallel scheduler. */
|
||||
readonly [TOOL_REGISTRY_SCHEDULER]: ToolRegistryScheduler = {
|
||||
prepare: exec => this.prepareScheduledExecution(exec),
|
||||
dispatch: exec => this.dispatchScheduledExecution(exec),
|
||||
finalize: (exec, result) => this.finalizeScheduledExecution(exec, result),
|
||||
finish: (exec, result) => this.finishScheduledExecution(exec, result),
|
||||
}
|
||||
|
||||
/** Context deferred by a running tool body, keyed by its scheduler-owned execution. */
|
||||
private deferredContexts = new WeakMap<ToolRunContext, HookContext[]>()
|
||||
private global = new Map<string, ToolDefinition>()
|
||||
private scoped = new Map<ScopeKey, Map<string, ToolDefinition>>()
|
||||
/** Compiled restriction filters, per scope (see {@link restrict}). */
|
||||
@@ -682,6 +752,24 @@ export class ToolRegistry extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Classify a pending call through the caller's visible tool definition. Only
|
||||
* an exact `true` is parallel; unknown, hidden, undeclared, invalid, or
|
||||
* throwing classifiers are exclusive.
|
||||
* @param exec - call name, parsed arguments, and optional agent scope.
|
||||
* @returns the fail-closed scheduling mode.
|
||||
*/
|
||||
executionMode(exec: ToolExecutionInput): ToolExecutionMode {
|
||||
const tool = this.get(exec.name, exec.agent)
|
||||
if (!tool?.isConcurrencySafe) return { kind: 'exclusive' }
|
||||
try {
|
||||
const concurrencySafe: unknown = tool.isConcurrencySafe(exec.arguments)
|
||||
return concurrencySafe === true ? { kind: 'parallel' } : { kind: 'exclusive' }
|
||||
} catch {
|
||||
return { kind: 'exclusive' }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute through pre-policy, guards, around-dispatch, post-policy, and final
|
||||
* notification. Tool and listener failures resolve as materialized error
|
||||
@@ -692,6 +780,28 @@ export class ToolRegistry extends Service {
|
||||
* @returns the materialized final result.
|
||||
*/
|
||||
async execute(exec: ToolExecutionInput): Promise<ToolExecutionResult> {
|
||||
return this.prepareExecution(exec, prepared => this.completeScheduledExecution(prepared))
|
||||
}
|
||||
|
||||
private async completeScheduledExecution(prepared: ScheduledToolPreparation): Promise<ToolExecutionResult> {
|
||||
switch (prepared.kind) {
|
||||
case 'dispatch': {
|
||||
const dispatched = await this.dispatchScheduledExecution(prepared.exec)
|
||||
return dispatched.kind === 'post-result'
|
||||
? await this.finalizeScheduledExecution(prepared.exec, dispatched.result)
|
||||
: this.finishScheduledExecution(prepared.exec, dispatched.result)
|
||||
}
|
||||
case 'post-result':
|
||||
return await this.finalizeScheduledExecution(prepared.exec, prepared.result)
|
||||
case 'final-result':
|
||||
return this.finishScheduledExecution(prepared.exec, prepared.result)
|
||||
/* v8 ignore next -- closed-union exhaustiveness guard */
|
||||
default:
|
||||
return assertNever(prepared, 'scheduled tool preparation')
|
||||
}
|
||||
}
|
||||
|
||||
private createExecution(exec: ToolExecutionInput): ScheduledToolPreparation | { kind: 'ready'; exec: ToolRunContext } {
|
||||
const deferredContexts: HookContext[] = []
|
||||
const token = createExecutionToken()
|
||||
const callId = exec.callId
|
||||
@@ -710,105 +820,143 @@ export class ToolRegistry extends Service {
|
||||
deferredContexts.push(context)
|
||||
},
|
||||
}
|
||||
let execution: ToolRunContext
|
||||
try {
|
||||
const detached = snapshotJsonValue(exec.arguments)
|
||||
if (detached === undefined) {
|
||||
throw new TypeError('tool execution arguments must be losslessly JSON-serializable')
|
||||
}
|
||||
execution = {
|
||||
...base,
|
||||
arguments: deepFreeze(detached),
|
||||
}
|
||||
const execution: ToolRunContext = { ...base, arguments: deepFreeze(detached) }
|
||||
this.deferredContexts.set(execution, deferredContexts)
|
||||
return { kind: 'ready', exec: execution }
|
||||
} catch (error: unknown) {
|
||||
execution = { ...base, arguments: undefined }
|
||||
const result = this.materializeFinalResult(toolErrorResult(error))
|
||||
this.notifyResult(execution, result)
|
||||
return result
|
||||
const execution: ToolRunContext = { ...base, arguments: undefined }
|
||||
return { kind: 'final-result', exec: execution, result: toolErrorResult(error) }
|
||||
}
|
||||
let result: ToolExecutionResult
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the ordered pre-execute and monotonic guard stages for the scheduler.
|
||||
* @param input - the caller-supplied execution input.
|
||||
* @returns the prepared execution plus the next scheduler stage.
|
||||
* @internal
|
||||
*/
|
||||
private async prepareScheduledExecution(input: ToolExecutionInput): Promise<ScheduledToolPreparation> {
|
||||
return this.prepareExecution(input, prepared => prepared)
|
||||
}
|
||||
|
||||
private async prepareExecution<T>(
|
||||
input: ToolExecutionInput,
|
||||
next: (prepared: ScheduledToolPreparation) => T | PromiseLike<T>,
|
||||
): Promise<T> {
|
||||
const created = this.createExecution(input)
|
||||
if (created.kind !== 'ready') return next(created)
|
||||
const exec = created.exec
|
||||
try {
|
||||
result = this.materializeFinalResult(await this.executePipeline(execution, deferredContexts))
|
||||
const carrier = scopeTarget(this, exec.agent)
|
||||
const gate = await this.ctx.waterfall(
|
||||
carrier, 'tools/pre-execute', exec,
|
||||
() => Promise.resolve<PreToolDecision>({ kind: 'allow' }),
|
||||
)
|
||||
const decision = gate.kind === 'ask' ? await this.serviceAsk(exec, gate) : gate
|
||||
const denialReason = decision.kind === 'allow'
|
||||
? this.guardReason(exec)
|
||||
: decision.reason
|
||||
if (denialReason !== undefined) {
|
||||
return await next({
|
||||
kind: 'post-result',
|
||||
exec,
|
||||
result: {
|
||||
content: [{ type: 'text', text: `Error: ${denialReason}` }],
|
||||
isError: true,
|
||||
},
|
||||
})
|
||||
}
|
||||
return await next({ kind: 'dispatch', exec })
|
||||
} catch (error: unknown) {
|
||||
// Outer backstop: a throwing pre/post-execute listener, guard, or the
|
||||
// waterfall machinery becomes an isError result, never a turn failure.
|
||||
result = this.materializeFinalResult(toolErrorResult(error))
|
||||
return next({ kind: 'final-result', exec, result: toolErrorResult(error) })
|
||||
}
|
||||
this.notifyResult(execution, result)
|
||||
return result
|
||||
}
|
||||
|
||||
/** Run the transformable pipeline; {@link execute} owns final normalization and notification. */
|
||||
private async executePipeline(exec: ToolRunContext, deferredContexts: HookContext[]): Promise<ToolExecutionResult> {
|
||||
// --- Gate: tools/pre-execute. An `ask` resolves through the optional
|
||||
// approval seam (or degrades to deny) before the monotonic guards run. The
|
||||
// carrier keys dispatch by exec.agent, so an `agent.ctx` listener gates only
|
||||
// its own agent's calls (agent-less calls are subject-less).
|
||||
const carrier = scopeTarget(this, exec.agent)
|
||||
const gate = await this.ctx.waterfall(
|
||||
carrier, 'tools/pre-execute', exec,
|
||||
() => Promise.resolve<PreToolDecision>({ kind: 'allow' }),
|
||||
)
|
||||
const decision = gate.kind === 'ask' ? await this.serviceAsk(exec, gate) : gate
|
||||
const denialReason = decision.kind === 'allow'
|
||||
? this.guardReason(exec)
|
||||
: decision.reason
|
||||
if (denialReason !== undefined) {
|
||||
// Every non-grant, including a failed/unavailable approval request, takes
|
||||
// the same deny path and still reaches post-policy plus result observers.
|
||||
const denied: ToolExecutionResult = {
|
||||
content: [{ type: 'text', text: `Error: ${denialReason}` }],
|
||||
isError: true,
|
||||
}
|
||||
return await this.postExecute(exec, denied)
|
||||
}
|
||||
|
||||
// --- Around-dispatch: tools/execute. The base `next` is the dispatch-
|
||||
// with-normalization thunk — the tool body's own try/catch turns a throw
|
||||
// into an isError result so a wrapper (and post-execute) can inspect it;
|
||||
// an unknown tool routes through the same catch. A `tools/execute` listener
|
||||
// (e.g. a timeout plugin) wraps this thunk: it may replace `exec.signal`
|
||||
// before delegating and inspect the normalized result after. Dispatched with the
|
||||
// same carrier as the gate, so an `agent.ctx` wrapper wraps only its own
|
||||
// agent's calls. ---
|
||||
const result = await this.ctx.waterfall(
|
||||
carrier, 'tools/execute', exec,
|
||||
async (): Promise<ToolExecutionResult> => {
|
||||
try {
|
||||
// Resolve through the CALLER's visible view ({@link get}): a scoped
|
||||
// tool shadows its global name-twin for that agent, and a
|
||||
// restricted-away global tool is exactly as absent as a nonexistent
|
||||
// one — same UNKNOWN_TOOL result, no capability leak in the error.
|
||||
const tool = this.get(exec.name, exec.agent)
|
||||
if (!tool) throw new ToolNotFoundError(exec.name)
|
||||
// Normalize the two `execute` return shapes: a bare ContentBlock[] (no
|
||||
// meta) or a { content, meta } object (a tool attaching a private
|
||||
// presentation payload). An array IS the content; the object carries it.
|
||||
const returned = await tool.execute(exec.arguments, exec)
|
||||
const content = Array.isArray(returned) ? returned : returned.content
|
||||
const meta = Array.isArray(returned) ? undefined : returned.meta
|
||||
return { content, isError: false, ...meta !== undefined ? { meta } : {} }
|
||||
} catch (error: unknown) {
|
||||
return toolErrorResult(error)
|
||||
/**
|
||||
* Run around-dispatch and the tool body. Tool and unknown-tool failures still
|
||||
* receive post-execute; pipeline failures are already final.
|
||||
* @param exec - the prepared execution.
|
||||
* @returns whether the result still needs post-execute.
|
||||
* @internal
|
||||
*/
|
||||
private async dispatchScheduledExecution(exec: ToolRunContext): Promise<ScheduledToolDispatch> {
|
||||
try {
|
||||
const carrier = scopeTarget(this, exec.agent)
|
||||
const result = await this.ctx.waterfall(
|
||||
carrier, 'tools/execute', exec,
|
||||
async (): Promise<ToolExecutionResult> => {
|
||||
try {
|
||||
const tool = this.get(exec.name, exec.agent)
|
||||
if (!tool) throw new ToolNotFoundError(exec.name)
|
||||
const returned = await tool.execute(exec.arguments, exec)
|
||||
const content = Array.isArray(returned) ? returned : returned.content
|
||||
const meta = Array.isArray(returned) ? undefined : returned.meta
|
||||
return { content, isError: false, ...meta !== undefined ? { meta } : {} }
|
||||
} catch (error: unknown) {
|
||||
return toolErrorResult(error)
|
||||
}
|
||||
},
|
||||
)
|
||||
const deferredContexts = this.deferredContexts.get(exec)
|
||||
/* v8 ignore next -- dispatch only receives executions minted by this registry's prepare stage */
|
||||
if (deferredContexts === undefined) throw new Error('tool registry scheduler invariant violated: unprepared execution')
|
||||
const resultWithDeferredContexts: ToolExecutionResult = deferredContexts.length === 0
|
||||
? result
|
||||
: {
|
||||
...result,
|
||||
additionalContexts: [
|
||||
...deferredContexts,
|
||||
...result.additionalContexts ?? [],
|
||||
],
|
||||
}
|
||||
},
|
||||
)
|
||||
const resultWithDeferredContexts: ToolExecutionResult = deferredContexts.length === 0
|
||||
? result
|
||||
: {
|
||||
...result,
|
||||
additionalContexts: [
|
||||
...deferredContexts,
|
||||
...result.additionalContexts ?? [],
|
||||
],
|
||||
}
|
||||
return await this.postExecute(exec, resultWithDeferredContexts)
|
||||
return { kind: 'post-result', result: resultWithDeferredContexts }
|
||||
} catch (error: unknown) {
|
||||
return { kind: 'final-result', result: toolErrorResult(error) }
|
||||
}
|
||||
}
|
||||
|
||||
/** Notify final-result observers without giving them a mutation/error channel into the outcome. */
|
||||
/**
|
||||
* Run ordered post-execute, then materialize and notify the final outcome.
|
||||
* @param exec - the prepared execution.
|
||||
* @param result - dispatch/pre result that still needs post-execute.
|
||||
* @returns the materialized final result.
|
||||
* @internal
|
||||
*/
|
||||
private async finalizeScheduledExecution(exec: ToolRunContext, result: ToolExecutionResult): Promise<ToolExecutionResult> {
|
||||
try {
|
||||
return this.finishScheduledExecution(exec, await this.postExecute(exec, result))
|
||||
} catch (error: unknown) {
|
||||
return this.finishScheduledExecution(exec, toolErrorResult(error))
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Materialize and notify a final result that must bypass post-execute.
|
||||
* @param exec - the prepared execution.
|
||||
* @param result - final result.
|
||||
* @returns the materialized final result.
|
||||
* @internal
|
||||
*/
|
||||
private finishScheduledExecution(exec: ToolRunContext, result: ToolExecutionResult): ToolExecutionResult {
|
||||
let finalResult: ToolExecutionResult
|
||||
try {
|
||||
finalResult = this.materializeFinalResult(result)
|
||||
} catch (error: unknown) {
|
||||
finalResult = this.materializeFinalResult(toolErrorResult(error))
|
||||
}
|
||||
this.notifyResult(exec, finalResult)
|
||||
return finalResult
|
||||
}
|
||||
|
||||
/** Notify observers without exposing a mutation or error channel into the outcome. */
|
||||
private notifyResult(exec: ToolExecution, result: ToolExecutionResult): void {
|
||||
// The pipeline is over: freeze the remaining mutable signal slot so every
|
||||
// observer sees the SAME WeakMap-keyable execution without a mutation race.
|
||||
// Freeze the remaining mutable signal slot before observers receive the
|
||||
// shared WeakMap-keyable execution object.
|
||||
Object.freeze(exec)
|
||||
const callbacks = this.ctx.events.dispatch('emit', [
|
||||
scopeTarget(this, exec.agent), 'tools/result', exec, result,
|
||||
|
||||
@@ -283,6 +283,14 @@ export interface DefineToolOptions<S extends SchemaSpec> {
|
||||
* is never sent to the model.
|
||||
*/
|
||||
readonly timeoutMs?: number
|
||||
/**
|
||||
* Optional pure synchronous classifier for sibling overlap. It receives typed
|
||||
* arguments after soft validation; invalid input returns `false` without
|
||||
* invoking it. See {@link ToolDefinition.isConcurrencySafe}.
|
||||
* @param args - typed validated arguments.
|
||||
* @returns whether this call may join a parallel group.
|
||||
*/
|
||||
isConcurrencySafe?(args: InferArgs<S>): boolean
|
||||
/**
|
||||
* Tool execution function. `args` is typed as {@link InferArgs<S>} — zero
|
||||
* casts needed. Returns either a bare {@link ContentBlock}`[]` (model-facing
|
||||
@@ -315,7 +323,7 @@ export interface DefineToolOptions<S extends SchemaSpec> {
|
||||
* @param options - the tool's name, description, typed parameter schema,
|
||||
* execute body, and optional presenters.
|
||||
* @returns a registry-ready definition with strict execution validation and
|
||||
* soft presenter validation for replay compatibility.
|
||||
* soft presenter and classifier validation for replay compatibility.
|
||||
*/
|
||||
export function defineTool<S extends SchemaSpec>(options: DefineToolOptions<S>): ToolDefinition {
|
||||
// Object-literal execute methods don't use `this`; the reference is safe.
|
||||
@@ -325,6 +333,8 @@ export function defineTool<S extends SchemaSpec>(options: DefineToolOptions<S>):
|
||||
const userPresentCall = options.presentCall
|
||||
// eslint-disable-next-line @typescript-eslint/unbound-method
|
||||
const userPresentResult = options.presentResult
|
||||
// eslint-disable-next-line @typescript-eslint/unbound-method
|
||||
const userIsConcurrencySafe = options.isConcurrencySafe
|
||||
if (options.timeoutMs !== undefined && (!Number.isFinite(options.timeoutMs) || options.timeoutMs <= 0)) {
|
||||
throw new Error(`defineTool(${options.name}): timeoutMs must be a positive finite number`)
|
||||
}
|
||||
@@ -359,5 +369,12 @@ export function defineTool<S extends SchemaSpec>(options: DefineToolOptions<S>):
|
||||
return userPresentResult(args as InferArgs<S>, result)
|
||||
}
|
||||
}
|
||||
// Invalid arguments fail closed without invoking the typed classifier.
|
||||
if (userIsConcurrencySafe) {
|
||||
tool.isConcurrencySafe = (args: unknown): boolean => {
|
||||
if (validateArgs(options.parameters, args).length > 0) return false
|
||||
return userIsConcurrencySafe(args as InferArgs<S>)
|
||||
}
|
||||
}
|
||||
return tool
|
||||
}
|
||||
|
||||
136
packages/core/tools/tests/execution-mode.spec.ts
Normal file
136
packages/core/tools/tests/execution-mode.spec.ts
Normal file
@@ -0,0 +1,136 @@
|
||||
/** Covers fail-closed per-call classification and model-schema isolation. */
|
||||
|
||||
import { describe, expect, expectTypeOf, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, {
|
||||
defineTool,
|
||||
type ToolDefinition,
|
||||
type ToolExecutionInput,
|
||||
type ToolExecutionMode,
|
||||
} from '@deepseek-ai/dsh-tools'
|
||||
|
||||
async function setup() {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
return ctx
|
||||
}
|
||||
|
||||
function exec(name: string, args: unknown): ToolExecutionInput {
|
||||
return { callId: CallId('c1'), name, arguments: args }
|
||||
}
|
||||
|
||||
describe('ToolRegistry.executionMode', () => {
|
||||
it('returns parallel only for an explicit true classifier', async () => {
|
||||
const ctx = await setup()
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'safe',
|
||||
description: 'parallel-safe',
|
||||
parameters: {},
|
||||
isConcurrencySafe: () => true,
|
||||
async execute() { return [] },
|
||||
}))
|
||||
expect(ctx.tools.executionMode(exec('safe', {}))).toEqual({ kind: 'parallel' })
|
||||
})
|
||||
|
||||
it('defaults to exclusive for a tool with no isConcurrencySafe declaration', async () => {
|
||||
const ctx = await setup()
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'plain',
|
||||
description: 'no declaration',
|
||||
parameters: {},
|
||||
async execute() { return [] },
|
||||
}))
|
||||
expect(ctx.tools.executionMode(exec('plain', {}))).toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
it('returns exclusive for an unknown tool', async () => {
|
||||
const ctx = await setup()
|
||||
expect(ctx.tools.executionMode(exec('nonexistent', {}))).toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
it('returns exclusive when the classifier returns false for these args', async () => {
|
||||
const ctx = await setup()
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'rw',
|
||||
description: 'read or write',
|
||||
parameters: { mode: { type: 'string', required: true } },
|
||||
isConcurrencySafe: args => args.mode === 'read',
|
||||
async execute() { return [] },
|
||||
}))
|
||||
expect(ctx.tools.executionMode(exec('rw', { mode: 'read' }))).toEqual({ kind: 'parallel' })
|
||||
expect(ctx.tools.executionMode(exec('rw', { mode: 'write' }))).toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
it('classifies invalid defineTool arguments as exclusive without throwing', async () => {
|
||||
const ctx = await setup()
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'needs-mode',
|
||||
description: 'requires mode',
|
||||
parameters: { mode: { type: 'string', required: true } },
|
||||
isConcurrencySafe: () => true,
|
||||
async execute() { return [] },
|
||||
}))
|
||||
expect(ctx.tools.executionMode(exec('needs-mode', {}))).toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
it('treats a throwing raw classifier as exclusive', async () => {
|
||||
const ctx = await setup()
|
||||
const raw: ToolDefinition = {
|
||||
name: 'thrower',
|
||||
description: 'classifier throws',
|
||||
parameters: { type: 'object', properties: {} },
|
||||
isConcurrencySafe() { throw new Error('boom') },
|
||||
async execute() { return [] },
|
||||
}
|
||||
ctx.tools.register(raw)
|
||||
expect(ctx.tools.executionMode(exec('thrower', {}))).toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
it('treats a truthy non-boolean raw result as exclusive', async () => {
|
||||
const ctx = await setup()
|
||||
const raw = {
|
||||
name: 'truthy',
|
||||
description: 'classifier returns a truthy string',
|
||||
parameters: { type: 'object', properties: {} },
|
||||
isConcurrencySafe() { return 'yes' },
|
||||
async execute() { return [] },
|
||||
} as unknown as ToolDefinition
|
||||
ctx.tools.register(raw)
|
||||
expect(ctx.tools.executionMode(exec('truthy', {}))).toEqual({ kind: 'exclusive' })
|
||||
})
|
||||
|
||||
it('passes parsed arguments directly to a raw definition', async () => {
|
||||
const ctx = await setup()
|
||||
let seen: unknown
|
||||
ctx.tools.register({
|
||||
name: 'raw-safe',
|
||||
description: 'raw',
|
||||
parameters: { type: 'object', properties: {} },
|
||||
isConcurrencySafe(args) { seen = args; return true },
|
||||
async execute() { return [] },
|
||||
})
|
||||
expect(ctx.tools.executionMode(exec('raw-safe', { anything: 1 }))).toEqual({ kind: 'parallel' })
|
||||
expect(seen).toEqual({ anything: 1 })
|
||||
})
|
||||
|
||||
it('isConcurrencySafe never reaches the model-facing schemas() projection', async () => {
|
||||
const ctx = await setup()
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'safe',
|
||||
description: 'parallel-safe',
|
||||
parameters: { x: { type: 'string', required: true } },
|
||||
isConcurrencySafe: () => true,
|
||||
async execute() { return [] },
|
||||
}))
|
||||
const schema = ctx.tools.schemas()[0] as unknown as Record<string, unknown>
|
||||
expect(Object.keys(schema).sort()).toEqual(['description', 'name', 'parameters'])
|
||||
expect(schema.isConcurrencySafe).toBeUndefined()
|
||||
})
|
||||
|
||||
it('ToolExecutionMode is the object-tagged union', () => {
|
||||
expectTypeOf<ToolExecutionMode>().toEqualTypeOf<{ kind: 'parallel' } | { kind: 'exclusive' }>()
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user