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

fix(ralph): harden execution boundaries

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-20 12:56:09 +08:00
parent c6f8247ac1
commit 60eea35df5
37 changed files with 535 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -34,7 +34,7 @@ Unknown options, malformed arguments, unsupported schemas, tripped caps, provide
## Run sequence
`start()` validates meta and parses the body, creates the worker, and returns a holder-owned `WorkflowRun`. Source mode installs TypeScript transforms through a data-URL bootstrap; built mode passes sibling `lib/worker.cjs` as a filesystem path because pkg's VFS hook expects CommonJS. Both work under ordinary Node. A ready/go handshake prevents a start-signal cancellation racing worker boot from executing the script's initial synchronous slice.
`start()` validates meta, parses the body, resolves a registered normalized provider route, and resolves any per-run total-child cap before creating a worker or publishing `workflow/start`. A requested `maxTotalAgents` must be a positive safe integer no greater than the engine's configured deployment ceiling. Source mode installs TypeScript transforms through a data-URL bootstrap; built mode passes sibling `lib/worker.cjs` as a filesystem path because pkg's VFS hook expects CommonJS. Both work under ordinary Node. A ready/go handshake prevents a start-signal cancellation racing worker boot from executing the script's initial synchronous slice.
For each `agent()` call:
@@ -81,7 +81,7 @@ The host keeps a ledger of forwarded child starts. A graceful worker supplies th
| `syncTimeoutMs` | `5000` | VM timeout for the script's initial synchronous slice. |
| `disposeGraceMs` | `5000` | Bound before force-settlement/termination and for public disposal. |
An owning consumer may set `WorkflowStartRequest.subagentProvider` for one run. This is an engine-level route, not a script hook or a model-facing option; the ordinary `workflow` tool leaves it unset.
An owning consumer may set `WorkflowStartRequest.subagentProvider` and `WorkflowStartRequest.maxTotalAgents` for one run. These are engine-level policy, not script hooks or model-facing options; the ordinary `workflow` tool leaves both unset. A per-run total-child cap may lower but never raise the configured `maxTotalAgents` ceiling.
## Model Experience

35
packages/workflow/workflow-workerthread/src/index.ts
View File

@@ -73,6 +73,36 @@ function assertBodyParses(body: string, name: string): void {
}
}
/** Resolve one run's provider route before publishing work. */
function resolveSubagentProvider(ctx: Context, configured: string, override: string | undefined): string {
const provider = override ?? configured
if (provider.length === 0 || provider !== provider.trim()) {
throw new WorkflowError(
'workflow subagentProvider must be a non-empty normalized string',
'INVALID_ARGUMENT',
)
}
if (ctx.subagents.getProvider(provider) === undefined) {
throw new WorkflowError(`no subagent provider registered for "${provider}"`, 'AGENT_START')
}
return provider
}
/** Resolve one run's total-child cap against the engine deployment ceiling. */
function resolveMaxTotalAgents(requested: number | undefined, ceiling: number): number {
if (requested === undefined) return ceiling
if (!Number.isSafeInteger(requested) || requested < 1) {
throw new WorkflowError('workflow maxTotalAgents must be a positive safe integer', 'INVALID_ARGUMENT')
}
if (requested > ceiling) {
throw new WorkflowError(
`workflow maxTotalAgents ${requested} exceeds the engine ceiling ${ceiling}`,
'INVALID_ARGUMENT',
)
}
return requested
}
/**
* The worker-thread engine service. `start()` validates the script up front
* (meta + a host-side body parse) and returns a {@link WorkflowRun} whose
@@ -113,13 +143,15 @@ class WorkerWorkflowEngine extends WorkflowService {
start(request: WorkflowStartRequest): WorkflowRun {
const meta = validateMeta(request.meta)
assertBodyParses(request.script, meta.name)
const subagentProvider = resolveSubagentProvider(this.ctx, this.config.provider, request.subagentProvider)
const maxTotalAgents = resolveMaxTotalAgents(request.maxTotalAgents, this.config.maxTotalAgents)
const id = WorkflowRunId(randomUUID())
const info: WorkflowRunInfo = { id, meta }
const limits: WorkerLimits = {
maxConcurrentAgents: this.config.maxConcurrentAgents === 0
? Math.min(16, Math.max(1, availableParallelism() - 2))
: this.config.maxConcurrentAgents,
maxTotalAgents: this.config.maxTotalAgents,
maxTotalAgents,
maxItemsPerCall: this.config.maxItemsPerCall,
syncTimeoutMs: this.config.syncTimeoutMs,
}
@@ -137,7 +169,6 @@ class WorkerWorkflowEngine extends WorkflowService {
// the now-inactive engine fiber and break the seam's holder-owned lifetime.
const runCtx = this.ctx
const subagents = runCtx.subagents
const subagentProvider = request.subagentProvider ?? this.config.provider
const workerRun = new WorkerRun(
runCtx,
subagents,

2
packages/workflow/workflow-workerthread/src/runtime.ts
View File

@@ -255,7 +255,7 @@ export class WorkflowExecution {
const opts = this.readAgentOptions(rawOpts)
if (this.started >= this.limits.maxTotalAgents) {
throw new WorkflowError(
`this run reached its total agent cap (${this.limits.maxTotalAgents}) — a runaway-loop backstop; raise maxTotalAgents in the engine config if the scale is intentional`,
`this run reached its total agent cap (${this.limits.maxTotalAgents}) — a runaway-loop backstop; raise the applicable maxTotalAgents limit if the scale is intentional`,
'AGENT_CAP',
)
}

1
packages/workflow/workflow-workerthread/tests/session.spec.ts
View File

@@ -392,6 +392,7 @@ describe('runWorkerSession over an in-process MessageChannel', () => {
const result = await host.result()
expect(result.stopReason).toBe('error')
expect(result.error).toContain('total agent cap (2)')
expect(result.error).toContain('applicable maxTotalAgents limit')
expect(result.agentsStarted).toBe(2)
host.close()
})

8
packages/workflow/workflow-workerthread/tests/source-worker.compat.spec.ts
View File

@@ -8,6 +8,7 @@ import { expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentProvider } from '@deepseek-ai/dsh-subagent'
import WorkerWorkflowEngine from '../src/index.ts'
import { SessionId } from '@deepseek-ai/dsh-session'
@@ -18,6 +19,13 @@ vi.setConfig({ testTimeout: 30_000 })
it('runs the default config through the source worker', async () => {
const ctx = new Context()
const subagents = await ctx.plugin(SubagentService)
const provider: SubagentProvider = {
name: 'spawn',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: true },
inheritsParentContext: false,
start: () => Promise.reject(new Error('source-worker compat script must not start a child')),
}
ctx.subagents.registerProvider(provider)
const engine = await ctx.plugin(WorkerWorkflowEngine, {})
const parent = { id: SessionId('workflow-compat-parent'), options: {} } as unknown as Agent
try {

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

@@ -6,7 +6,7 @@ import Loader from '@cordisjs/plugin-loader'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentCapabilities, SubagentProvider, SubagentResult, SubagentRun, SubagentStartRequest } from '@deepseek-ai/dsh-subagent'
import type { WorkflowMeta, WorkflowResult, WorkflowResultInfo, WorkflowRunInfo } from '@deepseek-ai/dsh-workflow'
import type { WorkflowMeta, WorkflowResult, WorkflowResultInfo, WorkflowRun, WorkflowRunInfo } from '@deepseek-ai/dsh-workflow'
import * as workerEngineModule from '../src/index.ts'
import WorkerWorkflowEngine, { type Config } from '../src/index.ts'
import { HostToWorkerType, WorkerToHostType } from '../src/protocol.ts'
@@ -252,6 +252,77 @@ describe('dsh-workflow-workerthread', () => {
expect(provider.runs).toHaveLength(1)
})
it('rejects invalid start-request provider routes before publishing a run', async () => {
const { ctx, parent } = await setup()
let starts = 0
ctx.on('workflow/start', () => { starts += 1 })
const messages: string[] = []
for (const subagentProvider of ['', 'missing']) {
let run: WorkflowRun | undefined
let thrown: unknown
try {
run = ctx.workflows.start({
...scripted("return 'must not start'"),
parent,
subagentProvider,
})
} catch (error: unknown) {
thrown = error
}
await run?.dispose()
messages.push(thrown instanceof Error ? thrown.message : '')
}
expect(messages).toEqual([
'workflow subagentProvider must be a non-empty normalized string',
'no subagent provider registered for "missing"',
])
expect(starts).toBe(0)
})
it('rejects invalid per-run total-agent caps before publishing a run', async () => {
const { ctx, parent } = await setup({ config: { maxTotalAgents: 2 } })
let starts = 0
ctx.on('workflow/start', () => { starts += 1 })
const errors: unknown[] = []
for (const maxTotalAgents of [0, 1.5, Number.NaN, 3]) {
try {
const handle = ctx.workflows.start({
...scripted("return 'must not start'"),
parent,
maxTotalAgents,
})
await handle.dispose()
} catch (error: unknown) {
errors.push(error)
}
}
expect(errors.slice(0, 3)).toEqual(Array(3).fill(expect.objectContaining({
code: 'INVALID_ARGUMENT',
message: 'workflow maxTotalAgents must be a positive safe integer',
})))
expect(errors[3]).toMatchObject({
code: 'INVALID_ARGUMENT',
message: 'workflow maxTotalAgents 3 exceeds the engine ceiling 2',
})
expect(starts).toBe(0)
})
it('enforces a per-run total-agent cap below the engine ceiling', async () => {
const { ctx, parent } = await setup({ config: { maxTotalAgents: 2 } })
const handle = ctx.workflows.start({
...scripted("await agent('first'); await agent('second'); return 'unreachable'"),
parent,
maxTotalAgents: 1,
})
const result = await handle.result
expect(result.stopReason).toBe('error')
expect(result.agentsStarted).toBe(1)
expect(result.error).toContain('total agent cap (1)')
await handle.dispose()
})
it('a fatal hook error inside the worker kills the script and reports the error', async () => {
const { ctx, parent } = await setup()
const result = await run(ctx, parent, scripted("return await parallel([() => agent('x', { isolation: 'worktree' })])"))
@@ -259,11 +330,18 @@ describe('dsh-workflow-workerthread', () => {
expect(result.error).toContain('"isolation" is deferred')
})
it('a provider start failure crosses back as a fatal AGENT_START error', async () => {
it('rejects an unregistered configured provider before publishing a run', async () => {
const { ctx, parent } = await setup({ config: { provider: 'nonexistent' } })
const result = await run(ctx, parent, scripted("return await pipeline([1], () => agent('p'))"))
expect(result.stopReason).toBe('error')
expect(result.error).toContain('agent() could not start a child')
let thrown: unknown
try {
ctx.workflows.start({ ...scripted("return 'must not start'"), parent })
} catch (error: unknown) {
thrown = error
}
expect(thrown).toMatchObject({
code: 'AGENT_START',
message: 'no subagent provider registered for "nonexistent"',
})
})
it('waits for async provider start before announcing a result that settled early', async () => {