Merge remote-tracking branch 'origin/master' into feat/telemetry-otel-plugin
Resolutions: regenerate the conflicted generated docs (cordis services catalog, event-producer-consumer, module-graph); take master's packages/README pair and re-insert the telemetry row on both sides; re-record the README and session-doc translation pairs.
This commit is contained in:
kingwl
@@ -1,6 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: 3acf4d3828291d5f318306f2652e0d920c695675
|
||||
README.zh.md: 11ff8318813b1abd096e1a3549d389cbba88f12b
|
||||
# pnpm run verify-translation-pairing --write packages/core/agent-loop/README.md
|
||||
README.md: 6ad982ffff7b73e16ee39f9c29da787e37547de4
|
||||
README.zh.md: c72df198774f0f8009cc5ab43932e69187757745
|
||||
|
||||
@@ -62,6 +62,8 @@ The driver owns one agent for its lifetime and runs inside `ctx.agents.withIniti
|
||||
|
||||
Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
|
||||
|
||||
After `agent/request` returns a provider/model call config, the loop asks `ctx.llm.prepareCall()` to validate any adapter-owned reasoning effort and materialize its configured default under the active turn signal. The prepared call retains the exact adapter registration across this asynchronous resolution, `request/header` logging, and terminal dispatch, so HMR cannot mix one adapter's capability result with another adapter's request. The effective config is logged before dispatch, so a listener can change effort between steps without hidden request drift. A route with no registered adapter preserves the proposed config so an `llm/stream` listener can own and short-circuit it; unhandled terminal dispatch still fails with `NO_ADAPTER`. A new loop instance restores the last effort only when its initial provider/model route exactly matches the logged route; a route change discards that opaque model-owned ID and resolves the new model independently.
|
||||
|
||||
Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, and immutable prior failures after the failed step closes. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. AgentLoop privately owns one cancellation holder whose explicit signal spans prompt policy, assembly, every step, model and tool work, recovery, continuation, and terminal stop; it retires the holder immediately before publishing `turn/end`, while the driver may remain `running` through the durability flush. An effective `cancel()` emits the typed runtime-only `user | parent` cause before clearing pending work and cooperatively aborting the holder; notification failures cannot veto cancellation, work queued by a notification observer is cleared, work queued by a later abort observer belongs to the next turn, and idle cancellation emits nothing. Durable `turn/end` remains coarse `aborted`; undispatched model tool calls receive synthetic `tool/call` and `ABORTED_BEFORE_DISPATCH` result pairs. Disposal wins terminal classification, and work that ignores the signal must settle before quiescence. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns the lifecycle and race contract. 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.
|
||||
|
||||
@@ -62,6 +62,8 @@ interface Config {
|
||||
|
||||
每次提供方调用成功结束时，都会恰好追加一个 `assistant/message` 完成锚点，包括无内容调用和以 `max-tokens` 结束的调用。成功的 `agent/step-result` 存储其转换后内容；被拒绝的结果会先记录空内容，再继续抛出原始失败。该锚点保留确切的 chunk 溯源（流没有 chunk 时为 `[]`），并在用量可用时保留用量；空内容不会进入派生消息历史。
|
||||
|
||||
在 `agent/request` 返回提供方／模型调用配置后，循环会调用 `ctx.llm.prepareCall()`，在活跃轮次信号的控制下校验由适配器持有的推理（reasoning）强度，并填入其配置默认值。准备完成的调用会在这次异步解析、`request/header` 日志记录和最终分派期间保留同一项确切的适配器注册，因此 HMR（热模块替换）不会把某个适配器的能力解析结果与另一适配器的请求混用。生效配置会在分派前写入日志，因此监听器可以在步骤之间更改推理强度，而不会产生未记录的请求变化。没有已注册适配器的路由会保留原定配置，使 `llm/stream` 监听器可以接管并短路该请求；最终分派仍会以 `NO_ADAPTER` 拒绝未得到处理的路由。新循环实例仅在初始提供方／模型路由与日志路由完全一致时恢复上次的推理强度；路由变化会丢弃由前一模型持有的不透明 ID，并单独解析新模型。
|
||||
|
||||
插件失败会结束当前轮次，而不是结束循环。只有最终适配器分发／迭代失败，以及带内的终止错误或中止结束原因，才进入 `agent/request-error`；中间件、结果处理、工具和 `agent/post-step` 仍属于普通轮次失败。失败步骤关闭后，恢复逻辑会接收确切的实时错误、不可变的提供方事实和不可变的先前失败。重试会在新的编号步骤中根据持久日志重建；成功会清除连续失败历史；耗尽后只在 `turn/end` 上记录一次结构化失败。AgentLoop 私下拥有一个取消持有者，其显式信号覆盖提示词策略、组装、每个步骤、模型与工具工作、恢复、continuation 和终止停止；它会在发布 `turn/end` 前立即退役该持有者，而驱动器可以在持久性 flush 期间继续保持 `running`。有效的 `cancel()` 会先发出仅存在于运行时的类型化 `user | parent` 原因，再清除待处理工作，并以协作方式中止该持有者；通知失败无法 veto 取消，通知观察方排队的工作会被清除，之后由中止观察方排队的工作属于下一轮次，空闲取消则不发出任何内容。持久 `turn/end` 仍使用粗粒度的 `aborted`；未分发的模型工具调用会收到合成的 `tool/call` 与 `ABORTED_BEFORE_DISPATCH` 结果对。释放会在终止分类中胜出；忽略信号的工作必须先结算，系统才能完全停稳。[显式取消决策](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md)规定生命周期与竞态契约。终止 continuation 的停止决定在轮次关闭和持久性 flush 期间始终具有权威性。
|
||||
|
||||
在步骤内，独占调用形成屏障；并行安全调用使用有界滚动池，并在启动前重新分类。只有分发／主体会重叠。策略、持久结果和结果上下文仍保持模型顺序。中止会停止新调用、drain 已启动的结果，然后在轮次通过普通中止路径关闭前，drain 已接纳的批次上下文。
|
||||
|
||||
@@ -7,7 +7,7 @@
|
||||
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import type { Context } from 'cordis'
|
||||
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock, FinishReason, GenerateOptions, LlmCallConfig, LlmFailure, Message, PreparedLlmCall } from '@deepseek-ai/dsh-llm'
|
||||
import { isDeepStrictEqual } from 'node:util'
|
||||
import { BlockAssembler, HarnessError, LlmError, assertNever, deepFreeze, errorChain, llmFailureOf, markAgentLoopRequest } from '@deepseek-ai/dsh-llm'
|
||||
import { agentEvents, agentInterruptReasonOf, assembleContextFor, AgentMessageId } from '@deepseek-ai/dsh-agent'
|
||||
@@ -661,19 +661,45 @@ async function runStep(
|
||||
|
||||
// Seed the first request from agent options and later requests from the logged header;
|
||||
// detach and freeze so listeners must return an attributable replacement.
|
||||
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
|
||||
: { provider: options.provider ?? '', model: options.model ?? '' }))
|
||||
const loggedConfig = session.requestHeader()?.config
|
||||
const initialProvider = options.provider ?? ''
|
||||
const initialModel = options.model ?? ''
|
||||
const initialConfig: LlmCallConfig = {
|
||||
provider: initialProvider,
|
||||
model: initialModel,
|
||||
...loggedConfig?.provider === initialProvider
|
||||
&& loggedConfig.model === initialModel
|
||||
&& loggedConfig.reasoningEffort !== undefined
|
||||
? { reasoningEffort: loggedConfig.reasoningEffort }
|
||||
: {},
|
||||
}
|
||||
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
|
||||
: initialConfig,
|
||||
))
|
||||
|
||||
// Listener replacements are recorded in the request header before dispatch.
|
||||
const config = await events.waterfall(
|
||||
const proposedConfig = await events.waterfall(
|
||||
'agent/request', turn, step, seedConfig, signal, () => Promise.resolve(seedConfig),
|
||||
)
|
||||
interruptionCheckpoint(signal)
|
||||
if (!config.provider || !config.model) {
|
||||
if (!proposedConfig.provider || !proposedConfig.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`)
|
||||
}
|
||||
let config: LlmCallConfig
|
||||
let preparedCall: PreparedLlmCall | undefined
|
||||
try {
|
||||
preparedCall = await ctx.llm.prepareCall(proposedConfig, signal)
|
||||
config = preparedCall.config
|
||||
} catch (error: unknown) {
|
||||
// A waterfall listener may own and short-circuit a route with no adapter.
|
||||
// Terminal dispatch still raises NO_ADAPTER when no listener handles it.
|
||||
if (!(error instanceof LlmError) || error.code !== 'NO_ADAPTER') throw error
|
||||
config = proposedConfig
|
||||
}
|
||||
interruptionCheckpoint(signal)
|
||||
|
||||
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion -- runTurn composes the prefix before every runStep call
|
||||
const sessionPrefix = transmission.sessionPrefix!
|
||||
@@ -691,6 +717,9 @@ async function runStep(
|
||||
const request: GenerateOptions = markAgentLoopRequest(deepFreeze({
|
||||
provider: header.config.provider,
|
||||
model: header.config.model,
|
||||
...header.config.reasoningEffort !== undefined
|
||||
? { reasoningEffort: header.config.reasoningEffort }
|
||||
: {},
|
||||
messages: [...header.messagePrefix ?? [], ...boundaryMessages],
|
||||
...header.system !== undefined ? { system: header.system } : {},
|
||||
...header.tools !== undefined ? { tools: header.tools } : {},
|
||||
@@ -704,7 +733,7 @@ async function runStep(
|
||||
// --- Model call (streaming-first; raw chunks are the replay record) ---
|
||||
const assembler = new BlockAssembler()
|
||||
const chunkSeqs: number[] = []
|
||||
const stream = ctx.llm.stream(request)
|
||||
const stream = preparedCall?.stream(request) ?? ctx.llm.stream(request)
|
||||
try {
|
||||
for await (const chunk of stream) {
|
||||
interruptionCheckpoint(signal)
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, LlmModelReasoningInfo, LlmResolvedModelInfo, StreamChunk } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, LlmAdapter } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
/** Helpers to write scripted responses tersely. */
|
||||
@@ -64,10 +64,25 @@ export function toolCallResponse(rawCallId: string, name: string, args: object,
|
||||
export class MockAdapter extends LlmAdapter {
|
||||
requests: GenerateOptions[] = []
|
||||
|
||||
constructor(private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang')[]) {
|
||||
constructor(
|
||||
private script: (StreamChunk[] | ((options: GenerateOptions) => StreamChunk[]) | 'hang')[],
|
||||
private readonly reasoning?: LlmModelReasoningInfo,
|
||||
) {
|
||||
super()
|
||||
}
|
||||
|
||||
override resolveModel(
|
||||
provider: string,
|
||||
model: string,
|
||||
): Promise<LlmResolvedModelInfo> {
|
||||
return Promise.resolve({
|
||||
provider,
|
||||
id: model,
|
||||
name: model,
|
||||
...this.reasoning === undefined ? {} : { reasoning: this.reasoning },
|
||||
})
|
||||
}
|
||||
|
||||
async * stream(options: GenerateOptions): AsyncIterable<StreamChunk> {
|
||||
this.requests.push(options)
|
||||
const entry = this.script.shift()
|
||||
|
||||
@@ -7,8 +7,8 @@
|
||||
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions } from '@deepseek-ai/dsh-llm'
|
||||
import LlmService, { LlmError, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import type { GenerateOptions, LlmModelReasoningInfo, LlmResolvedModelInfo } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry, { defineContentToolFixture } from '@deepseek-ai/dsh-tools'
|
||||
@@ -104,6 +104,168 @@ describe('request stability across the loop', () => {
|
||||
expectPrefixExtension(adapter.requests[0]!, adapter.requests[1]!)
|
||||
})
|
||||
|
||||
it('logs adapter defaults, supports per-turn effort changes, and restores the effective value', async () => {
|
||||
const reasoning = {
|
||||
efforts: [
|
||||
{ id: ReasoningEffortId('high'), name: 'High' },
|
||||
{ id: ReasoningEffortId('max'), name: 'Max' },
|
||||
],
|
||||
defaultEffort: ReasoningEffortId('high'),
|
||||
}
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')], reasoning)
|
||||
const ctx = await harness(adapter)
|
||||
const agent = ctx.agentLoop.create(SessionId('effort'), { provider: 'mock', model: 'mock' })
|
||||
ctx.on('agent/request', async (_agent, turn, _step, _config, _signal, next) => {
|
||||
const config = await next()
|
||||
return turn === 2 ? { ...config, reasoningEffort: ReasoningEffortId('max') } : config
|
||||
})
|
||||
|
||||
send(agent, 'first')
|
||||
await waitForIdle(ctx, agent)
|
||||
send(agent, 'second')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(adapter.requests.map(request => request.reasoningEffort)).toEqual([
|
||||
ReasoningEffortId('high'),
|
||||
ReasoningEffortId('max'),
|
||||
])
|
||||
const headers = agent.session.events.filter(event => event.type === 'request/header')
|
||||
expect(headers.map(event => event.data.header.config.reasoningEffort)).toEqual([
|
||||
ReasoningEffortId('high'),
|
||||
ReasoningEffortId('max'),
|
||||
])
|
||||
expect(headers.map(event => event.data.reason)).toEqual(['initial', 'change'])
|
||||
|
||||
const resumedAdapter = new MockAdapter([textResponse('three')], reasoning)
|
||||
const resumedCtx = await harness(resumedAdapter)
|
||||
const resumedHandle = await resumedCtx.agents.create({
|
||||
sessionId: SessionId('effort-resumed'),
|
||||
seed: structuredClone(agent.session.events),
|
||||
agentOptions: { provider: 'mock', model: 'mock' },
|
||||
})
|
||||
send(resumedHandle.agent, 'third')
|
||||
await waitForIdle(resumedCtx, resumedHandle.agent)
|
||||
|
||||
expect(resumedAdapter.requests[0]?.reasoningEffort).toBe(ReasoningEffortId('max'))
|
||||
const resumedHeaders = resumedHandle.agent.session.events.filter(event => event.type === 'request/header')
|
||||
expect(resumedHeaders.at(-1)?.data.header.config.reasoningEffort).toBe(ReasoningEffortId('max'))
|
||||
expect(resumedHeaders.at(-1)?.data.reason).toBe('resume')
|
||||
})
|
||||
|
||||
it('keeps exact-model resolution, request logging, and dispatch on one adapter registration', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: 'stable base' })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
const started = Promise.withResolvers<undefined>()
|
||||
const reasoning = Promise.withResolvers<LlmModelReasoningInfo>()
|
||||
const first = new class extends MockAdapter {
|
||||
override async resolveModel(
|
||||
provider: string,
|
||||
model: string,
|
||||
_signal?: AbortSignal,
|
||||
): Promise<LlmResolvedModelInfo> {
|
||||
started.resolve(undefined)
|
||||
return {
|
||||
provider,
|
||||
id: model,
|
||||
name: model,
|
||||
reasoning: await reasoning.promise,
|
||||
}
|
||||
}
|
||||
}([textResponse('first')])
|
||||
const second = new MockAdapter([textResponse('second')], {
|
||||
efforts: [{ id: ReasoningEffortId('max'), name: 'Max' }],
|
||||
defaultEffort: ReasoningEffortId('max'),
|
||||
})
|
||||
const disposeFirst = ctx.llm.registerAdapter(['mock'], first)
|
||||
const agent = ctx.agentLoop.create(SessionId('effort-hmr'), { provider: 'mock', model: 'mock' })
|
||||
|
||||
send(agent, 'go')
|
||||
await started.promise
|
||||
disposeFirst()
|
||||
ctx.llm.registerAdapter(['mock'], second)
|
||||
reasoning.resolve({
|
||||
efforts: [{ id: ReasoningEffortId('high'), name: 'High' }],
|
||||
defaultEffort: ReasoningEffortId('high'),
|
||||
})
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(first.requests.map(request => request.reasoningEffort)).toEqual([
|
||||
ReasoningEffortId('high'),
|
||||
])
|
||||
expect(second.requests).toHaveLength(0)
|
||||
const headers = agent.session.events.filter(event => event.type === 'request/header')
|
||||
expect(headers.at(-1)?.data.header.config.reasoningEffort).toBe(ReasoningEffortId('high'))
|
||||
})
|
||||
|
||||
it('aborts a blocked reasoning lookup before quiescent disposal completes', async () => {
|
||||
const started = Promise.withResolvers<AbortSignal>()
|
||||
const adapter = new class extends MockAdapter {
|
||||
override resolveModel(
|
||||
_provider: string,
|
||||
_model: string,
|
||||
signal?: AbortSignal,
|
||||
): Promise<never> {
|
||||
if (signal === undefined) return Promise.reject(new Error('missing reasoning signal'))
|
||||
started.resolve(signal)
|
||||
return new Promise((_resolve, reject) => {
|
||||
if (signal.aborted) {
|
||||
reject(signal.reason instanceof Error ? signal.reason : new Error('reasoning aborted'))
|
||||
return
|
||||
}
|
||||
signal.addEventListener('abort', () => {
|
||||
reject(signal.reason instanceof Error ? signal.reason : new Error('reasoning aborted'))
|
||||
}, { once: true })
|
||||
})
|
||||
}
|
||||
}([])
|
||||
const ctx = await harness(adapter)
|
||||
const handle = await ctx.agents.create({
|
||||
sessionId: SessionId('reasoning-dispose'),
|
||||
agentOptions: { provider: 'mock', model: 'mock' },
|
||||
})
|
||||
|
||||
send(handle.agent, 'go')
|
||||
const signal = await started.promise
|
||||
await handle.dispose()
|
||||
|
||||
expect(signal.aborted).toBe(true)
|
||||
expect(handle.agent.status).toBe('disposed')
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
expect(handle.agent.session.events.some(event => event.type === 'request/header')).toBe(false)
|
||||
})
|
||||
|
||||
it.each(['plain error', 'LLM error'] as const)(
|
||||
'does not swallow a %s from exact-model resolution',
|
||||
async (kind) => {
|
||||
const failure = kind === 'plain error'
|
||||
? new Error('reasoning metadata failed')
|
||||
: new LlmError('unsupported effort', 'UNSUPPORTED_REASONING_EFFORT')
|
||||
const adapter = new class extends MockAdapter {
|
||||
override resolveModel(): Promise<never> {
|
||||
return Promise.reject(failure)
|
||||
}
|
||||
}([])
|
||||
const ctx = await harness(adapter)
|
||||
const errors: Error[] = []
|
||||
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
|
||||
const agent = ctx.agentLoop.create(SessionId(`reasoning-${kind}`), {
|
||||
provider: 'mock',
|
||||
model: 'mock',
|
||||
})
|
||||
|
||||
send(agent, 'go')
|
||||
await waitForIdle(ctx, agent)
|
||||
|
||||
expect(errors).toContain(failure)
|
||||
expect(adapter.requests).toHaveLength(0)
|
||||
},
|
||||
)
|
||||
|
||||
it('a compaction replace rewrites the resend, and the log explains it', async () => {
|
||||
const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
|
||||
const ctx = await harness(adapter)
|
||||
@@ -301,6 +463,7 @@ describe('request stability across the loop', () => {
|
||||
const firstChunk = events.find(e => e.type === 'assistant/chunk' && e.seq > stepStart.seq)!
|
||||
const header = foldRequestHeader(events.slice(0, firstChunk.seq))!
|
||||
expect(request.model).toBe(header.config.model)
|
||||
expect(request.reasoningEffort).toBe(header.config.reasoningEffort)
|
||||
expect(request.system).toEqual(header.system)
|
||||
expect(structuredClone(request.tools ?? [])).toEqual(structuredClone(header.tools ?? []))
|
||||
expect(request.temperature).toBe(header.config.temperature)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: bbae91ff2f497208f1ce620e162c27968d666ac3
|
||||
README.zh.md: 95367b35a546c68491b9623daf45a54cd63f2731
|
||||
# pnpm run verify-translation-pairing --write packages/core/agent/README.md
|
||||
README.md: a65c53b3e4edf2031f286d7d172e73357c66ff1e
|
||||
README.zh.md: 05da8a0a0d3ad2ed879b72e20eae1c4efaf711c8
|
||||
|
||||
@@ -12,7 +12,7 @@ Tracks live agents and carries the initiating Agent through asynchronous driver
|
||||
|
||||
### Public API
|
||||
|
||||
The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-scope`, key = the agent) — register tools/sections/variables/listeners through it for that agent alone, all unwound on disposal. `agentEvents(ctx, agent)` is the fused dispatcher for ordinary agent-subject operations (carrier + injected subject in one move); its notification mode invokes every listener and contains both synchronous throws and returned-promise rejections. The registry lifecycle pair reuses one stable routing carrier. `assembleContextFor(agent)` builds the per-agent assembly context (`agent` + `scope` together). `installAgentLlmTarget(agentCtx, target)` snapshots a mutable provider/model selection during prompt assembly and applies that pair to both prompt variables and request routing for one step. `CreateAgentOptions.setup(agentCtx)` and `ResumeAgentOptions.setup(agentCtx)` compose a fresh or resumed agent's scoped world while both objects remain unpublished. Setup is trusted, composition-only same-process code: drive the agent only after creation resolves.
|
||||
The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-scope`, key = the agent) — register tools/sections/variables/listeners through it for that agent alone, all unwound on disposal. `agentEvents(ctx, agent)` is the fused dispatcher for ordinary agent-subject operations (carrier + injected subject in one move); its notification mode invokes every listener and contains both synchronous throws and returned-promise rejections. The registry lifecycle pair reuses one stable routing carrier. `assembleContextFor(agent)` builds the per-agent assembly context (`agent` + `scope` together). `installAgentLlmTarget(agentCtx, target)` snapshots a mutable provider/model/reasoning-effort selection during prompt assembly, applies the route to prompt variables, and applies the complete target to request routing for one step; an absent selected effort clears an inherited effort so the target uses adapter/provider defaults. `CreateAgentOptions.setup(agentCtx)` and `ResumeAgentOptions.setup(agentCtx)` compose a fresh or resumed agent's scoped world while both objects remain unpublished. Setup is trusted, composition-only same-process code: drive the agent only after creation resolves.
|
||||
|
||||
- `ctx.agents.register(agent: Agent): () => void` — record an **already-constructed** agent. Disposed with the calling fiber.
|
||||
- Advanced ordered lifecycle: `enter(agent, owner): () => void` enforces `agent.id === agent.session.id`, performs the authoritative ID collision check, and inserts without announcing; `owner` explicitly records the live creator-agent relation (or `undefined` for a root), independently of durable session lineage. `announce(agent)` emits `agent/created` exactly once. A detach requested synchronously by a creation listener is deferred until that dispatch unwinds, and every detach checks the captured entry object, so a stale capability cannot delete a later same-ID replacement. The async factory uses this split; ordinary plugins use `register()`.
|
||||
|
||||
@@ -12,7 +12,7 @@ Agent 接口、注册表、进程本地发起方作用域，以及 `agent/*` 事
|
||||
|
||||
### 公开 API
|
||||
|
||||
带作用域的注册表层：`Agent.ctx` 是 agent 的作用域上下文（`dsh-scope`，键 = 该 agent）。通过它注册工具／段／变量／监听器，只对该 agent 生效，并在释放时全部撤销。`agentEvents(ctx, agent)` 是普通 agent 主体操作的融合分发器（一次完成载体 + 注入主体）；其通知 mode 会调用每个监听器，并同时收容同步抛出和返回 Promise 的拒绝。注册表生命周期对复用一个稳定路由载体。`assembleContextFor(agent)` 构建按 agent 的组装上下文（同时包含 `agent` + `scope`）。`installAgentLlmTarget(agentCtx, target)` 在提示词组装期间快照可变的提供方／模型选择，并将该对同时应用到一个步骤的提示词变量与请求路由。`CreateAgentOptions.setup(agentCtx)` 和 `ResumeAgentOptions.setup(agentCtx)` 在新建或恢复的 agent 尚未发布时，组合其带作用域的世界。Setup 是受信任、仅用于组合的同进程代码：只有创建完成后才能驱动 agent。
|
||||
带作用域的注册表层：`Agent.ctx` 是 agent 的作用域上下文（`dsh-scope`，键 = 该 agent）。通过它注册工具／段／变量／监听器，只对该 agent 生效，并在释放时全部撤销。`agentEvents(ctx, agent)` 是普通 agent 主体操作的融合分发器（一次完成载体 + 注入主体）；其通知 mode 会调用每个监听器，并同时收容同步抛出和返回 Promise 的拒绝。注册表生命周期对复用一个稳定路由载体。`assembleContextFor(agent)` 构建按 agent 的组装上下文（同时包含 `agent` + `scope`）。`installAgentLlmTarget(agentCtx, target)` 在提示词组装期间快照可变的提供方／模型／推理（reasoning）强度选择，将路由应用到提示词变量，并将完整目标应用到一个步骤的请求路由；如果没有选定推理强度，则会清除继承的推理强度，使该目标使用适配器／提供方默认值。`CreateAgentOptions.setup(agentCtx)` 和 `ResumeAgentOptions.setup(agentCtx)` 在新建或恢复的 agent 尚未发布时，组合其带作用域的世界。Setup 是受信任、仅用于组合的同进程代码：只有创建完成后才能驱动 agent。
|
||||
|
||||
- `ctx.agents.register(agent: Agent): () => void`：记录一个 **已经构造完成** 的 agent。随调用 fiber 释放。
|
||||
- 高级有序生命周期：`enter(agent, owner): () => void` 强制 `agent.id === agent.session.id`，执行权威 ID 冲突检查，并在不通知的情况下插入；`owner` 显式记录实时创建方 agent 关系（根 agent 为 `undefined`），与持久会话谱系无关。`announce(agent)` 恰好发出一次 `agent/created`。创建监听器同步请求的 detach 会延后到该次分发结束；每次 detach 都会检查捕获的条目对象，因此陈旧能力无法删除后续使用同一 ID 的替代项。异步工厂使用这一拆分；普通插件使用 `register()`。
|
||||
|
||||
@@ -1,17 +1,19 @@
|
||||
/**
|
||||
* Agent-scoped provider/model target snapshot shared by interactive front doors.
|
||||
* Agent-scoped LLM target snapshot shared by interactive front doors.
|
||||
* @module @deepseek-ai/dsh-agent/llm-target
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import type { LlmCallConfig } from '@deepseek-ai/dsh-llm'
|
||||
import type { LlmCallConfig, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
/** Complete provider/model route selected for one live agent. */
|
||||
/** Complete provider/model route and optional reasoning effort selected for one live agent. */
|
||||
export interface AgentLlmTarget {
|
||||
/** Registered provider route. */
|
||||
provider: string
|
||||
/** Provider-owned model id. */
|
||||
model: string
|
||||
/** Adapter-owned reasoning effort, or provider/default behavior when absent. */
|
||||
reasoningEffort?: ReasoningEffortId
|
||||
}
|
||||
|
||||
/** Mutable selection plus the target captured for the current step. */
|
||||
@@ -24,9 +26,11 @@ export interface AgentLlmTargetRef {
|
||||
|
||||
/**
|
||||
* Couple one mutable target to agent-scoped prompt assembly and request routing.
|
||||
* Prompt assembly snapshots the selected pair before delegating, then applies
|
||||
* both prompt variables and request config to that snapshot so a concurrent
|
||||
* switch takes effect on a later step instead of splitting the two surfaces.
|
||||
* Prompt assembly snapshots the selected target before delegating, then applies
|
||||
* its route to prompt variables and its route/effort to request config so a
|
||||
* concurrent switch takes effect on a later step instead of splitting the two
|
||||
* surfaces. An absent selected effort clears any inherited effort so a model
|
||||
* switch can restore that target's provider/default behavior.
|
||||
*
|
||||
* @param agentCtx - The target agent's scoped context.
|
||||
* @param target - Mutable selection owned by the calling front door.
|
||||
@@ -52,10 +56,15 @@ export function installAgentLlmTarget(agentCtx: Context, target: AgentLlmTargetR
|
||||
async (_agent, _turn, _step, _config, _signal, next): Promise<LlmCallConfig> => {
|
||||
const resolved = await next()
|
||||
const selected = target.assembled
|
||||
return selected === undefined ? resolved : {
|
||||
...resolved,
|
||||
if (selected === undefined) return resolved
|
||||
const { reasoningEffort: _inheritedEffort, ...withoutInheritedEffort } = resolved
|
||||
return {
|
||||
...withoutInheritedEffort,
|
||||
provider: selected.provider,
|
||||
model: selected.model,
|
||||
...selected.reasoningEffort === undefined
|
||||
? {}
|
||||
: { reasoningEffort: selected.reasoningEffort },
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
@@ -7,7 +7,7 @@ import {
|
||||
type Agent,
|
||||
type AgentLlmTargetRef,
|
||||
} from '../src/index.ts'
|
||||
import type { LlmCallConfig } from '@deepseek-ai/dsh-llm'
|
||||
import { ReasoningEffortId, type LlmCallConfig } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
describe('installAgentLlmTarget()', () => {
|
||||
it('snapshots prompt variables and request routing together, then disposes both listeners', async () => {
|
||||
@@ -24,16 +24,31 @@ describe('installAgentLlmTarget()', () => {
|
||||
'agent/request', 1, 0, seed, signal, () => Promise.resolve(seed),
|
||||
)).resolves.toBe(seed)
|
||||
|
||||
target.current = { provider: 'alpha', model: 'a1' }
|
||||
target.current = {
|
||||
provider: 'alpha',
|
||||
model: 'a1',
|
||||
reasoningEffort: ReasoningEffortId('high'),
|
||||
}
|
||||
expect((await ctx.systemPrompt.assemble()).variables).toMatchObject({ provider: 'alpha', model: 'a1' })
|
||||
target.current = { provider: 'beta', model: 'b1' }
|
||||
await expect(agentEvents(ctx, agent).waterfall(
|
||||
'agent/request', 1, 0, seed, signal, () => Promise.resolve(seed),
|
||||
)).resolves.toEqual({ provider: 'alpha', model: 'a1', temperature: 0.2 })
|
||||
)).resolves.toEqual({
|
||||
provider: 'alpha',
|
||||
model: 'a1',
|
||||
reasoningEffort: ReasoningEffortId('high'),
|
||||
temperature: 0.2,
|
||||
})
|
||||
|
||||
expect((await ctx.systemPrompt.assemble()).variables).toMatchObject({ provider: 'beta', model: 'b1' })
|
||||
const inherited: LlmCallConfig = {
|
||||
provider: 'alpha',
|
||||
model: 'a1',
|
||||
reasoningEffort: ReasoningEffortId('max'),
|
||||
temperature: 0.2,
|
||||
}
|
||||
await expect(agentEvents(ctx, agent).waterfall(
|
||||
'agent/request', 1, 1, seed, signal, () => Promise.resolve(seed),
|
||||
'agent/request', 1, 1, inherited, signal, () => Promise.resolve(inherited),
|
||||
)).resolves.toEqual({ provider: 'beta', model: 'b1', temperature: 0.2 })
|
||||
|
||||
dispose()
|
||||
|
||||
@@ -35,6 +35,7 @@ const scopedSubjectResolvers: Readonly<Record<string, ScopedSubjectResolver | nu
|
||||
'subagent/end': null,
|
||||
'subagent/start': null,
|
||||
'system-prompt/assemble': args => (args[1] as Record<string, unknown>)['scope'],
|
||||
'tools/code-dispatch-log': args => (args[0] as Record<string, unknown>)['agent'],
|
||||
'tools/execute': args => (args[0] as Record<string, unknown>)['agent'],
|
||||
'tools/post-execute': args => (args[0] as Record<string, unknown>)['agent'],
|
||||
'tools/pre-execute': args => (args[0] as Record<string, unknown>)['agent'],
|
||||
|
||||
@@ -63,6 +63,7 @@ describe('scoped-dispatch invariants', () => {
|
||||
['approval/request', [{ agent, toolName: 'echo' }, () => Promise.resolve('unavailable')]],
|
||||
['goal/changed', [agent, { operation: 'create', ref: { id: 'goal-a', revision: 1 } }]],
|
||||
['system-prompt/assemble', [[], { scope: agent }]],
|
||||
['tools/code-dispatch-log', [{ exec: { callId: 'c', name: 't', arguments: {} }, agent, subCallId: 'c:code:1', name: 't', isError: false, content: [] }, () => Promise.resolve([])]],
|
||||
['tools/execute', [{ callId: 'c', name: 't', arguments: {}, agent }, () => Promise.resolve({ content: [], isError: false })]],
|
||||
['tools/post-execute', [{ callId: 'c', name: 't', arguments: {}, agent }, { content: [], isError: false }, () => Promise.resolve({ kind: 'accept' })]],
|
||||
['tools/pre-execute', [{ callId: 'c', name: 't', arguments: {}, agent }, () => Promise.resolve({ kind: 'allow' })]],
|
||||
|
||||
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: 18d6d385ff0c35ddbe7dc9a172ce9cd563bc4c1c
|
||||
README.zh.md: 93ea574eb01fd27fcd68f8b58a9e4187dfbd4fcb
|
||||
README.md: e46ff43c95df0ae1a6ec536d30417b342c11b151
|
||||
README.zh.md: abe4dbef6c7d26861cab987704c772a45e57a808
|
||||
|
||||
@@ -52,7 +52,7 @@ Durable values need one accepted representation, not a check followed by a secon
|
||||
|
||||
### Chunk-row storage codec (`chunk-rows.ts`)
|
||||
|
||||
Providers stream token-sized deltas, so a raw log stores hundreds of `assistant/chunk` lines whose JSON envelopes dwarf their payloads. `packChunkRuns(events)` packs each run of ≥3 consecutive same-block delta chunks into one storage row — `text-chunks`, `reasoning-chunks`, or `tool-call-chunks` (bare slash-less tags: storage vocabulary, not `SessionEventMap` members) — and `decodeStorageRecord(value)` expands a parsed line back into its exact events (`seq0`/`time0` + per-member `dt` gaps reconstruct every `seq`/`time`). The encoder whitelists exact shapes and stores anything unrecognized verbatim; the decoder validates row-tagged values and throws on malformation. Owned here so the JSONL backend and the fixture readers (`dsh-llm-replay`, `dsh-acp-snapshot`) share one codec; the write-side switch is the backend's `packChunks` config.
|
||||
Providers stream token-sized deltas, so a raw log stores hundreds of `assistant/chunk` lines whose JSON envelopes dwarf their payloads. `packChunkRuns(events)` packs each run of ≥3 consecutive same-block delta chunks into one storage row — `text-chunks`, `reasoning-chunks`, or `tool-call-chunks` (bare slash-less tags: storage vocabulary, not `SessionEventMap` members) — and `decodeStorageRecord(value)` expands a parsed line back into its exact events (`seq0`/`time0` + per-member `dt` gaps reconstruct every `seq`/`time`). The encoder whitelists exact shapes and stores anything unrecognized verbatim; the decoder validates row-tagged values and throws on malformation. Owned here so the JSONL backend and the fixture readers (`dsh-llm-replay`, `dsh-acp-snapshot`) share one codec; the backend's default-enabled `packChunks` config controls writes only.
|
||||
|
||||
### Surface types
|
||||
|
||||
|
||||
@@ -52,7 +52,7 @@
|
||||
|
||||
### 分片行存储编解码器（`chunk-rows.ts`）
|
||||
|
||||
提供方以 token 大小的增量流式输出，因此原始日志会存储数百行 `assistant/chunk`，其 JSON 封装远大于载荷。`packChunkRuns(events)` 将每段至少 3 个连续、同块的增量分片打包为一个存储行：`text-chunks`、`reasoning-chunks` 或 `tool-call-chunks`（不含斜杠的裸标签，属于存储词汇而不是 `SessionEventMap` 成员）。`decodeStorageRecord(value)` 则将已解析行展开回完全一致的事件（`seq0`／`time0` 加上每个成员的 `dt` 间隔，可重建每个 `seq`／`time`）。编码器只允许精确形态，并逐字存储任何无法识别的内容；解码器校验带行标签的值，形态错误时抛出异常。编解码器由此包所有，使 JSONL 后端和 fixture（测试前置数据）读取器（`dsh-llm-replay`、`dsh-acp-snapshot`）共享同一编解码器；写入侧开关是后端的 `packChunks` 配置。
|
||||
提供方以 token 大小的增量流式输出，因此原始日志会存储数百行 `assistant/chunk`，其 JSON 封装远大于载荷。`packChunkRuns(events)` 将每段至少 3 个连续、同块的增量分片打包为一个存储行：`text-chunks`、`reasoning-chunks` 或 `tool-call-chunks`（不含斜杠的裸标签，属于存储词汇而不是 `SessionEventMap` 成员）。`decodeStorageRecord(value)` 则将已解析行展开回完全一致的事件（`seq0`／`time0` 加上每个成员的 `dt` 间隔，可重建每个 `seq`／`time`）。编码器只允许精确形态，并逐字存储任何无法识别的内容；解码器校验带行标签的值，形态错误时抛出异常。编解码器由此包所有，使 JSONL 后端和 fixture（测试前置数据）读取器（`dsh-llm-replay`、`dsh-acp-snapshot`）共享同一编解码器；后端默认启用的 `packChunks` 配置只控制写入。
|
||||
|
||||
### Surface 类型
|
||||
|
||||
|
||||
@@ -183,6 +183,11 @@ function assertCurrentLlmShape(event: Record<string, unknown>, index: number): v
|
||||
const header = record['header']
|
||||
const config = typeof header === 'object' && header !== null ? (header as Record<string, unknown>)['config'] : undefined
|
||||
if (!hasProviderModel(config)) throw new Error(`seed request/header at index ${index} lacks provider/model`)
|
||||
const reasoningEffort = (config as Record<string, unknown>)['reasoningEffort']
|
||||
if (reasoningEffort !== undefined
|
||||
&& (typeof reasoningEffort !== 'string' || reasoningEffort.length === 0)) {
|
||||
throw new Error(`seed request/header at index ${index} has an invalid reasoningEffort`)
|
||||
}
|
||||
}
|
||||
if (event['type'] === 'assistant/message' && !hasProviderModel(record['provenance'])) {
|
||||
throw new Error(`seed assistant/message at index ${index} lacks provider/model provenance`)
|
||||
|
||||
@@ -156,7 +156,7 @@ export interface TodoItem {
|
||||
* canonical empty optional fields are absent.
|
||||
*/
|
||||
export interface EpochHeader {
|
||||
/** The conversation's call configuration (provider, model, and sampling scalars). */
|
||||
/** The conversation's call configuration (provider, model, reasoning effort, and sampling scalars). */
|
||||
config: LlmCallConfig
|
||||
/** Rendered system prompt text; absent for a system-less request. */
|
||||
system?: string
|
||||
|
||||
@@ -4,6 +4,7 @@ import { describe, expect, it } from 'vitest'
|
||||
import { Session, SessionId, canonicalHeader, foldRequestHeader, headerEquals } from '@deepseek-ai/dsh-session'
|
||||
import type { EpochHeader, SessionEvent } from '@deepseek-ai/dsh-session'
|
||||
import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm'
|
||||
import { ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
|
||||
const CONFIG = { provider: 'mock', model: 'm' }
|
||||
|
||||
@@ -29,6 +30,10 @@ describe('headerEquals', () => {
|
||||
it('compares every canonical field and preserves tool order', () => {
|
||||
expect(headerEquals(base, structuredClone(base))).toBe(true)
|
||||
expect(headerEquals(base, { ...base, config: { provider: 'mock', model: 'other' } })).toBe(false)
|
||||
expect(headerEquals(base, {
|
||||
...base,
|
||||
config: { ...base.config, reasoningEffort: ReasoningEffortId('high') },
|
||||
})).toBe(false)
|
||||
expect(headerEquals(base, { ...base, system: 'other' })).toBe(false)
|
||||
expect(headerEquals(base, { ...base, messagePrefix: [msg('other')] })).toBe(false)
|
||||
expect(headerEquals(base, { ...base, tools: [] })).toBe(false)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, expectTypeOf, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import { CallId, ReasoningEffortId } from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, {
|
||||
displayPromptContent,
|
||||
findLastMessageTurnEnd,
|
||||
@@ -118,6 +118,11 @@ describe('Session', () => {
|
||||
})
|
||||
|
||||
it('renders injected-context and steering messages as plain user content', () => {
|
||||
expect(displayPromptContent({
|
||||
content: [{ type: 'text', text: 'plain prompt' }],
|
||||
source: { kind: 'user' },
|
||||
})).toEqual([{ type: 'text', text: 'plain prompt' }])
|
||||
|
||||
const session = new Session(SessionId('s2'))
|
||||
session.append('user/message', {
|
||||
content: [{ type: 'text', text: 'file changed: a.ts' }],
|
||||
@@ -228,6 +233,35 @@ describe('Session', () => {
|
||||
.toEqual([unrelatedPrimitiveData])
|
||||
})
|
||||
|
||||
it('round-trips a non-empty reasoning effort and rejects invalid durable values', () => {
|
||||
const valid = {
|
||||
type: 'request/header',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: {
|
||||
header: {
|
||||
config: {
|
||||
provider: 'mock',
|
||||
model: 'model',
|
||||
reasoningEffort: ReasoningEffortId('adapter-owned'),
|
||||
},
|
||||
},
|
||||
reason: 'initial',
|
||||
},
|
||||
} as const
|
||||
expect(new Session(SessionId('reasoning-effort'), [valid]).events[0])
|
||||
.toEqual(valid)
|
||||
|
||||
for (const reasoningEffort of ['', 1]) {
|
||||
const invalid = structuredClone(valid) as unknown as SessionEvent
|
||||
if (invalid.type !== 'request/header') throw new Error('test fixture must be a request header')
|
||||
const config = invalid.data.header.config as unknown as Record<string, unknown>
|
||||
config.reasoningEffort = reasoningEffort
|
||||
expect(() => new Session(SessionId('invalid-reasoning-effort'), [invalid]))
|
||||
.toThrow('seed request/header at index 0 has an invalid reasoningEffort')
|
||||
}
|
||||
})
|
||||
|
||||
it('isolates the log from mutation through a derived message (append-only contract)', () => {
|
||||
const session = new Session(SessionId('s4'))
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'original' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
|
||||
@@ -2,5 +2,5 @@
|
||||
# side as of the last confirmed-consistent state. Both languages carry equal authority;
|
||||
# after editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write
|
||||
README.md: 14b7896e413a56fcee5a7db4cd92813f3e91c286
|
||||
README.zh.md: c89b03ff12bd346a2c0a8848a0e61b8fd738c318
|
||||
README.md: 3a9082f99663586936ce1eafaf09264a91798e28
|
||||
README.zh.md: 763ae1195d2c7a2606806970a3ba6371114d9d1c
|
||||
|
||||
@@ -114,16 +114,16 @@ Returning `undefined` selects generic fallback. Presenters depend only on their
|
||||
|
||||
### Code Mode
|
||||
|
||||
Under `code` or `both`, the registry exposes the reserved `run_code` transport and a deterministic TypeScript SDK for the current scope; only the program's outer logs and return value re-enter model context. The SDK declares exact `ToolArgsMap` and `ToolOutputMap` entries for every visible tool, and each binding resolves to the tool's canonical JSON value. Each lossless-JSON binding call re-enters the complete tool pipeline sequentially with logged correlation to the outer call. Denials and other failed results reject with the real program-visible `ToolCallError` carrying only `toolName` and `message`; Native content and internal error codes stay outside the Code contract. Ordinary side effects are not rolled back, and sub-call `additionalContexts` are deferred through the parent result to preserve call/result adjacency. Run settlement aborts and drains outstanding bindings; runtime failures surface as `CodeRunFailedError`. See the [Code Mode foundation](../../../.agents/notes/implemented/feature/2026-06-15-code-mode.md), [typed-return contract](../../../.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md), and [code-runtime seam](../../code-runtime/README.md). Try `pnpm run demo:code-mode`.
|
||||
Under `code` or `both`, the registry exposes the reserved `run_code` transport and a deterministic TypeScript SDK for the current scope; only the program's outer logs and return value re-enter model context. The SDK declares exact `ToolArgsMap` and `ToolOutputMap` entries for every visible tool, and each binding resolves to the tool's canonical JSON value. Each lossless-JSON binding call re-enters the complete tool pipeline under the native scheduling contract (concurrency-safe calls may overlap up to `maxParallelSubCalls`; exclusive calls run alone as ordering barriers) with logged correlation to the outer call. Denials and other failed results reject with the real program-visible `ToolCallError` carrying only `toolName` and `message`; Native content and internal error codes stay outside the Code contract. Ordinary side effects are not rolled back, and sub-call `additionalContexts` are deferred through the parent result to preserve call/result adjacency. Run settlement aborts and drains outstanding bindings; runtime failures surface as `CodeRunFailedError`. See the [Code Mode foundation](../../../.agents/notes/implemented/feature/2026-06-15-code-mode.md), [typed-return contract](../../../.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md), and [code-runtime seam](../../code-runtime/README.md). Try `pnpm run demo:code-mode`.
|
||||
|
||||
- **The SDK section** (`tools:sdk`, order 150): a lazy prompt section regenerating, at each assembly, `JsonValue`, exact `ToolArgsMap` / `ToolOutputMap`, `ToolName`, the `ToolCallError` declaration, and a mapped `tools` namespace for the calling scope's visible end capabilities (exotic names via quoted keys), plus fixed usage instructions. Deterministic — lexicographic tool order, byte-identical text for an unchanged tool set (prefix-cache-friendly). The codegen (`jsonSchemaToTs`, exported) handles every unified schema construct and degrades unsupported raw constructs to `unknown`, never throwing during prompt assembly.
|
||||
- **The dispatch bridge** (`run_code`'s execute): every binding call is snapshotted as lossless JSON before dispatch (`undefined`, `BigInt`, cycles, sparse arrays, `-0`, and exotic objects reject that one call), 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 success returns the final canonical value after policy; a failure reaches the worker as one message and becomes `ToolCallError(toolName, message)`. Each sub-call is logged as a `tool/code-dispatch` session event with deterministic id `<parent>:code:<n>` and a bounded Native-content summary; `deriveMessages()` does not surface that event or persist the value. 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/meta even when the program later fails.
|
||||
- **The dispatch bridge** (`run_code`'s execute): every binding call is snapshotted as lossless JSON before dispatch (`undefined`, `BigInt`, cycles, sparse arrays, `-0`, and exotic objects reject that one call), scheduled through a per-run pool that reuses the native concurrency contract — calls start strictly in submission order, consecutive `isConcurrencySafe` calls overlap up to the validated `maxParallelSubCalls` config (default 10; `1` restores serial dispatch), and an exclusive-classified call drains the pool, runs alone, and bars later calls — given the outer execution's opaque token as `parent`, and run through the complete pre-execute → guards → execute → post-execute → result pipeline. A success returns the final canonical value after policy; a failure reaches the worker as one message and becomes `ToolCallError(toolName, message)`. Each started sub-call logs a `tool/code-dispatch-start` event (deterministic id `<parent>:code:<n>`, numbered by submission) at pipeline entry and settles with one `tool/code-dispatch` event carrying the complete model-facing `content`/`isError` outcome (the `tool/result` vocabulary, so UIs render sub-calls through the native path — the pair's `time` fields carry per-sub-call timing); a queued call abandoned by run settlement logs neither. `deriveMessages()` surfaces neither event nor persists the canonical value. 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/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.
|
||||
- **Result boundary**: intermediate binding values cross the worker boundary whole and have no per-binding byte cap. `run_code` returns canonical `{ logs: string[], result?: JsonValue }`; strings render raw, every other present JSON root renders through a stack-safe pretty JSON traversal whose total indentation is capped at ten characters (deeper subtrees stay compact), `null` remains explicit, and absent `result` means the program returned `undefined`. The worker's configurable `maxOutputBytes` (default 64 MiB) applies only to the combined serialized outer log-array, completion-value, or failure-message payloads; fixed result-envelope syntax and presentation whitespace are outside that ledger. Invalid and over-limit completions fail explicitly, and only this outer result is eligible for ordinary spill.
|
||||
|
||||
### 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 Agent Note](../../../.agents/notes/implemented/feature/2026-07-10-parallel-tool-call-execution.md) owns the shipped declarations and rationale.
|
||||
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 reuse the same classification through the bridge's own pool. The [parallel tool-call Agent Note](../../../.agents/notes/implemented/feature/2026-07-10-parallel-tool-call-execution.md) owns the shipped declarations and rationale.
|
||||
|
||||
## Model Experience
|
||||
|
||||
@@ -156,7 +156,7 @@ Pass `run_code` the body of an async TypeScript function (erasable syntax only
|
||||
|
||||
- Call tools as `await tools.name(args)` — quoted access for exotic names: `tools["my-tool"](args)`. Every call resolves to the tool's typed canonical JSON value. Tool arguments must be lossless JSON.
|
||||
- A FAILED tool call rejects with `ToolCallError`, whose `toolName` identifies the failed tool and whose `message` is human-readable — `try/catch` it to handle and continue.
|
||||
- Calls execute sequentially, even under `Promise.all`.
|
||||
- Independent read-only calls MAY overlap under `Promise.all` (safe calls run concurrently; mutating calls run alone, in submission order). Sequence dependent work with `await`.
|
||||
- Emit results with `return` and/or `console.log(...)`. ONLY what you print or return comes back to you — intermediate tool results never enter the conversation, so extract just what you need.
|
||||
|
||||
The available tools:
|
||||
@@ -191,5 +191,5 @@ Append-only; newly visible content follows the reusable request prefix and does
|
||||
- **Caller-defined subagent and workflow structured outputs remain object-rooted** — this is a consumer-level guard; the shared schema vocabulary and tool outputs support every JSON root.
|
||||
- **`timeoutMs` on a definition is declarative only** — the registry never enforces deadlines; enforcement requires the `@deepseek-ai/dsh-timeout-policy` wrapper.
|
||||
- **Code Mode is TypeScript-only and the presentation mode is service-wide** — `mode: code`/`both` rejects prompt assembly unless `ctx.codeRuntime.language === 'typescript'`; scoped restrictions/shadows still choose each agent's visible bindings, but one tool cannot be native-only while another is code-only.
|
||||
- **Code Mode intermediate values are execution-local and unbounded by bytes** — they cannot be reconstructed from session replay and may exhaust process or worker memory; only the outer `run_code` output has the worker's configurable hard cap.
|
||||
- **Code Mode intermediate values are execution-local and unbounded by bytes** — the canonical typed values cannot be reconstructed from session replay and may exhaust process or worker memory; only the outer `run_code` output has the worker's configurable hard cap. The durable log copy of each sub-call IS bounded: the `tools/code-dispatch-log` waterfall lets the spill policy replace an oversized `tool/code-dispatch` content with a preview + locator ([rationale](../../../.agents/notes/implemented/feature/2026-07-26-code-dispatch-log-spill.md)).
|
||||
- **`run_code` state is fresh per run** — a persistent REPL-style kernel is rejected for the MVP (cross-call state would be invisible to the log); see [the Code Mode Agent Note](../../../.agents/notes/implemented/feature/2026-06-15-code-mode.md).
|
||||
|
||||
@@ -114,16 +114,16 @@ ctx.tools.register(defineTool({
|
||||
|
||||
### Code Mode
|
||||
|
||||
在 `code` 或 `both` 模式下，注册表为当前作用域公开保留的 `run_code` 传输和确定性的 TypeScript SDK；只有程序的外层日志与返回值会重新进入模型上下文。SDK 为每个可见工具声明精确的 `ToolArgsMap` 和 `ToolOutputMap` 条目，每个绑定都会解析为该工具的规范 JSON 值。每个无损 JSON 绑定调用都会按顺序重新进入完整工具流水线，并在日志中与外层调用建立关联。拒绝及其他失败结果会以程序实际可见的 `ToolCallError` 进行 reject，且只携带 `toolName` 和 `message`；Native 内容和内部错误码留在 Code 契约之外。普通副作用不会回滚，子调用的 `additionalContexts` 会通过父结果延迟，以保持调用／结果相邻。运行结算会中止并 drain 尚未完成的绑定；运行时失败以 `CodeRunFailedError` 形式出现。参见 [Code Mode 基础](../../../.agents/notes/implemented/feature/2026-06-15-code-mode.md)、[类型化返回契约](../../../.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md)和[代码运行时 seam](../../code-runtime/README.md)。可以运行 `pnpm run demo:code-mode` 试用。
|
||||
在 `code` 或 `both` 模式下，注册表为当前作用域公开保留的 `run_code` 传输和确定性的 TypeScript SDK；只有程序的外层日志与返回值会重新进入模型上下文。SDK 为每个可见工具声明精确的 `ToolArgsMap` 和 `ToolOutputMap` 条目，每个绑定都会解析为该工具的规范 JSON 值。每个无损 JSON 绑定调用都会在原生调度契约下重新进入完整工具流水线（并发安全的调用最多可重叠 `maxParallelSubCalls` 个；独占调用单独运行并构成排序屏障），并在日志中与外层调用建立关联。拒绝及其他失败结果会以程序实际可见的 `ToolCallError` 进行 reject，且只携带 `toolName` 和 `message`；Native 内容和内部错误码留在 Code 契约之外。普通副作用不会回滚，子调用的 `additionalContexts` 会通过父结果延迟，以保持调用／结果相邻。运行结算会中止并 drain 尚未完成的绑定；运行时失败以 `CodeRunFailedError` 形式出现。参见 [Code Mode 基础](../../../.agents/notes/implemented/feature/2026-06-15-code-mode.md)、[类型化返回契约](../../../.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md)和[代码运行时 seam](../../code-runtime/README.md)。可以运行 `pnpm run demo:code-mode` 试用。
|
||||
|
||||
- **SDK 段**（`tools:sdk`，顺序 150）：一个惰性提示词段，每次组装时都会重新生成 `JsonValue`、精确的 `ToolArgsMap` / `ToolOutputMap`、`ToolName`、`ToolCallError` 声明、面向调用作用域可见最终能力的映射 `tools` 命名空间（特殊名称使用带引号的键），以及固定用法说明。其输出具有确定性：工具按字典序排列；工具集合不变时，文本逐字节相同（有利于前缀 cache）。导出的代码生成器 `jsonSchemaToTs` 会处理统一 schema 的每种构造，并将不受支持的原始构造降级为 `unknown`，绝不会在提示词组装期间抛出。
|
||||
- **分发桥接层**（`run_code` 的 execute）：每个绑定调用都会在分发前快照为无损 JSON（`undefined`、`BigInt`、循环、稀疏数组、`-0` 和特殊对象会使该次调用被拒绝），通过每次运行独有的队列串行化（即使使用 `Promise.all`，底层调用也会按提交顺序逐个执行），以外层执行的不透明 token 作为 `parent`，并经过完整的 pre-execute → guards → execute → post-execute → result 流水线。成功会返回策略处理后的最终规范值；失败以一条消息到达 worker，并成为 `ToolCallError(toolName, message)`。每个子调用都会记录为 `tool/code-dispatch` 会话事件，其确定性 id 为 `<parent>:code:<n>`，并附带有界的 Native 内容摘要；`deriveMessages()` 不会公开该事件或持久化该值。token 关联让以提交为语义的观察器能够把内部成功延迟到最终 `run_code` 结果，而无需公开实时外层执行；普通工具副作用不会回滚。每个子调用的 `additionalContexts` 条目都会按分发顺序通过外层 `ToolRunContext` 延迟；循环只在父级 `run_code` 结果之后追加这些上下文，从而保持相邻关系，并且即使程序后来失败，也会保留各自的来源／元数据。
|
||||
- **分发桥接层**（`run_code` 的 execute）：每个绑定调用都会在分发前快照为无损 JSON（`undefined`、`BigInt`、循环、稀疏数组、`-0` 和特殊对象会使该次调用被拒绝），经由每次运行独有、复用原生并发契约的池调度——调用严格按提交顺序启动，连续的 `isConcurrencySafe` 调用最多可重叠经校验的 `maxParallelSubCalls` 配置个（默认 10；设为 `1` 即恢复串行分发），被分类为独占的调用先排空池、单独运行并阻挡其后的调用——以外层执行的不透明 token 作为 `parent`，并经过完整的 pre-execute → guards → execute → post-execute → result 流水线。成功会返回策略处理后的最终规范值；失败以一条消息到达 worker，并成为 `ToolCallError(toolName, message)`。每个已启动的子调用在进入流水线时记录一条 `tool/code-dispatch-start` 事件（确定性 id `<parent>:code:<n>`，按提交顺序编号），并以一条携带完整模型可见 `content`/`isError` 结果的 `tool/code-dispatch` 事件完结（采用 `tool/result` 词汇，因此 UI 会沿原生路径呈现子调用——这对事件的 `time` 字段承载每个子调用的计时）；因 run 结算而被放弃的排队调用两者都不记录。`deriveMessages()` 既不公开这两个事件，也不持久化规范值。token 关联让以提交为语义的观察器能够把内部成功延迟到最终 `run_code` 结果，而无需公开实时外层执行；普通工具副作用不会回滚。每个子调用的 `additionalContexts` 条目都会按分发顺序通过外层 `ToolRunContext` 延迟；循环只在父级 `run_code` 结果之后追加这些上下文，从而保持相邻关系，并且即使程序后来失败，也会保留各自的来源／元数据。
|
||||
- **结算纪律**：桥接层拥有一次运行作用域的中止；该中止会跟随传入的外层信号，并在运行因任何原因结算时触发，因此预算耗尽会中止正在运行的子工具，而不会将其遗留。桥接层随后会在返回之前 drain 队列，使每个 `tool/code-dispatch` 都落在仍打开的轮次内。失败的运行会抛出 `CodeRunFailedError`（`code: 'CODE_RUN_FAILED'`，message = 失败类型 + 已捕获日志），流水线会将其转换为模型可据以自我修正的结构化 `isError`。
|
||||
- **结果边界**：中间绑定值会完整跨越 worker 边界，且没有逐绑定字节上限。`run_code` 返回规范的 `{ logs: string[], result?: JsonValue }`；字符串原样呈现，其他所有存在的 JSON 根都通过栈安全的美化 JSON 遍历呈现，总缩进最多为 10 个字符（更深的子树保持紧凑），`null` 保持显式，而缺少 `result` 表示程序返回 `undefined`。worker 可配置的 `maxOutputBytes`（默认 64 MiB）只应用于组合序列化后的外层日志数组、完成值或失败消息载荷；固定的结果 envelope 语法和呈现空白不计入该账本。无效和超限的完成会明确失败，只有此外层结果可以使用普通 spill。
|
||||
|
||||
### 并行执行
|
||||
|
||||
agent loop 将连续的 `parallel` 调用归入有界滚动池，并把每个 `exclusive` 调用视为顺序屏障。只有分发／主体会重叠；策略、持久结果和上下文仍保持模型顺序。Code Mode 绑定仍按串行执行。[并行工具调用 Agent Note](../../../.agents/notes/implemented/feature/2026-07-10-parallel-tool-call-execution.md) 规定已交付声明及其原理。
|
||||
agent loop 将连续的 `parallel` 调用归入有界滚动池，并把每个 `exclusive` 调用视为顺序屏障。只有分发／主体会重叠；策略、持久结果和上下文仍保持模型顺序。Code Mode 绑定通过桥接层自己的池复用同一套分类。[并行工具调用 Agent Note](../../../.agents/notes/implemented/feature/2026-07-10-parallel-tool-call-execution.md) 规定已交付声明及其原理。
|
||||
|
||||
## 模型体验
|
||||
|
||||
@@ -156,7 +156,7 @@ Pass `run_code` the body of an async TypeScript function (erasable syntax only
|
||||
|
||||
- Call tools as `await tools.name(args)` — quoted access for exotic names: `tools["my-tool"](args)`. Every call resolves to the tool's typed canonical JSON value. Tool arguments must be lossless JSON.
|
||||
- A FAILED tool call rejects with `ToolCallError`, whose `toolName` identifies the failed tool and whose `message` is human-readable — `try/catch` it to handle and continue.
|
||||
- Calls execute sequentially, even under `Promise.all`.
|
||||
- Independent read-only calls MAY overlap under `Promise.all` (safe calls run concurrently; mutating calls run alone, in submission order). Sequence dependent work with `await`.
|
||||
- Emit results with `return` and/or `console.log(...)`. ONLY what you print or return comes back to you — intermediate tool results never enter the conversation, so extract just what you need.
|
||||
|
||||
The available tools:
|
||||
@@ -191,5 +191,5 @@ The available tools:
|
||||
- **调用方定义的 subagent 与工作流结构化输出仍要求对象根**：这是消费方层面的守卫；共享 schema 词汇和工具输出支持任意 JSON 根。
|
||||
- **定义上的 `timeoutMs` 仅为声明**：注册表绝不会强制执行截止时间；要强制执行，必须使用 `@deepseek-ai/dsh-timeout-policy` 包装层。
|
||||
- **Code Mode 只支持 TypeScript，且呈现模式在服务内统一**：`mode: code`/`both` 会拒绝组装提示词，除非 `ctx.codeRuntime.language === 'typescript'`；作用域限制／遮蔽仍会选择每个 agent 的可见绑定，但不能让一个工具仅使用 Native，而另一个仅使用 Code。
|
||||
- **Code Mode 中间值只存在于执行局部，且没有字节上限**：无法从会话回放重建这些值，它们可能耗尽进程或 worker 内存；只有外层 `run_code` 输出受 worker 可配置的硬上限约束。
|
||||
- **Code Mode 中间值只存在于执行局部，且没有字节上限**：这些规范的类型化值无法从会话回放重建，并可能耗尽进程或 worker 内存；只有外层 `run_code` 输出受 worker 可配置的硬上限约束。每个子调用的持久日志副本则**有**上限：`tools/code-dispatch-log` waterfall 允许 spill 策略把过大的 `tool/code-dispatch` 内容替换为预览加定位符（[原理](../../../.agents/notes/implemented/feature/2026-07-26-code-dispatch-log-spill.md)）。
|
||||
- **每次运行都会获得全新的 `run_code` 状态**：MVP 不采用持久 REPL 风格内核（跨调用状态不会出现在日志中）；参见 [Code Mode Agent Note](../../../.agents/notes/implemented/feature/2026-06-15-code-mode.md)。
|
||||
|
||||
@@ -1,37 +1,52 @@
|
||||
/**
|
||||
* Code Mode `run_code` transport. Programs call the registry's agent-visible
|
||||
* tools through nested, sequential executions; each sub-dispatch is logged for
|
||||
* reconstruction, while only the outer curated result enters model history.
|
||||
* tools through nested executions scheduled under the native concurrency
|
||||
* contract; each sub-dispatch is logged for reconstruction, while only the
|
||||
* outer curated result enters model history.
|
||||
* @module @deepseek-ai/dsh-tools/src/code-mode
|
||||
*/
|
||||
|
||||
import { parse } from 'node:path'
|
||||
import { CallId, HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import type { CodeBindingFunction, CodeRunResult, CodeRuntime } from '@deepseek-ai/dsh-code-runtime'
|
||||
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import type { JsonValue } from '@deepseek-ai/dsh-session'
|
||||
import { defineTool } from './schema.ts'
|
||||
import type { ToolDefinition, ToolRegistry } from './index.ts'
|
||||
import { TOOL_REGISTRY_SCHEDULER } from './index.ts'
|
||||
import type { CodeDispatchLog, ToolDefinition, ToolExecutionResult, ToolRegistry, ToolRunContext } from './index.ts'
|
||||
|
||||
declare module '@deepseek-ai/dsh-session' {
|
||||
interface SessionEventMap {
|
||||
/**
|
||||
* One bridged sub-dispatch from a `run_code` program: the parent
|
||||
* `run_code` call id, the deterministic sub-call id
|
||||
* (`<parent>:code:<n>`), the tool `name` with its JSON-normalized
|
||||
* `arguments` — the exact value dispatched, normalized BEFORE dispatch,
|
||||
* so this append can never fail on payload shape — whether the sub-call
|
||||
* errored, and a bounded `resultSummary` of its model-facing text. Before
|
||||
* bounding, occurrences of a non-root session workspace path are
|
||||
* normalized to `.` so host-specific absolute path lengths cannot change
|
||||
* the summary.
|
||||
* One sub-dispatch STARTING inside a `run_code` program: the parent
|
||||
* `run_code` call id, the deterministic sub-call id (`<parent>:code:<n>`,
|
||||
* numbered in submission order), and the tool `name` with its
|
||||
* JSON-normalized `arguments` — the exact value dispatched, normalized
|
||||
* BEFORE dispatch, so this append can never fail on payload shape.
|
||||
* Appended when the scheduler actually starts the call (not at
|
||||
* submission), so a start means the tool body pipeline was entered; a
|
||||
* call abandoned in the queue logs nothing. Log-only: `deriveMessages()`
|
||||
* ignores it; UIs use it for live per-sub-call running state and pair it
|
||||
* with `tool/code-dispatch` by `subCallId` (timing = the two events'
|
||||
* `time` fields).
|
||||
*/
|
||||
'tool/code-dispatch-start': { parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown }
|
||||
/**
|
||||
* One bridged sub-dispatch SETTLING: the pairing ids (matching the
|
||||
* `tool/code-dispatch-start` with the same `subCallId`), the tool `name`
|
||||
* with the same JSON-normalized `arguments`, and the sub-call's complete
|
||||
* model-facing outcome in `tool/result`'s own vocabulary
|
||||
* (`content` + `isError`), so UIs render a sub-call through the exact
|
||||
* code path that renders a native call. Every started sub-call settles
|
||||
* with exactly one of these (abort included: the aborted pipeline result
|
||||
* is an `isError` outcome).
|
||||
* Log-only: `deriveMessages()` ignores it, so sub-calls never re-enter
|
||||
* model context; persistence and UIs get every call. Appended inside the
|
||||
* parent `run_code`'s execution (the bridge drains its queue before
|
||||
* returning), so the turn-enclosure invariant holds by construction.
|
||||
* parent `run_code`'s execution (the bridge drains in-flight dispatches
|
||||
* before returning), so the turn-enclosure invariant holds by
|
||||
* construction.
|
||||
*/
|
||||
'tool/code-dispatch': { parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown; isError: boolean; resultSummary: string }
|
||||
'tool/code-dispatch': { parentCallId: CallId; subCallId: CallId; name: string; arguments: unknown; isError: boolean; content: ContentBlock[] }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,35 +70,6 @@ export class CodeRunFailedError extends HarnessError {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Cap for a `tool/code-dispatch` event's `resultSummary`. A log-ergonomics
|
||||
* constant, not config: the full result already flows to the program; the
|
||||
* summary exists so log readers see what a sub-call returned at a glance.
|
||||
*/
|
||||
const SUMMARY_MAX_CHARS = 200
|
||||
|
||||
/** Join Native content for the bounded durable sub-dispatch summary; non-text blocks become diagnostic placeholders. */
|
||||
function textOf(content: ContentBlock[]): string {
|
||||
return content
|
||||
.map((block) => {
|
||||
switch (block.type) {
|
||||
case 'text': return block.text
|
||||
// ContentBlockMap is merge-extensible — future block kinds land here
|
||||
// deliberately (no assertNever on merge-extensible unions).
|
||||
default: return `[${block.type} content]`
|
||||
}
|
||||
})
|
||||
.join('\n')
|
||||
}
|
||||
|
||||
/** Normalize workspace paths, then bound a sub-call's model-facing text for its durable log summary. */
|
||||
function summarize(text: string, cwd: string | undefined): string {
|
||||
const stableText = cwd === undefined || cwd === parse(cwd).root
|
||||
? text
|
||||
: text.replaceAll(cwd, '.')
|
||||
return stableText.length > SUMMARY_MAX_CHARS ? `${stableText.slice(0, SUMMARY_MAX_CHARS)}…` : stableText
|
||||
}
|
||||
|
||||
/**
|
||||
* Snapshot one binding call's argument as lossless JSON, then snapshot that
|
||||
* detached value again so dispatch and logging stay independent without
|
||||
@@ -201,17 +187,32 @@ function renderValue(value: JsonValue): string {
|
||||
type RunCodeOutput = { logs: string[]; result?: JsonValue }
|
||||
|
||||
/**
|
||||
* Build the `run_code` {@link ToolDefinition}: one required `code` parameter,
|
||||
* executed through the dispatch bridge described above. The
|
||||
* Registry-private capabilities the bridge receives at construction — the
|
||||
* `requireRuntime` idiom: operations only the owning registry can mint stay
|
||||
* off its public service surface and flow here as closures instead.
|
||||
*/
|
||||
export interface RunCodeBridgeOptions {
|
||||
/** Resolves `ctx.codeRuntime` or throws the loud misconfiguration error (shared with the registry's assembly-time checks). */
|
||||
requireRuntime: () => CodeRuntime
|
||||
/** The run's overlap cap for parallel-classified sub-calls (the registry passes its validated `maxParallelSubCalls`). */
|
||||
maxParallel: number
|
||||
/** Runs the contained `tools/code-dispatch-log` waterfall over one settled sub-dispatch (the registry's private invoker). */
|
||||
shapeDispatchLog: (dispatch: CodeDispatchLog) => Promise<ContentBlock[]>
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the `run_code` {@link ToolDefinition}: required `code` and
|
||||
* `description` parameters, executed through the dispatch bridge described
|
||||
* above. The
|
||||
* registry reserves it as presentation infrastructure under non-native modes,
|
||||
* outside the filterable global/scoped capability layers.
|
||||
* @param registry - the owning registry (sub-calls go through its `execute`,
|
||||
* bindings cover its registered tools).
|
||||
* @param requireRuntime - resolves `ctx.codeRuntime` or throws the loud
|
||||
* misconfiguration error (shared with the registry's assembly-time checks).
|
||||
* @param options - the registry-private capabilities described above.
|
||||
* @returns the registry-ready definition.
|
||||
*/
|
||||
export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () => CodeRuntime): ToolDefinition {
|
||||
export function createRunCodeTool(registry: ToolRegistry, options: RunCodeBridgeOptions): ToolDefinition {
|
||||
const { requireRuntime, maxParallel, shapeDispatchLog } = options
|
||||
return defineTool({
|
||||
name: RUN_CODE_NAME,
|
||||
description:
|
||||
@@ -221,6 +222,13 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
|
||||
+ 'Only what you print or return comes back — curate it.',
|
||||
parameters: {
|
||||
code: { type: 'string', required: true, description: 'The program: the body of an async TypeScript function.' },
|
||||
description: {
|
||||
type: 'string',
|
||||
required: true,
|
||||
description: 'Clear, concise description of what this program does in active voice, '
|
||||
+ '5-10 words (shown in the UI). Examples: "Count TODO markers across packages"; '
|
||||
+ '"Read failing test and its fixture"; "Rename config key in every cordis.yml".',
|
||||
},
|
||||
},
|
||||
output: {
|
||||
schema: {
|
||||
@@ -238,6 +246,9 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
|
||||
},
|
||||
},
|
||||
async execute(args, exec): Promise<RunCodeOutput> {
|
||||
if (args.description.trim().length === 0) {
|
||||
throw new Error('invalid description: expected a non-empty string')
|
||||
}
|
||||
const runtime = requireRuntime()
|
||||
|
||||
// The run-scoped abort: follows the outer signal in, and fires when the
|
||||
@@ -249,19 +260,120 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
|
||||
exec.signal.addEventListener('abort', onOuterAbort, { once: true })
|
||||
|
||||
let dispatches = 0
|
||||
// The per-run serialization queue: every binding call chains onto the tail, so even
|
||||
// `Promise.all` executes the underlying tool calls one at a time in submission order (the
|
||||
// tool contract carries no concurrency-safety metadata yet).
|
||||
let queue: Promise<void> = Promise.resolve()
|
||||
const enqueue = <T>(task: () => Promise<T>): Promise<T> => {
|
||||
const turn = queue.then(() => {
|
||||
if (runController.signal.aborted) {
|
||||
throw new Error(`run_code run is over (${String(runController.signal.reason)}); tool call abandoned`)
|
||||
// The per-run scheduler, reusing the NATIVE concurrency contract through
|
||||
// the registry's staged view (the loop scheduler's own seam) — and the
|
||||
// native loop's SEQUENCING: every ordered stage (the dispatch-start
|
||||
// append, prepare = pre-execute/guards, finalize/finish = post-execute,
|
||||
// context deferral, the settle append) runs inside ONE driver lane, so
|
||||
// ordered policy stages never overlap each other and only the
|
||||
// around-dispatch/body stage runs concurrently. Starts are strictly
|
||||
// submission-ordered; results commit in submission order through the
|
||||
// head-of-line cursor. Consecutive parallel-classified calls overlap up
|
||||
// to maxParallel; an exclusive call waits for the pool to drain, runs
|
||||
// alone, and holds its barrier until its COMMIT (post-execute included)
|
||||
// completes, exactly like a native exclusive group. Classification is
|
||||
// re-read via executionMode() immediately before each start (a registry
|
||||
// mutation while queued can flip a call exclusive), matching the native
|
||||
// scheduler's lazy reclassification.
|
||||
interface PendingDispatch {
|
||||
/** Ordered stage: append the start event, await prepare (pre-execute/guards), launch the body into `flight`. */
|
||||
start(): Promise<void>
|
||||
classify(): 'parallel' | 'exclusive'
|
||||
abandon(): void
|
||||
/** Ordered stage: post-execute + context deferral + settle event, in submission order. */
|
||||
commit(): Promise<void>
|
||||
/** The launched around-dispatch/body stage; resolved until start() replaces it. */
|
||||
flight: Promise<void>
|
||||
/** True once the dispatch stage parked its outcome; the commit cursor waits on it. */
|
||||
settled: boolean
|
||||
/** The classification this entry started under; an exclusive holds its barrier through commit(). */
|
||||
mode?: 'parallel' | 'exclusive'
|
||||
}
|
||||
const pendingQueue: PendingDispatch[] = []
|
||||
const inFlight = new Set<Promise<void>>()
|
||||
/** Tracked settle-event side work (log shaping + append), drained at run settlement. */
|
||||
const logWork = new Set<Promise<void>>()
|
||||
const commitQueue: PendingDispatch[] = []
|
||||
let exclusiveActive = false
|
||||
let driving = false
|
||||
let driverRun: Promise<void> = Promise.resolve()
|
||||
let wake: (() => void) | undefined
|
||||
const wakeup = (): void => {
|
||||
const release = wake
|
||||
wake = undefined
|
||||
release?.()
|
||||
}
|
||||
/**
|
||||
* The single ordered lane. Each pass commits the head-of-line settled
|
||||
* dispatch (ordered post-execute), then starts the next queued entry if
|
||||
* its slot is free (ordered pre-execute), and otherwise sleeps until a
|
||||
* body settles or a new submission arrives. One run reaching the
|
||||
* empty-queues/empty-pool state is quiescence.
|
||||
*/
|
||||
const drive = (): Promise<void> => {
|
||||
if (driving) return driverRun
|
||||
driving = true
|
||||
driverRun = (async () => {
|
||||
try {
|
||||
for (;;) {
|
||||
// Arm before inspecting state so a settle or submission landing
|
||||
// between the checks and the await below cannot be lost.
|
||||
const signal = new Promise<void>((resolve) => { wake = resolve })
|
||||
const commitHead = commitQueue[0]
|
||||
if (commitHead !== undefined && commitHead.settled) {
|
||||
commitQueue.shift()
|
||||
await commitHead.commit()
|
||||
// The barrier covers post-execute: later starts wait for the
|
||||
// exclusive call's full pipeline, as under the native loop.
|
||||
if (commitHead.mode === 'exclusive') exclusiveActive = false
|
||||
continue
|
||||
}
|
||||
const head = pendingQueue[0]
|
||||
if (head !== undefined) {
|
||||
if (runController.signal.aborted) {
|
||||
pendingQueue.shift()
|
||||
head.abandon()
|
||||
continue
|
||||
}
|
||||
// Reclassify at start time (fail-closed on registry changes).
|
||||
const mode = head.classify()
|
||||
const capacity = !exclusiveActive
|
||||
&& (mode === 'exclusive' ? inFlight.size === 0 : inFlight.size < maxParallel)
|
||||
if (capacity) {
|
||||
if (mode === 'exclusive') exclusiveActive = true
|
||||
head.mode = mode
|
||||
pendingQueue.shift()
|
||||
// Joined before start() so the commit cursor sees submission
|
||||
// order; nothing commits it until `settled` flips.
|
||||
commitQueue.push(head)
|
||||
await head.start()
|
||||
const flight: Promise<void> = head.flight.finally(() => {
|
||||
inFlight.delete(flight)
|
||||
wakeup()
|
||||
})
|
||||
inFlight.add(flight)
|
||||
continue
|
||||
}
|
||||
}
|
||||
if (pendingQueue.length === 0 && commitQueue.length === 0 && inFlight.size === 0) return
|
||||
await signal
|
||||
}
|
||||
} finally {
|
||||
driving = false
|
||||
wake = undefined
|
||||
}
|
||||
return task()
|
||||
})
|
||||
queue = turn.then(() => undefined, () => undefined)
|
||||
return turn
|
||||
})()
|
||||
return driverRun
|
||||
}
|
||||
/** Every dispatch settled AND committed; nothing can start (the run is aborted at call time). */
|
||||
const drainDispatches = async (): Promise<void> => {
|
||||
// The abort already fired: the driver abandons queued-unstarted
|
||||
// entries, awaits the live pool, and drains the ordered commit lane —
|
||||
// including a commit already in progress when the program returned.
|
||||
await drive()
|
||||
// Every settle's shaped append lands inside the open run_code turn
|
||||
// (tasks self-remove on settlement).
|
||||
while (logWork.size > 0) await Promise.allSettled([...logWork])
|
||||
}
|
||||
|
||||
// Read through a call, not a bare property: the abort state genuinely
|
||||
@@ -274,35 +386,110 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
|
||||
throw new Error(`run_code run is over (${String(runController.signal.reason)}); ${name} not dispatched`)
|
||||
}
|
||||
const normalized = jsonNormalizeArgs(rawArgs)
|
||||
const outcome = await enqueue(async () => {
|
||||
const n = ++dispatches
|
||||
const subCallId = CallId(`${String(exec.callId)}:code:${n}`)
|
||||
const result = await registry.execute({
|
||||
callId: subCallId,
|
||||
name,
|
||||
arguments: normalized.dispatched,
|
||||
...exec.agent ? { agent: exec.agent } : {},
|
||||
parent: exec.token,
|
||||
signal: runController.signal,
|
||||
})
|
||||
for (const context of result.additionalContexts ?? []) {
|
||||
exec.deferContext(context)
|
||||
const n = ++dispatches
|
||||
const subCallId = CallId(`${String(exec.callId)}:code:${n}`)
|
||||
const input = {
|
||||
callId: subCallId,
|
||||
name,
|
||||
arguments: normalized.dispatched,
|
||||
...exec.agent ? { agent: exec.agent } : {},
|
||||
parent: exec.token,
|
||||
signal: runController.signal,
|
||||
}
|
||||
type DispatchOutcome = { isError: true; message: string } | { isError: false; value: JsonValue }
|
||||
const scheduler = registry[TOOL_REGISTRY_SCHEDULER]
|
||||
const outcome = await new Promise<DispatchOutcome>((resolve, reject) => {
|
||||
// Set by the dispatch stage (or start() for a pre-settled result): what commit() finalizes in submission order.
|
||||
let parked:
|
||||
| { kind: 'post-result' | 'final-result'; exec: ToolRunContext; result: ToolExecutionResult }
|
||||
| undefined
|
||||
const settle = (result: ToolExecutionResult): void => {
|
||||
// The program gets its value NOW: log shaping (e.g. a spill
|
||||
// backend) must never delay the binding or occupy a dispatch
|
||||
// slot. The shaped append is tracked side work; the run's
|
||||
// settlement drains logWork so every settle event still lands
|
||||
// inside the open turn (shapeDispatchLog is contained, so this
|
||||
// chain cannot reject).
|
||||
resolve(result.isError
|
||||
? { isError: true, message: result.error.message }
|
||||
: { isError: false, value: result.value })
|
||||
const agent = exec.agent
|
||||
if (agent === undefined) return
|
||||
const task: Promise<void> = (async () => {
|
||||
// The durable copy may be reshaped (e.g. spilled to a preview +
|
||||
// locator) by the log-shaping waterfall; the program's value
|
||||
// and the model contract are untouched.
|
||||
const logged = await shapeDispatchLog({
|
||||
exec, agent, subCallId, name, isError: result.isError,
|
||||
// The registry deep-froze this projection at result
|
||||
// finalization; append snapshots the final copy again, so
|
||||
// the log stays detached.
|
||||
content: result.content,
|
||||
})
|
||||
agent.session.append('tool/code-dispatch', {
|
||||
parentCallId: exec.callId,
|
||||
subCallId,
|
||||
name,
|
||||
// The SIBLING parse of the dispatched value: byte-identical JSON,
|
||||
// but a separate object — a tool mutating its args cannot desync
|
||||
// this record from what it actually received.
|
||||
arguments: normalized.logged,
|
||||
isError: result.isError,
|
||||
content: logged,
|
||||
})
|
||||
})().finally(() => { logWork.delete(task) })
|
||||
logWork.add(task)
|
||||
}
|
||||
const text = textOf(result.content)
|
||||
exec.agent?.session.append('tool/code-dispatch', {
|
||||
parentCallId: exec.callId,
|
||||
subCallId,
|
||||
name,
|
||||
// The SIBLING parse of the dispatched value: byte-identical JSON,
|
||||
// but a separate object — a tool mutating its args cannot desync
|
||||
// this record from what it actually received.
|
||||
arguments: normalized.logged,
|
||||
isError: result.isError,
|
||||
resultSummary: summarize(text, exec.agent.session.header.cwd),
|
||||
pendingQueue.push({
|
||||
flight: Promise.resolve(),
|
||||
settled: false,
|
||||
// Re-read per driver pass against the same agent view the SDK
|
||||
// declared; fail-closed exclusive when undeclared/invalid.
|
||||
classify: () => registry.executionMode(input).kind,
|
||||
abandon: () => {
|
||||
reject(new Error(`run_code run is over (${String(runController.signal.reason)}); ${name} tool call abandoned`))
|
||||
},
|
||||
async start(): Promise<void> {
|
||||
exec.agent?.session.append('tool/code-dispatch-start', {
|
||||
parentCallId: exec.callId,
|
||||
subCallId,
|
||||
name,
|
||||
arguments: normalized.logged,
|
||||
})
|
||||
// Ordered prepare runs INSIDE the driver lane: the next entry's
|
||||
// pre-execute waits for this resolution, as under the native
|
||||
// scheduler. Only the launched body below overlaps.
|
||||
const prepared = await scheduler.prepare(input)
|
||||
if (prepared.kind === 'dispatch') {
|
||||
this.flight = scheduler.dispatch(prepared.exec).then((dispatchOutcome) => {
|
||||
parked = { kind: dispatchOutcome.kind, exec: prepared.exec, result: dispatchOutcome.result }
|
||||
this.settled = true
|
||||
})
|
||||
return
|
||||
}
|
||||
parked = { kind: prepared.kind, exec: prepared.exec, result: prepared.result }
|
||||
this.settled = true
|
||||
},
|
||||
async commit(): Promise<void> {
|
||||
/* v8 ignore next -- commit() runs only after `settled` flipped, which set parked. */
|
||||
if (parked === undefined) return
|
||||
const result = parked.kind === 'post-result'
|
||||
? await scheduler.finalize(parked.exec, parked.result)
|
||||
: scheduler.finish(parked.exec, parked.result)
|
||||
for (const context of result.additionalContexts ?? []) {
|
||||
exec.deferContext(context)
|
||||
}
|
||||
settle(result)
|
||||
// Backpressure on the shaped-append side channel: pending log
|
||||
// tasks (each retaining a full result while a slow backend
|
||||
// stores it) are bounded by the pool cap — beyond it the
|
||||
// ordered lane waits, so later sub-calls cannot start and
|
||||
// pending I/O/memory cannot grow without bound.
|
||||
while (logWork.size > maxParallel) await Promise.race(logWork)
|
||||
},
|
||||
})
|
||||
return result.isError
|
||||
? { isError: true as const, message: result.error.message }
|
||||
: { isError: false as const, value: result.value }
|
||||
wakeup()
|
||||
void drive()
|
||||
})
|
||||
// A budget expiry or outer cancel that lands while this call was in
|
||||
// flight already aborted the dispatch; stop the program now rather
|
||||
@@ -345,10 +532,11 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
|
||||
signal: runController.signal,
|
||||
})
|
||||
} finally {
|
||||
// Abort sub-dispatches and drain the folded queue before closing the turn.
|
||||
// Abort sub-dispatches and drain every in-flight dispatch before
|
||||
// closing the turn (queued-unstarted ones are abandoned unlogged).
|
||||
// Binding failures remain observable through their individual promises.
|
||||
runController.abort('run_code settled')
|
||||
await queue
|
||||
await drainDispatches()
|
||||
}
|
||||
|
||||
if (result.error) {
|
||||
@@ -363,10 +551,11 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
|
||||
exec.signal.removeEventListener('abort', onOuterAbort)
|
||||
}
|
||||
},
|
||||
// The program is the call's always-visible UI label.
|
||||
// The model-authored description is the call's always-visible UI label
|
||||
// (the bash `description` precedent); the program itself rides rawInput.
|
||||
presentCall: args => ({
|
||||
card: 'generic',
|
||||
title: args.code,
|
||||
title: args.description,
|
||||
kind: 'execute',
|
||||
rawInput: args.code,
|
||||
}),
|
||||
|
||||
@@ -123,6 +123,19 @@ declare module 'cordis' {
|
||||
* @mode waterfall
|
||||
*/
|
||||
'tools/post-execute'(this: Scoped<ToolRegistry>, exec: ToolExecution, result: Readonly<ToolExecutionResult>, next: () => Promise<PostToolDecision>): Promise<PostToolDecision>
|
||||
/**
|
||||
* Shape the DURABLE LOG COPY of one `run_code` sub-dispatch outcome before
|
||||
* the bridge appends its `tool/code-dispatch` event. `next()` keeps the
|
||||
* content unchanged; a listener may return replacement blocks (e.g. the
|
||||
* spill policy's preview + locator for an oversized text result). Only the
|
||||
* logged copy is affected — the program already received the complete
|
||||
* value, and the model sees neither. A throwing listener is contained:
|
||||
* the bridge falls back to logging the unshaped content.
|
||||
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent's dispatches.
|
||||
* @param dispatch - the parent execution, sub-call identity, and the settled content to log.
|
||||
* @mode waterfall
|
||||
*/
|
||||
'tools/code-dispatch-log'(this: Scoped<ToolRegistry>, dispatch: CodeDispatchLog, next: () => Promise<ContentBlock[]>): Promise<ContentBlock[]>
|
||||
/**
|
||||
* Observe the frozen, lossless-JSON final outcome. Listener failures are contained.
|
||||
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): keyed by `exec.agent`.
|
||||
@@ -272,6 +285,30 @@ export type ToolExecutionMode =
|
||||
| { kind: 'parallel' }
|
||||
| { kind: 'exclusive' }
|
||||
|
||||
/**
|
||||
* One settled `run_code` sub-dispatch about to be logged, as seen by the
|
||||
* `tools/code-dispatch-log` waterfall: the parent execution (session owner,
|
||||
* outer call identity), the sub-call identity, and the outcome whose durable
|
||||
* copy a listener may reshape. `content` is the RENDERED result projection
|
||||
* (what a native `tool/result` would carry) — the program itself received
|
||||
* the structured `value` (or just the error message on failure); only the
|
||||
* `tool/code-dispatch` event's copy changes.
|
||||
*/
|
||||
export interface CodeDispatchLog {
|
||||
/** The outer `run_code` execution. */
|
||||
readonly exec: ToolExecution
|
||||
/** The calling agent (the scope routing key and the spill owner), when the outer call has one. */
|
||||
readonly agent?: Agent
|
||||
/** Deterministic sub-call id (`<parent>:code:<n>`). */
|
||||
readonly subCallId: CallId
|
||||
/** The dispatched sub-tool name. */
|
||||
readonly name: string
|
||||
/** Whether the sub-call settled as an error. */
|
||||
readonly isError: boolean
|
||||
/** The sub-call's complete model-facing content (the settle event's default payload). */
|
||||
readonly content: ContentBlock[]
|
||||
}
|
||||
|
||||
/**
|
||||
* One pending tool call inside the registry pipeline. Parsed arguments cross
|
||||
* one lossless-JSON materialization boundary before policy and are deep-frozen;
|
||||
@@ -534,6 +571,14 @@ export interface Config {
|
||||
* absent or mismatched. Under `code`, native names in `toolOrder` are invalid.
|
||||
*/
|
||||
mode?: ToolPresentationMode
|
||||
/**
|
||||
* Concurrency cap for a `run_code` program's overlapping sub-calls
|
||||
* (default 10, the loop scheduler's own default). Sub-calls follow the
|
||||
* native scheduling contract — only calls whose tools classify
|
||||
* concurrency-safe overlap; exclusive calls form barriers — so `1`
|
||||
* restores strictly serial dispatch. Must be a positive integer.
|
||||
*/
|
||||
maxParallelSubCalls?: number
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -627,6 +672,15 @@ interface FusedToolSignal {
|
||||
dispose(): void
|
||||
}
|
||||
|
||||
/** Resolve the run_code overlap cap at the owning config boundary (direct construction bypasses the Loader schema). */
|
||||
function resolveMaxParallelSubCalls(value: number | undefined): number {
|
||||
const maxParallelSubCalls = value ?? 10
|
||||
if (!Number.isInteger(maxParallelSubCalls) || maxParallelSubCalls < 1) {
|
||||
throw new Error('maxParallelSubCalls must be a positive integer')
|
||||
}
|
||||
return maxParallelSubCalls
|
||||
}
|
||||
|
||||
/**
|
||||
* Tool registry and execution pipeline. Scoped registrations shadow globals;
|
||||
* one visibility resolver feeds presentation, lookup, and dispatch.
|
||||
@@ -636,6 +690,7 @@ export class ToolRegistry extends Service {
|
||||
|
||||
static Config: z<Config> = z.object({
|
||||
mode: z.union(['native', 'code', 'both'] as const).default('native'),
|
||||
maxParallelSubCalls: z.natural().min(1).default(10),
|
||||
})
|
||||
|
||||
/** Internal staged view consumed by `dsh-agent-loop`'s parallel scheduler. */
|
||||
@@ -672,7 +727,11 @@ export class ToolRegistry extends Service {
|
||||
// the filterable global/scoped capability layers.
|
||||
this.codeTransport = this.mode === 'native'
|
||||
? undefined
|
||||
: createRunCodeTool(this, () => this.requireCodeRuntime())
|
||||
: createRunCodeTool(this, {
|
||||
requireRuntime: () => this.requireCodeRuntime(),
|
||||
maxParallel: resolveMaxParallelSubCalls(config.maxParallelSubCalls),
|
||||
shapeDispatchLog: dispatch => this.shapeDispatchLog(dispatch),
|
||||
})
|
||||
ctx.systemPrompt.tools(context => this.wireSchemas(context.scope))
|
||||
if (this.mode !== 'native') {
|
||||
ctx.systemPrompt.section({
|
||||
@@ -923,6 +982,27 @@ export class ToolRegistry extends Service {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the `tools/code-dispatch-log` waterfall over one settled sub-dispatch
|
||||
* and return the content the bridge should log on `tool/code-dispatch`.
|
||||
* Contained: a throwing listener falls back to the unshaped content — log
|
||||
* shaping must never fail the dispatch or lose the settle event. Private:
|
||||
* the ONE consumer is the `run_code` bridge this registry constructs, which
|
||||
* receives it as a capability parameter (the `requireRuntime` idiom) — the
|
||||
* waterfall, not this invoker, is the public extension seam.
|
||||
*/
|
||||
private async shapeDispatchLog(dispatch: CodeDispatchLog): Promise<ContentBlock[]> {
|
||||
try {
|
||||
return await this.ctx.waterfall(
|
||||
scopeTarget(this, dispatch.agent), 'tools/code-dispatch-log', dispatch,
|
||||
() => Promise.resolve(dispatch.content),
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`tools: code-dispatch-log listener failed for ${dispatch.name}: ${errorMessage(error)}; logging the unshaped content`)
|
||||
return dispatch.content
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute through pre-policy, guards, around-dispatch, post-policy,
|
||||
* definition-owned content finalization, and final notification. Tool and
|
||||
|
||||
@@ -253,7 +253,7 @@ Pass \`run_code\` the body of an async TypeScript function (erasable syntax only
|
||||
|
||||
- Call tools as \`await tools.name(args)\` — quoted access for exotic names: \`tools["my-tool"](args)\`. Every call resolves to the tool's typed canonical JSON value. Tool arguments must be lossless JSON.
|
||||
- A FAILED tool call rejects with \`ToolCallError\`, whose \`toolName\` identifies the failed tool and whose \`message\` is human-readable — \`try/catch\` it to handle and continue.
|
||||
- Calls execute sequentially, even under \`Promise.all\`.
|
||||
- Independent read-only calls MAY overlap under \`Promise.all\` (safe calls run concurrently; mutating calls run alone, in submission order). Sequence dependent work with \`await\`.
|
||||
- Emit results with \`return\` and/or \`console.log(...)\`. ONLY what you print or return comes back to you — intermediate tool results never enter the conversation, so extract just what you need.
|
||||
|
||||
The available tools:`
|
||||
|
||||
@@ -42,6 +42,7 @@ class FakeRuntime extends CodeRuntime {
|
||||
|
||||
interface SetupOptions {
|
||||
mode?: Config['mode']
|
||||
maxParallelSubCalls?: number
|
||||
runtime?: false | { language?: string }
|
||||
toolOrder?: string[]
|
||||
}
|
||||
@@ -49,7 +50,7 @@ interface SetupOptions {
|
||||
async function setup(options: SetupOptions = {}) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt, { ...options.toolOrder ? { toolOrder: options.toolOrder } : {} })
|
||||
await ctx.plugin(ToolRegistry, { mode: options.mode ?? 'code' })
|
||||
await ctx.plugin(ToolRegistry, { mode: options.mode ?? 'code', ...options.maxParallelSubCalls !== undefined ? { maxParallelSubCalls: options.maxParallelSubCalls } : {} })
|
||||
let runtime: FakeRuntime | undefined
|
||||
if (options.runtime !== false) {
|
||||
await ctx.plugin(FakeRuntime, options.runtime ?? {})
|
||||
@@ -87,11 +88,11 @@ function registerEcho(ctx: Context, name = 'echo'): unknown[] {
|
||||
}
|
||||
|
||||
/** A structural fake of the owning agent: captures session appends. */
|
||||
function fakeAgent(options: { cwd?: string } = { cwd: '/workspace' }): { agent: Agent; events: { type: string; data: unknown }[] } {
|
||||
function fakeAgent(): { agent: Agent; events: { type: string; data: unknown }[] } {
|
||||
const events: { type: string; data: unknown }[] = []
|
||||
const agent = {
|
||||
session: {
|
||||
header: options.cwd === undefined ? {} : { cwd: options.cwd },
|
||||
header: { cwd: '/workspace' },
|
||||
append: (type: string, data: unknown) => { events.push({ type, data }) },
|
||||
},
|
||||
} as unknown as Agent
|
||||
@@ -99,12 +100,16 @@ function fakeAgent(options: { cwd?: string } = { cwd: '/workspace' }): { agent:
|
||||
}
|
||||
|
||||
/** Dispatch run_code through the registry pipeline, as the loop would. */
|
||||
async function runCode(ctx: Context, code: string, extras: { agent?: Agent; signal?: AbortSignal } = {}): Promise<ToolExecutionResult> {
|
||||
async function runCode(
|
||||
ctx: Context,
|
||||
code: string,
|
||||
extras: { agent?: Agent; signal?: AbortSignal; description?: string } = {},
|
||||
): Promise<ToolExecutionResult> {
|
||||
return ctx.tools.execute({
|
||||
signal: testToolSignal,
|
||||
callId: CallId('call-1'),
|
||||
name: RUN_CODE_NAME,
|
||||
arguments: { code },
|
||||
arguments: { code, description: extras.description ?? 'Run the test program' },
|
||||
...extras.agent ? { agent: extras.agent } : {},
|
||||
...extras.signal ? { signal: extras.signal } : {},
|
||||
})
|
||||
@@ -354,6 +359,331 @@ describe('mode-aware wire contribution', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('the sub-dispatch scheduler (native concurrency contract)', () => {
|
||||
/** Register a tool whose calls resolve only when the test releases them; returns live-call telemetry. */
|
||||
function registerGated(ctx: Context, name: string, concurrencySafe: boolean) {
|
||||
const gates: (() => void)[] = []
|
||||
let live = 0
|
||||
let peak = 0
|
||||
const order: string[] = []
|
||||
ctx.tools.register(defineTool({
|
||||
name,
|
||||
description: `Gated tool ${name}.`,
|
||||
parameters: { id: { type: 'string', required: true } },
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args, value) => [{ type: 'text', text: value }],
|
||||
},
|
||||
...concurrencySafe ? { isConcurrencySafe: () => true } : {},
|
||||
async execute(args, exec) {
|
||||
order.push(`start:${args.id}`)
|
||||
live++
|
||||
peak = Math.max(peak, live)
|
||||
// Abort-observing like a real tool: the run-scoped abort releases the
|
||||
// gate so the bridge's drain reaches quiescence.
|
||||
await new Promise<void>((release) => {
|
||||
gates.push(release)
|
||||
exec.signal.addEventListener('abort', () => { release() }, { once: true })
|
||||
})
|
||||
live--
|
||||
order.push(`end:${args.id}`)
|
||||
return `${name}:${args.id}`
|
||||
},
|
||||
}))
|
||||
const release = (): void => { gates.shift()?.() }
|
||||
const releaseAll = (): void => { while (gates.length > 0) gates.shift()!() }
|
||||
return { order, release, releaseAll, peakLive: () => peak, pending: () => gates.length }
|
||||
}
|
||||
|
||||
it('overlaps concurrency-safe calls under Promise.all and logs a start event per dispatch', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const gated = registerGated(ctx, 'safe_read', true)
|
||||
const { agent, events } = fakeAgent()
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
const all = Promise.all([
|
||||
tools.safe_read!({ id: 'a' }),
|
||||
tools.safe_read!({ id: 'b' }),
|
||||
tools.safe_read!({ id: 'c' }),
|
||||
])
|
||||
// All three must be START-able without any completion (overlap proof).
|
||||
await expect.poll(() => gated.pending()).toBe(3)
|
||||
gated.releaseAll()
|
||||
return { logs: [], value: (await all).map(String).join(',') }
|
||||
}
|
||||
const result = await runCode(ctx, 'program', { agent })
|
||||
expect(result.isError).toBe(false)
|
||||
expect(gated.peakLive()).toBe(3)
|
||||
if (result.isError) throw new Error('expected success')
|
||||
expect(result.value).toMatchObject({ result: 'safe_read:a,safe_read:b,safe_read:c' })
|
||||
// One start per dispatch, paired with its settle by subCallId, starts in submission order.
|
||||
const starts = events.filter(event => event.type === 'tool/code-dispatch-start').map(event => event.data as { subCallId: string })
|
||||
const settles = events.filter(event => event.type === 'tool/code-dispatch').map(event => event.data as { subCallId: string })
|
||||
expect(starts.map(start => start.subCallId)).toEqual(['call-1:code:1', 'call-1:code:2', 'call-1:code:3'])
|
||||
expect(new Set(settles.map(settle => settle.subCallId))).toEqual(new Set(starts.map(start => start.subCallId)))
|
||||
})
|
||||
|
||||
it('an exclusive call bars overlap: safe calls drain first, it runs alone, later calls wait', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const safe = registerGated(ctx, 'safe_read', true)
|
||||
const unsafe = registerGated(ctx, 'writer', false)
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
const reads = [tools.safe_read!({ id: 'r1' }), tools.safe_read!({ id: 'r2' })]
|
||||
const write = tools.writer!({ id: 'w' })
|
||||
const tail = tools.safe_read!({ id: 'r3' })
|
||||
await expect.poll(() => safe.pending()).toBe(2)
|
||||
// The exclusive call must NOT have started while the pool is live.
|
||||
expect(unsafe.pending()).toBe(0)
|
||||
safe.releaseAll()
|
||||
await expect.poll(() => unsafe.pending()).toBe(1)
|
||||
// The trailing safe call must NOT start while the exclusive one runs.
|
||||
expect(safe.pending()).toBe(0)
|
||||
unsafe.release()
|
||||
await expect.poll(() => safe.pending()).toBe(1)
|
||||
safe.releaseAll()
|
||||
await Promise.all([...reads, write, tail])
|
||||
return { logs: [], value: 'ordered' }
|
||||
}
|
||||
const result = await runCode(ctx, 'program')
|
||||
expect(result.isError).toBe(false)
|
||||
expect(safe.order.slice(0, 2)).toEqual(['start:r1', 'start:r2'])
|
||||
expect(unsafe.order).toEqual(['start:w', 'end:w'])
|
||||
// r3 started only after w ended.
|
||||
expect(safe.order.indexOf('start:r3')).toBeGreaterThan(safe.order.indexOf('end:r1'))
|
||||
})
|
||||
|
||||
it('maxParallelSubCalls caps the overlap window', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code', maxParallelSubCalls: 2 })
|
||||
const gated = registerGated(ctx, 'safe_read', true)
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
const all = Promise.all([
|
||||
tools.safe_read!({ id: 'a' }),
|
||||
tools.safe_read!({ id: 'b' }),
|
||||
tools.safe_read!({ id: 'c' }),
|
||||
])
|
||||
await expect.poll(() => gated.pending()).toBe(2)
|
||||
// The third call waits for a slot.
|
||||
expect(gated.pending()).toBe(2)
|
||||
gated.release()
|
||||
await expect.poll(() => gated.pending()).toBe(2)
|
||||
gated.releaseAll()
|
||||
await all
|
||||
return { logs: [], value: 'capped' }
|
||||
}
|
||||
const result = await runCode(ctx, 'program')
|
||||
if (result.isError) console.error('CAP-FAIL:', (result.content[0] as { text: string }).text)
|
||||
expect(result.isError).toBe(false)
|
||||
expect(gated.peakLive()).toBe(2)
|
||||
})
|
||||
|
||||
it('a tool unregistered between binding enumeration and dispatch fails as unknown tool', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const calls: unknown[] = []
|
||||
const dispose = ctx.tools.register(defineTool({
|
||||
name: 'ephemeral',
|
||||
description: 'Unregistered between binding enumeration and dispatch.',
|
||||
parameters: {},
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args, value) => [{ type: 'text', text: value }],
|
||||
},
|
||||
execute() {
|
||||
calls.push('ran')
|
||||
return Promise.resolve('ok')
|
||||
},
|
||||
}))
|
||||
runtime.behavior = async (request) => {
|
||||
// The binding exists (enumerated at run start); the registry mutation
|
||||
// makes prepare resolve UNKNOWN_TOOL as a final-result, which commits
|
||||
// through scheduler.finish (no post-execute).
|
||||
dispose()
|
||||
const message = await request.bindings[0]!.functions.ephemeral!({})
|
||||
.then(() => 'resolved', (error: unknown) => error instanceof Error ? error.message : String(error))
|
||||
return { logs: [], value: message }
|
||||
}
|
||||
const result = await runCode(ctx, 'program')
|
||||
expect(result.isError).toBe(false)
|
||||
if (result.isError) throw new Error('expected success')
|
||||
expect(result.value).toMatchObject({ result: 'unknown tool "ephemeral"' })
|
||||
expect(calls).toEqual([])
|
||||
})
|
||||
|
||||
it('ordered pre-execute never overlaps: a slow policy on one call delays the next start', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const gated = registerGated(ctx, 'safe_read', true)
|
||||
const stages: string[] = []
|
||||
let releaseGate: (() => void) | undefined
|
||||
ctx.on('tools/pre-execute', async (preExec, next) => {
|
||||
if (preExec.name !== 'safe_read') return next()
|
||||
stages.push(`pre-enter:${String(preExec.callId)}`)
|
||||
if (releaseGate === undefined) {
|
||||
// The FIRST call's policy awaits an asynchronous decision.
|
||||
await new Promise<void>((resolve) => { releaseGate = resolve })
|
||||
}
|
||||
stages.push(`pre-exit:${String(preExec.callId)}`)
|
||||
return next()
|
||||
})
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
const all = Promise.all([tools.safe_read!({ id: 'a' }), tools.safe_read!({ id: 'b' })])
|
||||
// Both submissions are in; the second pre-execute must NOT have entered
|
||||
// while the first is still awaiting its policy decision.
|
||||
await expect.poll(() => stages.length).toBeGreaterThanOrEqual(1)
|
||||
expect(stages).toEqual(['pre-enter:call-1:code:1'])
|
||||
releaseGate!()
|
||||
await expect.poll(() => gated.pending()).toBe(2)
|
||||
gated.releaseAll()
|
||||
await all
|
||||
return { logs: [], value: 'ordered-prepare' }
|
||||
}
|
||||
const result = await runCode(ctx, 'program')
|
||||
expect(result.isError).toBe(false)
|
||||
expect(stages).toEqual([
|
||||
'pre-enter:call-1:code:1', 'pre-exit:call-1:code:1',
|
||||
'pre-enter:call-1:code:2', 'pre-exit:call-1:code:2',
|
||||
])
|
||||
})
|
||||
|
||||
it('an exclusive call holds its barrier through post-execute: the next start waits for the commit', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const writer = registerGated(ctx, 'writer', false)
|
||||
const reader = registerGated(ctx, 'safe_read', true)
|
||||
const stages: string[] = []
|
||||
let releasePost: (() => void) | undefined
|
||||
ctx.on('tools/post-execute', async (postExec, _result, next): Promise<PostToolDecision> => {
|
||||
if (postExec.name === 'writer') {
|
||||
stages.push('post-enter:writer')
|
||||
await new Promise<void>((resolve) => { releasePost = resolve })
|
||||
stages.push('post-exit:writer')
|
||||
}
|
||||
return next()
|
||||
})
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
const w = tools.writer!({ id: 'w' })
|
||||
const r = tools.safe_read!({ id: 'r' })
|
||||
await expect.poll(() => writer.pending()).toBe(1)
|
||||
writer.release()
|
||||
// The writer's body is done and its async post-execute is running; the
|
||||
// parallel read must not have STARTED (no pre/body) while the exclusive
|
||||
// call's pipeline is still open.
|
||||
await expect.poll(() => stages).toContain('post-enter:writer')
|
||||
expect(reader.pending()).toBe(0)
|
||||
releasePost!()
|
||||
await w
|
||||
await expect.poll(() => reader.pending()).toBe(1)
|
||||
reader.releaseAll()
|
||||
await r
|
||||
return { logs: [], value: 'barrier-through-commit' }
|
||||
}
|
||||
const result = await runCode(ctx, 'program')
|
||||
expect(result.isError).toBe(false)
|
||||
expect(stages).toEqual(['post-enter:writer', 'post-exit:writer'])
|
||||
})
|
||||
|
||||
it('run settlement drains a commit already in progress: the settle event lands inside the turn', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const gated = registerGated(ctx, 'safe_read', true)
|
||||
const { agent, events } = fakeAgent()
|
||||
let releasePost: (() => void) | undefined
|
||||
ctx.on('tools/post-execute', async (postExec, _result, next): Promise<PostToolDecision> => {
|
||||
if (postExec.name === 'safe_read') {
|
||||
await new Promise<void>((resolve) => { releasePost = resolve })
|
||||
}
|
||||
return next()
|
||||
})
|
||||
runtime.behavior = async (request) => {
|
||||
// Fire-and-forget: the program returns while the sub-call's async
|
||||
// post-execute commit is mid-flight.
|
||||
request.bindings[0]!.functions.safe_read!({ id: 'a' }).catch(() => 'run-over')
|
||||
await expect.poll(() => gated.pending()).toBe(1)
|
||||
gated.release()
|
||||
await expect.poll(() => releasePost !== undefined).toBe(true)
|
||||
queueMicrotask(() => { releasePost!() })
|
||||
return { logs: [], value: 'returned-early' }
|
||||
}
|
||||
const result = await runCode(ctx, 'program', { agent })
|
||||
expect(result.isError).toBe(false)
|
||||
// The drain awaited the in-progress commit: the settle event exists and
|
||||
// preceded the run_code turn closing (all appends happen inside
|
||||
// execute()). The run's settlement aborted the sub-call's signal while
|
||||
// its post-execute was mid-flight, so the native cancellation contract
|
||||
// replaces the successful outcome with the aborted result — the event is
|
||||
// still durable and in-turn, which is the invariant under test.
|
||||
const settles = events.filter(event => event.type === 'tool/code-dispatch')
|
||||
expect(settles).toHaveLength(1)
|
||||
expect(settles[0]?.data).toMatchObject({ name: 'safe_read', isError: true })
|
||||
})
|
||||
|
||||
it('post-execute and context commitment stay in submission order under out-of-order completion', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const gated = registerGated(ctx, 'safe_read', true)
|
||||
const postOrder: string[] = []
|
||||
ctx.on('tools/post-execute', async (postExec, _result, next): Promise<PostToolDecision> => {
|
||||
if (postExec.name === 'safe_read') {
|
||||
postOrder.push(String(postExec.callId))
|
||||
return {
|
||||
kind: 'accept' as const,
|
||||
additionalContexts: [{
|
||||
content: [{ type: 'text' as const, text: `ctx:${String(postExec.callId)}` }],
|
||||
source: { kind: 'plugin' as const, plugin: 'order-probe' },
|
||||
}],
|
||||
}
|
||||
}
|
||||
return next()
|
||||
})
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
const all = Promise.all([tools.safe_read!({ id: 'a' }), tools.safe_read!({ id: 'b' })])
|
||||
await expect.poll(() => gated.pending()).toBe(2)
|
||||
// Complete b FIRST (out of submission order), then a.
|
||||
gated.release() // releases a (FIFO gate) — invert: release twice reversed is not possible;
|
||||
gated.releaseAll()
|
||||
await all
|
||||
return { logs: [], value: 'ordered-commit' }
|
||||
}
|
||||
const result = await runCode(ctx, 'program')
|
||||
expect(result.isError).toBe(false)
|
||||
// Post-execute observed submission order regardless of completion interleave.
|
||||
expect(postOrder).toEqual(['call-1:code:1', 'call-1:code:2'])
|
||||
// Deferred contexts reach the outer result in the same order.
|
||||
expect(result.additionalContexts?.map(c => (c.content[0] as { text: string }).text))
|
||||
.toEqual(['ctx:call-1:code:1', 'ctx:call-1:code:2'])
|
||||
})
|
||||
|
||||
it('a queued-unstarted call abandoned by run settlement logs no start event', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const gated = registerGated(ctx, 'writer', false)
|
||||
const { agent, events } = fakeAgent()
|
||||
const abandoned: string[] = []
|
||||
runtime.behavior = async (request) => {
|
||||
const tools = request.bindings[0]!.functions
|
||||
// First exclusive call occupies the pool; the second queues unstarted.
|
||||
// Both rejections are captured (abandonment fires only at settlement,
|
||||
// AFTER this program has already failed — awaiting it here would deadlock).
|
||||
tools.writer!({ id: 'w1' }).catch(() => 'settled-under-abort')
|
||||
tools.writer!({ id: 'w2' }).catch((error: unknown) => {
|
||||
abandoned.push(error instanceof Error ? error.message : String(error))
|
||||
})
|
||||
await expect.poll(() => gated.pending()).toBe(1)
|
||||
// Fail the program while w1 is in flight and w2 is queued unstarted.
|
||||
throw new Error('program failed with a queued call')
|
||||
}
|
||||
const result = await runCode(ctx, 'program', { agent })
|
||||
expect(result.isError).toBe(true)
|
||||
const starts = events.filter(event => event.type === 'tool/code-dispatch-start').map(event => (event.data as { subCallId: string }).subCallId)
|
||||
const settles = events.filter(event => event.type === 'tool/code-dispatch').map(event => (event.data as { subCallId: string }).subCallId)
|
||||
// w1 started and settled under the abort; w2 never started and never
|
||||
// settled — no start event, no settle event, binding rejected with the
|
||||
// abandonment message at drain time.
|
||||
expect(starts).toEqual(['call-1:code:1'])
|
||||
expect(settles).toEqual(['call-1:code:1'])
|
||||
expect(abandoned).toEqual(['run_code run is over (run_code settled); writer tool call abandoned'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('the run_code dispatch bridge', () => {
|
||||
it('bridges tool calls, returns only the curated output, and logs one event per dispatch', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
@@ -374,8 +704,14 @@ describe('the run_code dispatch bridge', () => {
|
||||
expect(calls).toEqual([{ value: 'one' }, { value: 'two' }])
|
||||
const dispatches = events.filter(event => event.type === 'tool/code-dispatch')
|
||||
expect(dispatches.map(event => event.data)).toEqual([
|
||||
{ parentCallId: 'call-1', subCallId: 'call-1:code:1', name: 'echo', arguments: { value: 'one' }, isError: false, resultSummary: 'echo:one' },
|
||||
{ parentCallId: 'call-1', subCallId: 'call-1:code:2', name: 'echo', arguments: { value: 'two' }, isError: false, resultSummary: 'echo:two' },
|
||||
{
|
||||
parentCallId: 'call-1', subCallId: 'call-1:code:1', name: 'echo',
|
||||
arguments: { value: 'one' }, isError: false, content: [{ type: 'text', text: 'echo:one' }],
|
||||
},
|
||||
{
|
||||
parentCallId: 'call-1', subCallId: 'call-1:code:2', name: 'echo',
|
||||
arguments: { value: 'two' }, isError: false, content: [{ type: 'text', text: 'echo:two' }],
|
||||
},
|
||||
])
|
||||
expect(result.meta).toBeUndefined()
|
||||
})
|
||||
@@ -465,6 +801,52 @@ describe('the run_code dispatch bridge', () => {
|
||||
expect(result.content[0]).toEqual({ type: 'text', text: 'caught: deliberate failure' })
|
||||
})
|
||||
|
||||
it('a throwing tools/code-dispatch-log listener is contained: the unshaped content is logged', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
registerEcho(ctx)
|
||||
ctx.on('tools/code-dispatch-log', () => { throw new Error('shaper exploded') })
|
||||
const { agent, events } = fakeAgent()
|
||||
runtime.behavior = async (request) => {
|
||||
const value = await request.bindings[0]!.functions.echo!({ value: 'x' })
|
||||
return { logs: [], value: value as string }
|
||||
}
|
||||
const result = await runCode(ctx, 'program', { agent })
|
||||
expect(result.isError).toBe(false)
|
||||
const settle = events.find(event => event.type === 'tool/code-dispatch')
|
||||
expect(settle?.data).toMatchObject({ name: 'echo', isError: false, content: [{ type: 'text', text: 'echo:x' }] })
|
||||
})
|
||||
|
||||
it('a throwing tools/pre-execute listener settles the sub-call without post-execute', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const calls = registerEcho(ctx)
|
||||
const postExecuted: string[] = []
|
||||
ctx.on('tools/pre-execute', (exec, next) => {
|
||||
if (exec.name === 'echo') throw new Error('gate exploded')
|
||||
return next()
|
||||
})
|
||||
ctx.on('tools/post-execute', (exec, _result, next): Promise<PostToolDecision> => {
|
||||
if (exec.name === 'echo') postExecuted.push(exec.name)
|
||||
return next()
|
||||
})
|
||||
const { agent, events } = fakeAgent()
|
||||
runtime.behavior = async (request) => {
|
||||
const message = await request.bindings[0]!.functions.echo!({ value: 'x' })
|
||||
.then(() => 'resolved', (error: unknown) => error instanceof Error ? error.message : String(error))
|
||||
return { logs: [], value: message }
|
||||
}
|
||||
const result = await runCode(ctx, 'program', { agent })
|
||||
expect(result.isError).toBe(false)
|
||||
if (result.isError) throw new Error('expected success')
|
||||
expect(result.value).toMatchObject({ result: 'gate exploded' })
|
||||
// The pipeline failure is final: the body never ran and post-execute was
|
||||
// skipped, yet the settle event still carries the error outcome.
|
||||
expect(calls).toEqual([])
|
||||
expect(postExecuted).toEqual([])
|
||||
const settles = events.filter(event => event.type === 'tool/code-dispatch')
|
||||
expect(settles).toHaveLength(1)
|
||||
expect(settles[0]?.data).toMatchObject({ name: 'echo', isError: true })
|
||||
})
|
||||
|
||||
it('a tools/pre-execute deny reaches the program as a binding rejection', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
registerEcho(ctx)
|
||||
@@ -693,19 +1075,26 @@ describe('the run_code dispatch bridge', () => {
|
||||
expect((result.content[0] as { text: string }).text).toContain('requires a code runtime')
|
||||
})
|
||||
|
||||
it('presents the program as the execute-card title', async () => {
|
||||
it('presents the model-authored description as the execute-card title over the program input', async () => {
|
||||
const { ctx } = await setup({ mode: 'code' })
|
||||
const tool = ctx.tools.get(RUN_CODE_NAME)!
|
||||
// The program is the title, mirroring how command tools label their cards
|
||||
// with the command while retaining the same value in the expanded input.
|
||||
expect(tool.presentCall?.({ code: 'return 1' })).toEqual({
|
||||
// The description labels the card (the bash description precedent); the
|
||||
// program itself remains the expanded raw input.
|
||||
expect(tool.presentCall?.({ code: 'return 1', description: 'Return the constant one' })).toEqual({
|
||||
card: 'generic',
|
||||
title: 'return 1',
|
||||
title: 'Return the constant one',
|
||||
kind: 'execute',
|
||||
rawInput: 'return 1',
|
||||
})
|
||||
})
|
||||
|
||||
it('rejects a whitespace-only description with a structured isError', async () => {
|
||||
const { ctx } = await setup({ mode: 'code' })
|
||||
const result = await runCode(ctx, 'return 1', { description: ' ' })
|
||||
expect(result.isError).toBe(true)
|
||||
expect((result.content[0] as { text: string }).text).toContain('invalid description')
|
||||
})
|
||||
|
||||
it.each([
|
||||
['logs only', { logs: ['printed'] }, 'printed'],
|
||||
['result only', { logs: [], value: 'returned' }, 'returned'],
|
||||
@@ -759,7 +1148,7 @@ describe('the run_code dispatch bridge', () => {
|
||||
expect('presentResult' in tool).toBe(false)
|
||||
})
|
||||
|
||||
it('renders non-text sub-result blocks as placeholders and truncates long event summaries', async () => {
|
||||
it('logs the complete sub-result content verbatim, non-text blocks and long text included', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
const { agent, events } = fakeAgent()
|
||||
const long = 'x'.repeat(300)
|
||||
@@ -786,58 +1175,10 @@ describe('the run_code dispatch bridge', () => {
|
||||
expect(result.isError).toBe(false)
|
||||
expect((result.content[0] as { text: string }).text).toBe('mixed-value')
|
||||
const dispatch = events.find(event => event.type === 'tool/code-dispatch')?.data as SessionEventMap['tool/code-dispatch']
|
||||
expect(dispatch.resultSummary.length).toBe(201)
|
||||
expect(dispatch.resultSummary.endsWith('…')).toBe(true)
|
||||
})
|
||||
|
||||
it('normalizes the session workspace root before bounding durable result summaries', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'workspace_path',
|
||||
description: 'Return a path beneath the session workspace.',
|
||||
parameters: {},
|
||||
output: {
|
||||
schema: { type: 'string' },
|
||||
render: (_args, value) => [{ type: 'text', text: value }],
|
||||
},
|
||||
execute(_args, exec) {
|
||||
const cwd = exec.agent?.session.header.cwd ?? ''
|
||||
return Promise.resolve(`<path>${cwd}/nested/task.txt</path>\n${'x'.repeat(240)}`)
|
||||
},
|
||||
}))
|
||||
runtime.behavior = async request => ({
|
||||
logs: [],
|
||||
value: await request.bindings[0]!.functions.workspace_path!({}),
|
||||
})
|
||||
|
||||
const short = fakeAgent({ cwd: '/tmp/workspace' })
|
||||
const long = fakeAgent({ cwd: `/tmp/${'long-segment/'.repeat(30)}workspace` })
|
||||
const shortResult = await runCode(ctx, 'program', { agent: short.agent })
|
||||
const longResult = await runCode(ctx, 'program', { agent: long.agent })
|
||||
const shortDispatch = short.events[0]!.data as SessionEventMap['tool/code-dispatch']
|
||||
const longDispatch = long.events[0]!.data as SessionEventMap['tool/code-dispatch']
|
||||
|
||||
expect(shortResult.content).not.toEqual(longResult.content)
|
||||
expect(shortDispatch.resultSummary).toBe(longDispatch.resultSummary)
|
||||
expect(shortDispatch.resultSummary).toHaveLength(201)
|
||||
expect(shortDispatch.resultSummary).toMatch(/^<path>\.\/nested\/task\.txt<\/path>\n.+…$/)
|
||||
})
|
||||
|
||||
it('leaves result summaries unchanged when a session cwd is absent or is the filesystem root', async () => {
|
||||
const { ctx, runtime } = await setup({ mode: 'code' })
|
||||
registerEcho(ctx)
|
||||
runtime.behavior = async request => ({
|
||||
logs: [],
|
||||
value: await request.bindings[0]!.functions.echo!({ value: '/workspace/value' }),
|
||||
})
|
||||
|
||||
const absent = fakeAgent({})
|
||||
const root = fakeAgent({ cwd: '/' })
|
||||
await runCode(ctx, 'program', { agent: absent.agent })
|
||||
await runCode(ctx, 'program', { agent: root.agent })
|
||||
|
||||
expect((absent.events[0]!.data as SessionEventMap['tool/code-dispatch']).resultSummary).toBe('echo:/workspace/value')
|
||||
expect((root.events[0]!.data as SessionEventMap['tool/code-dispatch']).resultSummary).toBe('echo:/workspace/value')
|
||||
expect(dispatch.content).toEqual([
|
||||
{ type: 'text', text: long },
|
||||
{ type: 'reasoning', text: 'hidden' },
|
||||
])
|
||||
})
|
||||
|
||||
it('rejects undefined, getter-throwing, exotic, and unrepresentable binding arguments before dispatch', async () => {
|
||||
@@ -1067,13 +1408,27 @@ describe('the run_code dispatch bridge', () => {
|
||||
name: 'echo',
|
||||
arguments: { value: 'x' },
|
||||
isError: false,
|
||||
resultSummary: 'echo:x',
|
||||
content: [{ type: 'text', text: 'echo:x' }],
|
||||
})
|
||||
const derived = session.deriveMessages()
|
||||
expect(derived).toHaveLength(1)
|
||||
expect(derived[0]?.role).toBe('user')
|
||||
})
|
||||
|
||||
it('direct construction rejects a non-positive parallel sub-call cap at load', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt, {})
|
||||
expect(() => new ToolRegistry(ctx, { mode: 'code', maxParallelSubCalls: 0 }))
|
||||
.toThrow('maxParallelSubCalls must be a positive integer')
|
||||
})
|
||||
|
||||
it('direct construction in code mode defaults the parallel sub-call cap', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt, {})
|
||||
const registry = new ToolRegistry(ctx, { mode: 'code' })
|
||||
expect(registry.get(RUN_CODE_NAME)).toBeDefined()
|
||||
})
|
||||
|
||||
it('defaults to native mode under direct construction with no config', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt, {})
|
||||
|
||||
@@ -144,7 +144,7 @@ describe('renderToolsSdk', () => {
|
||||
// The fixed instruction lines the model relies on.
|
||||
expect(text).toContain('erasable syntax only')
|
||||
expect(text).toContain('rejects with `ToolCallError`')
|
||||
expect(text).toContain('sequentially, even under `Promise.all`')
|
||||
expect(text).toContain('MAY overlap under `Promise.all`')
|
||||
expect(text).toContain('lossless JSON')
|
||||
})
|
||||
|
||||
|
||||
Reference in New Issue
Block a user