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

fix(scope): close remaining ownership boundaries

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-12 03:51:55 +08:00
parent 3dca90261c
commit 36b8370027
79 changed files with 3957 additions and 817 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
packages/workflow/workflow-workerthread/README.md
View File

@@ -35,9 +35,9 @@ Values LEAVING the script (hook options/schemas, the script's return) are materi
## Cancellation, death, disposal
Per-run limits: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` posts the cancel to the worker (its hooks start throwing `CANCELLED`; the script dies at its next await) and cancels every host-side child NOW on **both seam channels** — the shared request signal aborts AND each registered child's explicit `cancel()` is called host-side, because the seam leaves a provider free to honor either channel and a worker wedged in a synchronous spin could not relay its own per-child cancel RPCs (those later land as idempotent no-ops). The grace then arms: a run still unsettled `disposeGraceMs` later force-settles `cancelled` and the worker is **terminated**. A cancellation that lands before the body runs (the ready→go handshake) reports `cancelled` without executing anything; a worker `result` racing an in-flight host cancellation reports `cancelled` too (first-wins settlement — the seam-visible result had not settled when cancellation was requested); post-cancel `phase`/`log` narration is suppressed host-side, while cancelled children still deliver their paired `agent-end`.
Per-run limits: a concurrency semaphore (`maxConcurrentAgents`), a total-`agent()` cap (`maxTotalAgents`), and a per-call item cap (`maxItemsPerCall`), all config. `cancel()` posts the cancel to the worker (its hooks start throwing `CANCELLED`; the script dies at its next await) and cancels every host-side child NOW on **both seam channels** — the shared request signal aborts AND each registered child's explicit `cancel()` is called host-side, because the seam leaves a provider free to honor either channel and a worker wedged in a synchronous spin could not relay its own per-child cancel RPCs (those later land as idempotent no-ops). Each provider-owned cancel callback is exception-contained independently, so a broken child cannot prevent peer cancellation or workflow settlement. The grace then arms: a run still unsettled `disposeGraceMs` later force-settles `cancelled` and the worker is **terminated**. A cancellation that lands before the body runs (the ready→go handshake) reports `cancelled` without executing anything; a worker `result` racing an in-flight host cancellation reports `cancelled` too (first-wins settlement — the seam-visible result had not settled when cancellation was requested); post-cancel `phase`/`log` narration is suppressed host-side, while cancelled children still deliver their paired `agent-end`.
A worker that dies unexpectedly (an OOM, a script reaching `process.exit` through the documented vm escape) settles the run `stopReason: 'error'` with the exit diagnostics — or `'cancelled'` when a cancel was in flight — and the host-side child registry is what winds every surviving child down. `dispose()` = cancel + immediate host-driven disposal of every registered child (a wedged worker can relay no dispose RPC, so child teardown overlaps the grace instead of starting after it; the worker's own dispose RPCs join the same per-child disposal) + bounded wait (result, then child-registry quiescence, capped by the grace) + unconditional `worker.terminate()`: the thread never outlives its run. Once a run settles, stray children a script fired without awaiting are cancelled too, and `dispose()` waits for their disposal (bounded by the grace) before returning. `agent-start`/`agent-end` pairing is host-guaranteed the same way: forwarded starts live in a ledger, worker-reported ends pair them on the graceful paths, and the termination paths (grace force-settle, worker death) synthesize the missing ends (outcome `cancelled`) before the run settles — a start still in flight across the force-settle can surface after `workflow/end`, immediately paired the same way.
A worker that dies unexpectedly (an OOM, a script reaching `process.exit` through the documented vm escape) settles the run `stopReason: 'error'` with the exit diagnostics — or `'cancelled'` when a cancel was in flight — and the host-side child registry is what winds every surviving child down. `dispose()` = cancel + immediate host-driven disposal of every registered child (a wedged worker can relay no dispose RPC, so child teardown overlaps the grace instead of starting after it; the worker's own dispose RPCs join the same per-child disposal) + bounded wait (result, then child-registry quiescence, capped by the grace) + unconditional `worker.terminate()`: the thread never outlives its run. Before an ordinary run settlement becomes observable, the host cancels every stray child on both channels too—even a fire-and-forget run still waiting on `started`, for which the worker has no handle yet—and `dispose()` then waits for their disposal (bounded by the grace) before returning. `agent-start`/`agent-end` pairing is host-guaranteed the same way: forwarded starts live in a ledger, worker-reported ends pair them on the graceful paths, and the termination paths (grace force-settle, worker death) synthesize the missing ends (outcome `cancelled`) before the run settles — a start still in flight across the force-settle can surface after `workflow/end`, immediately paired the same way.
**Engine-specific limitations**: worker startup is paid per run; on a termination path `agentsStarted` reports the HOST-observed count (accepted `child-start`s — calls still queued worker-side for a concurrency slot are unknowable then); and a returned promise or thenable resolves per JavaScript semantics BEFORE materialization — that is what makes an un-awaited `return agent('x')` work — with the value-boundary guard applying to the resolution.

1
packages/workflow/workflow-workerthread/package.json
View File

@@ -30,6 +30,7 @@
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-brand": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-subagent": "^0.0.1",
"@deepseek-ai/dsh-tools": "^0.0.1",
"@deepseek-ai/dsh-workflow": "^0.0.1",

64
packages/workflow/workflow-workerthread/src/host.ts
View File

@@ -15,10 +15,13 @@
* terminated — the real kill an in-process engine could not perform).
*
* Children live in a host-side registry (callId → run) as soon as the provider
* accepts them, so cancellation reaches even a pre-publication attempt. The
* host observes `result` immediately but acknowledges the child to the worker
* only after `started` fulfills; readiness failure is a start error and the
* host disposes the attempt because the worker never received a handle. The
* accepts them, so cancellation reaches even a pre-publication attempt. Both
* explicit run cancellation and the shared request signal are driven when the
* workflow is cancelled OR normally settles, so a fire-and-forget child cannot
* survive merely by honoring only one channel. The host observes `result`
* immediately but acknowledges the child to the worker only after `started`
* fulfills; readiness failure is a start error and the host disposes the
* attempt because the worker never received a handle. The
* worker drives disposal by RPC on the graceful path, `dispose()` host-drives
* every registered child's disposal immediately (a wedged worker can relay no
* dispose RPC, and child teardown must overlap the grace, not start after it),
@@ -44,6 +47,7 @@ import type { WorkerOptions } from 'node:worker_threads'
import type { Context } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { assertNever } from '@deepseek-ai/dsh-llm'
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
import type { SubagentRun } from '@deepseek-ai/dsh-subagent'
import type { WorkflowAgentEndInfo, WorkflowAgentInfo, WorkflowMeta, WorkflowResult, WorkflowRun, WorkflowRunId } from '@deepseek-ai/dsh-workflow'
import { renderThrown } from './realm.ts'
@@ -180,11 +184,10 @@ export class WorkerRun implements WorkflowRun {
if (this.settled || this.cancelReason !== undefined) return
this.cancelReason = reason ?? 'workflow cancelled'
this.post(HostToWorkerType.Cancel, { reason: this.cancelReason })
this.controller.abort(this.cancelReason)
// The explicit channel is driven host-side, not left to the worker: a
// provider honoring only run.cancel() must not wait on a wedged worker's
// ChildCancel relay (those later RPCs land as idempotent no-ops).
for (const run of this.children.values()) run.cancel(this.cancelReason)
this.cancelChildren(this.cancelReason)
this.graceTimer = setTimeout(() => {
// The worker may no longer speak (it is about to be terminated): pair
// every stranded start before the run settles, so ends precede
@@ -274,7 +277,10 @@ export class WorkerRun implements WorkflowRun {
this.onChildStart(message.callId, message.request)
break
case WorkerToHostType.ChildCancel:
this.children.get(message.callId)?.cancel(message.reason)
{
const run = this.children.get(message.callId)
if (run !== undefined) this.cancelChild(run, message.reason)
}
break
case WorkerToHostType.ChildDispose:
this.onChildDispose(message.callId)
@@ -322,11 +328,19 @@ export class WorkerRun implements WorkflowRun {
const forwardResult = run.result.then<() => void, () => void>(
(result) => {
try {
const snapshot: ChildResult = structuredClone({
output: result.output,
...result.structured !== undefined ? { structured: result.structured } : {},
stopReason: result.stopReason,
// Capture every provider-owned field once, then materialize the
// worker-bound value in one lossless traversal. A stateful accessor
// cannot validate one result and send another, and an exotic value is
// rejected before any prototype-erasing clone.
const output = result.output
const structured = result.structured
const stopReason = result.stopReason
const snapshot = snapshotJsonValue<ChildResult>({
output,
...structured !== undefined ? { structured } : {},
stopReason,
})
if (snapshot === undefined) throw new TypeError('child result is not losslessly JSON-serializable')
return () => { this.post(HostToWorkerType.ChildSettled, { callId, result: snapshot }) }
} catch (error: unknown) {
const rendered = `workflow child result could not cross the worker boundary: ${renderThrown(error)}`
@@ -416,18 +430,34 @@ export class WorkerRun implements WorkflowRun {
/** Abort + dispose every registered child (worker death / final teardown); disposal is contained, not awaited. */
private reapChildren(reason: string): void {
this.controller.abort(this.cancelReason ?? reason)
const cancellation = this.cancelReason ?? reason
this.cancelChildren(cancellation)
for (const [callId, run] of [...this.children]) {
run.cancel(this.cancelReason ?? reason)
void this.disposeChild(callId, run)
}
}
/** Drive both cancellation channels for every child already accepted by the host. */
private cancelChildren(reason: string): void {
this.controller.abort(reason)
for (const run of this.children.values()) this.cancelChild(run, reason)
}
/** Contain one provider-owned cancel callback so every peer still receives cancellation. */
private cancelChild(run: SubagentRun, reason?: string): void {
try {
run.cancel(reason)
} catch (error: unknown) {
this.ctx.logger.warn(`workflow-workerthread: child cancel failed: ${renderThrown(error)}`)
}
}
private onResult(result: WorkflowResult): void {
// The worker's settle-reap already child-cancel()s every stray; this
// abort fires the seam signal too, for providers that only honor the
// request signal (both channels, on every path).
if (this.cancelReason === undefined) this.controller.abort('workflow settled')
// The worker cancels handles it already received, but a fire-and-forget
// child may still be waiting on readiness and therefore have no worker
// handle. Drive BOTH provider-permitted channels from the host before the
// workflow becomes externally settled.
if (this.cancelReason === undefined) this.cancelChildren('workflow settled')
if (this.cancelReason !== undefined && result.stopReason !== 'cancelled') {
// The script settled while our cancel was crossing the thread boundary
// — the seam-visible result had NOT settled when cancellation was

134
packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts
View File

@@ -366,15 +366,68 @@ describe('dsh-workflow-workerthread', () => {
expect((result.value as { message: string }).message).toContain('backend exploded')
})
it('maps an uncloneable ready-child result to fatal AGENT_RESULT instead of wedging the bridge', async () => {
it('maps a non-JSON ready-child result to fatal AGENT_RESULT instead of wedging the bridge', async () => {
const { ctx, parent } = await setup({
reply: () => ({ output: [], structured: () => { /* deliberately not cloneable */ }, stopReason: 'completed' }),
reply: () => ({ output: [], structured: () => { /* deliberately outside lossless JSON */ }, stopReason: 'completed' }),
})
const result = await run(ctx, parent, scripted(`
try { await agent('p'); return 'unreachable' } catch (e) { return { code: e.code, message: e.message } }
`))
expect(result.value).toMatchObject({ code: 'AGENT_RESULT' })
expect((result.value as { message: string }).message).toContain('could not cross the worker boundary')
expect((result.value as { message: string }).message).toContain('subagent result must be losslessly JSON-serializable')
})
it('contains a non-JSON result even if the injected subagent service violates its normalization contract', async () => {
// SubagentService normally rejects this before the workflow sees it. Stub
// the injected seam itself so the host's defensive worker-boundary guard
// remains independently covered rather than becoming dead, untested code.
const { ctx, parent } = await setup()
const invalid = {
output: [],
structured: () => { /* deliberately outside lossless JSON */ },
stopReason: 'completed',
} as unknown as SubagentResult
const start = vi.spyOn(ctx.subagents, 'start').mockReturnValue({
id: AgentId('raw-invalid-child'),
started: Promise.resolve(),
result: Promise.resolve(invalid),
cancel: () => { /* already settled */ },
dispose: () => Promise.resolve(),
})
const result = await run(ctx, parent, scripted(`
try { await agent('p'); return 'unreachable' } catch (e) { return { code: e.code, message: e.message } }
`))
expect(start).toHaveBeenCalledOnce()
expect(result.value).toMatchObject({ code: 'AGENT_RESULT' })
expect((result.value as { message: string }).message)
.toContain('workflow child result could not cross the worker boundary')
})
it('reads each resolved child-result field once before crossing the worker boundary', async () => {
let structuredReads = 0
class DriftedStructured { readonly value = 'drifted' }
const { ctx, parent } = await setup({
reply: () => ({
output: [],
get structured() {
structuredReads += 1
return structuredReads === 1 ? { value: 'accepted' } : new DriftedStructured()
},
stopReason: 'completed',
}),
})
const result = await run(ctx, parent, scripted(`
const found = await agent('p', {
schema: { type: 'object', properties: { value: { type: 'string' } }, required: ['value'] }
})
return found.value
`))
expect(result.value).toBe('accepted')
expect(structuredReads).toBe(1)
})
it('a child whose dispose() throws synchronously cannot wedge the script (the host acks anyway)', async () => {
@@ -703,6 +756,81 @@ describe('dsh-workflow-workerthread', () => {
await handle.dispose()
})
it('the settle-reap explicitly cancels a readiness-pending stray before workflow/end', async () => {
const { ctx, parent, provider } = await setup({ manual: true, deferStart: true })
const childLifecycle: string[] = []
let cancellationAtWorkflowEnd: string | undefined
ctx.on('workflow/agent-start', () => { childLifecycle.push('start') })
ctx.on('workflow/agent-end', () => { childLifecycle.push('end') })
ctx.on('workflow/end', () => {
cancellationAtWorkflowEnd = provider.runs[0]?.cancelled
})
const handle = ctx.workflows.start({
...scripted(`
agent('readiness-pending stray')
return 'done'
`),
parent,
})
const result = await handle.result
expect(result.stopReason).toBe('completed')
expect(provider.runs).toHaveLength(1)
expect(provider.runs[0]!.request.signal?.aborted).toBe(true)
expect(provider.runs[0]!.request.signal?.reason).toBe('workflow settled')
expect(provider.runs[0]!.cancelled).toBe('workflow settled')
expect(cancellationAtWorkflowEnd).toBe('workflow settled')
expect(childLifecycle).toEqual([])
await handle.dispose()
expect(provider.runs[0]!.disposeCalls).toBe(1)
})
it('contains a throwing child cancel and still settles after cancelling peer strays', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
let starts = 0
const cancelled: string[] = []
const warnings: string[] = []
ctx.logger.warn = ((message: unknown) => { warnings.push(String(message)) }) as typeof ctx.logger.warn
const provider: SubagentProvider = {
name: 'throwing-cancel',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
inheritsParentContext: false,
start: () => {
const index = starts++
return {
id: AgentId(`throwing-cancel-${index}`),
started: new Promise(() => { /* readiness stays pending */ }),
result: new Promise(() => { /* cancellation callback owns settlement */ }),
cancel: (reason?: string) => {
if (index === 0) throw new Error('cancel callback broke')
cancelled.push(`${index}:${reason ?? 'cancelled'}`)
},
dispose: () => Promise.resolve(),
}
},
}
ctx.subagents.registerProvider(provider)
await ctx.plugin(WorkerWorkflowEngine, { provider: 'throwing-cancel', maxConcurrentAgents: 2 })
const handle = ctx.workflows.start({
...scripted(`
agent('first stray')
agent('second stray')
return 'done'
`),
parent: fakeParent(),
})
const result = await handle.result
expect(result.stopReason).toBe('completed')
expect(starts).toBe(2)
expect(cancelled).toContain('1:workflow settled')
expect(warnings.some(message => message.includes('cancel callback broke'))).toBe(true)
await handle.dispose()
})
it("cancel() drives each child's explicit cancel() host-side: a wedged worker cannot delay it", async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)

3
packages/workflow/workflow-workerthread/tsconfig.json
View File

@@ -26,6 +26,9 @@
{
"path": "../../llm/llm"
},
{
"path": "../../core/session"
},
{
"path": "../../subagent/subagent"
},