Skip to main content
← Back to list
01Issue
FeatureShippedSwamp CLI
Assigneesstack72

Relationships

#981 Worker fleets phase 4a: dispatch runner extraction — child process per dispatch

Opened by stack72 · 7/5/2026· Shipped 7/6/2026

Summary

Phase 4a of worker fleets. Today, dispatches execute in-process inside the worker, and the environment snapshot is applied by mutating process-global `Deno.env`. This makes concurrent dispatches unsafe — the serial constraint (`busy` flag) exists because of this global mutation.

After this PR, each dispatch runs in a dispatch runner — a child process of the same swamp binary. The environment snapshot is applied as the child's spawn environment (no global mutation). The supervisor (the worker process) bridges capability RPCs between the child and the orchestrator. A runner crash fails only that dispatch, not the worker.

This PR ships at capacity 1 — one runner at a time, same as today. Phase 4b adds `--concurrency N` slots on top. The split de-risks the architectural change: if 4a ships but 4b slips, every worker still gets crash isolation and clean env handling.

Design reference: `worker-fleets.md` section "Phase 4 — Capacity: slots and dispatch runners" → subsection (a).

Changes

1. Stdio RPC transport

New file: `src/domain/remote/stdio_transport.ts`

A new `RpcTransport` implementation that sends/receives JSON frames over stdin/stdout:

```typescript export class StdioTransport implements RpcTransport { send(data: string): void { // Write a length-prefixed frame to stdout // Format: 4-byte big-endian length + UTF-8 JSON payload // This avoids newline-based framing which breaks if JSON contains newlines } }

export function createStdioReader( stdin: ReadableStream, onFrame: (data: string) => void, onClose: () => void, ): void { // Read length-prefixed frames from stdin } ```

The framing must be length-prefixed, not newline-delimited — JSON payloads can contain newlines (e.g., in method output strings). The `RpcChannel` is already transport-abstracted, so the stdio transport plugs in identically to the WebSocket transport.

Tests: `stdio_transport_test.ts` — frame round-trip, cancel propagation, close-aborts-pending, binary safety (embedded newlines in payload).

2. Dispatch runner entry point

New file: `src/worker/exec_dispatch.ts`

A hidden CLI entry point (`swamp worker exec-dispatch`) that runs exactly one dispatch:

  1. Bootstrap: Read `RunnerBootstrapParams` from stdin (first frame): session credential, data-plane URL, cache dir, and the `DispatchParams` to execute
  2. Set up context: Create a `DataPlaneClient` with the session credential for direct HTTP requests to the orchestrator. Create a `WorkerBundleCache` from the cache dir. The process environment IS the merged environment (supervisor set it at spawn).
  3. Set up RPC: Create an `RpcChannel` over the `StdioTransport`. Register capability verb handlers that forward to the orchestrator (via the supervisor's bridge — the child doesn't know about this indirection).
  4. Execute: Load the model (from bundle cache or `builtin:` registry), run the method with the remote `MethodContext`, stream run events back over the RPC channel.
  5. Exit: Return the `DispatchResult` as the final RPC response and exit.

Critical: The runner must redirect LogTape and console output to stderr before loading any model code. stdout carries RPC frames only. Method-spawned subprocesses must use piped stdio capture (`stdout: "piped"`), never `stdout: "inherit"`, or they corrupt the frame stream.

Register as: A hidden subcommand in the worker command group (not shown in help, not documented).

3. Supervisor: spawn runner per dispatch

File: `src/worker/dispatch_handler.ts`

Replace the in-process execution with child-process spawning:

```typescript // Instead of: const restoreEnv = applyEnvironmentOverlay(params.environmentSnapshot); try { return await handleDispatch(rawParams, ctx, options); } finally { restoreEnv(); }

// Now: const child = new Deno.Command(Deno.execPath(), { args: ["worker", "exec-dispatch"], stdin: "piped", stdout: "piped", stderr: "piped", env: mergeEnvironment(workerBaseEnv, params.environmentSnapshot, denylist), }); ```

The supervisor:

  • Spawns the child with the merged environment as spawn env
  • Sends `RunnerBootstrapParams` as the first frame over the child's stdin
  • Creates an `RpcChannel` over the stdio transport to the child
  • Bridges capability RPCs: calls arriving from the child (e.g., `capability.queryData`) are forwarded to the orchestrator over the enrolled control socket; responses are routed back to the child by call ID
  • Forwards run events: stream events from the child are forwarded to the orchestrator
  • Forwards cancel: `rpc.cancel` from the orchestrator is forwarded to the child; if the child doesn't respond within `CANCEL_GRACE_MS`, kill it
  • Captures stderr: child's stderr goes to worker logs (not to the orchestrator)
  • Handles crash: if the child exits non-zero or the stdio channel closes unexpectedly, report a dispatch `error` result — the worker stays enrolled

Delete `applyEnvironmentOverlay`: The global env mutation is no longer needed. Remove lines 64-84 of dispatch_handler.ts and all call sites (lines 223, 228). The environment is now applied at spawn time.

4. Environment merging at spawn

New file or extend: `src/domain/remote/environment_snapshot.ts`

Add a `mergeEnvironment` function:

```typescript export function mergeEnvironment( base: Record<string, string>, snapshot: Record<string, string>, denylist: Set, ): Record<string, string> { const merged = { ...base }; for (const [key, value] of Object.entries(snapshot)) { if (!isDenied(key, denylist)) { merged[key] = value; } } return merged; } ```

The denylist is the existing one from `environment_snapshot.ts` (HOME, PATH, USER, SWAMP_*, DENO_*, etc.). The base is the worker's own process environment (`Deno.env.toObject()`). The result is passed as `env` to `Deno.Command`.

5. Capability bridge in supervisor

File: `src/worker/dispatch_handler.ts` (or a new `src/worker/runner_bridge.ts`)

The bridge multiplexes child → orchestrator RPCs over the single enrolled control socket:

  1. Child sends `rpc.request` with method `capability.queryData` and a child-scoped call ID
  2. Supervisor receives it via the stdio RPC channel
  3. Supervisor forwards it to the orchestrator via the enrolled WebSocket RPC channel, using a new globally-unique call ID
  4. Orchestrator responds
  5. Supervisor routes the response back to the child using the original child-scoped call ID

The bridge must map child call IDs to orchestrator call IDs and route responses correctly. Since capacity 1 means one runner at a time, there's no cross-runner ID collision in this PR — but the mapping should be designed to handle multiple runners (phase 4b).

The capability verbs that need bridging (all 14 from the protocol):

  • `getData`, `queryData`, `listVersions`, `persistResource`, `persistFile`, `appendData`, `deleteData`, `resolveSecret`, `putSecret`, `readDefinition`, `readOutput`, `resolveModel`, `getExtensionFile`, `log/event`

The child makes data-plane HTTP requests (persistResource, persistFile, appendData, getData for bytes) directly to the orchestrator using the session credential — these don't go through the bridge. Only the metadata-RPC verbs (queryData, resolveSecret, etc.) go through the bridge.

6. Data-plane direct access from runner

The child process receives the session credential and data-plane URL in its bootstrap params. It creates its own `DataPlaneClient` and makes HTTP requests directly to the orchestrator. The `recordFirstWrite` ordering is preserved because the data-plane on the orchestrator side still awaits `mark_writes` before persisting — the request comes from the child's IP instead of the supervisor's, but the session credential ties it to the same dispatch.

7. Retire in-process execution path

The serial in-process execution path (`applyEnvironmentOverlay` → run method in worker process → `restoreEnvironment`) is deleted. Capacity 1 is one runner at a time — every dispatch has identical isolation semantics regardless of capacity. The `busy` flag and `draining` flag in the dispatch handler now gate runner spawning, not in-process execution.

Invariants to preserve

  1. `recordFirstWrite` ordering. The data plane awaits `mark_writes` before the first durable write. The child's direct HTTP requests use the same session credential bound to the same dispatch lease.
  2. stdout is frames only. Any non-frame output to the runner's stdout corrupts the RPC channel. LogTape, console, and subprocess output must go to stderr.
  3. Workers stay repo-less. The runner child is also repo-less — it gets its context from bootstrap params and the RPC bridge, not from a local repo.
  4. Capacity 1 semantics. With one runner at a time, the behavior must be byte-for-byte identical to today's serial execution (same outputs, same errors, same lease transitions). The only difference: crash isolation (a method that crashes takes out the runner, not the worker).
  5. Cancel propagation. `rpc.cancel` from the orchestrator must reach the runner. If the runner ignores it, the supervisor kills it after `CANCEL_GRACE_MS`.
  6. Environment denylist pinned to protocol version. The denylist in `environment_snapshot.ts` applies at spawn time, same vars as before.

Tests required

Stdio transport tests (`stdio_transport_test.ts`):

  • Frame round-trip (send → receive)
  • Length-prefixed framing handles embedded newlines
  • Cancel propagation
  • Close aborts pending calls

Runner bootstrap tests:

  • Bootstrap params deserialize correctly
  • LogTape redirected to stderr before model code loads
  • Method execution produces correct output

Supervisor/bridge tests:

  • Capability RPC forwarded child → orchestrator → child
  • Run events forwarded child → orchestrator
  • Cancel forwarded orchestrator → child
  • Runner crash → dispatch error, worker stays enrolled
  • Environment isolation: child sees snapshot env, worker process env unchanged

Integration test:

  • Same dispatch result as in-process execution (byte-for-byte output comparison)
  • Crash isolation: method that throws → dispatch fails, subsequent dispatch succeeds

Acceptance criteria

  1. A dispatch produces identical results to the pre-runner in-process path
  2. `Deno.env` is never mutated in the worker process during dispatch
  3. A runner crash fails only that dispatch — the worker stays enrolled and accepts the next dispatch
  4. All existing remote-execution integration tests pass
  5. `deno check`, `deno lint`, `deno fmt`, `deno run test`, `deno run compile` all pass

Repo conventions

  • New `.ts` files need the AGPLv3 header
  • TypeScript strict, no `any`, named exports only
  • Unit tests next to source
  • Use the `github-pr` skill for commits/PRs
  • No protocol version bump needed (the wire format between orchestrator and worker is unchanged — the runner is worker-internal)
02Bog Flow
✓OPEN✓TRIAGED✓IN PROGRESS✓SHIPPED+ 1 MOREASSIGNED+ 5 MOREREVIEW+ 4 MOREPR_MERGED+ 1 MORENOTIFICATION_SKIPPED

Shipped

7/6/2026, 2:01:00 AM

Click a lifecycle step above to view its details.

03Sludge Pulse
stack72 assigned stack727/5/2026, 9:37:09 PM

Sign in to post a ripple.