testbed/src

packages/testbed/src

Purpose

Public exports for launching and supervising connected-agent testbeds.

Public surface

`AgentName`

TypeAlias

export type AgentName = string & Brand.Brand<"AgentName">;

`AgentName`

Variable

export type AgentName = string & Brand.Brand<"AgentName">

`awaitAgentReadyByPolling`

Function

export function awaitAgentReadyByPolling(
  connections: PollingConnections,
  agentId: AgentId,
  timeoutMs: number,
  pollIntervalMs: number = DEFAULT_POLL_INTERVAL_MS,
): Effect.Effect<ReadyOutcome, never, never>

`createOpenClawAdapter`

Function

export function createOpenClawAdapter(
  input: OpenClawAdapterOptions,
): OpenClawAdapter

Creates an OpenClawAdapter. Omitted paths resolve from this package’s installed, versioned openclaw and @moltzap/openclaw-channel dependencies. Callers may override either path for local development.

`launchTestbed`

Function

export function launchTestbed(
  options: TestbedLaunchOptions,
): Effect.Effect<Testbed, RuntimeLaunchFailed, never>

Launch a testbed of N agents (sequentially by default; concurrency is opt-in), tearing down all already-started agents if any one fails. Sibling: launchTestbedWithProcessSignals adds SIGINT / SIGTERM handlers so Ctrl-C during startup interrupts cleanly via TestbedStartupInterrupted.

`launchTestbedWithProcessSignals`

Function

export function launchTestbedWithProcessSignals(
  options: TestbedProcessSignalOptions,
): Effect.Effect<
  Testbed,
  RuntimeLaunchFailed | TestbedStartupInterrupted,
  never
>

Wraps launchTestbed with OS-signal handlers so user Ctrl-C during startup interrupts cleanly instead of half-launching a testbed. Fails with:

TestbedStartupInterrupted — a signal arrives during testbed startup

`LogSlice`

Interface

export interface LogSlice {
  /** stdout+stderr bytes starting from the requested offset. */
  readonly text: string;
  /** Byte offset to pass on the next call to continue reading. */
  readonly nextOffset: number;
}

`NanoclawAdapter`

Class

export class NanoclawAdapter implements Runtime {
  private state: AdapterState | null = null;

  constructor(private readonly options: NanoclawAdapterOptions) {}

  spawn(input: SpawnInput): Effect.Effect<void, SpawnFailed, never> {
    return this.launchRuntime(input).pipe(
      Effect.mapError((cause) => spawnFailed(input.agentName, cause)),
      Effect.provide(NodeContext.layer),
    );
  }

  waitUntilReady(timeoutMs: number): Effect.Effect<ReadyOutcome, never, never> {
    if (!this.state) {
      return Effect.succeed({ _tag: "Ready" as const });
    }
    const { handle, spawnInput } = this.state;
    return raceReadiness({
      serverReady: this.options.server.awaitAgentReady(
        spawnInput.agentId,
        timeoutMs,
      ),
      source: {
        pollExitCode: () => pollFiberExitCode(handle.exitFiber),
        stderr: () => getNanoclawRuntimeLogs(handle),
        timeoutMs,
      },
      teardown: () => this.teardown(),
    });
  }

  teardown(): Effect.Effect<void, never, never> {
    return this.state?.teardown ?? Effect.void;
  }

  getLogs(offset: number): LogSlice {
    if (!this.state) return { text: "", nextOffset: 0 };
    const full = getNanoclawRuntimeLogs(this.state.handle);
    const text = full.slice(offset);
    return { text, nextOffset: full.length };
  }

  getInboundMarker(): string {
    return "New messages";
  }

  private launchRuntime(input: SpawnInput) {
    const lease = { committed: false };
    return Effect.uninterruptibleMask((restore) =>
      Effect.scoped(
        Effect.gen(this, function* () {
          const handle = yield* Effect.acquireRelease(
            restore(
              acquireNanoclawRuntime(
                input,
                this.options.autoRegisterConversations ?? false,
              ),
            ),
            (acquired) =>
              lease.committed
                ? Effect.void
                : stopNanoclawRuntimeSafely(
                    acquired,
                    "failed to release uncommitted NanoClaw runtime",
                  ),
          );
          const teardown = yield* Effect.cached(
            stopNanoclawRuntimeSafely(
              handle,
              "failed to tear down NanoClaw runtime",
            ),
          );
          yield* Effect.sync(() => {
            this.state = { handle, spawnInput: input, teardown };
            lease.committed = true;
          });
        }),
      ),
    );
  }
}

Nanoclaw runtime adapter. Runs agent subprocesses inside Docker containers via the OneCLI gateway. Two-phase startup: ensure the runtime cache is installed, then launch. Inbound marker: New messages. The immutable cache key covers the pinned NanoClaw source, dependency lock, bundled channel/skill, and host ABI.

`NanoclawAdapterOptions`

Interface

export interface NanoclawAdapterOptions {
  readonly server: RuntimeServerHandle;

  /**
   * Registers conversations on first delivery so NanoClaw will process them.
   * Defaults to `false`; enable it only for disposable testbeds that should
   * accept conversations without a pre-provisioned NanoClaw registration.
   */
  readonly autoRegisterConversations?: boolean;
}

`OpenClawAdapter`

Class

export class OpenClawAdapter implements Runtime {
  private state: AdapterState | null = null;

  constructor(private readonly deps: OpenClawAdapterDeps) {}

  spawn(input: SpawnInput): Effect.Effect<void, SpawnFailed, never> {
    return startOpenClawAdapter(this.deps, input, (state) => {
      this.state = state;
    }).pipe(
      Effect.mapError((cause) => spawnFailed(input.agentName, cause)),
      Effect.provide(NodeContext.layer),
    );
  }

  waitUntilReady(timeoutMs: number): Effect.Effect<ReadyOutcome, never, never> {
    if (!this.state) {
      return Effect.succeed({ _tag: "Ready" as const });
    }
    const { process: proc, spawnInput, logBuffer } = this.state;
    return raceReadiness({
      serverReady: this.deps.server.awaitAgentReady(
        spawnInput.agentId,
        timeoutMs,
      ),
      source: {
        pollExitCode: () => pollExitCode(proc),
        stderr: () => logBuffer.value,
        timeoutMs,
      },
      teardown: () => this.teardown(),
    });
  }

  teardown(): Effect.Effect<void, never, never> {
    return Effect.uninterruptible(
      Effect.gen(this, function* () {
        const teardownState = yield* Effect.sync(() => {
          const state = this.state;
          if (!state || state.tornDown) return null;
          state.tornDown = true;
          return { process: state.process, stateDir: state.stateDir };
        });

        if (teardownState === null) return;

        const { process: proc, stateDir } = teardownState;
        yield* stopSpawnedOpenClawProcess(proc);
        yield* removeOpenClawStateDir(stateDir).pipe(
          Effect.provide(NodeContext.layer),
        );
      }),
    );
  }

  getLogs(offset: number): LogSlice {
    if (!this.state) return { text: "", nextOffset: 0 };
    const full = this.state.logBuffer.value;
    const text = full.slice(offset);
    return { text, nextOffset: full.length };
  }

  getInboundMarker(): string {
    return "inbound from agent:";
  }
}

OpenClaw runtime adapter. Spawns the OpenClaw gateway as a child process, configures it with a moltzap channel plugin, and reports readiness via the server-side WS authentication event. Readiness signal: server-side WS authentication event surfaces via deps.server.awaitAgentReady. Inbound traffic log marker: inbound from agent:. Errors flow into the testbed via SpawnFailed (boot) or RuntimeExitedBeforeReady / RuntimeReadyTimedOut (post-spawn, surfaced by processExitLoop).

`OpenClawAdapterDeps`

Interface

export interface OpenClawAdapterDeps {
  readonly server: RuntimeServerHandle;
  readonly openclawBin: string;
  readonly channelDistDir: string;
}

`OpenClawAdapterOptions`

Interface

export interface OpenClawAdapterOptions {
  readonly server: RuntimeServerHandle;
  readonly openclawBin?: string;
  readonly channelDistDir?: string;
}

`ReadyOutcome`

TypeAlias

export type ReadyOutcome =
  | { readonly _tag: "Ready" }

`Runtime`

Interface

export interface Runtime {
  spawn(input: SpawnInput): Effect.Effect<void, SpawnFailed, never>;

  /**
   * Blocks until the agent's subprocess has authenticated against the server
   * (confirmed by ConnectionManager entry) or timeout/exit.
   * On Timeout or ProcessExited, the adapter calls teardown internally
   * before returning.
   */
  waitUntilReady(timeoutMs: number): Effect.Effect<ReadyOutcome, never, never>;

  /** Idempotently stops the spawned runtime and removes its isolated state. */
  teardown(): Effect.Effect<void, never, never>;

  /** Returns stdout+stderr from the given byte offset. */
  getLogs(offset: number): LogSlice;

  /** Substring that proves inbound message delivery when matched against post-send logs. */
  getInboundMarker(): string;
}

Runtime interface contract for agent subprocess management. Five methods. spawn starts the subprocess. waitUntilReady blocks until the server’s ConnectionManager confirms authentication (or timeout/exit). teardown kills the process group and removes the working directory. getLogs returns accumulated output from a byte offset. getInboundMarker returns a substring that proves an inbound message was received by the runtime’s channel plugin.

`RuntimeExitedBeforeReady`

Class

export class RuntimeExitedBeforeReady extends Data.TaggedError(
  "RuntimeExitedBeforeReady",
)<{
  readonly agentName: string;
  readonly exitCode: number | null;
  readonly stderr: string;
  readonly message: string;
}> {}

Raised by startPendingRuntimeAgent when waitUntilReady returns ProcessExited. The process exited before reaching ready. stderr carries the full accumulated stdout+stderr at exit; exitCode is null only if the process exited via signal. Caller action: inspect stderr; check binary auth config.

`RuntimeKind`

TypeAlias

export type RuntimeKind = RuntimeSelection["kind"];

`RuntimeLaunchFailed`

TypeAlias

export type RuntimeLaunchFailed =
  | SpawnFailed
  | RuntimeReadyTimedOut
  | RuntimeExitedBeforeReady;

Union of every failure mode startRuntimeAgent and launchTestbed can produce. Use Effect.catchTags to branch by tag, or Effect.catchAll to handle uniformly. Note: TestbedStartupInterrupted lives in testbed.ts because it only arises in the signal-handling variant and carries the interrupting Signal.

`RuntimeReadyTimedOut`

Class

export class RuntimeReadyTimedOut extends Data.TaggedError(
  "RuntimeReadyTimedOut",
)<{
  readonly agentName: string;
  readonly timeoutMs: number;
  readonly message: string;
}> {}

Raised by startPendingRuntimeAgent when waitUntilReady returns Timeout. The process did not signal ready within timeoutMs and has been torn down before the failure reaches the caller. Caller action: increase readyTimeoutMs, or enable process-level diagnostics at the adapter boundary.

`RuntimeServerHandle`

Interface

export interface RuntimeServerHandle {
  /**
   * Resolves to `Ready` when the named agent has authenticated against the
   * server. Resolves to `Timeout` after `timeoutMs` if no authenticated
   * connection ever appears. Resolves to `ProcessExited` only if the
   * implementation can detect that the agent's owning process exited before
   * authenticating; otherwise `Timeout` covers that case (the runtime
   * adapters layer their own exit-detection on top via `Effect.race`).
   *
   * In-process implementations wire this through `awaitAgentReadyByPolling`.
   * Out-of-process implementations (e.g., a zapbot orchestrator talking to
   * a standalone moltzap-server) implement it directly, typically via a
   * presence-event subscription on the server's WebSocket API.
   */
  awaitAgentReady(
    agentId: AgentId,
    timeoutMs: number,
  ): Effect.Effect<ReadyOutcome, never, never>;
}

`RuntimeStartOptions`

TypeAlias

export type RuntimeStartOptions = RuntimeStartOptionsBase & RuntimeSelection;

interface TestbedLaunchOptionsBase {
  readonly server: RuntimeServerHandle;
  readonly agents: ReadonlyArray<TestbedAgentSpec>;
  readonly readyTimeoutMs: number;
  readonly concurrency?: number | "unbounded";
}

`ServerUrl`

TypeAlias

export type ServerUrl = string & Brand.Brand<"ServerUrl">;

`ServerUrl`

Variable

export type ServerUrl = string & Brand.Brand<"ServerUrl">

`SpawnFailed`

Class

export class SpawnFailed extends Data.TaggedError("SpawnFailed")<{
  readonly agentName: string;
  readonly message: string;
  readonly cause: Error;
}> {}

Raised by Runtime.spawn() in any adapter when the child process cannot be started — exec error, missing binary, port allocation failure, state-dir creation failure. cause carries the underlying Error. Caller action: surface to user. No retry — binary or config is wrong.

`SpawnInput`

Interface

export interface SpawnInput {
  readonly agentName: AgentName;
  readonly apiKey: AgentKey;
  readonly agentId: AgentId;
  readonly serverUrl: ServerUrl;
  readonly workspaceFiles?: ReadonlyArray<WorkspaceFile>;
  readonly modelId?: string;
}

`startRuntimeAgent`

Function

export function startRuntimeAgent(
  options: RuntimeStartOptions,
): Effect.Effect<Runtime, RuntimeLaunchFailed, never>

Spawn one runtime agent, wait for ready, release the startup cleanup scope and hand a long-lived Runtime back to the caller. Error channel is the union RuntimeLaunchFailed of the three shapes above. Sibling: launchTestbed for multi-agent coordinated startup. Fails with:

SpawnFailed — the child process cannot be started (exec error, bad binary, port allocation failure, state-dir error)
RuntimeReadyTimedOut — waitUntilReady exceeds readyTimeoutMs
RuntimeExitedBeforeReady — the process exits before signaling ready (inspect stderr)

`Testbed`

Interface

export interface Testbed {
  readonly agents: ReadonlyArray<TestbedAgent>;
  stopAll(): Effect.Effect<void, never, never>;
  getLogs(name: string): string;
}

`TestbedAgent`

Interface

export interface TestbedAgent {
  readonly name: string;
  readonly agentId: AgentId;
}

`TestbedAgentSpec`

Interface

export interface TestbedAgentSpec {
  readonly agentName: string;
  readonly apiKey: AgentKey;
  readonly agentId: AgentId;
  readonly serverUrl: string;
  readonly workspaceFiles?: ReadonlyArray<WorkspaceFile>;
  readonly modelId?: string;
}

`TestbedLaunchOptions`

TypeAlias

export type TestbedLaunchOptions = TestbedLaunchOptionsBase & RuntimeSelection;

export type TestbedProcessSignalOptions = TestbedLaunchOptions & {
  readonly signals?: ReadonlyArray<Signal>;
};

`TestbedProcessSignalOptions`

TypeAlias

export type TestbedProcessSignalOptions = TestbedLaunchOptions & {
  readonly signals?: ReadonlyArray<Signal>;
};

`TestbedStartupInterrupted`

Class

export class TestbedStartupInterrupted extends Data.TaggedError(
  "TestbedStartupInterrupted",
)<{
  readonly signal: Signal;
  readonly message: string;
}> {}

`WorkspaceFile`

Interface

export interface WorkspaceFile {
  readonly relativePath: string;
  readonly content: string;
}

Files

await-agent-ready.ts
errors.ts
nanoclaw-adapter.ts
openclaw-adapter.ts
runtime.ts
testbed.ts

​testbed/src

​Purpose

​Public surface

​Files

testbed/src

Purpose

Public surface

Files