Files
deepseek-harness/packages/core/tools/src/code-mode.ts
T
Tianyi Cui 443e2bc509 refactor(tools): shapeDispatchLog off the public registry surface
Responding to review on #661: a public method on the generic ToolRegistry
service whose only caller is the run_code bridge was ad-hoc surface
widening. The bridge now receives it as a registry-private capability
closure in RunCodeBridgeOptions (the requireRuntime idiom, alongside the
cap), the method is private, and it leaves the generated service
catalog/API surfaces.

The pattern is now named as a code smell where reviewers look: the
packages/AGENTS.md capability-interface rule gains the inverse-smell
clause (ceiling 660→675 — the list is at capacity and the clause needs
one sentence), and dsh-code-review's capability-fit check tells reviewers
to flag single-consumer public service methods and require the closure
form.
2026-07-26 22:52:15 +08:00

567 lines
27 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* Code Mode `run_code` transport. Programs call the registry's agent-visible
* 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 { 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 { 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 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 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; content: ContentBlock[] }
}
}
/** The model-facing name of the Code Mode tool. */
export const RUN_CODE_NAME = 'run_code'
/** The `tools:sdk` section order: inside the 100199 tool-guidance band, after per-tool guidance sections. */
export const SDK_SECTION_ORDER = 150
/**
* Thrown by `run_code` when the program run itself failed — a program
* exception, a budget expiry, an abort, or substrate death. Extends
* {@link HarnessError} (`code: 'CODE_RUN_FAILED'`); the registry's execution
* pipeline converts it into a structured `isError` result whose text carries
* the failure kind plus the captured logs, so the model can self-correct.
*/
export class CodeRunFailedError extends HarnessError {
constructor(message: string) {
super(message, 'CODE_RUN_FAILED')
this.name = 'CodeRunFailedError'
}
}
/**
* Snapshot one binding call's argument as lossless JSON, then snapshot that
* detached value again so dispatch and logging stay independent without
* reintroducing structured-clone's platform-specific nesting limit.
*/
function jsonNormalizeArgs(value: unknown): { dispatched: unknown; logged: unknown } {
let snapshot: JsonValue | undefined
try {
snapshot = snapshotJsonValue(value) as JsonValue | undefined
} catch (error: unknown) {
throw new Error(`tool arguments must be lossless JSON: ${error instanceof Error ? error.message : String(error)}`)
}
if (snapshot === undefined) {
throw new Error('tool arguments must be lossless JSON (call the tool with an arguments object, e.g. `{}`)')
}
const logged = snapshotJsonValue(snapshot)
/* v8 ignore next -- snapshot is already a detached lossless JSON value. */
if (logged === undefined) {
throw new Error('tool arguments could not be detached for durable logging')
}
return { dispatched: snapshot, logged }
}
/** Two-space JSON presentation, matching the existing shallow `run_code` text contract. */
const JSON_INDENT = ' '
/**
* ECMAScript caps `JSON.stringify`'s `space` string at ten characters. The
* renderer also caps TOTAL indentation there, compacting deeper subtrees, so
* formatted output remains linear in the canonical JSON size.
*/
const MAX_JSON_INDENT_CHARS = 10
/** A pending fragment in the iterative JSON presentation traversal. */
type JsonRenderTask =
| { kind: 'text'; text: string }
| { kind: 'value'; value: JsonValue; depth: number; compact: boolean }
/** Render one non-string JSON root without recursive traversal or unbounded indentation growth. */
function renderJsonValue(value: Exclude<JsonValue, string>): string {
const chunks: string[] = []
const tasks: JsonRenderTask[] = [{ kind: 'value', value, depth: 0, compact: false }]
for (let task = tasks.pop(); task !== undefined; task = tasks.pop()) {
if (task.kind === 'text') {
chunks.push(task.text)
continue
}
const current = task.value
if (current === null || typeof current === 'boolean' || typeof current === 'number') {
chunks.push(String(current))
continue
}
if (typeof current === 'string') {
chunks.push(JSON.stringify(current))
continue
}
const compact = task.compact || (task.depth + 1) * JSON_INDENT.length > MAX_JSON_INDENT_CHARS
const childDepth = task.depth + 1
if (Array.isArray(current)) {
chunks.push('[')
if (current.length === 0) {
chunks.push(']')
continue
}
tasks.push({ kind: 'text', text: compact ? ']' : `\n${JSON_INDENT.repeat(task.depth)}]` })
for (let index = current.length - 1; index >= 0; index--) {
const item = current[index]
/* v8 ignore next -- canonical JsonValue arrays are dense. */
if (item === undefined) throw new Error('cannot render a sparse JSON array')
tasks.push({ kind: 'value', value: item, depth: childDepth, compact })
tasks.push({
kind: 'text',
text: compact
? index === 0 ? '' : ','
: `${index === 0 ? '\n' : ',\n'}${JSON_INDENT.repeat(childDepth)}`,
})
}
continue
}
const keys = Object.keys(current)
chunks.push('{')
if (keys.length === 0) {
chunks.push('}')
continue
}
tasks.push({ kind: 'text', text: compact ? '}' : `\n${JSON_INDENT.repeat(task.depth)}}` })
for (let index = keys.length - 1; index >= 0; index--) {
const key = keys[index]
/* v8 ignore next -- the loop is bounded by the captured key count. */
if (key === undefined) throw new Error('cannot render a missing JSON object key')
const item = current[key]
/* v8 ignore next -- canonical JsonValue records contain no undefined properties. */
if (item === undefined) throw new Error('cannot render an undefined JSON object property')
tasks.push({ kind: 'value', value: item, depth: childDepth, compact })
tasks.push({
kind: 'text',
text: compact
? `${index === 0 ? '' : ','}${JSON.stringify(key)}:`
: `${index === 0 ? '\n' : ',\n'}${JSON_INDENT.repeat(childDepth)}${JSON.stringify(key)}: `,
})
}
}
return chunks.join('')
}
/** Render one present program completion value for the model-facing result text. */
function renderValue(value: JsonValue): string {
return typeof value === 'string' ? value : renderJsonValue(value)
}
/** Canonical value returned by the outer Code Mode transport. */
type RunCodeOutput = { logs: string[]; result?: JsonValue }
/**
* 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 options - the registry-private capabilities described above.
* @returns the registry-ready definition.
*/
export function createRunCodeTool(registry: ToolRegistry, options: RunCodeBridgeOptions): ToolDefinition {
const { requireRuntime, maxParallel, shapeDispatchLog } = options
return defineTool({
name: RUN_CODE_NAME,
description:
'Execute a TypeScript program against the available tools. Write the BODY of an '
+ 'async function (erasable syntax only; top-level `await` and `return` work) and '
+ 'call tools as `await tools.name(args)` per the declarations in the system prompt. '
+ '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: {
type: 'object',
additionalProperties: false,
properties: {
logs: { type: 'array', required: true, items: { type: 'string' } },
result: { type: 'json' },
},
},
render: (_args, value) => {
const rendered = value.result === undefined ? '' : renderValue(value.result)
const parts = [value.logs.join('\n'), rendered].filter(part => part.length > 0)
return [{ type: 'text', text: parts.length > 0 ? parts.join('\n') : '(run_code completed with no output)' }]
},
},
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
// run settles for ANY reason, so an in-flight sub-dispatch is aborted
// (its executor kills on this signal) instead of orphaned, and
// queued-unstarted dispatches are abandoned.
const runController = new AbortController()
const onOuterAbort = (): void => { runController.abort(exec.signal.reason) }
exec.signal.addEventListener('abort', onOuterAbort, { once: true })
let dispatches = 0
// 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 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
// changes across awaits, and a direct `.aborted` re-check after one
// would be narrowed away by control flow analysis.
const runOver = (): boolean => runController.signal.aborted
const binding = (name: string): CodeBindingFunction => async (rawArgs: unknown): Promise<JsonValue> => {
if (runOver()) {
throw new Error(`run_code run is over (${String(runController.signal.reason)}); ${name} not dispatched`)
}
const normalized = jsonNormalizeArgs(rawArgs)
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)
}
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)
},
})
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
// than hand it a result from a run that is over.
if (runOver()) {
throw new Error(`run_code run is over (${String(runController.signal.reason)}); ${name} result discarded`)
}
// The worker turns a binding rejection into ToolCallError and adds
// only the binding name. Native content and internal error metadata
// stay outside the program-facing failure contract.
if (outcome.isError) throw new Error(outcome.message)
return outcome.value
}
// Null-prototype + defineProperty, mirroring the worker-side namespace
// build: a registered tool named `__proto__` must become an ordinary
// own key (a plain-object assignment would hit the prototype setter,
// silently dropping the binding), and the runtime host resolves
// binding names as own properties only.
const functions: Record<string, CodeBindingFunction> = Object.create(null) as Record<string, CodeBindingFunction>
// Enumerate the CALLING AGENT's visible set (scoped tools join,
// restricted globals vanish) — the same view the SDK section declared,
// so a program can bind exactly what its prompt promised; sub-dispatch
// re-resolves per call through the same view (exec.agent threads down).
for (const schema of registry.schemas(exec.agent)) {
if (schema.name === RUN_CODE_NAME) continue
Object.defineProperty(functions, schema.name, { enumerable: true, value: binding(schema.name) })
}
try {
let result: CodeRunResult
try {
result = await runtime.run({
program: args.code,
bindings: [{
global: 'tools',
functions,
errorClass: { name: 'ToolCallError', memberNameProperty: 'toolName' },
}],
signal: runController.signal,
})
} finally {
// 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 drainDispatches()
}
if (result.error) {
const logsText = result.logs.length > 0 ? `\nCaptured output:\n${result.logs.join('\n')}` : ''
throw new CodeRunFailedError(`code run failed (${result.error.kind}): ${result.error.message}${logsText}`)
}
return {
logs: result.logs,
...result.value !== undefined ? { result: result.value } : {},
}
} finally {
exec.signal.removeEventListener('abort', onOuterAbort)
}
},
// 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.description,
kind: 'execute',
rawInput: args.code,
}),
// Deliberately no presentResult: the generic surface fallback keeps this
// title and reads durable result content without duplicating a large raw
// result into the host view payload.
})
}