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

Merge remote-tracking branch 'origin/master' into xtr/agent-loop-message-machine

Browse Source 
# Conflicts:
#	docs/config-catalog.md
#	docs/cordis-catalog/services.md
#	docs/core-data-structures/tools.i18n.yaml
#	packages/core/tools/README.i18n.yaml
This commit is contained in:
_Kerman
2026-07-27 01:44:13 +08:00
parent 107e05e79b fc33543285
commit 92ff4b0eea
76 changed files with 2157 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/tools/src/code-mode.ts
+61 -19
View File
@@ -13,7 +13,7 @@ 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 { ToolDefinition, ToolExecutionResult, ToolRegistry, ToolRunContext } from './index.ts'
import type { CodeDispatchLog, ToolDefinition, ToolExecutionResult, ToolRegistry, ToolRunContext } from './index.ts'
declare module '@deepseek-ai/dsh-session' {
interface SessionEventMap {
@@ -186,6 +186,20 @@ function renderValue(value: JsonValue): string {
/** 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
@@ -194,13 +208,11 @@ type RunCodeOutput = { logs: string[]; result?: JsonValue }
* 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 maxParallel - the run's overlap cap for parallel-classified
* sub-calls (the registry passes its validated `maxParallelSubCalls`).
* @param options - the registry-private capabilities described above.
* @returns the registry-ready definition.
*/
export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () => CodeRuntime, maxParallel: number): ToolDefinition {
export function createRunCodeTool(registry: ToolRegistry, options: RunCodeBridgeOptions): ToolDefinition {
const { requireRuntime, maxParallel, shapeDispatchLog } = options
return defineTool({
name: RUN_CODE_NAME,
description:
@@ -279,6 +291,8 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
}
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
@@ -357,6 +371,9 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
// 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
@@ -387,22 +404,41 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
| { kind: 'post-result' | 'final-result'; exec: ToolRunContext; result: ToolExecutionResult }
| undefined
const settle = (result: ToolExecutionResult): void => {
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,
// The registry deep-froze this projection at result finalization;
// append snapshots it again, so the log copy stays detached.
content: result.content,
})
// 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(),
@@ -450,6 +486,12 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
// failure cannot stop the turn through a recovering program.
if (result.concludesTurn) exec.concludeTurn()
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()
packages/core/tools/src/index.ts
+63 -1
View File
@@ -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;
@@ -704,7 +741,11 @@ export class ToolRegistry extends Service {
// the filterable global/scoped capability layers.
this.codeTransport = this.mode === 'native'
? undefined
: createRunCodeTool(this, () => this.requireCodeRuntime(), resolveMaxParallelSubCalls(config.maxParallelSubCalls))
: 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({
@@ -955,6 +996,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