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

fix(tools): bound the shaped-append side channel; total error containment; recorded spill snapshot

Browse Source 
Responding to ds-review-bot round 2 on #661:

- logWork is bounded: past maxParallelSubCalls pending shaped-append tasks
  the ordered commit lane holds (Promise.race drains one), so a slow spill
  backend backpressures the run instead of accumulating unbounded pending
  I/O and retained results. Tasks self-remove on settlement; run
  settlement still drains every task inside the open turn. New spill test
  drives three oversized reads against a hung backend at cap 1 and proves
  the third dispatch cannot start until a save drains.
- shapeDispatchLog's catch uses errorMessage() (total), so a thrown value
  with a throwing toString cannot escape the containment and lose the
  settle event.
- CodeDispatchLog.content documented as the RENDERED result projection
  (native tool/result vocabulary), not what the program received — the
  program gets the structured value; doc pair + type-equiv re-synced.
- New RECORDED tui-agent snapshot scenario code-mode-dispatch-spill: the
  real Loader-visible composition (worker runtime + spill-local + policy)
  drives an oversized bash sub-call end-to-end; replay proves the durable
  dispatch copy is bounded to preview + locator while the program value
  stays whole (the outer result carries just the line count).

Agent Note updated (both languages).
This commit is contained in:
Tianyi Cui
2026-07-26 18:29:09 +08:00
parent 442a3dd884
commit c3c10820ba
14 changed files with 384 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
packages/core/tools/src/code-mode.ts
View File

@@ -359,12 +359,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.
while (logWork.size > 0) {
const pending = [...logWork]
await Promise.allSettled(pending)
for (const done of pending) logWork.delete(done)
}
// 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
@@ -406,7 +403,7 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
: { isError: false, value: result.value })
const agent = exec.agent
if (agent === undefined) return
logWork.add((async () => {
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.
@@ -428,7 +425,8 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
isError: result.isError,
content: logged,
})
})())
})().finally(() => { logWork.delete(task) })
logWork.add(task)
}
pendingQueue.push({
flight: Promise.resolve(),
@@ -470,6 +468,12 @@ export function createRunCodeTool(registry: ToolRegistry, requireRuntime: () =>
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()

8
packages/core/tools/src/index.ts
View File

@@ -289,8 +289,10 @@ export type ToolExecutionMode =
* 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. The complete `content` is what the program
* already received; only the `tool/code-dispatch` event's copy changes.
* 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. */
@@ -991,7 +993,7 @@ export class ToolRegistry extends Service {
() => Promise.resolve(dispatch.content),
)
} catch (error: unknown) {
this.ctx.logger.warn(`tools: code-dispatch-log listener failed for ${dispatch.name}: ${String(error)}; logging the unshaped content`)
this.ctx.logger.warn(`tools: code-dispatch-log listener failed for ${dispatch.name}: ${errorMessage(error)}; logging the unshaped content`)
return dispatch.content
}
}

60
packages/spill/spill-policy/tests/spill-policy.spec.ts
View File

@@ -29,9 +29,12 @@ const testToolSignal = new AbortController().signal
class StubStore extends SpillStore {
saves: SaveTextSpill[] = []
fail = false
/** Per-save hang hook: each call awaits the returned promise before completing. */
gate: (() => Promise<void>) | undefined
async saveText(input: SaveTextSpill): Promise<SpillRef> {
if (this.fail) throw new Error('disk full')
await this.gate?.()
this.saves.push(input)
return {
locator: SpillLocator(`/spill/${input.suggestedName}`),
@@ -367,6 +370,63 @@ describe('the durable dispatch-log arm', () => {
expect(smallAfterHuge).toBe(true)
})
it('a sustained slow backend backpressures the run instead of accumulating unbounded log tasks', async () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
// Cap 1: once the hung shaped-append backlog exceeds the cap, the ordered
// lane holds inside the second commit, so the THIRD dispatch cannot start
// until a pending save drains — the bound is observable as its missing
// start event.
await ctx.plugin(ToolRegistry, { mode: 'code', maxParallelSubCalls: 1 })
await ctx.plugin(StubStore)
await ctx.plugin(SpillPolicy, { maxInlineBytes: 100 })
await ctx.plugin(WorkerCodeRuntime, {})
const store = ctx.spillStore as StubStore
const releases: (() => void)[] = []
store.gate = () => new Promise<void>((resolve) => { releases.push(resolve) })
const events: { type: string; data: unknown }[] = []
const agent = {
session: {
header: { id: SessionId('dispatch-spill-bound'), cwd: '/workspace' },
append: (type: string, data: unknown) => { events.push({ type, data }) },
},
}
ctx.tools.register(textTool('huge_read', 'H'.repeat(2_000)))
const started = (n: number): boolean => events.some(event => event.type === 'tool/code-dispatch-start'
&& (event.data as { subCallId: string }).subCallId.endsWith(`:code:${n}`))
const runPromise = ctx.tools.execute({
signal: testToolSignal,
callId: CallId('parent-bound'),
name: 'run_code',
arguments: {
code: 'await tools.huge_read({}); await tools.huge_read({}); await tools.huge_read({}); return "done"',
description: 'Three oversized reads against a hung backend',
},
agent: agent as never,
})
// Two hung saves = backlog above the cap: the lane must hold before
// starting dispatch 3.
await vi.waitFor(() => {
if (releases.length < 2) throw new Error('second hung save not reached yet')
})
expect(started(2)).toBe(true)
expect(started(3)).toBe(false)
releases.shift()!()
// Draining one pending save releases the lane; dispatch 3 starts.
await vi.waitFor(() => {
if (!started(3)) throw new Error('third dispatch not started yet')
})
while (releases.length > 0) releases.shift()!()
const result = await runPromise
expect(result.isError).toBe(false)
await vi.waitFor(() => {
if (releases.length > 0) { while (releases.length > 0) releases.shift()!() }
if (events.filter(event => event.type === 'tool/code-dispatch').length !== 3) {
throw new Error('settle events still pending')
}
})
})
it('a saveText failure keeps the complete content in the durable log (best-effort)', async () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)