Your first surface.

We are going to build fleet-top — a live top for your own machine. By the end of this page it will run in your browser: load averages and memory in the header, a table of processes that updates on its own, and a button that kills a process. No polling code, no reconnect handling, no hand-written RPC — you declare the shape of the live state once and read it.

We will get there in two hops. First we consume the surface in the same process, so you see it work with no network at all. Then we change one line and the same code is talking to a WebSocket. That is the whole point of a surface, and you will feel it by the end.

Set up

Declare the surface

A surface is a typed, reactive slice of live state, declared once with defineSurface. Ours has four members: two cells (load and memory — single values), one collection (the processes, keyed by pid), and one procedure (kill — the single mutation).

/**
 * fleet-top part 1 — the surface, declared once.
 *
 * A live `top` for one machine. Four primitives carry the whole tool:
 *
 *   - `load`      — Cell: the 1/5/15-minute load averages (a singleton).
 *   - `memory`    — Cell: bytes used / total (a singleton).
 *   - `processes` — Collection keyed by pid: one row per process.
 *   - `process.kill` — Procedure: the one mutation (send a signal to a pid).
 *
 * `defineSurface` turns this spec into `surface.group` (the flat Effect RPC
 * group the server binds handlers on and every link is built over — one member
 * per wire tag, `surface/<member>/<verb>`) plus `surface.descriptors` /
 * `surface.spec` for reflection. Nothing else in the app re-declares these
 * shapes — the inferred domain types at the bottom are the single source of
 * truth (`SurfaceTypes` lifts them straight out of the spec).
 *
 * The schemas are Effect Schemas. Two spellings are laws rather than taste:
 * an optional wire key is `Schema.optionalKey` (never `Schema.optional`, which
 * round-trips an explicit `undefined` through `null`), and a defaulted wire key
 * is `Schema.withDecodingDefaultKey` — as `kill`'s `signal` is below, so a
 * caller may omit it while the handler always receives a real signal name.
 */

import { defineSurface, type SurfaceTypes } from "@kolu/surface/define";
import { Effect, Schema } from "effect";

// ── Named schemas — referenced from more than one position ──────────────

const PidSchema = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0));

const ProcessSchema = Schema.Struct({
  /** Owning user (name on darwin/ps, uid-derived on linux). */
  user: Schema.String,
  /** Percent of one core during the last poll window. */
  cpuPct: Schema.Number,
  /** Resident memory as a percent of total. */
  memPct: Schema.Number,
  /** The command line (truncated). */
  command: Schema.String,
});

const LoadSchema = Schema.Struct({
  /** 1-minute, 5-minute, 15-minute load averages. */
  avg: Schema.Tuple([Schema.Number, Schema.Number, Schema.Number]),
  /** Logical CPU count — the "100% == this many busy cores" denominator. */
  cores: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)),
});

const MemorySchema = Schema.Struct({
  /** Bytes in use (total − available). */
  used: Schema.Number,
  /** Total physical memory in bytes. */
  total: Schema.Number,
});

export const DEFAULT_LOAD: typeof LoadSchema.Type = {
  avg: [0, 0, 0],
  cores: 0,
};
export const DEFAULT_MEMORY: typeof MemorySchema.Type = {
  used: 0,
  total: 0,
};

/** `kill`'s argument. `signal` absent on the wire ⇒ decoded as `"TERM"`.
 *  `withDecodingDefaultKey`, never `withDecodingDefault`: the key may be
 *  MISSING, never an explicit `undefined`. */
const KillInputSchema = Schema.Struct({
  pid: PidSchema,
  signal: Schema.Literals(["TERM", "KILL", "HUP", "INT"]).pipe(
    Schema.withDecodingDefaultKey(Effect.succeed("TERM" as const)),
  ),
});
const KillOutputSchema = Schema.Struct({ ok: Schema.Boolean });

// ── The surface ─────────────────────────────────────────────────────────

export const surface = defineSurface({
  cells: {
    load: { schema: LoadSchema, default: DEFAULT_LOAD },
    memory: { schema: MemorySchema, default: DEFAULT_MEMORY },
  },
  collections: {
    processes: { keySchema: PidSchema, schema: ProcessSchema },
  },
  procedures: {
    process: {
      // Imperative escape hatch: killing a pid is a command, not a keyed
      // upsert, so it doesn't fit the collection's mutation verbs.
      kill: { input: KillInputSchema, output: KillOutputSchema },
    },
  },
});

// ── Inferred domain types — single source of truth ──────────────────────

type SF = SurfaceTypes<typeof surface.spec>;

export type Pid = SF["collections"]["processes"]["Key"];
export type Process = SF["collections"]["processes"]["Value"];
export type Load = SF["cells"]["load"]["Value"];
export type Memory = SF["cells"]["memory"]["Value"];

/** `process.kill`'s argument as a CALLER spells it — the ENCODED side, where
 *  `signal` is optional — and its result. `SurfaceTypes` covers the four
 *  reactive primitives; a procedure's two sides are read straight off its own
 *  schemas. */
export type KillArgs = typeof KillInputSchema.Encoded;
export type KillResult = typeof KillOutputSchema.Type;

That one declaration is the contract. From it the framework derives the wire shape — a flat set of string-tagged members like surface/load/get — the server handlers, and the client hooks; you never hand-write any of them. (For why there are four member kinds and not one, see Why surfaces; you do not need it to finish here.)

Fill in the data

The surface says what the state is; ordinary code says where it comes from. src/server/top.ts starts a poll loop that reads the OS every second and writes the cells and the collection; src/server/proc.ts is the platform-specific bit that reads /proc on Linux or ps on macOS. Nothing framework-shaped is happening in there — it is just a timer that calls set and upsert. Open them if you are curious; you do not have to change a line.

See it work — in one process

Before any network, the surface is consumable in the same process. directDispatch takes the served surface itself and calls its handlers directly; buildSurfaceFace re-nests the flat wire tags into face.surface.load.get, so what you hold is the exact face a WebSocket consumer holds — with zero serialization in either direction.

/**
 * The first link: `directDispatch` — in-process, no wire.
 *
 * The wire links (`websocketLink`, `stdioLink`, `unixSocketLink`) SEPARATE the
 * serve side from the consume side. `directDispatch` FUSES them: it takes the
 * served handler record itself and calls handlers directly, so the face you hold
 * here is the exact face a WebSocket or ssh consumer holds — zero serialization,
 * in either direction, and `live` is constant-`true` honestly (there is no
 * transport that could half-open).
 *
 * `buildSurfaceFace` is the addressing layer: it re-nests the flat wire tags
 * (`surface/load/get`) into `face.surface.load.get`. Streaming verbs hand back a
 * lazy Effect `Stream`; unary verbs hand back a `Promise`. The face is
 * deliberately STRUCTURAL — per-member types live in the spec-derived bound
 * hooks `surfaceClient` builds (see `client/wire.ts`) — so a non-reactive reader
 * like this one names the member shape it is calling, once.
 *
 * Run with `pnpm run inproc`. This is the honest "hello world" of the stack:
 * define → implement → consume, with the transport collapsed to nothing.
 */

import {
  buildSurfaceFace,
  type StreamingProcedure,
} from "@kolu/surface/client";
import { directDispatch } from "@kolu/surface/links/direct";
import { Effect, Option, Stream } from "effect";
import type { Load, Memory, Pid } from "./common/surface";
import { surface } from "./common/surface";
import { createTop } from "./server/top";

/** A cell `get` and a collection `keys` both OPEN with the current snapshot,
 *  then stream deltas. In-process we only want that first frame, so run the
 *  stream's head — which interrupts the subscription the moment it lands. */
async function snapshot<T>(
  stream: Stream.Stream<T, unknown>,
  what: string,
): Promise<T> {
  const head = await Effect.runPromise(Stream.runHead(stream));
  if (Option.isNone(head)) {
    throw new Error(`${what}: stream closed before its snapshot frame`);
  }
  return head.value;
}

async function main(): Promise<void> {
  const top = createTop();
  top.start();

  // `directDispatch` takes the served surface (anything carrying `handlers`),
  // so the whole runtime goes in verbatim.
  const face = buildSurfaceFace(surface, directDispatch(top.runtime));
  const cell = <T>(name: "load" | "memory") =>
    face.surface[name]?.get as StreamingProcedure<undefined, T>;

  // Give the first poll a moment to land, then read the cells + collection.
  await new Promise((r) => setTimeout(r, 100));

  const load = await snapshot(cell<Load>("load")(undefined), "load");
  const memory = await snapshot(cell<Memory>("memory")(undefined), "memory");
  const pids = await snapshot(
    (face.surface.processes?.keys as StreamingProcedure<undefined, Pid[]>)(
      undefined,
    ),
    "processes.keys",
  );

  process.stdout.write(
    `load ${load.avg.join(" ")} over ${load.cores} cores · ` +
      `mem ${(memory.used / 1e9).toFixed(1)}/${(memory.total / 1e9).toFixed(1)} GB · ` +
      `${pids.length} processes\n`,
  );

  top.dispose();
}

main().catch((err) => {
  process.stderr.write(`fatal: ${(err as Error).message}\n`);
  process.exit(1);
});

Run it:

pnpm run inproc

You will see one line, printed from live readings of your machine:

load 0.42 0.38 0.31 over 8 cores · mem 9.2/16.0 GB · 431 processes

That is a working surface — defined, served, and consumed — with the transport collapsed to nothing. Notice the snapshot helper: a cell read is a stream whose first frame is the current snapshot, so running the stream’s head is the one-shot read — and it interrupts the subscription the moment that frame lands. That is not an in-process quirk; every link yields the same snapshot-then-deltas shape, which is what keeps a reconnect honest (see Reactive honesty).

Put it on a WebSocket

Now the same surface over the wire. src/server/main.ts serves the same createTop() engine — the same { group, handlers } pair — over a WebSocket. On the client, the only thing that changes from the in-process version is the link: websocketLink dials (async — building a protocol and its fibers is an effect), createLiveSignal takes the whole { dispatch, wire } it minted and adds the half-open watchdog, then surfaceClient binds the hooks.

/// <reference types="vite/client" />
/**
 * Client surface bundle — `surfaceClient` over a live WebSocket.
 *
 * Two steps. `websocketLink` DIALS: it owns the socket, the reconnect schedule,
 * and a URL thunk re-evaluated on every re-dial. Then `createLiveSignal` takes
 * the WHOLE `{ dispatch, wire }` the link minted together and adds the half-open
 * watchdog — a websocket can stay `open` while no bytes flow (the
 * green-over-a-dead-link lie), so `surfaceClient` REFUSES a bare wire dispatch
 * and takes only this watchdog-backed handle. Both live in `@kolu/surface`, so
 * this part needs no `@kolu/surface-app` dependency on the client side.
 *
 * The result — `app.cells.load`, `app.cells.memory`, `app.collections.processes`,
 * `app.procedures.process.kill` — is the exact surface `directDispatch` gave us
 * in-process (`inproc.ts`); only the transport changed.
 */

import { websocketLink } from "@kolu/surface/links/websocket";
import { createLiveSignal, surfaceClient } from "@kolu/surface/solid";
import { surface } from "../common/surface";

const wsUrl = `${location.protocol === "https:" ? "wss:" : "ws:"}//${location.host}/rpc/ws`;

export const link = await websocketLink({
  group: surface.group,
  url: () => wsUrl,
  // No close code retires this wire — every drop is transient, so the link
  // re-dials through all of them.
  isTerminalClose: () => false,
});

const transport = createLiveSignal(link, {});

if (import.meta.hot) {
  import.meta.hot.dispose(() => {
    transport.dispose();
    void link.dispose();
  });
}

export const app = surfaceClient(surface, transport);

The result — app.cells.load, app.collections.processes, app.procedures.process.kill — is the exact shape directDispatch gave us in-process. Only the transport changed.

Render it live

In the Solid component, each member becomes a bound .use() hook. There is no subscription lifecycle to manage: when a process appears or a load average changes, only the affected row re-renders.

const load = app.cells.load.use();
const memory = app.cells.memory.use();
const processes = app.collections.processes.use({
  onError: (err) => console.error("processes subscription failed", err),
});

// Busiest first — sort the live key set by the current cpu reading.
const rows = createMemo<Pid[]>(() =>
  [...processes.keys()].sort(
    (a, b) =>
      (processes.byKey(b)?.()?.cpuPct ?? 0) -
      (processes.byKey(a)?.()?.cpuPct ?? 0),
  ),
);

// A declared procedure is an EFFECT — a description until something runs it.
// A DOM handler is the edge where that happens.
const kill = (pid: Pid): void => {
  Effect.runFork(
    Effect.catchCause(
      app.procedures.process.kill({ pid, signal: "TERM" }),
      (cause) => Effect.sync(() => console.error("kill failed", cause)),
    ),
  );
};

The kill handler is a plain procedure call. The rest of App.tsx is ordinary Solid markup — a header and a table — reading load.value(), processes.byKey(pid)?.(), and so on.

Watch it move

Start the server and the client together:

pnpm run dev

Open http://localhost:5175. You will see your machine’s load and memory in the header and a live process table, busiest first, updating every second on its own. Find a process you own and click kill — the row vanishes as the next snapshot lands. You did not write a refresh; the surface pushed the change.

Next

Right now fleet-top dies when you close the terminal. In Make it a daemon you will turn it into a long-lived process that survives, supervises itself, and can be upgraded live while your browser stays connected.