fix(sdk,kernel): run authored agents in parallel under --local-agent

docs/SURFACE.md

@@ -361,6 +361,27 @@ f.llm(strings: TemplateStringsArray, ...values: unknown[]): Step<string>;
 
 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 <n>`, 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. 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 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;
 a declared `model` must be in that project's `models` array. A template call

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<AgentTransport>,
+        /// 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<String>,
         #[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:?}")]

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"
+    );
+}

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<Input = undefined>(
 
   const worker = authoredWorkerRunner(
     definition, journal, flowPath, journalSteps, waitOptions,
-    localAgentStream, budget, definition.header.budget, options.rootRunId,
+    localAgentStream, budget, definition.header.budget, options.rootRunId, options.workerCapacity,
   );
 
   /**
@@ -600,6 +606,7 @@ export async function executeAuthoredFlow<Input = undefined>(
   }
   if (bodyFailed) {
     try {
+      worker.stop(bodyFailure);
       await stopAuthoredOperations(authoredSteps, bodyFailure);
     } finally {
       lifecycle.close();
@@ -626,6 +633,7 @@ export async function executeAuthoredFlow<Input = undefined>(
       `flow "${definition.name}" returned without done()`,
     );
     try {
+      worker.stop(missingCompletion);
       await stopAuthoredOperations(authoredSteps, missingCompletion);
     } finally {
       lifecycle.close();

packages/sdk/src/authored-flow-lifecycle.ts

@@ -124,6 +124,9 @@ export class AuthoredFlowLifecycle {
   private readonly callbackFailures = new Map<OperationToken, unknown>();
   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<number, ReadonlySet<number>> } | 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<T extends OperationToken>(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<void> {
-    const observations: Promise<unknown>[] = [];
+    const operationsByRoot = new Map<number, OperationToken[]>();
     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<Promise<unknown>, Set<OperationToken>>();
+    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<unknown>[] = [];
+    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<number> {
     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<number, ReadonlySet<number>> {
+    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<number, Set<number>>();
+    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;
   }
 }
 

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 }),
     });

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);