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

Merge origin/master into codex/send-one-turn

Browse Source
This commit is contained in:
pku-xht
2026-07-20 11:52:30 +08:00
parent 915c727e14 023bed1a83
commit 0891465ee0
1250 changed files with 64146 additions and 16415 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

6
packages/core/agent-loop/src/constants.ts Normal file
View 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

214
packages/core/agent-loop/src/index.ts
View File

@@ -12,9 +12,9 @@ import { createScope } from '@deepseek-ai/dsh-scope'
import type { Scope } from '@deepseek-ai/dsh-scope'
import { agentEvents } from '@deepseek-ai/dsh-agent'
import type {
Agent,
AgentFactory,
AgentHandle,
AgentId,
AgentOptions,
CreateAgentOptions,
ResumeAgentOptions,
@@ -32,8 +32,7 @@ import {
ReactLoopAgent,
} from './agent.ts'
import type { PreparedReactLoopAgent } from './agent.ts'
export { ReactLoopAgent } from './agent.ts'
import { DEFAULT_MAX_PARALLEL_TOOL_CALLS } from './constants.ts'
/** Fiber states that cannot own or serve a new lifecycle. */
const INACTIVE_STATES: ReadonlySet<FiberState> = new Set([
@@ -42,10 +41,21 @@ const INACTIVE_STATES: ReadonlySet<FiberState> = new Set([
FiberState.FAILED,
])
/** Render an arbitrary thrown value without letting coercion escape containment. */
function renderThrown(value: unknown): string {
try {
return String(value)
} catch {
return '<unrenderable thrown value>'
}
}
/** Factory-level ownership of every preparing or live transaction. */
class FactoryOwnership {
private accepting = true
private readonly inactive = Promise.withResolvers<void>()
private transactions = new Set<AgentCreationTransaction>()
private startupTasks = new Set<Promise<void>>()
constructor(private readonly fiber: Context['fiber']) {}
@@ -58,21 +68,44 @@ class FactoryOwnership {
return () => { this.transactions.delete(transaction) }
}
/** Join config startup work that begins before an agent transaction exists. */
trackStartup(task: Promise<void>): void {
this.startupTasks.add(task)
const forget = () => { this.startupTasks.delete(task) }
void task.then(forget, forget)
}
/** Resolve `task`, or stop waiting when factory teardown begins. */
async waitWhileActive(task: Promise<void>): Promise<void> {
await Promise.race([task, this.inactive.promise])
}
async dispose(): Promise<void> {
this.accepting = false
this.inactive.resolve()
const reason = new Error('agent loop is not active')
await Promise.all(
[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
)
await Promise.all([
...[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
...this.startupTasks,
])
}
}
/** Build the public cancellation error while preserving a caller-supplied cause. */
function signalAbortError(id: AgentId, signal: AbortSignal): Error {
function signalAbortError(id: SessionId, signal: AbortSignal): Error {
if (signal.reason instanceof Error) return signal.reason
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
@@ -105,7 +138,7 @@ class AgentCreationTransaction {
private readonly loopCtx: Context,
private readonly ownerCtx: Context,
private readonly ownership: FactoryOwnership,
readonly id: AgentId,
readonly id: SessionId,
signal?: AbortSignal,
) {
ownerCtx.fiber.assertActive()
@@ -163,13 +196,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)
@@ -227,7 +260,7 @@ class AgentCreationTransaction {
this.publishing = true
try {
this.detachSession = agent.ctx.sessions.enter(session)
this.detachAgent = this.loopCtx.agents.enter(agent)
this.detachAgent = this.loopCtx.agents.enter(agent, this.ownerAgent)
agent.ctx.sessions.announce(session)
this.assertActive()
@@ -316,14 +349,35 @@ declare module 'cordis' {
interface Context {
agentLoop: AgentLoop
}
interface Events {
/**
* A declarative agent entry failed before it could publish a live agent.
* Consumers that buffer work for the configured identity use this
* transient signal to reject that work instead of waiting forever. Normal
* factory teardown suppresses failures from the cancelled startup attempt.
* @param sessionId - exact shared agent/session identity that failed startup.
* @param error - persistence, setup, or publication failure.
* @mode emit
*/
'agent-loop/config-start-failed'(sessionId: SessionId, error: unknown): void
}
}
/** 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 & {
/** Registry identity for the live agent. */
id: AgentId
/** Stable config label used in logs and as the fresh combined-id prefix. */
id: string
/** Optional stable identity; remounts resume its materialized history, while first use creates it fresh. */
sessionId?: SessionId
/** Optional workspace for a fresh session. */
cwd?: string
/** Persisted session to resume instead of creating a fresh session. */
@@ -331,14 +385,35 @@ export interface Config {
})[]
}
/** Concrete ReactLoopAgent factory and driver service. */
/** Reject self-contained identity conflicts before any configured agent starts. */
function validateConfiguredAgents(agents: Config['agents']): void {
const exactIdentities = new Map<SessionId, string>()
for (const { id, sessionId, resumeSessionId } of agents) {
const hasResumeId = resumeSessionId !== undefined && resumeSessionId !== ''
if (sessionId !== undefined && hasResumeId) {
throw new Error(`agent "${id}": sessionId and resumeSessionId are mutually exclusive`)
}
const exactIdentity = hasResumeId ? resumeSessionId : sessionId
if (exactIdentity === undefined) continue
const firstId = exactIdentities.get(exactIdentity)
if (firstId !== undefined) {
throw new Error(`agents "${firstId}" and "${id}" use duplicate exact session identity "${exactIdentity}"`)
}
exactIdentities.set(exactIdentity, id)
}
}
/** Concrete agent factory and driver service. */
export class AgentLoop extends Service implements AgentFactory {
static inject = ['agents', 'sessions', 'llm', 'tools', 'systemPrompt']
/** 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),
provider: z.string(),
model: z.string(),
cwd: z.string(),
resumeSessionId: z.string(),
@@ -346,31 +421,45 @@ 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()')
ctx.effect(() => ctx.agents.setFactory(this), 'agentLoop.setFactory()')
ctx.systemPrompt.variable('provider', context => context.agent?.options.provider)
ctx.systemPrompt.variable('model', context => context.agent?.options.model)
ctx.systemPrompt.variable('cwd', context => context.agent?.session.header.cwd)
for (const { id, cwd, resumeSessionId, ...options } of config.agents) {
for (const { id, sessionId, cwd, resumeSessionId, ...options } of config.agents) {
const meta = cwd === undefined ? {} : { cwd }
if (resumeSessionId === undefined || resumeSessionId === '') {
this.create(id, options, cwd === undefined ? {} : { cwd })
const configuredId = sessionId ?? SessionId(`${id}-session-${randomUUID()}`)
const persistence = sessionId === undefined ? undefined : ctx.get('sessionPersistence')
if (persistence === undefined) {
this.create(configuredId, options, meta)
} else {
const startup = this.restoreOrCreateConfigured(ctx, persistence, configuredId, options, meta).catch((error: unknown) => {
this.reportConfiguredStartupFailure(id, 'restore', configuredId, error)
})
this.ownership.trackStartup(startup)
}
continue
}
ctx.effect(() => {
const fiber = ctx.inject(['sessionPersistence'], (childCtx: Context) => {
void this.resumeWith(ctx, childCtx.sessionPersistence, {
agentId: id,
resumeSessionId,
agentOptions: options,
}).catch((error: unknown) => {
ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`)
this.reportConfiguredStartupFailure(id, 'resume', resumeSessionId, error)
})
})
return fiber.dispose
@@ -378,21 +467,84 @@ export class AgentLoop extends Service implements AgentFactory {
}
}
/** Report a contained declarative-start failure to identity-bound consumers. */
private reportConfiguredStartupFailure(
configId: string,
action: 'restore' | 'resume',
sessionId: SessionId,
error: unknown,
): void {
if (!this.ownership.isActive()) return
this.ctx.logger.warn(`agent "${configId}": config-driven ${action} of "${sessionId}" failed: ${renderThrown(error)}`)
const args: unknown[] = ['agent-loop/config-start-failed', sessionId, error]
for (const callback of this.ctx.events.dispatch('emit', args)) {
try {
const returned: unknown = callback(...args)
void Promise.resolve(returned).catch((listenerError: unknown) => {
this.ctx.logger.warn(`agent "${configId}": config-start-failed listener rejected: ${renderThrown(listenerError)}`)
})
} catch (listenerError: unknown) {
this.ctx.logger.warn(`agent "${configId}": config-start-failed listener threw: ${renderThrown(listenerError)}`)
}
}
}
/** Restore a materialized exact config identity on remount, or create it on first use. */
private async restoreOrCreateConfigured(
ownerCtx: Context,
persistence: SessionPersistence,
sessionId: SessionId,
agentOptions: AgentOptions,
meta: Pick<SessionHeader, 'cwd'>,
): Promise<void> {
await this.waitForDrainingConfiguredIdentity(ownerCtx, sessionId)
if (!this.ownership.isActive()) return
const exists = (await persistence.list()).some(header => header.id === sessionId)
if (!this.ownership.isActive()) return
if (exists) {
await this.resumeWith(ownerCtx, persistence, { resumeSessionId: sessionId, agentOptions })
return
}
this.create(sessionId, agentOptions, meta)
}
/** Wait for an already-disposed same-id lifecycle to finish registry teardown. */
private async waitForDrainingConfiguredIdentity(ownerCtx: Context, sessionId: SessionId): Promise<void> {
const current = ownerCtx.agents.get(sessionId)
if (current?.status !== 'disposed') return
const released = Promise.withResolvers<void>()
const checkReleased = (): void => {
if (ownerCtx.agents.get(sessionId) === undefined && ownerCtx.sessions.get(sessionId) === undefined) {
released.resolve()
}
}
const disposeAgentListener = ownerCtx.on('agent/disposed', checkReleased)
const disposeSessionListener = ownerCtx.on('session/disposed', checkReleased)
try {
checkReleased()
await this.ownership.waitWhileActive(released.promise)
} finally {
disposeAgentListener()
disposeSessionListener()
}
}
/**
* Create an agent on a fresh per-run session, owned by the accessing fiber.
* Constructor-driven config calls use the loop fiber itself.
* @param id - agent registry id.
* Create an agent and session under one caller-supplied identity, owned by
* the accessing fiber. Constructor-driven config calls mint a fresh combined
* id before entering this boundary.
* @param id - shared agent/session identity.
* @param options - concrete loop options.
* @param meta - optional fresh-session workspace metadata.
* @returns the published running agent.
*/
create(id: AgentId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): ReactLoopAgent {
create(id: SessionId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): Agent {
const loopCtx = this.runtime.ctx
const transaction = new AgentCreationTransaction(loopCtx, this.ctx, this.ownership, id)
try {
const sessionId = SessionId(`${id}-session-${randomUUID()}`)
const session = loopCtx.sessions.prepare(sessionId, { meta })
const agent = transaction.prepare(options, session)
const session = loopCtx.sessions.prepare(id, { meta })
const agent = transaction.prepare(options, session, this.maxParallelToolCalls)
transaction.publish('startup')
return agent
} catch (error: unknown) {
@@ -410,11 +562,12 @@ 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,
this.ownership,
options.agentId,
options.sessionId,
options.signal,
)
try {
@@ -422,7 +575,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')
@@ -454,11 +607,12 @@ 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,
this.ownership,
options.agentId,
options.resumeSessionId,
options.signal,
)
try {
@@ -473,7 +627,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')

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

@@ -1,15 +1,16 @@
/**
* Drives one agent across queued durable turns. Turn failures are contained so
* later work can run; the session log, not this driver, owns conversation state.
* See docs/rfc/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md.
* See .agents/notes/implemented/architecture/2026-06-18-agent-lifecycle-and-ownership-seams.md.
* @module dsh-agent-loop/loop
*/
import type { Context } from 'cordis'
import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
import { isDeepStrictEqual } from 'node:util'
import { BlockAssembler, HarnessError, assertNever, deepFreeze, isLlmAdapterFailure } from '@deepseek-ai/dsh-llm'
import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision, RequestError, RequestErrorDecision } from '@deepseek-ai/dsh-agent'
import { canonicalHeader } from '@deepseek-ai/dsh-session'
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
@@ -17,27 +18,32 @@ 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 type { ReactLoopAgent } from './agent.ts'
import { executeToolCalls } from './tool-calls.ts'
import type { Inbox } from './inbox.ts'
/** An Error with an optional machine-readable code (e.g., from LlmError or a throwing plugin). */
type CodedError = Error & { code?: string }
/** Normalize thrown values while preserving an existing error code. */
function toError(error: unknown): CodedError {
function toError(error: unknown): RequestError {
return error instanceof Error ? error : new HarnessError(String(error), 'UNKNOWN', { cause: error })
}
/** Distinguishes final model-request failures from failures in later step processing. */
class TerminalModelRequestFailure extends Error {
constructor(readonly requestError: RequestError) {
super(requestError.message, { cause: requestError })
this.name = 'TerminalModelRequestFailure'
}
}
/** Convert terminal failure finishes into step errors; unknown extensible finishes remain successful. */
function finishError(finish: FinishReason): CodedError | undefined {
function finishError(finish: FinishReason): RequestError | undefined {
switch (finish.kind) {
case 'error': {
const error: CodedError = new Error(finish.message)
const error: RequestError = new Error(finish.message)
if (finish.code !== undefined) error.code = finish.code
return error
}
case 'aborted': {
const error: CodedError = new Error('model stream aborted')
const error: RequestError = new Error('model stream aborted')
error.code = 'ABORTED'
return error
}
@@ -51,7 +57,7 @@ function finishError(finish: FinishReason): CodedError | undefined {
* Build the `{ message, code? }` part of an error payload, omitting the
* `code` key entirely when absent (exactOptionalPropertyTypes-correct).
*/
function errorData(err: CodedError): { message: string; code?: string } {
function errorData(err: RequestError): { message: string; code?: string } {
return { message: err.message, ...typeof err.code === 'string' ? { code: err.code } : {} }
}
@@ -72,6 +78,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. */
@@ -85,16 +93,23 @@ export interface LoopHandle {
clearCancel(): void
/** Settle idle waiters when pre-running cancellation skips a turn, without emitting `agent/status`. */
settleIdle(): void
/** Run an active tool-call batch, accepting post-tool context into the FIFO drained before settlement. */
readonly withToolBatch: <T>(run: (acceptContext: (context: HookContext) => void) => Promise<T>) => Promise<T>
}
/**
* Drive queued messages as independent durable turns until disposal. Plugin failures end the
* current turn without terminating the driver.
* @param ctx - the plugin context the loop reaches events (agent/…, session/flush) and services (systemPrompt, llm, tools) through.
* @param agent - the agent this invocation drives for its whole lifetime (its inbox, session, and options).
* Drive queued messages as independent durable turns until disposal. Plugin
* failures end the current turn without terminating the driver. The caller establishes the
* `ctx.agents.withInitiator()` boundary before entry; package-private
* orchestration recovers that exact Agent and captures its Session locally.
* @param ctx - the plugin context the loop reaches its initiating Agent,
* events (agent/…, session/flush), and services (systemPrompt, llm, tools)
* through.
* @param handle - the bridge to the agent's mutable state: status/abort setters plus the disposal and cancel-marker reads.
* @throws when no initiating Agent is active.
*/
export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopHandle): Promise<void> {
export async function runLoop(ctx: Context, handle: LoopHandle): Promise<void> {
const agent = ctx.agents.requireInitiator()
// Per-instance prefix and request-header state; conversation history remains in the session log.
const transmission = createTransmissionLog()
@@ -133,7 +148,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
const turn = lastTurnNumber(session) + 1
let terminalStopped = false
try {
terminalStopped = await runTurn(ctx, events, agent, handle, turn, transmission)
terminalStopped = await runTurn(ctx, events, handle, turn, transmission)
} catch (error: unknown) {
// Pre-turn failure has no durable boundary to close; report it without appending outside a turn.
const err = toError(error)
@@ -156,9 +171,17 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
}
async function runTurn(
ctx: Context, events: AgentEventDispatch, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
ctx: Context, events: AgentEventDispatch, handle: LoopHandle, turn: number, transmission: TransmissionLog,
): Promise<boolean> {
const agent = ctx.agents.requireInitiator()
const { session } = agent
const drainSteering = (): boolean => {
const messages = handle.inbox.drainSteering()
for (const message of messages) {
session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
}
return messages.length > 0
}
// Claim one queued message before opening its turn, but append it only after `turn/start`.
const message = handle.inbox.dequeueQueued()
@@ -168,6 +191,7 @@ async function runTurn(
let reason: TurnEndReason = { kind: 'completed' }
let step = 0
let requestRetryAttempt = 0
let stepOpen = false
let errorReported = false
let terminalStopped = false
@@ -180,7 +204,7 @@ async function runTurn(
}
// Record the durable turn failure once and contain the live error notification.
const failTurn = (err: CodedError): void => {
const failTurn = (err: RequestError): void => {
if (errorReported) return
errorReported = true
reason = { kind: 'error', step, ...errorData(err) }
@@ -237,7 +261,7 @@ async function runTurn(
// Steering from the previous round's continuation listeners joins before
// the request.
drainSteering(agent, handle.inbox, turn)
drainSteering()
// The step's AbortController exists BEFORE any async pre-step work so a
// dispose() or cancel() — in a synchronous turn-start listener or an
@@ -247,7 +271,7 @@ async function runTurn(
const abort = new AbortController()
handle.setAbort(abort)
// Assemble once before pre-step so pressure checks and the request share the same prompt.
// Assemble once before pre-step so listener work and the request share one prompt value.
const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent))
const fullSystemPrompt = renderPrompt(assembly)
@@ -258,9 +282,9 @@ async function runTurn(
break
}
// Compose the request-only prefix once per loop instance before pressure
// checks. It precedes all derived history and is recorded only in the
// request header, not as session history.
// Compose the request-only prefix once per loop instance before the first
// request boundary. It precedes all derived history and is recorded only
// in the request header, not as session history.
if (transmission.sessionPrefix === undefined) {
const emptyPrefix: Message[] = deepFreeze([])
const composed = await events.waterfall(
@@ -277,8 +301,8 @@ async function runTurn(
transmission.sessionPrefix = deepFreeze(structuredClone(composed))
}
// Await surface mutations outside the step; pressure checks receive the pending prefix.
await events.serial('agent/pre-step', turn, step, fullSystemPrompt, transmission.sessionPrefix, abort.signal)
// Await surface mutations outside the step before snapshotting history.
await events.serial('agent/pre-step', turn, step, abort.signal)
// Interruption landing during the pre-step seam: do not open an empty step.
if (handle.isCancelled() || handle.isDisposed()) {
@@ -308,14 +332,69 @@ async function runTurn(
break
}
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
let stepOutcome:
| { hadToolCalls: boolean; finish: FinishReason }
| { requestError: RequestError }
| { error: RequestError }
try {
stepOutcome = await runStep(
ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
ctx, events, handle, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
} finally {
if (error instanceof TerminalModelRequestFailure) {
stepOutcome = { requestError: error.requestError }
} else {
stepOutcome = { error: toError(error) }
}
}
if ('requestError' in stepOutcome) {
// Recovery observes a balanced failed step and the original provider
// error while the failed step's signal remains the active owner.
closeStep()
if (handle.isDisposed() || abort.signal.aborted) {
handle.setAbort(undefined)
reason = handle.isDisposed()
? { kind: 'disposed' }
: { kind: 'aborted', reason: String(abort.signal.reason) }
break
}
const defaultDecision: RequestErrorDecision = { action: 'fail' }
let recoveryDecision: RequestErrorDecision = defaultDecision
try {
recoveryDecision = await events.waterfall(
'agent/request-error', turn, step, stepOutcome.requestError,
requestRetryAttempt, abort.signal,
() => Promise.resolve(defaultDecision),
)
} catch (recoveryError: unknown) {
ctx.logger.warn(
`agent "${agent.id}": request recovery failed at turn ${turn}, step ${step}: ${toError(recoveryError).message}`,
)
}
handle.setAbort(undefined)
// Cancellation and disposal always win over either a recovery decision
// or a recovery-listener failure.
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (handle.isDisposed() || abort.signal.aborted) {
reason = handle.isDisposed()
? { kind: 'disposed' }
: { kind: 'aborted', reason: String(abort.signal.reason) }
break
}
switch (recoveryDecision.action) {
case 'retry':
requestRetryAttempt += 1
continue
case 'fail':
failTurn(stepOutcome.requestError)
break
/* v8 ignore next -- closed-union exhaustiveness guard */
default:
assertNever(recoveryDecision, 'agent request-error decision')
}
break
}
if ('error' in stepOutcome) {
@@ -323,7 +402,9 @@ async function runTurn(
// runLoop re-enqueues it as a queued message, so an abort-then-steer
// starts a fresh turn instead of being silently consumed.
closeStep()
handle.setAbort(undefined)
const { error } = stepOutcome
/* v8 ignore next -- narrow race: disposal while non-request step work throws. */
if (handle.isDisposed()) {
reason = { kind: 'disposed' }
} else if (abort.signal.aborted) {
@@ -335,14 +416,47 @@ async function runTurn(
break
}
requestRetryAttempt = 0
// Preserve max-token completion unless a later disposal, abort, or error wins.
const stepReason = stepFinishReason(stepOutcome.finish)
if (stepReason) reason = stepReason
// Steering that arrived during streaming/tool execution.
const steered = drainSteering(agent, handle.inbox, turn)
const steered = drainSteering()
try {
await events.serial('agent/post-step', turn, step, abort.signal)
} catch (error: unknown) {
stepOutcome = { error: toError(error) }
}
if ('error' in stepOutcome) {
closeStep()
handle.setAbort(undefined)
/* v8 ignore next -- narrow race: disposal while a post-step listener throws. */
if (handle.isDisposed()) {
reason = { kind: 'disposed' }
} else if (abort.signal.aborted) {
/* v8 ignore next -- signal.reason always set by cancellation or disposal. */
reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
} else {
failTurn(stepOutcome.error)
}
break
}
if (handle.isDisposed() || abort.signal.aborted) {
reason = handle.isDisposed()
? { kind: 'disposed' }
: { kind: 'aborted', reason: String(abort.signal.reason) }
closeStep()
handle.setAbort(undefined)
break
}
closeStep()
handle.setAbort(undefined)
const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
let decision: ContinuationDecision
@@ -429,15 +543,6 @@ async function runTurn(
return terminalStopped
}
/** Drain the steering queue into the session. Returns whether any arrived. */
function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boolean {
const messages = inbox.drainSteering()
for (const message of messages) {
agent.session.append('steering/message', { turn, content: message.content, source: message.source }, { surfaceOp: 'append' })
}
return messages.length > 0
}
/**
* Run one committed step: transform call config, log the request header, build
* the request from the cached prefix plus the step-boundary snapshot, stream and
@@ -447,7 +552,7 @@ function drainSteering(agent: ReactLoopAgent, inbox: Inbox, turn: number): boole
async function runStep(
ctx: Context,
events: AgentEventDispatch,
agent: ReactLoopAgent,
handle: LoopHandle,
turn: number,
step: number,
assembly: PromptAssembly,
@@ -456,6 +561,7 @@ async function runStep(
transmission: TransmissionLog,
signal: AbortSignal,
): Promise<{ hadToolCalls: boolean; finish: FinishReason }> {
const agent = ctx.agents.requireInitiator()
const { session, options } = agent
// Seed the first request from agent options and later requests from the logged header;
@@ -463,12 +569,12 @@ async function runStep(
const seedConfig: LlmCallConfig = deepFreeze(structuredClone(transmission.loggedHeader
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- loggedHeader ⟹ a snapshot is in the log
? session.requestHeader()!.config
: { model: options.model ?? '' }))
: { provider: options.provider ?? '', model: options.model ?? '' }))
// Listener replacements are recorded in the request header before dispatch.
const config = await events.waterfall('agent/request', turn, step, seedConfig, () => Promise.resolve(seedConfig))
if (!config.model) {
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
if (!config.provider || !config.model) {
throw new Error(`agent "${agent.id}" has no provider/model: set AgentOptions.provider and AgentOptions.model or supply both via the agent/request waterfall`)
}
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
@@ -485,6 +591,7 @@ async function runStep(
// Freeze the logged header plus boundary snapshot; the prefix precedes derived history.
const request: GenerateOptions = deepFreeze({
provider: header.config.provider,
model: header.config.model,
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
...header.system !== undefined ? { system: header.system } : {},
@@ -499,98 +606,96 @@ async function runStep(
// --- Model call (streaming-first; raw chunks are the replay record) ---
const assembler = new BlockAssembler()
const chunkSeqs: number[] = []
for await (const chunk of ctx.llm.stream(request)) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
chunkSeqs.push(chunkEvent.seq)
assembler.push(chunk)
const stream = ctx.llm.stream(request)
try {
for await (const chunk of stream) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const chunkEvent = session.append('assistant/chunk', { turn, step, chunk })
chunkSeqs.push(chunkEvent.seq)
assembler.push(chunk)
}
} catch (error: unknown) {
if (isLlmAdapterFailure(stream, error)) throw new TerminalModelRequestFailure(error)
throw error
}
// Normalize failure finish chunks into the same path as thrown stream errors.
const stepError = finishError(assembler.finish)
if (stepError) throw stepError
if (stepError) throw new TerminalModelRequestFailure(stepError)
const recordAssistantMessage = (
assembledContent: ContentBlock[],
message: Message,
preserveReplayState = true,
): void => {
session.append(
'assistant/message',
{
turn,
step,
content: message.content,
provenance: assistantProvenance(
header.config,
assembler.replayState,
preserveReplayState && isDeepStrictEqual(message.content, assembledContent),
),
...assembler.usage === undefined ? {} : { usage: assembler.usage },
},
{ surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
)
}
// A rejected result still records the successful provider call without retaining rejected output.
const processStepResult = async (assembledContent: ContentBlock[], message: Message): Promise<Message> => {
try {
return await events.waterfall(
'agent/step-result', turn, step, message, () => Promise.resolve(message),
)
} catch (error: unknown) {
recordAssistantMessage(assembledContent, { ...message, content: [] }, false)
throw error
}
}
if (assembler.finish.kind === 'max-tokens') {
let message: Message = withoutToolCalls(assembler.message())
message = withoutToolCalls(await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message)))
const assembled = assembler.message()
const assembledContent = structuredClone(assembled.content)
let message: Message = withoutToolCalls(assembled)
message = withoutToolCalls(await processStepResult(assembledContent, message))
// Preserve usage even when max-token truncation produced no content.
if (message.content.length > 0 || assembler.usage) {
// The finish chunk guarantees non-empty provenance here.
session.append(
'assistant/message',
{ turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
{ surfaceOp: 'append', sourceEventSeqs: chunkSeqs },
)
}
recordAssistantMessage(assembledContent, message)
return { hadToolCalls: false, finish: assembler.finish }
}
// Record the post-waterfall message that tool dispatch uses.
let message: Message = assembler.message()
message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
const assembled = assembler.message()
const assembledContent = structuredClone(assembled.content)
let message: Message = assembled
message = await processStepResult(assembledContent, message)
// Empty messages exist only to carry usage; omit empty provenance.
if (message.content.length > 0 || assembler.usage) {
session.append(
'assistant/message',
{ turn, step, content: message.content, ...(assembler.usage ? { usage: assembler.usage } : {}) },
{ surfaceOp: 'append', ...(chunkSeqs.length > 0 ? { sourceEventSeqs: chunkSeqs } : {}) },
)
}
// Every successful call records its completion anchor, including explicit
// empty chunk provenance for a contentless, usage-less provider response.
recordAssistantMessage(assembledContent, message)
// 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')
// Buffer context until all results are appended to preserve call/result adjacency.
const pendingContext: HookContext[] = []
for (const call of toolCalls) {
/* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
const callEvent = session.append('tool/call', { turn, step, callId: call.id, name: call.name, arguments: call.arguments })
let parsedArguments: unknown
try {
parsedArguments = call.arguments ? JSON.parse(call.arguments) : {}
} catch {
parsedArguments = call.arguments
}
// TODO(pre-tool-input-rewrite): Keep logged history and live presentation aligned;
// see docs/rfc/proposed/feature/2026-06-30-pre-tool-input-rewrite.md.
const result = await ctx.tools.execute({
callId: call.id,
name: call.name,
arguments: parsedArguments,
agent,
signal,
})
session.append('tool/result', {
turn, step,
// Correlation comes from the immutable execution input; the result does
// not duplicate this authoritative transcript identity.
callId: call.id,
content: result.content,
isError: result.isError,
...result.error ? { error: result.error } : {},
// Persist tool-owned presentation data for replay.
...result.meta !== undefined ? { meta: result.meta } : {},
}, { surfaceOp: 'append', sourceEventSeqs: [callEvent.seq] })
pendingContext.push(...result.additionalContexts ?? [])
// The signal may flip while the tool is awaited.
/* v8 ignore start -- signal.reason default unreachable: cancel()/disposal always set it */
// eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted'))
/* v8 ignore stop */
}
if (toolCalls.length === 0) return { hadToolCalls: false, finish: assembler.finish }
return handle.withToolBatch(async (acceptContext) => {
await executeToolCalls(
ctx, turn, step, toolCalls, signal, handle.maxParallelToolCalls, acceptContext,
)
return { hadToolCalls: true, finish: assembler.finish }
})
}
// Append buffered context after the complete result batch.
for (const context of pendingContext) {
agent.inject(context.content, {
source: context.source,
...context.envelope !== undefined ? { envelope: context.envelope } : {},
...context.meta !== undefined ? { meta: context.meta } : {},
})
/** Build durable assistant provenance, dropping replay state after any content rewrite. */
function assistantProvenance(config: LlmCallConfig, replayState: unknown, contentUnchanged: boolean): NonNullable<Message['provenance']> {
return {
provider: config.provider,
model: config.model,
...contentUnchanged && replayState !== undefined ? { replayState } : {},
}
return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish }
}
function withoutToolCalls(message: Message): Message {

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

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

253
packages/core/agent-loop/src/tool-calls.ts Normal file
View File

@@ -0,0 +1,253 @@
/**
* 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 advertised call records a balanced `tool/call`/`tool/result` pair. Calls
* skipped after abort receive synthetic error results so replay stays valid.
* @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'
/** 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
}
/** One scheduler group outcome, including a drained cancellation. */
interface GroupOutcome {
consumed: number
aborted: boolean
}
/**
* Schedule one assistant step's tool calls by their live concurrency mode.
* Started calls receive ordered results. Abort drains them, records synthetic
* results for unstarted calls, and returns with the signal still aborted after
* accepting started-call context into the batch FIFO owned by the caller.
* The committed step's AgentLoop driver boundary supplies the initiating Agent
* that becomes each explicit {@link ToolExecutionInput.agent}.
*
* @param ctx - loop context that owns the tool registry and carries the initiating Agent.
* @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,
turn: number,
step: number,
toolCalls: ToolCallBlock[],
signal: AbortSignal,
maxParallel: number,
acceptContext: (context: HookContext) => void,
): Promise<void> {
const agent = ctx.agents.requireInitiator()
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]
const outcome = await runGroup(
ctx, turn, step, group, mode, signal, maxParallel, acceptContext,
)
next += outcome.consumed
if (outcome.aborted) {
for (const call of planned.slice(next)) appendSkippedToolCall(session, turn, step, call.block)
return
}
}
}
/** 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, records results for skipped calls, and
* returns an aborted outcome.
*/
async function runGroup(
ctx: Context,
turn: number,
step: number,
group: PlannedCall[],
mode: ToolExecutionMode['kind'],
signal: AbortSignal,
maxParallel: number,
acceptContext: (context: HookContext) => void,
): Promise<GroupOutcome> {
const { session } = ctx.agents.requireInitiator()
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.
if (signal.aborted) aborted = true
await fillPool()
}
if (aborted) {
// Started calls and accepted context settle first; every remaining model
// call then receives an ordered synthetic result before the turn aborts.
for (const call of group.slice(started)) appendSkippedToolCall(session, turn, step, call.block)
return { consumed: group.length, aborted: true }
}
/* 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 { consumed: started, aborted: false }
}
/** Append the durable call/result pair for a model call skipped after cancellation. */
function appendSkippedToolCall(session: Session, turn: number, step: number, block: ToolCallBlock): void {
const callSeq = appendToolCall(session, turn, step, block)
appendToolResult(session, turn, step, block, {
content: [{ type: 'text', text: 'Error: tool call skipped because the step was aborted before execution' }],
isError: true,
error: { name: 'AbortError', code: 'ABORTED' },
}, callSeq)
}
/** 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] })
}