server-core/test-utils

packages/server/src/test-utils

Purpose

Shared server-core test utility exports.

Public surface

`AppEndpointHandlers`

TypeAlias

export type AppEndpointHandlers = {
  readonly [D in AnyAppCallbackRpcDefinition as D["name"]]: AppEndpointHandler<D>;
};

Mapped over the closed AnyAppCallbackRpcDefinition union, keyed by each definition’s wire name. Mandates one handler per task-callback RPC at construction time — adding a new entry to appCallbackMethods becomes a compile error at every endpoint construction site.

`AwaitNotificationError`

TypeAlias

export type AwaitNotificationError =
  | AwaitNotificationTimeoutError
  | AwaitNotificationClosedError;

/**
 * Stream-based one-shot waiter. Consumes `client.subscribe(def)` via
 * `Stream.runHead`, failing with `AwaitNotificationTimeoutError` on timeout
 * and `AwaitNotificationClosedError` when the transport closed before a
 * matching frame arrived. Distinguishing close from timeout keeps a dead
 * connection from masquerading as a missing notification.
 */
export function awaitOneNotification<D extends AnyNotificationDefinition>(
  client: Pick<TestAgentClient, "subscribe">,
  definition: D,
  timeoutMs: number = DEFAULT_AWAIT_NOTIFICATION_TIMEOUT_MS,
): Effect.Effect<NotificationDelivery<D>, AwaitNotificationError> {
  const closed = () =>
    new AwaitNotificationClosedError({
      definition: definition.name,
    });
  return client.subscribe(definition).pipe(
    Stream.map(
      (params): NotificationDelivery<D> => ({
        definition,
        method: definition.name,
        params,
      }),
    ),
    Stream.runHead,
    Effect.either,
    Effect.flatMap(
      Either.match({
        onLeft: () => Effect.fail(closed()),
        onRight: Option.match({
          onNone: () => Effect.fail(closed()),
          onSome: (notification) => Effect.succeed(notification),
        }),
      }),
    ),
    Effect.timeoutFail({
      duration: Duration.millis(timeoutMs),
      onTimeout: () =>
        new AwaitNotificationTimeoutError({
          definition: definition.name,
          durationMs: timeoutMs,
        }),
    }),
  );
}

`awaitOneNotification`

Function

export function awaitOneNotification<D extends AnyNotificationDefinition>(
  client: Pick<TestAgentClient, "subscribe">,
  definition: D,
  timeoutMs: number = DEFAULT_AWAIT_NOTIFICATION_TIMEOUT_MS,
): Effect.Effect<NotificationDelivery<D>, AwaitNotificationError>

Stream-based one-shot waiter. Consumes client.subscribe(def) via Stream.runHead, failing with AwaitNotificationTimeoutError on timeout and AwaitNotificationClosedError when the transport closed before a matching frame arrived. Distinguishing close from timeout keeps a dead connection from masquerading as a missing notification.

`closeAllClients`

Function

export function closeAllClients(): Effect.Effect<void, never>

`connectAppClient`

Function

export function connectAppClient(
  appId: AppId,
  appKey: AppKey,
  handlers: AppCallbackHandlers<AppCallbackContext>,
): Effect.Effect<TestAppClient, Error>

`ConnectedAgent`

Interface

export interface ConnectedAgent {
  client: TestAgentClient;
  agentId: AgentId;
  apiKey: AgentKey;
  name: string;
}

`connectTestClient`

Function

export function connectTestClient(opts: {
  agentId: AgentId;
  apiKey: AgentKey;
  wsUrl?: string;
}): Effect.Effect<TestAgentClient, Error>

`CoreSchemaSqlLoadError`

TypeAlias

export type CoreSchemaSqlLoadError =
  | CoreSchemaSqlAccessError
  | CoreSchemaSqlReadError;

const __dirname = dirname(fileURLToPath(import.meta.url));

`CoreTestDatabasePort`

Interface

export interface CoreTestDatabasePort {
  execute(sql: string): PromiseLike<unknown>;
  reset(): PromiseLike<void>;
}

Database operations available to consumers of the published test harness.

`CoreTestReadyOutcome`

TypeAlias

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

`CoreTestRuntimeServerHandle`

Interface

export interface CoreTestRuntimeServerHandle {
  awaitAgentReady(
    agentId: AgentId,
    timeoutMs: number,
  ): Effect.Effect<CoreTestReadyOutcome, never, never>;
}

Process capabilities needed by in-process runtime tests.

`CoreTestServer`

TypeAlias

export type CoreTestServer = CoreTestServerPort;

/**
 * Start a test server and expose its package-owned integration ports.
 * @param opts Test server configuration.
 * @returns A promise for the running server's integration ports.
 */
export function startCoreTestServer(opts: StartCoreTestServerOptions = {}) {
  return Effect.runPromise(
    startCoreTestServerEffect(opts).pipe(
      Effect.map((server) => server.testPort),
    ),
  );
}

Canonical published handle for a running core test server.

`CoreTestServerHandle`

Interface

export interface CoreTestServerHandle {
  baseUrl: string;
  wsUrl: string;
  db: EffectKysely<Database>;
  coreApp: CoreApp;

  /**
   * Pre-wired `RuntimeServerHandle` for runtime-adapter tests. Implements
   * `awaitAgentReady` by polling the live `ConnectionManager` — the same
   * pattern `@moltzap/testbed`'s `awaitAgentReadyByPolling` exports for
   * downstream in-process consumers. Out-of-process consumers (zapbot's
   * orchestrator) construct their own handle over WebSocket presence.
   */
  runtimeServer: CoreTestRuntimeServerHandle;

  /**
   * The auto-wired `InMemorySpanExporter`, or `null` when the caller
   * supplied a custom `spanProcessor`. Tests that want to inspect OTel
   * spans call `getFinishedSpans()` on this exporter and map them via
   * their own package-specific projection.
   */
  readonly spanExporter: InMemorySpanExporter | null;

  /** Published projection that keeps persistence and tracing vendors private. */
  readonly testPort: CoreTestServerPort;
}

`CoreTestServerPort`

Interface

export interface CoreTestServerPort {
  readonly baseUrl: string;
  readonly wsUrl: string;
  readonly db: CoreTestDatabasePort;
  readonly runtimeServer: CoreTestRuntimeServerHandle;
  readonly spanExporter: CoreTestSpanExporterPort | null;
}

Published server handle composed only from server-owned test ports.

`CoreTestSpan`

Interface

export interface CoreTestSpan {
  readonly name: string;
  readonly attributes: Readonly<Record<string, unknown>>;
}

Stable projection of a finished server trace span.

`CoreTestSpanExporterPort`

Interface

export interface CoreTestSpanExporterPort {
  getFinishedSpans(): ReadonlyArray<CoreTestSpan>;
  reset(): void;
}

Trace-capture operations available to test-harness consumers.

`createTestAgent`

Function

export function createTestAgent(
  name: string,
  opts?: CreateTestAgentOptions,
): Effect.Effect<TestAgent, never>

`DEFAULT_TEST_ADMIN_USER_ID`

Variable

export const DEFAULT_TEST_ADMIN_USER_ID: UserIdValue = Schema.decodeUnknownSync(
  UserId,
)("00000000-0000-4000-8000-00000000ad00")

`getBaseUrl`

Function

export function getBaseUrl(): string

`getCoreDb`

Function

export function getCoreDb(): EffectKysely<Database>

`getCoreEncryptionEnvelope`

Function

export function getCoreEncryptionEnvelope(): EnvelopeEncryption

`getWsUrl`

Function

export function getWsUrl(): string

`loadCoreSchemaSql`

Function

export function loadCoreSchemaSql(): Effect.Effect<
  string,
  CoreSchemaSqlLoadError
>

`makeFakeService`

Function

export const makeFakeService = <S extends object>(impl: Partial<S>): S

Build a typed test double for an interface S from a partial implementation. The cast is intentional: tests typically implement only the methods the system under test actually calls. Unused methods throw at runtime via the Proxy trap so a missing implementation becomes a clear test failure instead of undefined is not a function. Because the generic parameter S is invariant, TypeScript still enforces that every method you do implement matches the real signature — this is the compile-time contract-drift insurance. Adding a field to the real interface does NOT fail compilation (tests are a Partial), but changing an existing field’s signature does.

`makeHandlerAppEndpoint`

Function

export function makeHandlerAppEndpoint(args: {
  readonly id: ConnectionId;
  readonly handlers: AppEndpointHandlers;
}): AppEndpoint

Build an AppEndpoint whose outbound originator.call dispatches to in-process handlers instead of going over a WebSocket. The endpoint satisfies the same { connId, originator } shape a connected app’s arm carries so AppEndpointRegistry, AppRegistry, and sendRpcToClient see ONE shape.

originator.callback({ definition, params }) indexes handlers by definition.name. The mapped type guarantees every member of AnyAppCallbackRpcDefinition has a handler — no runtime “method not found” branch exists.
originator.notify / failAllPending are no-ops.
originator.handle / originator.resolve defect — an in-process endpoint never receives inbound frames; a call here is a wiring bug.

`makePgliteHarness`

Function

export function makePgliteHarness(): Effect.Effect<
  PgliteHarness,
  PgliteHarnessError
>

Spin up a fresh PGlite instance with the core schema loaded.

`PGLITE_HOOK_TIMEOUT_MS`

Variable

export const PGLITE_HOOK_TIMEOUT_MS = 30_000

Suggested timeout for pglite-backed beforeEach/afterEach hooks.

`PgliteHarness`

Interface

export interface PgliteHarness {
  /** Effect-Kysely-wrapped client. Yieldable as Effect via the toolkit. */
  readonly db: EffectKysely<Database>;

  /**
   * Run raw SQL. The harness uses this to load the schema; tests can use it
   * to seed extra rows after `make()` returns.
   */
  readonly exec: (sql: string) => Effect.Effect<unknown, PgliteExecError>;

  /** Tear down the in-memory instance. Call from `afterEach`. */
  readonly close: Effect.Effect<void, PgliteCloseError>;
}

`PgliteHarnessError`

TypeAlias

export type PgliteHarnessError =
  | CoreSchemaSqlLoadError
  | PgliteCreateError
  | PgliteExecError
  | PgliteCloseError;

const SQL_PREVIEW_MAX_CHARS = 160;

function sqlPreview(sql: string): string {
  return sql.replace(/\s+/g, " ").trim().slice(0, SQL_PREVIEW_MAX_CHARS);
}

`postJson`

Function

export function postJson(
  baseUrl: string,
  path: string,
  body: Record<string, unknown>,
): Effect.Effect<PostJsonResult, PostJsonError>

POST body as JSON to ${baseUrl}${path} and resolve with {status, json}. HTTP integration tests import this helper to avoid repeated request/JSON boilerplate.

`registerAgent`

Function

export function registerAgent(
  baseUrl: string,
  name: string,
  opts?: { description?: string; inviteCode?: string },
): Effect.Effect<TestAgent, Error>

`registerAndConnect`

Function

export function registerAndConnect(
  name: string,
): Effect.Effect<ConnectedAgent, Error>

`registerApp`

Function

export function registerApp(
  baseUrl: string,
  manifest: AppManifest,
  inviteCode?: string,
): Effect.Effect<
  { readonly appId: AppId; readonly appKey: AppKey },
  AppRegistrationError
>

`resetCoreTestDb`

Function

export function resetCoreTestDb()

`setupAgentGroup`

Function

export function setupAgentGroup(
  count: number,
  opts?: { groupName?: string },
): Effect.Effect<
  {
    agents: ConnectedAgent[];
    conversationId?: ConversationId;
    taskId?: TaskId;
  },
  Error
>

Create N agents, all connected. Optionally create a group conversation.

`setupAgentPair`

Function

export function setupAgentPair(): Effect.Effect<
  { alice: ConnectedAgent; bob: ConnectedAgent },
  Error
>

Create two agents, both connected. No contacts needed (core has open access).

`startCoreTestServer`

Function

export function startCoreTestServer(opts: StartCoreTestServerOptions = {})

Start a test server and expose its package-owned integration ports. Returns: A promise for the running server’s integration ports.

`startCoreTestServerEffect`

Function

export function startCoreTestServerEffect(
  opts: StartCoreTestServerOptions = {},
)

`startCoreTestServerFull`

Function

export function startCoreTestServerFull(opts: StartCoreTestServerOptions = {})

`StartCoreTestServerOptions`

Interface

export interface StartCoreTestServerOptions {
  readonly pgHost?: string;
  readonly pgPort?: number;
  readonly encryption?: boolean;
  readonly registrationSecret?: string;
  readonly adminUserId?: UserId;
}

`stopCoreTestServer`

Function

export function stopCoreTestServer()

`trackClient`

Function

export function trackClient(client: TestAgentClient | TestAppClient): void

Files

app-endpoint.ts
core-schema-sql.ts
fakes.ts
helpers.ts
index.ts
pglite-harness.ts
ports.ts
server.ts

​server-core/test-utils

​Purpose

​Public surface

​Files

server-core/test-utils

Purpose

Public surface

Files