From acddf5dc7d3cb99cc0cda98652f24048b72087b4 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Tue, 22 Sep 2026 11:11:17 -0700 Subject: [PATCH 1/6] fix(sdk): run concurrent agent and LLM steps under --local-agent instead of parking them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The local agent worker registered capacity 1 (the LLM worker too), and the kernel never queues for a busy worker: the second concurrent f.agent/f.llm in an authored body was admitted, found no free worker, and parked the run ("no worker is attached for step type agent"). - Local agent and LLM workers now hold DEFAULT_LOCAL_AGENT_CAPACITY (4) dispatches; `flows run|resume --local-agent --agent-capacity ` (1-32) sets it. The flag without --local-agent, or with --cloud, is refused. - The authored body sizes its admission to that capacity (worker-slots.ts): calls beyond it wait in-process, FIFO, for a slot before run.start, so the body never asks the kernel for more than the worker holds. Threaded through the durable root, the node runtime request, and resume. Agents sharing a working directory still run one at a time for artifact attribution (worker-cli.ts serializedByDirectory); they now complete instead of parking, and LLM steps overlap. Real agent overlap needs f.agent's cwd, which the kernel refuses today — a separate fix. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/SURFACE.md | 15 +++ packages/sdk/src/authored-flow-executor.ts | 8 +- packages/sdk/src/authored-node-entry.ts | 5 +- packages/sdk/src/authored-node-runner.ts | 3 +- packages/sdk/src/authored-root.ts | 4 + packages/sdk/src/authored-worker-step.ts | 10 +- packages/sdk/src/cli-commands.ts | 6 ++ packages/sdk/src/cli.ts | 21 ++++- packages/sdk/src/cli/direct-run.ts | 7 +- packages/sdk/src/cli/run.ts | 15 ++- packages/sdk/src/llm-worker.ts | 9 +- packages/sdk/src/local-agent.ts | 7 +- packages/sdk/src/worker-slots.ts | 40 ++++++++ .../tests/authored-parallel-agents.test.ts | 94 +++++++++++++++++++ packages/sdk/tests/relay-cli-surface.test.ts | 4 +- packages/sdk/tests/worker-slots.test.ts | 74 +++++++++++++++ 16 files changed, 302 insertions(+), 20 deletions(-) create mode 100644 packages/sdk/src/worker-slots.ts create mode 100644 packages/sdk/tests/authored-parallel-agents.test.ts create mode 100644 packages/sdk/tests/worker-slots.test.ts diff --git a/docs/SURFACE.md b/docs/SURFACE.md index d2385f550..ca9eb793e 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -361,6 +361,21 @@ f.llm(strings: TemplateStringsArray, ...values: unknown[]): Step; Run with `flows run chain.flow.ts --input '{}' --local-agent`. This attaches both an agent worker and a workspace-free LLM worker for the authored body. + +Each worker holds 4 dispatches at once. Set a different number (1–32) with +`--agent-capacity `, which applies to `run` and `resume`. The agent and LLM +workers are counted separately. + +A body that starts more concurrent `f.agent` or `f.llm` calls than the capacity +does not fail: the extra calls wait in-process for a free slot. Without that +wait, they would be submitted with no worker free, and the kernel would park +them. + +Agents that share a working directory still run one at a time. This lets the +worker attribute each file change to the step that made it. So today +concurrent `f.agent` calls do not overlap, although they no longer park. +Concurrent `f.llm` calls do overlap. + The LLM step remains `type: llm` in the journal. It uses the same CLI resolution, authentication probes, and exact `flows.json` model allow-list as agent steps; a declared `model` must be in that project's `models` array. A template call diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index 25b979b6b..d58b8abe5 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -154,6 +154,12 @@ export interface ExecuteAuthoredFlowOptions { readonly onWait?: RunLifecycleOptions['onWait']; readonly onProgress?: (event: ProgressEvent) => void; readonly localAgentStream?: string; + /** + * How many agent (and, separately, LLM) dispatches the attached local + * workers hold at once. Set, it caps this body's concurrent `f.agent` / + * `f.llm` child runs to match; unset, calls are admitted as they arrive. + */ + readonly workerCapacity?: number; /** Durable kernel root that owns this body's child admission identities. */ readonly rootRunId?: string; /** Installed flow-extension plugins, in lock order, so `f.hook` can AND-compose them. */ @@ -245,7 +251,7 @@ export async function executeAuthoredFlow( const worker = authoredWorkerRunner( definition, journal, flowPath, journalSteps, waitOptions, - localAgentStream, budget, definition.header.budget, options.rootRunId, + localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity, ); /** diff --git a/packages/sdk/src/authored-node-entry.ts b/packages/sdk/src/authored-node-entry.ts index 1909ea5f1..b03bdef73 100644 --- a/packages/sdk/src/authored-node-entry.ts +++ b/packages/sdk/src/authored-node-entry.ts @@ -6,6 +6,7 @@ import { executeAuthoredFlow } from './authored-flow-executor.js'; import { loadPinnedAuthoredSource } from './authored-source-authority.js'; import { assertAuthoredNodeVersion, parseAuthoredParentPid } from './authored-runtime-capability.js'; import { AuthoredFlowExecutionError, AuthoredHumanParked } from './authored-flow-error.js'; +import { isAgentCapacity } from './worker-slots.js'; import type { AuthoredRootMetadata } from './authored-root.js'; let channelKey: string | undefined, sequence = 0; @@ -48,7 +49,7 @@ try { send({ type: 'ready', runtime: { kind: 'node', version: process.versions.node, executableSha256: hash(process.execPath), payloadSha256: hash(process.argv[1]!) } }); const request = await new Promise<{ channelKey: string; metadata: AuthoredRootMetadata; socketPath: string; - rootRunId: string; dataDir: string; localAgentStream?: string }>((resolve, reject) => { + rootRunId: string; dataDir: string; localAgentStream?: string; workerCapacity?: number }>((resolve, reject) => { let buffer = ''; process.stdin.setEncoding('utf8'); const onData = (chunk: string): void => { @@ -66,6 +67,7 @@ try { controller.signal.throwIfAborted(); const loaded = await loadPinnedAuthoredSource(request.metadata, true); if (request.localAgentStream !== request.metadata.localAgentStream) throw new Error('authored root local agent surface mismatch'); + if (request.workerCapacity !== undefined && !isAgentCapacity(request.workerCapacity)) throw new Error('invalid authored worker capacity'); client = new JournalClient(request.socketPath); await client.connect(); await client.hello('flows-authored-node'); const result = await executeAuthoredFlow(loaded.handle, client, @@ -74,6 +76,7 @@ try { flowPath: request.metadata.flowPath, rootRunId: request.rootRunId, extensions: loaded.extensions, localAgentStream: request.localAgentStream, signal: controller.signal, + ...(request.workerCapacity === undefined ? {} : { workerCapacity: request.workerCapacity }), onProgress: event => send({ type: 'progress', event }), onWait: event => send({ type: 'wait', event }), }); diff --git a/packages/sdk/src/authored-node-runner.ts b/packages/sdk/src/authored-node-runner.ts index 1dd97944a..a06f77661 100644 --- a/packages/sdk/src/authored-node-runner.ts +++ b/packages/sdk/src/authored-node-runner.ts @@ -130,7 +130,8 @@ export async function runAuthoredInNode( clearTimeout(startupTimer); // Keep stdin open: EOF tells the child its lease-owning parent died. child.stdin!.write(JSON.stringify({ channelKey, metadata, socketPath, rootRunId, - dataDir: options.dataDir, localAgentStream: options.localAgentStream }) + '\n'); + dataDir: options.dataDir, localAgentStream: options.localAgentStream, + workerCapacity: options.workerCapacity }) + '\n'); } else if (!ready || result) throw new Error('unexpected authored runtime message'); else if (message.type === 'progress') options.onProgress?.(message.event); else if (message.type === 'wait') options.onWait?.(message.event); diff --git a/packages/sdk/src/authored-root.ts b/packages/sdk/src/authored-root.ts index 0b6513042..55f0dda95 100644 --- a/packages/sdk/src/authored-root.ts +++ b/packages/sdk/src/authored-root.ts @@ -51,6 +51,8 @@ export interface DurableAuthoredOptions { readonly dataDir: string; readonly admissionKey?: string; readonly localAgentStream?: string; + /** Concurrency of the attached local workers; see `ExecuteAuthoredFlowOptions.workerCapacity`. */ + readonly workerCapacity?: number; readonly lifecycle?: RunLifecycleOptions; } @@ -201,6 +203,7 @@ async function driveRoot( if (process.versions['bun'] !== undefined) { return runAuthoredInNode(metadata, journal.socketPath, dispatch.run_id, { dataDir: options.dataDir, localAgentStream: options.localAgentStream, + ...(options.workerCapacity === undefined ? {} : { workerCapacity: options.workerCapacity }), ...options.lifecycle, signal, }); } @@ -213,6 +216,7 @@ async function driveRoot( dataDir: options.dataDir, flowPath: metadata.flowPath, localAgentStream: options.localAgentStream, + ...(options.workerCapacity === undefined ? {} : { workerCapacity: options.workerCapacity }), rootRunId: dispatch.run_id, extensions: loaded.extensions, ...options.lifecycle, diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts index bd85fac07..278064fa6 100644 --- a/packages/sdk/src/authored-worker-step.ts +++ b/packages/sdk/src/authored-worker-step.ts @@ -14,6 +14,7 @@ import { snapshotJsonValue } from './json-value.js'; import { authoredChildAdmissionKey } from './authored-admission.js'; import { alsoRecord, recordAuthoredChild } from './authored-step-index.js'; import type { StepFailedDetails } from './failure-kinds.js'; +import { WorkerSlots } from './worker-slots.js'; const WORKSPACE_PERMISSION_ANNOTATION = /:\s*(readonly|readwrite)\s*$/i; @@ -22,8 +23,12 @@ export function authoredWorkerRunner( definition: { name: string }, journal: JournalClient, flowPath: string, journalSteps: AuthoredFlowJournalStep[], waitOptions: RunLifecycleOptions, localAgentStream?: string, budget?: AuthoredBudget, headerBudget?: unknown, - rootRunId?: string, + rootRunId?: string, workerCapacity?: number, ) { + // Sized to the attached local workers, so concurrent calls wait here for a + // slot instead of being admitted and parked for want of a free worker. + const slots = workerCapacity === undefined ? undefined + : { agent: new WorkerSlots(workerCapacity), llm: new WorkerSlots(workerCapacity) }; const context: AuthoredStepContext = { ...(rootRunId === undefined ? {} : { rootRunId }), ...(waitOptions.dataDir === undefined ? {} : { dataDir: waitOptions.dataDir }), @@ -113,9 +118,10 @@ export function authoredWorkerRunner( return readCompletedStepOutput(journal, outcome.run_id, id, journalSteps, context); }; const admissionKey = authoredChildAdmissionKey(rootRunId, id); - return budget === undefined + const admit = async () => budget === undefined ? consume(await journal.runStart(spec, undefined, admissionKey)) : budget.execute(journal, spec, consume, admissionKey); + return slots === undefined ? admit() : slots[step.type === 'llm' ? 'llm' : 'agent'].run(admit); } return { diff --git a/packages/sdk/src/cli-commands.ts b/packages/sdk/src/cli-commands.ts index 0bf944748..bb8ce66f6 100644 --- a/packages/sdk/src/cli-commands.ts +++ b/packages/sdk/src/cli-commands.ts @@ -1,5 +1,6 @@ import { DEFAULT_DATA_DIR } from './daemon-connection.js'; import { DEFAULT_RUN_LIMIT } from './cli/cloud-read.js'; +import { DEFAULT_LOCAL_AGENT_CAPACITY, MAX_LOCAL_AGENT_CAPACITY } from './worker-slots.js'; import type { ParsedArgs } from './cli.js'; /** @@ -88,6 +89,11 @@ const LOCAL_EXECUTION_OPTIONS = [ JSON_OPTION, DATA_DIR_OPTION, { flags: '--local-agent', description: 'Run agent steps in this process instead of a worker' }, + { + flags: '--agent-capacity ', + description: `With --local-agent, how many agent steps (and, separately, LLM steps) run at once (1-${MAX_LOCAL_AGENT_CAPACITY})`, + defaultValue: String(DEFAULT_LOCAL_AGENT_CAPACITY), + }, { flags: '--no-spawn', description: 'Require a running relayflowd rather than starting one' }, { flags: '--no-observer-link', description: 'Do not mint an observer link for this run' }, { diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts index 202a69a4a..5deb026b0 100644 --- a/packages/sdk/src/cli.ts +++ b/packages/sdk/src/cli.ts @@ -44,6 +44,7 @@ import { runHnMonitor } from './cli/hn-monitor.js'; import { runTickRunner } from './cli/tick-runner.js'; import { DEFAULT_DATA_DIR } from './daemon-connection.js'; import { CLI_VERB_NAMES } from './cli-commands.js'; +import { isAgentCapacity } from './worker-slots.js'; import { mintObserverUrl, resolveObserverLinkEnv, @@ -80,8 +81,8 @@ export type ParsedArgs = | { command: 'schedules'; json: boolean } | { command: 'unschedule'; scheduleId: string; json: boolean } | { command: 'check'; json: boolean; watch: boolean; value: string } - | { command: 'run'; bucket: string | undefined; reuseFromRunId: string | undefined; localAgent: boolean; dataDir: string; input: string | undefined; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } - | { command: 'resume'; localAgent: boolean; dataDir: string; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } + | { command: 'run'; bucket: string | undefined; reuseFromRunId: string | undefined; localAgent: boolean; agentCapacity: number | undefined; dataDir: string; input: string | undefined; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } + | { command: 'resume'; localAgent: boolean; agentCapacity: number | undefined; dataDir: string; json: boolean; spawn: boolean; noObserverLink: boolean; allowHumanInfluenced: boolean; value: string } | { command: 'answer'; dataDir: string; json: boolean; spawn: boolean; note: string | undefined; by: string | undefined; runId: string; waitId: string; answer: boolean } | RunsArgs | LogsArgs @@ -295,6 +296,7 @@ export async function runCli( onPtyReady: (path: string) => io.stderr(`PTY ${path}`), ...(parsed.command === 'run' && parsed.reuseFromRunId !== undefined ? { reuseFromRunId: parsed.reuseFromRunId } : {}), localAgent: parsed.localAgent, + ...(parsed.agentCapacity === undefined ? {} : { agentCapacity: parsed.agentCapacity }), onProgress: showProgress, onWait: (progress: RunProgress) => { emitWait(progress, io); @@ -581,6 +583,7 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { let syncCode = false; let noConnect = false; let localAgent = false; + let agentCapacity: number | undefined; let allowHumanInfluenced = false; let dataDir = DEFAULT_DATA_DIR; let sawDataDir = false; @@ -612,6 +615,14 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { localAgent = true; continue; } + if (argument === '--agent-capacity') { + // Decimal digits only: `Number` would also read "0x4", "4e0" and " 4". + const value = args[++index]; + if (command === 'check' || agentCapacity !== undefined || value === undefined || !/^[0-9]+$/.test(value)) return undefined; + agentCapacity = Number(value); + if (!isAgentCapacity(agentCapacity)) return undefined; + continue; + } if (argument === '--watch') { if (command !== 'check' || watch) return undefined; watch = true; @@ -669,6 +680,8 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { positionals.push(argument); } if (positionals.length !== 1) return undefined; + // It sizes the in-process worker, so without `--local-agent` it describes nothing. + if (agentCapacity !== undefined && !localAgent) return undefined; if (bucket !== undefined && (cloud || !parseDigestReference(positionals[0]!))) return undefined; if (cloud) { @@ -688,8 +701,8 @@ function parseArgs(args: readonly string[]): ParsedArgs | undefined { return command === 'check' ? { command, json, watch, value: positionals[0]! } : command === 'run' - ? { command, bucket, reuseFromRunId, localAgent, dataDir, input, json, spawn, noObserverLink, allowHumanInfluenced, value: positionals[0]! } - : { command, localAgent, dataDir, json, spawn, noObserverLink, allowHumanInfluenced, value: positionals[0]! }; + ? { command, bucket, reuseFromRunId, localAgent, agentCapacity, dataDir, input, json, spawn, noObserverLink, allowHumanInfluenced, value: positionals[0]! } + : { command, localAgent, agentCapacity, dataDir, json, spawn, noObserverLink, allowHumanInfluenced, value: positionals[0]! }; } /** diff --git a/packages/sdk/src/cli/direct-run.ts b/packages/sdk/src/cli/direct-run.ts index f79a34ee8..98743efd1 100644 --- a/packages/sdk/src/cli/direct-run.ts +++ b/packages/sdk/src/cli/direct-run.ts @@ -4,6 +4,7 @@ import { authoredLocalAgentStream } from '../authored-admission.js'; import { McpPreflightError } from './check-typescript.js'; import { attachLocalAgent } from '../local-agent.js'; import { LlmWorker } from '../llm-worker.js'; +import { DEFAULT_LOCAL_AGENT_CAPACITY } from '../worker-slots.js'; import { AuthoredFlowExecutionError, } from '../authored-flow-executor.js'; @@ -67,19 +68,20 @@ export async function runDirectFlow( let localLlm: LlmWorker | undefined; let llmClient: JournalClient | undefined; let llmFailure: unknown; + const workerCapacity = options.agentCapacity ?? DEFAULT_LOCAL_AGENT_CAPACITY; const admissionIdentity = options.authoredAdmissionKey ?? process.env['RELAYFLOW_AUTHORED_ADMISSION_KEY'] ?? randomUUID(); try { if (options.localAgent) { localAgent = await attachLocalAgent( - client, dataDir, options.onPtyReady, authoredLocalAgentStream(admissionIdentity), + client, dataDir, options.onPtyReady, authoredLocalAgentStream(admissionIdentity), workerCapacity, ); // A session owns one worker registration. Keep the workspace-free LLM // worker on its own connection so it cannot replace the agent worker. llmClient = new JournalClient(socketPath); await llmClient.connect(); await llmClient.hello('flows-local-llm'); - localLlm = new LlmWorker(llmClient, `${localAgent.stream}-llm`); + localLlm = new LlmWorker(llmClient, `${localAgent.stream}-llm`, workerCapacity); localLlm.on('error', error => { llmFailure = error; client.close(); }); await localLlm.attach(); } @@ -89,6 +91,7 @@ export async function runDirectFlow( dataDir, admissionKey: admissionIdentity, localAgentStream: localAgent?.stream, + ...(localAgent === undefined ? {} : { workerCapacity }), lifecycle: { onProgress: options.onProgress, ...(options.signal !== undefined ? { signal: options.signal } : {}), diff --git a/packages/sdk/src/cli/run.ts b/packages/sdk/src/cli/run.ts index 8b3386949..475bd3702 100644 --- a/packages/sdk/src/cli/run.ts +++ b/packages/sdk/src/cli/run.ts @@ -26,6 +26,7 @@ import type { RunStatus, } from '../protocol.js'; import type { StepType } from '../spec.js'; +import { DEFAULT_LOCAL_AGENT_CAPACITY } from '../worker-slots.js'; import type { LoweredCompletionReason } from '../authored-flow-executor.js'; import { checkFlow, @@ -91,6 +92,8 @@ export interface RunLifecycleOptions { reuseFromRunId?: string; onProgress?: (event: ProgressEvent) => void; localAgent?: boolean; + /** `--agent-capacity`: the local workers' concurrency; the default is `DEFAULT_LOCAL_AGENT_CAPACITY`. */ + agentCapacity?: number; signal?: AbortSignal; onWait?: (progress: RunProgress) => void; /** @@ -148,7 +151,9 @@ async function executeCheckedFlow( const spec = toKernelSpec(checked.flow!); // Use the checked CLI/model and declared surfaces unchanged. The worker // advertises its existing pins; the daemon still owns surface matching. - if (options.localAgent) localAgent = await attachLocalAgent(client, dataDir, options.onPtyReady); + if (options.localAgent) { + localAgent = await attachLocalAgent(client, dataDir, options.onPtyReady, undefined, options.agentCapacity); + } if (options.localAgent && spec.steps.some(step => step.type === 'agent' && communicationInstruction(step.instruction))) { const { attachCommunicationWorkers } = await import('../communication/local.js'); communicationWorkers = await attachCommunicationWorkers(spec, socketPath, dataDir); @@ -192,6 +197,7 @@ export async function resumeFlow( let authoredAgent: Awaited> | undefined; let authoredLlm: LlmWorker | undefined; let authoredLlmClient: JournalClient | undefined; + const workerCapacity = options.agentCapacity ?? DEFAULT_LOCAL_AGENT_CAPACITY; try { const authoredRoot = await readAuthoredRootMetadata(client, runId); if (authoredRoot !== undefined) { @@ -203,17 +209,18 @@ export async function resumeFlow( } if (options.localAgent) { authoredAgent = await attachLocalAgent( - client, dataDir, options.onPtyReady, authoredRoot.localAgentStream, + client, dataDir, options.onPtyReady, authoredRoot.localAgentStream, workerCapacity, ); authoredLlmClient = new JournalClient(socketPath); await authoredLlmClient.connect(); await authoredLlmClient.hello('flows-authored-resume-llm'); - authoredLlm = new LlmWorker(authoredLlmClient, `${authoredAgent.stream}-llm`); + authoredLlm = new LlmWorker(authoredLlmClient, `${authoredAgent.stream}-llm`, workerCapacity); await authoredLlm.attach(); } const result = await resumeDurableAuthoredFlow(runId, client, { dataDir, localAgentStream: authoredAgent?.stream, + ...(authoredAgent === undefined ? {} : { workerCapacity }), lifecycle: options, }); if (result === undefined) throw new Error('authored root disappeared during resume'); @@ -224,7 +231,7 @@ export async function resumeFlow( // second call the earlier rebase left is a stale reference from before // the helper fanout renamed the API. if (options.localAgent) { - authoredAgent = await attachLocalAgent(client, dataDir, options.onPtyReady); + authoredAgent = await attachLocalAgent(client, dataDir, options.onPtyReady, undefined, options.agentCapacity); const entries = (await client.journalRead(runId, 1)).entries as Array<{ entry_type: string; payload?: { spec?: import('../spec.js').KernelRunSpec } }>; const spec = entries.find(entry => entry.entry_type === 'run.spawned')?.payload?.spec; if (spec?.steps.some(step => step.type === 'agent' && communicationInstruction(step.instruction))) { diff --git a/packages/sdk/src/llm-worker.ts b/packages/sdk/src/llm-worker.ts index e1f514bc8..35895079c 100644 --- a/packages/sdk/src/llm-worker.ts +++ b/packages/sdk/src/llm-worker.ts @@ -1,4 +1,5 @@ import { workerSpend } from './worker-spend.js'; +import { DEFAULT_LOCAL_AGENT_CAPACITY } from './worker-slots.js'; import type { WorkerCliResult } from './worker-cli.js'; import { EventEmitter } from 'node:events'; import type { JournalClient } from './journal-client.js'; @@ -17,7 +18,11 @@ export class LlmWorker extends EventEmitter { private closing = false; private readonly inFlight = new Set>(); - constructor(private readonly client: JournalClient, private readonly workerId: string) { + constructor( + private readonly client: JournalClient, + private readonly workerId: string, + private readonly capacity: number = DEFAULT_LOCAL_AGENT_CAPACITY, + ) { super(); } @@ -25,7 +30,7 @@ export class LlmWorker extends EventEmitter { if (this.attached || this.closing) throw new Error('llm worker: cannot attach twice or after close'); this.client.on('step.dispatch', this.onDispatch); try { - await this.client.workerAttach(this.workerId, ['llm'], { workspace: [], streams: [] }, 1); + await this.client.workerAttach(this.workerId, ['llm'], { workspace: [], streams: [] }, this.capacity); this.attached = true; } catch (error) { this.client.off('step.dispatch', this.onDispatch); diff --git a/packages/sdk/src/local-agent.ts b/packages/sdk/src/local-agent.ts index 40708575a..0c39c29a9 100644 --- a/packages/sdk/src/local-agent.ts +++ b/packages/sdk/src/local-agent.ts @@ -1,6 +1,7 @@ import { randomUUID } from 'node:crypto'; import type { JournalClient } from './journal-client.js'; import { AgentWorker } from './worker.js'; +import { DEFAULT_LOCAL_AGENT_CAPACITY } from './worker-slots.js'; /** A local worker for stream-only steps; no workspace recovery is claimed. */ export async function attachLocalAgent( @@ -8,6 +9,7 @@ export async function attachLocalAgent( dataDir?: string, onPtyReady?: (path: string) => void, requestedStream?: string, + capacity: number = DEFAULT_LOCAL_AGENT_CAPACITY, ): Promise<{ stream: string; readonly failure: unknown; @@ -18,7 +20,10 @@ export async function attachLocalAgent( const stream = requestedStream ?? `local-agent-${randomUUID()}`; const worker = new AgentWorker(client, { workerId: stream, - capacity: 1, + // More than one: independent agent steps run side by side instead of the + // second parking behind the first. Authored bodies size their admission to + // this same number (worker-slots.ts), so they never ask for more. + capacity, dataDir, onPtyReady, pins: { workspace: [], streams: [{ stream, read_offset: 0 }] }, }); diff --git a/packages/sdk/src/worker-slots.ts b/packages/sdk/src/worker-slots.ts new file mode 100644 index 000000000..b7bcac61e --- /dev/null +++ b/packages/sdk/src/worker-slots.ts @@ -0,0 +1,40 @@ +/** The local agent worker's default concurrency, and the ceiling `--agent-capacity` accepts. */ +export const DEFAULT_LOCAL_AGENT_CAPACITY = 4; +export const MAX_LOCAL_AGENT_CAPACITY = 32; + +/** A positive integer no larger than the ceiling; anything else is not a capacity. */ +export function isAgentCapacity(value: unknown): value is number { + return Number.isInteger(value) && (value as number) >= 1 && (value as number) <= MAX_LOCAL_AGENT_CAPACITY; +} + +/** + * First-come admission to a worker that holds `capacity` dispatches at once. + * + * The kernel never queues a step for a busy worker: an attempt with no worker + * below capacity parks the run ("no worker is attached for step type …"). An + * authored body opens one child run per `f.agent`/`f.llm`, so `Promise.all` + * over more calls than the worker holds would park the overflow. Holding the + * overflow here, before `run.start`, keeps the kernel's admission exact and + * turns "too many at once" into "wait for a slot". + */ +export class WorkerSlots { + private held = 0; + private readonly waiting: Array<() => void> = []; + + constructor(readonly capacity: number) { + if (!isAgentCapacity(capacity)) throw new RangeError(`worker capacity must be an integer from 1 to ${MAX_LOCAL_AGENT_CAPACITY} (got ${capacity})`); + } + + async run(work: () => Promise): Promise { + if (this.held < this.capacity) this.held++; + // A released slot is handed straight to the next waiter, so `held` never + // dips below capacity while anyone is queued and no later caller can jump it. + else await new Promise(resolve => this.waiting.push(resolve)); + try { + return await work(); + } finally { + const next = this.waiting.shift(); + if (next !== undefined) next(); else this.held--; + } + } +} diff --git a/packages/sdk/tests/authored-parallel-agents.test.ts b/packages/sdk/tests/authored-parallel-agents.test.ts new file mode 100644 index 000000000..cb1c9610f --- /dev/null +++ b/packages/sdk/tests/authored-parallel-agents.test.ts @@ -0,0 +1,94 @@ +import { readFileSync, writeFileSync } from 'node:fs'; +import { join, resolve } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { flow } from '@relayflows/surface'; +import { AuthoredFlowExecutionError, executeAuthoredFlow } from '../src/authored-flow-executor.js'; +import { attachLocalAgent } from '../src/local-agent.js'; +import { JournalClient } from '../src/journal-client.js'; +import { LlmWorker } from '../src/llm-worker.js'; +import { socketPathFor } from '../src/daemon-connection.js'; +import { chainFixture } from './flow-chain-fixture.js'; + +const closes: Array<() => Promise> = []; +afterEach(async () => { + for (const close of closes.splice(0).reverse()) await close(); +}); + +/** The fixture's wrapper, but each session holds for a while and journals when it ran. */ +async function slowAgents(capacity: number) { + const fixture = chainFixture(); + closes.push(() => fixture.close()); + const spans = join(fixture.root, 'spans.jsonl'); + writeFileSync(fixture.wrapper, `#!/usr/bin/env node +import { receiveWrapperRequest } from ${JSON.stringify(resolve('../../testdata/preflight/wrapper-session.mjs'))}; +import { appendFileSync } from 'node:fs'; +if (process.argv[2] === 'auth') process.exit(0); +const request = await receiveWrapperRequest(); +if (request) { + const start = Date.now(); + await new Promise(done => setTimeout(done, 400)); + appendFileSync(${JSON.stringify(spans)}, JSON.stringify({ start, end: Date.now() }) + '\\n'); + process.stdout.write('done'); +} +`); + const client = await fixture.connect(); + const agent = await attachLocalAgent(client, undefined, undefined, undefined, capacity); + closes.push(() => agent.close()); + // One worker registration per session, so the LLM worker gets its own connection. + const llmClient = new JournalClient(socketPathFor(fixture.data)); + await llmClient.connect(); + await llmClient.hello('parallel-llm-worker'); + closes.push(async () => { llmClient.close(); }); + const llm = new LlmWorker(llmClient, `${agent.stream}-llm`, capacity); + await llm.attach(); + closes.push(() => llm.close()); + const readSpans = () => readFileSync(spans, 'utf8').trim().split('\n') + .map(line => JSON.parse(line) as { start: number; end: number }); + return { fixture, client, agent, readSpans }; +} + +function peakOverlap(spans: Array<{ start: number; end: number }>): number { + return Math.max(...spans.map(({ start }) => spans.filter(other => other.start <= start && start < other.end).length)); +} + +const threeReviewers = flow('three-reviewers', async f => { + await Promise.all(['a', 'b', 'c'].map(lens => f.agent(`review-${lens}`, { task: `Review for ${lens}` }))); + f.done('success'); +}); + +const threeSummaries = flow('three-summaries', async f => { + await Promise.all(['a', 'b', 'c'].map(topic => f.llm`Summarize ${topic}`)); + f.done('success'); +}); + +describe('authored steps under local workers with capacity', () => { + it('runs more concurrent f.llm calls than the worker holds side by side, never more than its capacity', async () => { + const { fixture, client, agent, readSpans } = await slowAgents(2); + const result = await executeAuthoredFlow(threeSummaries, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, workerCapacity: 2, + }); + expect(result.completionReason).toBe('success'); + expect(result.journalSteps.filter(step => step.id.startsWith('llm-'))).toHaveLength(3); + expect(peakOverlap(readSpans())).toBe(2); + }); + + it('completes more concurrent f.agent calls than the worker holds: the overflow waits for a slot instead of parking', async () => { + const { fixture, client, agent } = await slowAgents(2); + const result = await executeAuthoredFlow(threeReviewers, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, workerCapacity: 2, + }); + expect(result.completionReason).toBe('success'); + expect(result.journalSteps.filter(step => step.id.startsWith('agent-'))).toHaveLength(3); + // No overlap is asserted: agents sharing a working directory still take + // turns for artifact attribution (worker-cli.ts serializedByDirectory). + }); + + it('parks the overflow when the body is not told the capacity (the defect this closes)', async () => { + const { fixture, client, agent } = await slowAgents(1); + const run = executeAuthoredFlow(threeReviewers, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, + }); + await expect(run).rejects.toSatisfy(error => + error instanceof AuthoredFlowExecutionError && error.code === 'agent_parked'); + }); +}); diff --git a/packages/sdk/tests/relay-cli-surface.test.ts b/packages/sdk/tests/relay-cli-surface.test.ts index ef8b02b6f..49ea41196 100644 --- a/packages/sdk/tests/relay-cli-surface.test.ts +++ b/packages/sdk/tests/relay-cli-surface.test.ts @@ -105,14 +105,14 @@ const INVOCATIONS: readonly { verb: string; argv: readonly string[]; variant: Pa { verb: 'resume', argv: ['resume', RUN_ID], variant: 'resume' }, { verb: 'resume', - argv: ['resume', '--json', '--data-dir', '.relayflowd', '--local-agent', '--no-spawn', + argv: ['resume', '--json', '--data-dir', '.relayflowd', '--local-agent', '--agent-capacity', '8', '--no-spawn', '--no-observer-link', '--allow-human-influenced', RUN_ID], variant: 'resume', }, { verb: 'run', argv: ['run', 'flow.yaml'], variant: 'run' }, { verb: 'run', - argv: ['run', '--json', '--data-dir', '.relayflowd', '--local-agent', '--no-spawn', + argv: ['run', '--json', '--data-dir', '.relayflowd', '--local-agent', '--agent-capacity', '8', '--no-spawn', '--no-observer-link', '--allow-human-influenced', '--input', '{"a":1}', 'review.flow.ts'], variant: 'run', }, diff --git a/packages/sdk/tests/worker-slots.test.ts b/packages/sdk/tests/worker-slots.test.ts new file mode 100644 index 000000000..a1866a9be --- /dev/null +++ b/packages/sdk/tests/worker-slots.test.ts @@ -0,0 +1,74 @@ +import { describe, expect, it } from 'vitest'; +import { parseCliArgs } from '../src/cli.js'; +import { DEFAULT_LOCAL_AGENT_CAPACITY, MAX_LOCAL_AGENT_CAPACITY, WorkerSlots } from '../src/worker-slots.js'; + +function deferred() { + let resolve!: () => void; + const promise = new Promise(done => { resolve = done; }); + return { promise, resolve }; +} + +describe('WorkerSlots', () => { + it('holds at most `capacity` at once and admits the rest in arrival order', async () => { + const slots = new WorkerSlots(2); + const gates = [deferred(), deferred(), deferred(), deferred()]; + const started: number[] = []; + let live = 0; + let peak = 0; + const runs = gates.map((gate, index) => slots.run(async () => { + started.push(index); + peak = Math.max(peak, ++live); + await gate.promise; + live--; + return index; + })); + await Promise.resolve(); + expect(started).toEqual([0, 1]); + gates[1]!.resolve(); + await runs[1]; + expect(started).toEqual([0, 1, 2]); + gates[0]!.resolve(); gates[2]!.resolve(); gates[3]!.resolve(); + expect(await Promise.all(runs)).toEqual([0, 1, 2, 3]); + expect(started).toEqual([0, 1, 2, 3]); + expect(peak).toBe(2); + }); + + it('frees the slot when the work throws', async () => { + const slots = new WorkerSlots(1); + await expect(slots.run(async () => { throw new Error('boom'); })).rejects.toThrow('boom'); + expect(await slots.run(async () => 'next')).toBe('next'); + }); + + it('refuses a capacity that is not an integer from 1 to the ceiling', () => { + for (const bad of [0, -1, 1.5, MAX_LOCAL_AGENT_CAPACITY + 1, Number.NaN]) { + expect(() => new WorkerSlots(bad)).toThrow(RangeError); + } + }); +}); + +describe('--agent-capacity', () => { + it('sizes the local workers on run and resume', () => { + expect(parseCliArgs(['run', '--local-agent', '--agent-capacity', '8', 'flow.yaml'])) + .toMatchObject({ command: 'run', localAgent: true, agentCapacity: 8 }); + expect(parseCliArgs(['resume', '--local-agent', '--agent-capacity', '1', 'run-1'])) + .toMatchObject({ command: 'resume', agentCapacity: 1 }); + expect(parseCliArgs(['run', '--local-agent', 'flow.yaml'])).toMatchObject({ agentCapacity: undefined }); + expect(DEFAULT_LOCAL_AGENT_CAPACITY).toBeGreaterThan(1); + }); + + it('is refused as an invocation when it is not a capacity or has no local worker to size', () => { + for (const argv of [ + ['run', '--local-agent', '--agent-capacity', '0', 'flow.yaml'], + ['run', '--local-agent', '--agent-capacity', String(MAX_LOCAL_AGENT_CAPACITY + 1), 'flow.yaml'], + ['run', '--local-agent', '--agent-capacity', '2.5', 'flow.yaml'], + ['run', '--local-agent', '--agent-capacity', '0x4', 'flow.yaml'], + ['run', '--local-agent', '--agent-capacity', 'flow.yaml'], + ['run', '--local-agent', '--agent-capacity', '2', '--agent-capacity', '3', 'flow.yaml'], + ['run', '--agent-capacity', '2', 'flow.yaml'], + ['run', '--cloud', '--agent-capacity', '2', 'flow.yaml'], + ['check', '--agent-capacity', '2', 'flow.yaml'], + ]) { + expect(parseCliArgs(argv), argv.join(' ')).toBeUndefined(); + } + }); +}); From ab065fcb41538c78cf61c7e2338b16328056a0b1 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Tue, 22 Sep 2026 11:31:15 -0700 Subject: [PATCH 2/6] fix(kernel,sdk): carry an agent step's cwd so agents in separate directories run side by side `f.agent({ cwd })` and YAML `cwd` were documented and threaded by the SDK, but the kernel's StepSpec had no such field, so every such step was refused with `invalid_spec: unknown field "cwd"`. With every agent sharing the runner's directory, the worker's per-directory serialization made concurrent agents take turns even with spare capacity. - Kernel: optional absolute `cwd` on agent steps, carried and dispatched like `model`. A relative path is refused (RelativeStepCwd). Omitting it serializes the step exactly as before; the step spec hash for a cwd-less step is pinned to main's value. - SDK: the authored surface resolves a relative `cwd` against the runner's directory before submission. - Test: two agents in distinct directories overlap at capacity 2 against a real daemon (fails on a kernel without this change). Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/SURFACE.md | 11 +++-- kernel/relayflowd-core/src/spec.rs | 16 ++++++ kernel/relayflowd-core/src/spec/tests.rs | 49 +++++++++++++++++++ packages/sdk/src/authored-worker-step.ts | 4 +- .../tests/authored-parallel-agents.test.ts | 17 ++++++- 5 files changed, 92 insertions(+), 5 deletions(-) diff --git a/docs/SURFACE.md b/docs/SURFACE.md index ca9eb793e..7ee714298 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -372,9 +372,14 @@ wait, they would be submitted with no worker free, and the kernel would park them. Agents that share a working directory still run one at a time. This lets the -worker attribute each file change to the step that made it. So today -concurrent `f.agent` calls do not overlap, although they no longer park. -Concurrent `f.llm` calls do overlap. +worker attribute each file change to the step that made it. To run agents side +by side, give each its own directory with `cwd` (for example one git worktree +per agent): `f.agent("api", { task, cwd: "/repo/.wt/api" })`. The kernel carries +`cwd` on the agent step and the worker starts the CLI there. It must be +absolute in the kernel spec; the TypeScript surface resolves a relative `cwd` +against the runner's directory, while a relative `cwd` in YAML is refused. +Setting `cwd` is part of the step's spec hash; omitting it hashes exactly as +before. Concurrent `f.llm` calls always overlap. The LLM step remains `type: llm` in the journal. It uses the same CLI resolution, authentication probes, and exact `flows.json` model allow-list as agent steps; diff --git a/kernel/relayflowd-core/src/spec.rs b/kernel/relayflowd-core/src/spec.rs index bc5f4dae3..94f48a3b0 100644 --- a/kernel/relayflowd-core/src/spec.rs +++ b/kernel/relayflowd-core/src/spec.rs @@ -163,6 +163,14 @@ impl RunSpec { if cli.as_ref().is_some_and(|value| value.trim().is_empty()) { return Err(SpecError::EmptyStepCli(step.id.clone())); } + if let StepKind::Agent { cwd: Some(cwd), .. } = &step.kind + && !cwd.starts_with('/') + { + return Err(SpecError::RelativeStepCwd { + step: step.id.clone(), + cwd: cwd.clone(), + }); + } if let StepKind::Agent { surfaces, .. } = &step.kind { for workspace in &surfaces.workspace { if path_surface_identity(&workspace.surface).is_none() { @@ -314,6 +322,7 @@ const STEP_AGENT_FIELDS: &[&str] = &[ "cli", "model", "transport", + "cwd", "recovery_mode", "surfaces", "permissions", @@ -429,6 +438,11 @@ pub enum StepKind { /// choice so the worker can honor it deterministically. #[serde(default, skip_serializing_if = "Option::is_none")] transport: Option, + /// Absolute working directory the worker starts the CLI in. Carried + /// and dispatched like `model`: the kernel never enters it, and + /// omitting it serializes the step exactly as before. + #[serde(default, skip_serializing_if = "Option::is_none")] + cwd: Option, #[serde(default)] recovery_mode: RecoveryMode, /// Declared mutable surfaces (RFC Appendix A rule 1) — names only. @@ -739,6 +753,8 @@ pub enum SpecError { EmptyStepId, #[error("step {0} cli cannot be empty")] EmptyStepCli(String), + #[error("agent step {step} cwd must be an absolute path, got {cwd:?}")] + RelativeStepCwd { step: String, cwd: String }, #[error("agent step {step} declares non-canonical external surface {path:?}")] InvalidExternalSurface { step: String, path: String }, #[error("agent step {step} declares non-canonical workspace surface {surface:?}")] diff --git a/kernel/relayflowd-core/src/spec/tests.rs b/kernel/relayflowd-core/src/spec/tests.rs index b8443bfff..d41a233b6 100644 --- a/kernel/relayflowd-core/src/spec/tests.rs +++ b/kernel/relayflowd-core/src/spec/tests.rs @@ -329,3 +329,52 @@ fn preflight_data_is_fail_closed() { Err(SpecError::EmptyStepCli("a".to_owned())) ); } + +#[test] +fn agent_cwd_is_carried_and_must_be_absolute() { + let with_cwd = RunSpec::parse(&json!({ + "steps": [{"id": "a", "type": "agent", "instruction": "i", "cwd": "/repo/.wt/a"}] + })) + .unwrap(); + assert!(with_cwd.validate().is_ok()); + assert_eq!( + serde_json::to_value(&with_cwd.steps[0]).unwrap()["cwd"], + json!("/repo/.wt/a") + ); + + let relative = RunSpec::parse(&json!({ + "steps": [{"id": "a", "type": "agent", "instruction": "i", "cwd": "wt/a"}] + })) + .unwrap(); + assert_eq!( + relative.validate(), + Err(SpecError::RelativeStepCwd { + step: "a".to_owned(), + cwd: "wt/a".to_owned() + }) + ); + + // `cwd` is agent-only: an llm step still refuses it as an unknown field. + assert!(matches!( + RunSpec::parse(&json!({ + "steps": [{"id": "a", "type": "llm", "prompt": "p", "cwd": "/repo"}] + })), + Err(SpecError::UnknownField { .. }) + )); +} + +#[test] +fn agent_without_cwd_hashes_as_before() { + let spec = RunSpec::parse(&json!({ + "steps": [{"id": "a", "type": "agent", "instruction": "i", "cli": "claude"}] + })) + .unwrap(); + let serialized = serde_json::to_value(&spec.steps[0]).unwrap(); + assert!(serialized.get("cwd").is_none()); + // Pinned to the value main computes: journals recorded before `cwd` + // existed must still memoize against this step. + assert_eq!( + crate::memoization::step_spec_hash(&spec.steps[0]), + "f5e8e24cd23fb3ee42ae0e8ddb7d68b0c869720fed13adce3bd5b23a8a06704f" + ); +} diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts index 278064fa6..176544307 100644 --- a/packages/sdk/src/authored-worker-step.ts +++ b/packages/sdk/src/authored-worker-step.ts @@ -1,3 +1,4 @@ +import { resolve } from 'node:path'; import type { AuthoredBudget } from './authored-budget.js'; import { parseBudget } from './budget.js'; import type { AgentOptions, AgentResult, LlmOptions, NamedGate } from '@relayflows/surface'; @@ -173,7 +174,8 @@ export function authoredWorkerRunner( ...(options.workspace === undefined ? {} : { surfaces: { workspace: [{ surface: options.workspace }] } }), ...(options.cli === undefined ? {} : { cli: options.cli }), ...(options.model === undefined ? {} : { model: options.model }), - ...(options.cwd === undefined ? {} : { cwd: options.cwd }), + // The kernel takes only an absolute cwd; a relative one means the runner's own directory. + ...(options.cwd === undefined ? {} : { cwd: resolve(options.cwd) }), ...(options.transport === undefined ? {} : { transport: options.transport }), ...(verification === undefined ? {} : { verification }), }); diff --git a/packages/sdk/tests/authored-parallel-agents.test.ts b/packages/sdk/tests/authored-parallel-agents.test.ts index cb1c9610f..8ef84f13a 100644 --- a/packages/sdk/tests/authored-parallel-agents.test.ts +++ b/packages/sdk/tests/authored-parallel-agents.test.ts @@ -1,4 +1,4 @@ -import { readFileSync, writeFileSync } from 'node:fs'; +import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { join, resolve } from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; import { flow } from '@relayflows/surface'; @@ -83,6 +83,21 @@ describe('authored steps under local workers with capacity', () => { // turns for artifact attribution (worker-cli.ts serializedByDirectory). }); + it('runs agents in distinct working directories side by side (the kernel carries cwd)', async () => { + const { fixture, client, agent, readSpans } = await slowAgents(2); + const trees = ['a', 'b'].map(name => join(fixture.root, 'trees', name)); + for (const tree of trees) mkdirSync(tree, { recursive: true }); + const inTrees = flow('two-trees', async f => { + await Promise.all(trees.map((cwd, index) => f.agent(`tree-${index}`, { task: 'work here', cwd }))); + f.done('success'); + }); + const result = await executeAuthoredFlow(inTrees, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, workerCapacity: 2, + }); + expect(result.completionReason).toBe('success'); + expect(peakOverlap(readSpans())).toBe(2); + }); + it('parks the overflow when the body is not told the capacity (the defect this closes)', async () => { const { fixture, client, agent } = await slowAgents(1); const run = executeAuthoredFlow(threeReviewers, client, undefined, { From 3985c2702a8340a08ffaba01c78ba54865fb5d2e Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Tue, 22 Sep 2026 12:11:58 -0700 Subject: [PATCH 3/6] fix(sdk): make the authored completion gate near-linear so it cannot starve lease renewal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 114-step task-graph run with parallel agents finished its body, then spent ~180 s in the completion gate without yielding. The root's lease renewal (timer-driven, same event loop) never fired, the lease lapsed, and the run was reaped as `crashed` although every step had succeeded. Profiled on a synthetic task-graph flow (map of subtask promises, each awaiting Promise.all of its deps, then a chain of steps), two terms were super-linear: 1. `aggregatesFor` asked `dependsOn(member, invocation)` per operation × per invocation × per combinator member; each call re-walked the member's whole ancestry (the single-entry cache never hit). Ancestries include every promise a step created while waiting (journal polling), so long steps made it far worse. Now one batch pass (`promise-ancestry.ts`): the upward closure of all members is visited once, cycles condensed with an iterative Tarjan, invocation sets propagated as bitsets. Same answers as the pairwise walk; cached per graph version. 2. `observeCallbackFailures` / `derivedWorkInFlight` scanned every tracked promise once per operation, and observed each settled promise once per operation whose roots overlap. Now one pass indexed by root, one observer per promise crediting every owning operation. The single-entry `dependenciesOf` cache is now also keyed by graph version, so it can no longer serve a walk from before the graph grew. Gate on the synthetic shape, 40/80/160 operations: 316/2 871/26 342 ms before, 20/36/68 ms after. Run-level gap between body return and completion (real daemon, 162 steps, 2 s steps): 4 468 ms before, 92 ms after. Tests: a gate-cost regression over a 160-operation task graph (fails at ~25 s before, bound 1 s), and a direct test of the reachability pass against the pairwise walk (cycles, >32 targets, a 200 000-deep chain). `causesOf` matches the accessor #553 adds, so that branch's addition collapses on rebase. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/sdk/src/authored-flow-lifecycle.ts | 80 ++++++++++++--- packages/sdk/src/authored-promise-graph.ts | 65 ++++++++---- packages/sdk/src/promise-ancestry.ts | 98 +++++++++++++++++++ .../sdk/tests/authored-flow-operation.test.ts | 40 ++++++++ packages/sdk/tests/promise-ancestry.test.ts | 44 +++++++++ 5 files changed, 294 insertions(+), 33 deletions(-) create mode 100644 packages/sdk/src/promise-ancestry.ts create mode 100644 packages/sdk/tests/promise-ancestry.test.ts diff --git a/packages/sdk/src/authored-flow-lifecycle.ts b/packages/sdk/src/authored-flow-lifecycle.ts index 88ca4c708..31fe2caea 100644 --- a/packages/sdk/src/authored-flow-lifecycle.ts +++ b/packages/sdk/src/authored-flow-lifecycle.ts @@ -124,6 +124,9 @@ export class AuthoredFlowLifecycle { private readonly callbackFailures = new Map(); private completionAsyncId: number | undefined; private closed = false; + /** Bumped by every invocation and combinator registration; with the graph's version it keys `groupsByInvocation`. */ + private registrations = 0; + private groupsByInvocation: { readonly key: string; readonly groups: Map> } | undefined; constructor() { this.graph = new AuthoredPromiseGraph( @@ -168,6 +171,7 @@ export class AuthoredFlowLifecycle { operationAggregates.add(aggregateId); } this.promiseAllGroups.push({ aggregate: aggregateId, members: memberIds }); + this.registrations++; } registerInvocation( @@ -178,6 +182,7 @@ export class AuthoredFlowLifecycle { const operationInvocations = this.invocations.get(operation); if (operationInvocations === undefined) this.invocations.set(operation, [invocation]); else operationInvocations.push(invocation); + this.registrations++; this.graph.registerRoot(asyncId); return invocation; } @@ -244,21 +249,43 @@ export class AuthoredFlowLifecycle { * Must be called before the gate awaits anything. */ derivedWorkInFlight(operations: readonly T[]): T[] { + const inFlight = this.graph.rootsInFlight(); return operations.filter((operation) => - this.graph.inFlightFrom(this.rootsFor(operation)).length > 0); + [...this.rootsFor(operation)].some((root) => inFlight.has(root))); } + /** + * Observe each settled derived promise once, crediting a rejection to every + * operation whose roots it derives from. Observing it once per operation + * instead — each after a scan of every tracked promise — was the gate's + * second quadratic term: roots of a task graph overlap heavily, so every + * operation re-observed most of the flow. + */ async observeCallbackFailures(operations: readonly OperationToken[]): Promise { - const observations: Promise[] = []; + const operationsByRoot = new Map(); for (const operation of operations) { - for (const promise of this.graph.settledFrom(this.rootsFor(operation))) { - observations.push(nativePromiseThen.call( - promise, - () => undefined, - (error: unknown) => { this.recordCallbackFailure(operation, error); }, - )); + for (const root of this.rootsFor(operation)) { + const owners = operationsByRoot.get(root); + if (owners === undefined) operationsByRoot.set(root, [operation]); + else owners.push(operation); } } + const owners = new Map, Set>(); + for (const { root, handle } of this.graph.settledWithRoots()) { + const rootOwners = operationsByRoot.get(root); + if (rootOwners === undefined) continue; + let promiseOwners = owners.get(handle); + if (promiseOwners === undefined) owners.set(handle, promiseOwners = new Set()); + for (const operation of rootOwners) promiseOwners.add(operation); + } + const observations: Promise[] = []; + for (const [promise, promiseOwners] of owners) { + observations.push(nativePromiseThen.call( + promise, + () => undefined, + (error: unknown) => { for (const operation of promiseOwners) this.recordCallbackFailure(operation, error); }, + )); + } await Promise.all(observations); } @@ -280,6 +307,7 @@ export class AuthoredFlowLifecycle { this.invocations.clear(); this.promiseAllAggregates.clear(); this.promiseAllGroups.length = 0; + this.groupsByInvocation = undefined; this.callbackFailures.clear(); this.activeResolverProbes.length = 0; uninstallPromiseAllObserver(); @@ -301,17 +329,41 @@ export class AuthoredFlowLifecycle { private aggregatesFor(operation: OperationToken): Set { const aggregates = new Set(this.promiseAllAggregates.get(operation) ?? []); + const groups = this.aggregatesByInvocation(); for (const invocation of this.invocations.get(operation) ?? []) { - for (const group of this.promiseAllGroups) { - if ( - group.members.size > 0 - && [...group.members].some((member) => this.graph.dependsOn(member, invocation.asyncId)) - ) { + for (const aggregate of groups.get(invocation.asyncId) ?? []) aggregates.add(aggregate); + } + return aggregates; + } + + /** + * For each invocation, the aggregates of every combinator group with a + * member that depends on it. Asked per group member per invocation, this was + * the gate's quadratic hot path: each question re-walked the member's whole + * ancestry. It is now one batch pass, recomputed only when the graph or the + * registrations have changed since the last gate question. + */ + private aggregatesByInvocation(): Map> { + const key = `${this.graph.version}:${this.registrations}`; + if (this.groupsByInvocation?.key === key) return this.groupsByInvocation.groups; + const targets: number[] = []; + for (const operationInvocations of this.invocations.values()) { + for (const invocation of operationInvocations) targets.push(invocation.asyncId); + } + const members = [...new Set(this.promiseAllGroups.flatMap((group) => [...group.members]))]; + const reached = this.graph.dependenciesAmong(members, targets); + const groups = new Map>(); + for (const group of this.promiseAllGroups) { + for (const member of group.members) { + for (const target of reached.get(member) ?? []) { + let aggregates = groups.get(target); + if (aggregates === undefined) groups.set(target, aggregates = new Set()); aggregates.add(group.aggregate); } } } - return aggregates; + this.groupsByInvocation = { key, groups }; + return groups; } } diff --git a/packages/sdk/src/authored-promise-graph.ts b/packages/sdk/src/authored-promise-graph.ts index a96d85c34..7168550ba 100644 --- a/packages/sdk/src/authored-promise-graph.ts +++ b/packages/sdk/src/authored-promise-graph.ts @@ -1,4 +1,5 @@ import { createHook, executionAsyncId, type AsyncHook } from 'node:async_hooks'; +import { reachableTargets } from './promise-ancestry.js'; /** * The promise graph an authored flow body actually creates. @@ -35,7 +36,9 @@ export class AuthoredPromiseGraph { private readonly attributedRoots = new Map(); private readonly roots = new Set(); private readonly pending = new Set(); - private dependencyCache: { readonly start: number; readonly found: ReadonlySet } | undefined; + private dependencyCache: { readonly start: number; readonly found: ReadonlySet; readonly version: number } | undefined; + /** Bumped by every graph change, so derived answers can be cached between changes. */ + private changes = 0; constructor( private readonly inScope: () => boolean, @@ -45,6 +48,7 @@ export class AuthoredPromiseGraph { init: (asyncId, type, triggerAsyncId, resource) => { if (type !== 'PROMISE' || typeof resource !== 'object' || resource === null) return; if (!this.inScope()) return; + this.changes++; this.triggers.set(asyncId, triggerAsyncId); this.creationContexts.set(asyncId, executionAsyncId()); this.promiseIds.set(resource, asyncId); @@ -59,6 +63,7 @@ export class AuthoredPromiseGraph { promiseResolve: (asyncId) => { this.onPromiseResolve(asyncId); if (!this.triggers.has(asyncId)) return; + this.changes++; const cause = executionAsyncId(); if (cause !== asyncId) this.resolutionCauses.set(asyncId, cause); this.pending.delete(asyncId); @@ -98,6 +103,7 @@ export class AuthoredPromiseGraph { /** Mark a promise whose descendants belong to an authored operation. */ registerRoot(asyncId: number): void { if (asyncId <= 0 || this.roots.has(asyncId)) return; + this.changes++; this.roots.add(asyncId); const parked = this.unattributedChildren.get(asyncId); if (parked === undefined) return; @@ -105,31 +111,58 @@ export class AuthoredPromiseGraph { for (const child of parked) this.attribute(child, asyncId); } - /** Promises derived from `roots` that have not settled. */ - inFlightFrom(roots: ReadonlySet): number[] { - const found: number[] = []; + /** The roots that still have derived promises in flight. One pass over `pending`. */ + rootsInFlight(): Set { + const found = new Set(); for (const asyncId of this.pending) { const root = this.attributedRoots.get(asyncId); - if (root !== undefined && roots.has(root)) found.push(asyncId); + if (root !== undefined) found.add(root); } return found; } - /** Settled promises derived from `roots`, with their handles. */ - settledFrom(roots: ReadonlySet): Promise[] { - const found: Promise[] = []; + /** Every settled derived promise with its root, in attribution order. One pass. */ + settledWithRoots(): Array<{ readonly root: number; readonly handle: Promise }> { + const found: Array<{ root: number; handle: Promise }> = []; for (const [asyncId, root] of this.attributedRoots) { - if (!roots.has(root) || this.pending.has(asyncId)) continue; + if (this.pending.has(asyncId)) continue; const handle = this.handles.get(asyncId); - if (handle !== undefined) found.push(handle); + if (handle !== undefined) found.push({ root, handle }); } return found; } + /** The three recorded edges out of one promise, for walks that stop early. */ + causesOf(asyncId: number): number[] { + const causes: number[] = []; + for (const edge of [ + this.triggers.get(asyncId), + this.resolutionCauses.get(asyncId), + this.creationContexts.get(asyncId), + ]) { + if (edge !== undefined && edge !== asyncId) causes.push(edge); + } + return causes; + } + dependsOn(descendant: number, ancestor: number): boolean { return this.dependenciesOf(descendant).has(ancestor); } + /** A counter that changes whenever the graph does; equal values mean equal answers. */ + get version(): number { + return this.changes; + } + + /** + * `dependsOn(start, target)` for every pair at once: for each start, the + * targets it depends on. One pass over the shared ancestry instead of one + * walk per pair (see promise-ancestry.ts). + */ + dependenciesAmong(starts: readonly number[], targets: readonly number[]): Map> { + return reachableTargets(starts, targets, (node) => this.causesOf(node)); + } + clear(): void { this.triggers.clear(); this.creationContexts.clear(); @@ -175,22 +208,16 @@ export class AuthoredPromiseGraph { */ private dependenciesOf(start: number): ReadonlySet { const cached = this.dependencyCache; - if (cached !== undefined && cached.start === start) return cached.found; + if (cached !== undefined && cached.start === start && cached.version === this.changes) return cached.found; const found = new Set(); const stack = [start]; while (stack.length > 0) { const current = stack.pop()!; if (found.has(current)) continue; found.add(current); - for (const edge of [ - this.triggers.get(current), - this.resolutionCauses.get(current), - this.creationContexts.get(current), - ]) { - if (edge !== undefined && edge !== current) stack.push(edge); - } + for (const edge of this.causesOf(current)) stack.push(edge); } - this.dependencyCache = { start, found }; + this.dependencyCache = { start, found, version: this.changes }; return found; } } diff --git a/packages/sdk/src/promise-ancestry.ts b/packages/sdk/src/promise-ancestry.ts new file mode 100644 index 000000000..3646879b9 --- /dev/null +++ b/packages/sdk/src/promise-ancestry.ts @@ -0,0 +1,98 @@ +/** + * Which of a fixed set of `targets` each of `starts` can reach by following + * `edgesOf` — the batch form of asking `dependsOn(start, target)` for every + * pair, with the same answer (a node reaches itself). + * + * The completion gate used to ask that pair-by-pair, re-walking each start's + * whole ancestry every time: operations × combinator members × tracked + * promises. A flow of ~110 steps whose agents ran for twenty minutes produced + * enough tracked promises (journal polling while each step waited) that the + * gate ran for three minutes without yielding, starving the root's lease + * renewal until the run was reaped as crashed. + * + * Here the upward closure of every start is visited once, cycles (a promise + * resolved from a context created after it) are condensed with an iterative + * Tarjan pass, and target sets flow as bitsets from ancestors to descendants + * in the order Tarjan emits components. Cost is O((V + E) · ⌈T / 32⌉) for the + * visited subgraph, whatever the number of starts. + */ +export function reachableTargets( + starts: readonly number[], + targets: readonly number[], + edgesOf: (node: number) => readonly number[], +): Map> { + const bit = new Map(); + for (const target of targets) if (!bit.has(target)) bit.set(target, bit.size); + const words = Math.max(1, Math.ceil(bit.size / 32)); + + const index = new Map(); + const low = new Map(); + const onStack = new Set(); + const stack: number[] = []; + const component = new Map(); + const labels: Uint32Array[] = []; + + const closeComponent = (root: number): void => { + const label = new Uint32Array(words); + const members: number[] = []; + let member: number; + do { + member = stack.pop()!; + onStack.delete(member); + members.push(member); + } while (member !== root); + const id = labels.length; + for (const node of members) component.set(node, id); + for (const node of members) { + const own = bit.get(node); + if (own !== undefined) label[own >>> 5]! |= 1 << (own & 31); + // Every edge leaving the component points at one Tarjan already closed. + for (const edge of edgesOf(node)) { + const target = component.get(edge); + if (target === undefined || target === id) continue; + const inherited = labels[target]!; + for (let word = 0; word < words; word++) label[word]! |= inherited[word]!; + } + } + labels.push(label); + }; + + // Iterative Tarjan: a deep promise chain must not overflow the JS stack. + for (const start of starts) { + if (index.has(start)) continue; + const frames: Array<{ node: number; edges: readonly number[]; next: number }> = []; + const open = (node: number): void => { + index.set(node, index.size); + low.set(node, index.get(node)!); + stack.push(node); + onStack.add(node); + frames.push({ node, edges: edgesOf(node), next: 0 }); + }; + open(start); + while (frames.length > 0) { + const frame = frames.at(-1)!; + if (frame.next < frame.edges.length) { + const edge = frame.edges[frame.next++]!; + if (!index.has(edge)) open(edge); + else if (onStack.has(edge)) low.set(frame.node, Math.min(low.get(frame.node)!, index.get(edge)!)); + continue; + } + frames.pop(); + const parent = frames.at(-1); + if (parent !== undefined) low.set(parent.node, Math.min(low.get(parent.node)!, low.get(frame.node)!)); + if (low.get(frame.node) === index.get(frame.node)) closeComponent(frame.node); + } + } + + const reached = new Map>(); + for (const start of starts) { + if (reached.has(start)) continue; + const label = labels[component.get(start)!]!; + const found = new Set(); + for (const [target, position] of bit) { + if ((label[position >>> 5]! & (1 << (position & 31))) !== 0) found.add(target); + } + reached.set(start, found); + } + return reached; +} diff --git a/packages/sdk/tests/authored-flow-operation.test.ts b/packages/sdk/tests/authored-flow-operation.test.ts index 11e62f0f6..16aff1729 100644 --- a/packages/sdk/tests/authored-flow-operation.test.ts +++ b/packages/sdk/tests/authored-flow-operation.test.ts @@ -152,6 +152,46 @@ it('completes the gate in linear time over a body with 30000 ordinary awaits', a } }); +// The task-graph shape (flows examples/task-graph): a map of subtask promises, +// each awaiting Promise.all of its dependencies, then a chain of steps, with +// ordinary awaits standing in for the journal polling a long step does while it +// waits. The gate used to ask dependsOn(member, invocation) per operation per +// combinator member, re-walking the member's whole ancestry each time, and to +// observe every settled derived promise once per operation: cubic. A 114-step +// agent run spent ~180 s in it without yielding, starving the root's lease +// renewal until the run was reaped as crashed. Measured on this shape (160 +// operations): 26 342 ms before, 68 ms after (40 / 80 / 160 operations: +// 316 / 2 871 / 26 342 ms before, 20 / 36 / 68 ms after). The bound fails +// anything quadratic in operations and will not flake on a slow host. +it('completes the gate in near-linear time over a large task graph of Promise.all groups', async () => { + const lifecycle = new AuthoredFlowLifecycle(); + const operations: AuthoredFlowOperation[] = []; + const step = async (): Promise => { + const authored = operation('run', lifecycle, operations.length + 1); + operations.push(authored as AuthoredFlowOperation); + for (let poll = 0; poll < 40; poll++) await Promise.resolve(poll); + await authored.step; + }; + try { + await lifecycle.runBody(async () => { + const done = new Map>(); + const subtask = async (id: number): Promise => { + await Promise.all([done.get(id - 1), done.get(id - 2)].filter(Boolean)); + for (let index = 0; index < 4; index++) await step(); + }; + for (let id = 0; id < 40; id++) done.set(id, subtask(id)); + await Promise.all(done.values()); + lifecycle.markCompletion(); + }); + expect(operations).toHaveLength(160); + const startedAt = Date.now(); + await verifyAuthoredOperations('task-graph-gate', operations, lifecycle); + expect(Date.now() - startedAt).toBeLessThan(1_000); + } finally { + lifecycle.close(); + } +}); + // P1-B, second half: the graph used to retain every promise created ANYWHERE in // the process for the life of the flow (a probe measured 20 002 unrelated // promises held by strong reference). Only promises created inside the flow's diff --git a/packages/sdk/tests/promise-ancestry.test.ts b/packages/sdk/tests/promise-ancestry.test.ts new file mode 100644 index 000000000..35aee6308 --- /dev/null +++ b/packages/sdk/tests/promise-ancestry.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from 'vitest'; +import { reachableTargets } from '../src/promise-ancestry.js'; + +/** Edges point from a promise to what it depends on, as the promise graph's do. */ +function graph(edges: Record) { + return (node: number): readonly number[] => edges[node] ?? []; +} + +/** The pairwise definition the batch pass must agree with. */ +function walk(start: number, edgesOf: (node: number) => readonly number[]): Set { + const found = new Set(); + const stack = [start]; + while (stack.length > 0) { + const node = stack.pop()!; + if (found.has(node)) continue; + found.add(node); + stack.push(...edgesOf(node)); + } + return found; +} + +describe('reachableTargets', () => { + it('matches a pairwise walk, a node reaching itself, across a cycle', () => { + // 5 -> 4 -> 3 -> 4 (cycle via a resolution cause), 3 -> 1, 5 -> 2; 6 is unrelated. + const edgesOf = graph({ 5: [4, 2], 4: [3], 3: [4, 1], 6: [7] }); + const targets = [1, 2, 3, 4, 5, 6, 7, 99]; + const reached = reachableTargets([5, 4, 3, 6, 1], targets, edgesOf); + for (const start of [5, 4, 3, 6, 1]) { + const expected = [...walk(start, edgesOf)].filter((node) => targets.includes(node)); + expect([...reached.get(start)!].sort()).toEqual(expected.sort()); + } + expect(reached.get(5)).toEqual(new Set([1, 2, 3, 4, 5])); + expect(reached.get(4)).toEqual(new Set([1, 3, 4])); + }); + + it('handles more than 32 targets and a chain too deep for recursion', () => { + const depth = 200_000; + const edgesOf = (node: number): readonly number[] => (node > 0 ? [node - 1] : []); + const targets = Array.from({ length: 70 }, (_, index) => index * 1_000); + const reached = reachableTargets([depth, 35_500], targets, edgesOf); + expect(reached.get(depth)!.size).toBe(70); + expect([...reached.get(35_500)!].sort((a, b) => a - b)).toEqual(targets.slice(0, 36)); + }); +}); From 37f8c745887024b0f54c8d9bc5d51446329ac98e Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Tue, 22 Sep 2026 12:36:12 -0700 Subject: [PATCH 4/6] =?UTF-8?q?fix(sdk):=20address=20#554=20review=20?= =?UTF-8?q?=E2=80=94=20check-time=20cwd,=20cancellable=20slot=20queue,=20o?= =?UTF-8?q?verlapping=20trees=20serialize?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Declarative agent `cwd` must be absolute at `flows check` (validate.ts), matching the kernel's run.start refusal instead of passing check and failing later. Refused rather than resolved against the checking directory, which would make the compiled spec and its hash depend on where the check ran; `f.agent` still resolves before compiling. (Devin) - WorkerSlots.close(reason) rejects queued and later admissions; the authored executor calls it on body failure and on a missing done(), before stopping operations. A queued call is already `running`, so operation cancellation could not reach it and it was admitted after the flow had failed. Test: capacity 1, three agents, body fails -> one ran, two never started (three ran without the fix). (Codex P1) - The per-directory artifact queue keys on the realpath and serializes any two trees where one contains the other: a symlink alias of the same directory, and a nested tree (an agent in /repo while another writes /repo/.wt/x), since the snapshot walks the whole subtree. Sibling trees still run side by side. Tests for both cases against a real daemon. (Codex P2) Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/sdk/src/authored-flow-executor.ts | 2 + packages/sdk/src/authored-worker-step.ts | 5 ++ packages/sdk/src/validate.ts | 7 +++ packages/sdk/src/worker-cli.ts | 27 +++++++--- packages/sdk/src/worker-slots.ts | 21 ++++++-- .../sdk/tests/agent-cwd-validation.test.ts | 34 ++++++++++++ .../tests/authored-parallel-agents.test.ts | 52 ++++++++++++++++++- packages/sdk/tests/worker-slots.test.ts | 16 ++++++ 8 files changed, 151 insertions(+), 13 deletions(-) create mode 100644 packages/sdk/tests/agent-cwd-validation.test.ts diff --git a/packages/sdk/src/authored-flow-executor.ts b/packages/sdk/src/authored-flow-executor.ts index d58b8abe5..7aab04482 100644 --- a/packages/sdk/src/authored-flow-executor.ts +++ b/packages/sdk/src/authored-flow-executor.ts @@ -606,6 +606,7 @@ export async function executeAuthoredFlow( } if (bodyFailed) { try { + worker.stop(bodyFailure); await stopAuthoredOperations(authoredSteps, bodyFailure); } finally { lifecycle.close(); @@ -632,6 +633,7 @@ export async function executeAuthoredFlow( `flow "${definition.name}" returned without done()`, ); try { + worker.stop(missingCompletion); await stopAuthoredOperations(authoredSteps, missingCompletion); } finally { lifecycle.close(); diff --git a/packages/sdk/src/authored-worker-step.ts b/packages/sdk/src/authored-worker-step.ts index 176544307..d7793a352 100644 --- a/packages/sdk/src/authored-worker-step.ts +++ b/packages/sdk/src/authored-worker-step.ts @@ -126,6 +126,11 @@ export function authoredWorkerRunner( } return { + /** Refuse every agent/LLM call still waiting for a worker slot (body teardown). */ + stop(reason: unknown): void { + slots?.agent.close(reason); + slots?.llm.close(reason); + }, async agent(id: string, options: AgentOptions, verification?: NamedGate): Promise { if (options.workspace !== undefined && localAgentStream !== undefined) { throw new AuthoredFlowExecutionError('unsupported_workspace_permission', diff --git a/packages/sdk/src/validate.ts b/packages/sdk/src/validate.ts index 09b733c1a..cc6a3ca2d 100644 --- a/packages/sdk/src/validate.ts +++ b/packages/sdk/src/validate.ts @@ -486,6 +486,13 @@ class Validator { } this.validateCli(st.cli, at); this.validateModel(st.model, at); + // The kernel refuses a relative cwd at run.start; refuse it here so `flows + // check` and the run agree. Not resolved against the checking directory: + // that would make the compiled spec (and its hash) depend on where the + // check ran. `f.agent` resolves before compiling, so it never lands here. + if (st.cwd !== undefined && (!isNonEmptyString(st.cwd) || !st.cwd.startsWith('/'))) { + this.fail(`${at}.cwd: expected an absolute path (got ${JSON.stringify(st.cwd)})`); + } if (st.surfaces !== undefined) this.validateSurfaces(st.surfaces, `${at}.surfaces`); if (st.permissions !== undefined) this.validatePermissions(st.permissions, `${at}.permissions`); } diff --git a/packages/sdk/src/worker-cli.ts b/packages/sdk/src/worker-cli.ts index 4b95d30dc..eeb59c269 100644 --- a/packages/sdk/src/worker-cli.ts +++ b/packages/sdk/src/worker-cli.ts @@ -121,7 +121,7 @@ export async function runAgentCli( const artifactRoot = mode === 'agent' ? resolve(cwd ?? process.cwd()) : undefined; return artifactRoot === undefined ? execute() - : serializedByDirectory(artifactRoot, async () => { + : serializedByDirectory(realPath(artifactRoot)!, async () => { const before = await snapshotWorkspaceFiles(artifactRoot); const result = await execute(); const kernelData = sidechannel === undefined ? undefined : realPath(resolve(sidechannel.dataDir)); @@ -220,14 +220,25 @@ function realPath(path: string | undefined): string | undefined { try { return realpathSync(path); } catch { return path; } } -/** One agent at a time per canonical working directory, for the artifact interval. */ -const directoryQueues = new Map>(); +/** + * One agent at a time per overlapping working tree, for the artifact interval. + * + * `directory` must be canonical (symlink-free): `/repo` and a `/tmp/link` to it + * are one tree. Two trees overlap when one contains the other, because the + * snapshot walks the whole subtree — an agent in `/repo` would otherwise be + * credited with files an agent in `/repo/.wt/api` wrote meanwhile. Disjoint + * trees (sibling worktrees) run side by side. Each run waits for every + * earlier-registered overlapping run, so order within an overlap is arrival. + */ +const directoryRuns = new Set<{ readonly directory: string; readonly settled: Promise }>(); function serializedByDirectory(directory: string, task: () => Promise): Promise { - const previous = directoryQueues.get(directory) ?? Promise.resolve(); - const run = previous.then(task, task); - const settled = run.then(() => undefined, () => undefined); - directoryQueues.set(directory, settled); - void settled.then(() => { if (directoryQueues.get(directory) === settled) directoryQueues.delete(directory); }); + const blockers = [...directoryRuns] + .filter(other => under(other.directory, directory) || under(directory, other.directory)) + .map(other => other.settled); + const run = Promise.all(blockers).then(task); + const entry = { directory, settled: run.then(() => undefined, () => undefined) }; + directoryRuns.add(entry); + void entry.settled.then(() => { directoryRuns.delete(entry); }); return run; } diff --git a/packages/sdk/src/worker-slots.ts b/packages/sdk/src/worker-slots.ts index b7bcac61e..6f638d019 100644 --- a/packages/sdk/src/worker-slots.ts +++ b/packages/sdk/src/worker-slots.ts @@ -19,22 +19,37 @@ export function isAgentCapacity(value: unknown): value is number { */ export class WorkerSlots { private held = 0; - private readonly waiting: Array<() => void> = []; + private readonly waiting: Array<{ resolve: () => void; reject: (reason: unknown) => void }> = []; + private closed: { readonly reason: unknown } | undefined; constructor(readonly capacity: number) { if (!isAgentCapacity(capacity)) throw new RangeError(`worker capacity must be an integer from 1 to ${MAX_LOCAL_AGENT_CAPACITY} (got ${capacity})`); } async run(work: () => Promise): Promise { + if (this.closed !== undefined) throw this.closed.reason; if (this.held < this.capacity) this.held++; // A released slot is handed straight to the next waiter, so `held` never // dips below capacity while anyone is queued and no later caller can jump it. - else await new Promise(resolve => this.waiting.push(resolve)); + else await new Promise((resolve, reject) => this.waiting.push({ resolve, reject })); try { return await work(); } finally { const next = this.waiting.shift(); - if (next !== undefined) next(); else this.held--; + if (next !== undefined) next.resolve(); else this.held--; } } + + /** + * Refuse every queued and future admission with `reason`; work already + * holding a slot is not interrupted. The authored body calls this when it + * fails: a queued call is already `running` as an operation, so operation + * cancellation cannot reach it, and without this it would still be admitted + * and run after the flow had failed. + */ + close(reason: unknown): void { + if (this.closed !== undefined) return; + this.closed = { reason }; + for (const waiter of this.waiting.splice(0)) waiter.reject(reason); + } } diff --git a/packages/sdk/tests/agent-cwd-validation.test.ts b/packages/sdk/tests/agent-cwd-validation.test.ts new file mode 100644 index 000000000..dcddf1874 --- /dev/null +++ b/packages/sdk/tests/agent-cwd-validation.test.ts @@ -0,0 +1,34 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, resolve } from 'node:path'; +import { spawnSync } from 'node:child_process'; +import { afterEach, describe, expect, it } from 'vitest'; +import { validateSpec } from '../src/index.js'; + +const roots: string[] = []; +afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); + +const agent = (cwd: unknown) => ({ version: '0.1.0', steps: [{ id: 'a', type: 'agent', instruction: 'x', cwd }] }); + +// The kernel refuses a relative agent cwd at run.start (spec.rs RelativeStepCwd). +// A declarative flow must be refused by `flows check` the same way, not accepted +// and then refused minutes later. +describe('declarative agent cwd', () => { + it('accepts an absolute cwd and refuses a relative or empty one', () => { + expect(validateSpec(agent('/repo/.wt/api')).ok).toBe(true); + for (const cwd of ['worktrees/api', './api', '']) { + const result = validateSpec(agent(cwd)); + expect(result.ok).toBe(false); + expect(result.errors.join(' ')).toContain('steps[0].cwd: expected an absolute path'); + } + }); + + it('is refused by `flows check` on a YAML flow before anything runs', () => { + const root = mkdtempSync(join(tmpdir(), 'agent-cwd-')); roots.push(root); + const flow = join(root, 'flow.yaml'); + writeFileSync(flow, 'version: 0.1.0\nsteps:\n - id: a\n type: agent\n instruction: x\n cwd: worktrees/api\n'); + const checked = spawnSync(process.execPath, [resolve('dist/cli.js'), 'check', flow], { cwd: root, encoding: 'utf8' }); + expect(checked.status).toBe(2); + expect(checked.stdout + checked.stderr).toContain('cwd: expected an absolute path'); + }); +}); diff --git a/packages/sdk/tests/authored-parallel-agents.test.ts b/packages/sdk/tests/authored-parallel-agents.test.ts index 8ef84f13a..692b9cc68 100644 --- a/packages/sdk/tests/authored-parallel-agents.test.ts +++ b/packages/sdk/tests/authored-parallel-agents.test.ts @@ -1,4 +1,4 @@ -import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; +import { mkdirSync, readFileSync, symlinkSync, writeFileSync } from 'node:fs'; import { join, resolve } from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; import { flow } from '@relayflows/surface'; @@ -81,7 +81,8 @@ describe('authored steps under local workers with capacity', () => { expect(result.journalSteps.filter(step => step.id.startsWith('agent-'))).toHaveLength(3); // No overlap is asserted: agents sharing a working directory still take // turns for artifact attribution (worker-cli.ts serializedByDirectory). - }); + // They share this package's directory, so each snapshot walks it: allow time. + }, 20_000); it('runs agents in distinct working directories side by side (the kernel carries cwd)', async () => { const { fixture, client, agent, readSpans } = await slowAgents(2); @@ -98,6 +99,53 @@ describe('authored steps under local workers with capacity', () => { expect(peakOverlap(readSpans())).toBe(2); }); + // A snapshot walks the whole tree, so a symlink alias of the same directory, + // or a directory nested inside another agent's, must still take turns. + it.each([ + ['a symlink alias of the same directory', (tree: string, root: string) => { + const link = join(root, 'alias'); + symlinkSync(tree, link); + return link; + }], + ['a directory nested inside the other', (tree: string) => { + const nested = join(tree, '.wt', 'inner'); + mkdirSync(nested, { recursive: true }); + return nested; + }], + ])('serializes agents whose cwd is %s', async (_case, second) => { + const { fixture, client, agent, readSpans } = await slowAgents(2); + const tree = join(fixture.root, 'trees', 'shared'); + mkdirSync(tree, { recursive: true }); + const trees = [tree, second(tree, fixture.root)]; + const overlapping = flow('overlapping-trees', async f => { + await Promise.all(trees.map((cwd, index) => f.agent(`tree-${index}`, { task: 'work here', cwd }))); + f.done('success'); + }); + const result = await executeAuthoredFlow(overlapping, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, workerCapacity: 2, + }); + expect(result.completionReason).toBe('success'); + expect(peakOverlap(readSpans())).toBe(1); + }); + + it('never starts queued agents once the body has failed', async () => { + const { fixture, client, agent, readSpans } = await slowAgents(1); + const failing = flow('fails-while-queued', async f => { + await Promise.all([ + ...['a', 'b', 'c'].map(lens => f.agent(`review-${lens}`, { task: `Review for ${lens}` })), + new Promise((_, reject) => setTimeout(() => reject(new Error('body failed')), 100)), + ]); + f.done('success'); + }); + await expect(executeAuthoredFlow(failing, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, workerCapacity: 1, + })).rejects.toThrow('body failed'); + // Teardown waits for the one agent already holding the slot; the two + // queued behind it are refused, not admitted after the flow has failed. + await new Promise(done => setTimeout(done, 1_000)); + expect(readSpans()).toHaveLength(1); + }); + it('parks the overflow when the body is not told the capacity (the defect this closes)', async () => { const { fixture, client, agent } = await slowAgents(1); const run = executeAuthoredFlow(threeReviewers, client, undefined, { diff --git a/packages/sdk/tests/worker-slots.test.ts b/packages/sdk/tests/worker-slots.test.ts index a1866a9be..a593e2415 100644 --- a/packages/sdk/tests/worker-slots.test.ts +++ b/packages/sdk/tests/worker-slots.test.ts @@ -33,6 +33,22 @@ describe('WorkerSlots', () => { expect(peak).toBe(2); }); + it('close refuses queued and later calls but lets the running one finish', async () => { + const slots = new WorkerSlots(1); + const gate = deferred(); + const started: string[] = []; + const running = slots.run(async () => { started.push('a'); await gate.promise; return 'a'; }); + const queued = slots.run(async () => { started.push('b'); return 'b'; }); + await Promise.resolve(); + const reason = new Error('body failed'); + slots.close(reason); + await expect(queued).rejects.toBe(reason); + await expect(slots.run(async () => 'c')).rejects.toBe(reason); + gate.resolve(); + expect(await running).toBe('a'); + expect(started).toEqual(['a']); + }); + it('frees the slot when the work throws', async () => { const slots = new WorkerSlots(1); await expect(slots.run(async () => { throw new Error('boom'); })).rejects.toThrow('boom'); From 09f892a28666f0a2884a5b6da2b484e3eb0fa8c2 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Tue, 22 Sep 2026 12:49:22 -0700 Subject: [PATCH 5/6] fix(sdk): hand a freed worker slot over on a later macrotask; canonicalize missing cwd tails Cursor on #554: - When the agent holding a slot rejects and that rejection fails the body, the failure reaches WorkerSlots.close only through microtasks, but the slot was handed to the next waiter synchronously in `finally`, so one queued agent still reached run.start after the flow had failed. The handoff now happens on setImmediate and re-checks `closed`; a waiter taken off the queue before close() is refused there and the slot passed on, so it can neither run nor hang. Test: capacity 1, three agents, the first session fails -> one agent ever ran (two ran before this change). - The artifact lock's key fell back to the unresolved path when realpath failed, so `/link/new` (not created yet) and `/link` compared as disjoint. canonicalTree resolves the deepest existing ancestor and keeps the missing tail. Unit test with a symlinked ancestor. Co-Authored-By: Claude Opus 5.5 (1M context) --- packages/sdk/src/worker-cli.ts | 25 ++++++++++- packages/sdk/src/worker-slots.ts | 43 +++++++++++++++++-- .../tests/authored-parallel-agents.test.ts | 19 +++++++- packages/sdk/tests/canonical-tree.test.ts | 25 +++++++++++ packages/sdk/tests/worker-slots.test.ts | 18 ++++++++ 5 files changed, 123 insertions(+), 7 deletions(-) create mode 100644 packages/sdk/tests/canonical-tree.test.ts diff --git a/packages/sdk/src/worker-cli.ts b/packages/sdk/src/worker-cli.ts index eeb59c269..321cc38c3 100644 --- a/packages/sdk/src/worker-cli.ts +++ b/packages/sdk/src/worker-cli.ts @@ -1,5 +1,5 @@ import { realpathSync } from 'node:fs'; -import { resolve, sep } from 'node:path'; +import { basename, dirname, join, resolve, sep } from 'node:path'; import { diffWorkspaceFiles, snapshotWorkspaceFiles } from './agent-artifacts.js'; import { claudeResultOutcome, decodeProviderResult, decodeWrapperResult, requirePricedUsage } from './worker-usage.js'; import { openSidechannel, type SidechannelContext } from './pty-sidechannel.js'; @@ -121,7 +121,7 @@ export async function runAgentCli( const artifactRoot = mode === 'agent' ? resolve(cwd ?? process.cwd()) : undefined; return artifactRoot === undefined ? execute() - : serializedByDirectory(realPath(artifactRoot)!, async () => { + : serializedByDirectory(canonicalTree(artifactRoot), async () => { const before = await snapshotWorkspaceFiles(artifactRoot); const result = await execute(); const kernelData = sidechannel === undefined ? undefined : realPath(resolve(sidechannel.dataDir)); @@ -214,6 +214,27 @@ function under(path: string | undefined, directory: string | undefined): boolean return path === directory || path.startsWith(directory.endsWith(sep) ? directory : directory + sep); } +/** + * Symlink-free form of an absolute path that may not exist yet: the deepest + * existing ancestor is resolved and the missing tail is kept. A plain + * `realPath` fallback would leave `/link/new` unresolved while `/link` + * resolves, and the overlap check would then treat them as disjoint trees. + */ +export function canonicalTree(path: string): string { + const missing: string[] = []; + let current = path; + for (;;) { + try { + return join(realpathSync(current), ...missing.reverse()); + } catch { + const parent = dirname(current); + if (parent === current) return path; + missing.push(basename(current)); + current = parent; + } + } +} + /** Symlink-free form of a path, or the path itself when it cannot be resolved. */ function realPath(path: string | undefined): string | undefined { if (path === undefined) return undefined; diff --git a/packages/sdk/src/worker-slots.ts b/packages/sdk/src/worker-slots.ts index 6f638d019..24b9ffb77 100644 --- a/packages/sdk/src/worker-slots.ts +++ b/packages/sdk/src/worker-slots.ts @@ -31,13 +31,50 @@ export class WorkerSlots { if (this.held < this.capacity) this.held++; // A released slot is handed straight to the next waiter, so `held` never // dips below capacity while anyone is queued and no later caller can jump it. - else await new Promise((resolve, reject) => this.waiting.push({ resolve, reject })); + else { + await new Promise((resolve, reject) => this.waiting.push({ resolve, reject })); + // Woken with the slot, but the body may have failed in between (see release). + const closed = this.closedReason(); + if (closed !== undefined) { + this.release(); + throw closed.reason; + } + } try { return await work(); } finally { - const next = this.waiting.shift(); - if (next !== undefined) next.resolve(); else this.held--; + this.release(); + } + } + + /** Read through a call so a check after an `await` is not narrowed away. */ + private closedReason(): { readonly reason: unknown } | undefined { + return this.closed; + } + + /** + * Hand the slot to the next waiter on a later macrotask, not synchronously. + * When the work that held it rejected and that rejection fails the body, the + * failure reaches `close` through microtasks only (operation → Promise.all → + * body → executor). Waking the waiter synchronously let it admit a child run + * before `close` could refuse it; deferring lets the teardown land first. + */ + private release(): void { + const next = this.waiting.shift(); + if (next === undefined) { + this.held--; + return; } + setImmediate(() => { + if (this.closed === undefined) { + next.resolve(); + return; + } + // Closed meanwhile. `close` could not see this waiter (already taken off + // the queue), so refuse it here, and pass the slot on instead of leaking it. + next.reject(this.closed.reason); + this.release(); + }); } /** diff --git a/packages/sdk/tests/authored-parallel-agents.test.ts b/packages/sdk/tests/authored-parallel-agents.test.ts index 692b9cc68..c6b10a2f4 100644 --- a/packages/sdk/tests/authored-parallel-agents.test.ts +++ b/packages/sdk/tests/authored-parallel-agents.test.ts @@ -15,19 +15,23 @@ afterEach(async () => { }); /** The fixture's wrapper, but each session holds for a while and journals when it ran. */ -async function slowAgents(capacity: number) { +async function slowAgents(capacity: number, failFirstSession = false) { const fixture = chainFixture(); closes.push(() => fixture.close()); const spans = join(fixture.root, 'spans.jsonl'); writeFileSync(fixture.wrapper, `#!/usr/bin/env node import { receiveWrapperRequest } from ${JSON.stringify(resolve('../../testdata/preflight/wrapper-session.mjs'))}; -import { appendFileSync } from 'node:fs'; +import { appendFileSync, existsSync, writeFileSync } from 'node:fs'; if (process.argv[2] === 'auth') process.exit(0); const request = await receiveWrapperRequest(); if (request) { const start = Date.now(); await new Promise(done => setTimeout(done, 400)); appendFileSync(${JSON.stringify(spans)}, JSON.stringify({ start, end: Date.now() }) + '\\n'); + ${failFirstSession ? `if (!existsSync(${JSON.stringify(join(fixture.root, 'failed-once'))})) { + writeFileSync(${JSON.stringify(join(fixture.root, 'failed-once'))}, ''); + process.exit(1); + }` : ''} process.stdout.write('done'); } `); @@ -146,6 +150,17 @@ describe('authored steps under local workers with capacity', () => { expect(readSpans()).toHaveLength(1); }); + // #554 (Cursor): the holder's own failure fails the body; the next queued + // agent must not be handed the slot before teardown refuses it. + it('never starts a queued agent when the agent holding the only slot fails', async () => { + const { fixture, client, agent, readSpans } = await slowAgents(1, true); + await expect(executeAuthoredFlow(threeReviewers, client, undefined, { + flowPath: fixture.flowPath, localAgentStream: agent.stream, workerCapacity: 1, + })).rejects.toBeDefined(); + await new Promise(done => setTimeout(done, 1_000)); + expect(readSpans()).toHaveLength(1); + }); + it('parks the overflow when the body is not told the capacity (the defect this closes)', async () => { const { fixture, client, agent } = await slowAgents(1); const run = executeAuthoredFlow(threeReviewers, client, undefined, { diff --git a/packages/sdk/tests/canonical-tree.test.ts b/packages/sdk/tests/canonical-tree.test.ts new file mode 100644 index 000000000..7a70b0488 --- /dev/null +++ b/packages/sdk/tests/canonical-tree.test.ts @@ -0,0 +1,25 @@ +import { mkdirSync, mkdtempSync, realpathSync, rmSync, symlinkSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join, sep } from 'node:path'; +import { afterEach, expect, it } from 'vitest'; +import { canonicalTree } from '../src/worker-cli.js'; + +const roots: string[] = []; +afterEach(() => { for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); }); + +// #554 (Cursor): the artifact lock compares canonical trees. A cwd that does +// not exist yet must still resolve through its existing ancestor's symlinks, +// or `/link/new` and `/link` compare as disjoint and are not serialized. +it('resolves a missing tail through the deepest existing ancestor', () => { + const root = realpathSync(mkdtempSync(join(tmpdir(), 'canonical-tree-'))); roots.push(root); + const tree = join(root, 'tree'); + mkdirSync(tree); + const link = join(root, 'link'); + symlinkSync(tree, link); + + expect(canonicalTree(link)).toBe(tree); + const nested = canonicalTree(join(link, 'wt', 'not-yet')); + expect(nested).toBe(join(tree, 'wt', 'not-yet')); + // Nested under the resolved ancestor, so the lock sees the overlap. + expect(nested.startsWith(canonicalTree(link) + sep)).toBe(true); +}); diff --git a/packages/sdk/tests/worker-slots.test.ts b/packages/sdk/tests/worker-slots.test.ts index a593e2415..e7c1b304c 100644 --- a/packages/sdk/tests/worker-slots.test.ts +++ b/packages/sdk/tests/worker-slots.test.ts @@ -26,6 +26,8 @@ describe('WorkerSlots', () => { expect(started).toEqual([0, 1]); gates[1]!.resolve(); await runs[1]; + // The freed slot is handed over on the next macrotask (see WorkerSlots.release). + await new Promise(done => setImmediate(done)); expect(started).toEqual([0, 1, 2]); gates[0]!.resolve(); gates[2]!.resolve(); gates[3]!.resolve(); expect(await Promise.all(runs)).toEqual([0, 1, 2, 3]); @@ -49,6 +51,22 @@ describe('WorkerSlots', () => { expect(started).toEqual(['a']); }); + // The Cursor case on #554: the holder's own rejection fails the body, and + // that failure reaches close() through microtasks only. A synchronous + // handoff let the next waiter start before close() could refuse it. + it('does not start a waiter when the holder fails and close follows through microtasks', async () => { + const slots = new WorkerSlots(1); + const started: string[] = []; + const reason = new Error('holder failed'); + const holder = slots.run(async () => { started.push('a'); throw reason; }); + const waiter = slots.run(async () => { started.push('b'); return 'b'; }); + // As the executor does: the body's failure, observed a few microtasks later, closes the slots. + void holder.catch(async (error: unknown) => { await Promise.resolve(); await Promise.resolve(); slots.close(error); }); + await expect(holder).rejects.toBe(reason); + await expect(waiter).rejects.toBe(reason); + expect(started).toEqual(['a']); + }); + it('frees the slot when the work throws', async () => { const slots = new WorkerSlots(1); await expect(slots.run(async () => { throw new Error('boom'); })).rejects.toThrow('boom'); From 18a95dbb60a22ac3d6328faa95fab5518ee49ee5 Mon Sep 17 00:00:00 2001 From: Relayflow Lead Date: Tue, 22 Sep 2026 13:42:36 -0700 Subject: [PATCH 6/6] docs(sdk): advertise --agent-capacity in usage; qualify concurrent f.llm overlap Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/SURFACE.md | 3 ++- packages/sdk/src/cli.ts | 6 +++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/docs/SURFACE.md b/docs/SURFACE.md index 7ee714298..19fda3355 100644 --- a/docs/SURFACE.md +++ b/docs/SURFACE.md @@ -379,7 +379,8 @@ per agent): `f.agent("api", { task, cwd: "/repo/.wt/api" })`. The kernel carries absolute in the kernel spec; the TypeScript surface resolves a relative `cwd` against the runner's directory, while a relative `cwd` in YAML is refused. Setting `cwd` is part of the step's spec hash; omitting it hashes exactly as -before. Concurrent `f.llm` calls always overlap. +before. Concurrent `f.llm` calls have no directory lock. They overlap up to the +configured capacity, and calls beyond it wait for a slot. The LLM step remains `type: llm` in the journal. It uses the same CLI resolution, authentication probes, and exact `flows.json` model allow-list as agent steps; diff --git a/packages/sdk/src/cli.ts b/packages/sdk/src/cli.ts index 5deb026b0..b483e1a28 100644 --- a/packages/sdk/src/cli.ts +++ b/packages/sdk/src/cli.ts @@ -112,13 +112,13 @@ const USAGE = [ 'flows run @sha256: [--bucket ] [--data-dir ] [--json]', 'flows check [--watch] [--json] ', 'flows serve-webhook --data-dir --port

[--allow [,]]', - 'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir

] [--local-agent] [--reuse-from ] ', + 'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent [--agent-capacity ]] [--reuse-from ] ', 'flows run --cloud [--json] [--wait] [--sync-code] [--no-connect] ', 'flows run --cloud [--json] [--wait] [--sync-code] [--no-connect] --input ', 'flows sync [--json] [--dry-run] [--dir ] ', - 'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent] --input ', + 'flows run [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent [--agent-capacity ]] --input ', 'flows tick start --schedule-id --interval-ms [--epoch-ms ] [--max-catch-up ] [--poll-interval-ms ] [--data-dir ] ', - 'flows resume [--allow-human-influenced] [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent] ', + 'flows resume [--allow-human-influenced] [--json] [--no-spawn] [--no-observer-link] [--data-dir ] [--local-agent [--agent-capacity ]] ', 'flows answer [--json] [--no-spawn] [--data-dir ] [--note ] [--by ] ', 'flows replay [--allow-human-influenced] [--json] [--data-dir ] [--at ]', 'flows status [--json] [--data-dir ] [--tail ] []',