feat(runner): WS protocol server with handshake

apps/runner scaffold (app-tier, walking-skeleton story 04): WS server
speaking @repo/core-runner-protocol. Every inbound/outbound frame is
envelope-wrapped and zod-parsed; hello/ready handshake gates on the
workspace-scoped token (constant-time compare, redacted token on
rejection replies); named error events for version/schema/auth
rejections. Config via env only (token never argv); port announced on
stdout for the story-06 provisioner. Runtime deps: ws (the standard
Node WS server; ADR-022 traces do not apply to app-tier) and zod.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016j8z4VHjedXDTjEDNg7qHK

apps/runner/src/config.test.ts

@@ -0,0 +1,80 @@
+import { describe, expect, it } from "vitest";
+import { loadConfigFromEnv } from "@/config";
+
+const REQUIRED_ENV = {
+  RUNNER_TOKEN: "ws-token-1",
+  RUNNER_WORKSPACE_DIR: "/tmp/veect-ws",
+};
+
+describe("loadConfigFromEnv", () => {
+  it("applies defaults for host, port, and heartbeat", () => {
+    const config = loadConfigFromEnv({ ...REQUIRED_ENV });
+    expect(config).toEqual({
+      host: "127.0.0.1",
+      port: 0,
+      token: "ws-token-1",
+      workspaceDir: "/tmp/veect-ws",
+      heartbeatMs: 1000,
+    });
+  });
+
+  it("reads explicit values from the environment", () => {
+    const config = loadConfigFromEnv({
+      ...REQUIRED_ENV,
+      RUNNER_HOST: "0.0.0.0",
+      RUNNER_PORT: "8123",
+      RUNNER_HEARTBEAT_MS: "250",
+    });
+    expect(config.host).toBe("0.0.0.0");
+    expect(config.port).toBe(8123);
+    expect(config.heartbeatMs).toBe(250);
+  });
+
+  it("treats empty strings as unset for optional values", () => {
+    const config = loadConfigFromEnv({
+      ...REQUIRED_ENV,
+      RUNNER_HOST: "",
+      RUNNER_PORT: "",
+      RUNNER_HEARTBEAT_MS: "",
+    });
+    expect(config.host).toBe("127.0.0.1");
+    expect(config.port).toBe(0);
+    expect(config.heartbeatMs).toBe(1000);
+  });
+
+  it("rejects a missing or empty token", () => {
+    expect(() => loadConfigFromEnv({ RUNNER_WORKSPACE_DIR: "/tmp/x" })).toThrow(
+      /RUNNER_TOKEN/,
+    );
+    expect(() =>
+      loadConfigFromEnv({ RUNNER_TOKEN: "", RUNNER_WORKSPACE_DIR: "/tmp/x" }),
+    ).toThrow(/RUNNER_TOKEN/);
+  });
+
+  it("rejects a missing or empty workspace dir", () => {
+    expect(() => loadConfigFromEnv({ RUNNER_TOKEN: "t" })).toThrow(
+      /RUNNER_WORKSPACE_DIR/,
+    );
+    expect(() =>
+      loadConfigFromEnv({ RUNNER_TOKEN: "t", RUNNER_WORKSPACE_DIR: "" }),
+    ).toThrow(/RUNNER_WORKSPACE_DIR/);
+  });
+
+  it("rejects non-integer numeric values", () => {
+    expect(() =>
+      loadConfigFromEnv({ ...REQUIRED_ENV, RUNNER_PORT: "abc" }),
+    ).toThrow(/RUNNER_PORT/);
+    expect(() =>
+      loadConfigFromEnv({ ...REQUIRED_ENV, RUNNER_HEARTBEAT_MS: "1.5" }),
+    ).toThrow(/RUNNER_HEARTBEAT_MS/);
+  });
+
+  it("rejects out-of-range ports", () => {
+    expect(() =>
+      loadConfigFromEnv({ ...REQUIRED_ENV, RUNNER_PORT: "70000" }),
+    ).toThrow();
+    expect(() =>
+      loadConfigFromEnv({ ...REQUIRED_ENV, RUNNER_PORT: "-1" }),
+    ).toThrow();
+  });
+});

apps/runner/src/config.ts

@@ -0,0 +1,66 @@
+import { z } from "zod";
+
+/**
+ * Runner process configuration, sourced from the environment by the
+ * provisioner (walking-skeleton story 06) or by the integration tests
+ * that spawn the runner directly.
+ *
+ * - `RUNNER_TOKEN` (required) — the workspace-scoped auth token every
+ *   protocol envelope is gated on. A secret: env only, never argv.
+ * - `RUNNER_WORKSPACE_DIR` (required) — the directory this runner owns;
+ *   the clone lands at `<workspaceDir>/repo`.
+ * - `RUNNER_PORT` (default `0`) — WS listen port; `0` binds an ephemeral
+ *   port which `main.ts` announces on stdout (`event: "listening"`).
+ * - `RUNNER_HOST` (default `127.0.0.1`) — WS listen host.
+ * - `RUNNER_HEARTBEAT_MS` (default `1000`) — staged-progress heartbeat
+ *   interval while a stage is running.
+ */
+const runnerConfigSchema = z
+  .object({
+    host: z.string().min(1),
+    port: z.number().int().min(0).max(65535),
+    token: z.string().min(1),
+    workspaceDir: z.string().min(1),
+    heartbeatMs: z.number().int().positive(),
+  })
+  .strict();
+
+export type RunnerConfig = z.infer<typeof runnerConfigSchema>;
+
+function parseIntegerEnv(
+  name: string,
+  raw: string | undefined,
+  fallback: number,
+): number {
+  if (raw === undefined || raw === "") return fallback;
+  const value = Number(raw);
+  if (!Number.isInteger(value)) {
+    throw new Error(`${name} must be an integer, received "${raw}"`);
+  }
+  return value;
+}
+
+export function loadConfigFromEnv(env: NodeJS.ProcessEnv): RunnerConfig {
+  const token = env.RUNNER_TOKEN;
+  if (token === undefined || token === "") {
+    throw new Error("RUNNER_TOKEN is required (workspace-scoped auth token)");
+  }
+  const workspaceDir = env.RUNNER_WORKSPACE_DIR;
+  if (workspaceDir === undefined || workspaceDir === "") {
+    throw new Error("RUNNER_WORKSPACE_DIR is required");
+  }
+  return runnerConfigSchema.parse({
+    host:
+      env.RUNNER_HOST === undefined || env.RUNNER_HOST === ""
+        ? "127.0.0.1"
+        : env.RUNNER_HOST,
+    port: parseIntegerEnv("RUNNER_PORT", env.RUNNER_PORT, 0),
+    token,
+    workspaceDir,
+    heartbeatMs: parseIntegerEnv(
+      "RUNNER_HEARTBEAT_MS",
+      env.RUNNER_HEARTBEAT_MS,
+      1000,
+    ),
+  });
+}

apps/runner/src/log.test.ts

@@ -0,0 +1,38 @@
+import { describe, expect, it, vi } from "vitest";
+import { createLogger } from "@/log";
+
+describe("createLogger", () => {
+  it("emits structured JSON lines to the injected sink", () => {
+    const lines: string[] = [];
+    const log = createLogger((line) => lines.push(line));
+
+    log.info("listening", { host: "127.0.0.1", port: 4321 });
+    log.error("fatal", { error: "boom" });
+
+    expect(lines.map((line) => JSON.parse(line))).toEqual([
+      { level: "info", event: "listening", host: "127.0.0.1", port: 4321 },
+      { level: "error", event: "fatal", error: "boom" },
+    ]);
+  });
+
+  it("omits fields cleanly when none are given", () => {
+    const lines: string[] = [];
+    const log = createLogger((line) => lines.push(line));
+    log.info("connection-open");
+    expect(JSON.parse(lines[0] ?? "")).toEqual({
+      level: "info",
+      event: "connection-open",
+    });
+  });
+
+  it("writes to stdout by default", () => {
+    const write = vi
+      .spyOn(process.stdout, "write")
+      .mockImplementation(() => true);
+    createLogger().info("listening", { port: 1 });
+    expect(write).toHaveBeenCalledWith(
+      `${JSON.stringify({ level: "info", event: "listening", port: 1 })}\n`,
+    );
+    write.mockRestore();
+  });
+});

apps/runner/src/log.ts

@@ -0,0 +1,34 @@
+/**
+ * Minimal JSON-lines logger for the runner process.
+ *
+ * The runner logs to stdout so the provisioner (story 06) and the
+ * integration tests can read a structured stream. The sink is injectable
+ * so in-process tests can capture every emitted line and assert
+ * credential-leak absence (spec §15): nothing in this module — and by
+ * convention nothing passed to it — may ever contain a secret. Callers
+ * log event names and safe fields only, never raw protocol payloads.
+ */
+export type LogSink = (line: string) => void;
+
+export interface Logger {
+  info(event: string, fields?: Record<string, unknown>): void;
+  error(event: string, fields?: Record<string, unknown>): void;
+}
+
+const stdoutSink: LogSink = (line) => {
+  process.stdout.write(`${line}\n`);
+};
+
+export function createLogger(sink: LogSink = stdoutSink): Logger {
+  const emit = (
+    level: "info" | "error",
+    event: string,
+    fields?: Record<string, unknown>,
+  ) => {
+    sink(JSON.stringify({ level, event, ...fields }));
+  };
+  return {
+    info: (event, fields) => emit("info", event, fields),
+    error: (event, fields) => emit("error", event, fields),
+  };
+}

apps/runner/src/main.ts

@@ -0,0 +1,34 @@
+import { mkdir } from "node:fs/promises";
+import { loadConfigFromEnv } from "./config";
+import { createLogger } from "./log";
+import { startRunnerServer } from "./server";
+
+/**
+ * Runner process bootstrap: env config → workspace dir → WS server.
+ *
+ * Announces the bound port on stdout as a JSON line
+ * (`{"level":"info","event":"listening","host":...,"port":...}`) — the
+ * contract the provisioner (story 06) and the spawn-based integration
+ * tests use to discover an ephemeral port (`RUNNER_PORT=0`).
+ */
+async function main(): Promise<void> {
+  const log = createLogger();
+  const config = loadConfigFromEnv(process.env);
+  await mkdir(config.workspaceDir, { recursive: true });
+  const server = await startRunnerServer({ ...config, log });
+  log.info("listening", { host: server.host, port: server.port });
+
+  const shutdown = (signal: string): void => {
+    log.info("shutdown", { signal });
+    void server.close().then(() => process.exit(0));
+  };
+  process.on("SIGTERM", () => shutdown("SIGTERM"));
+  process.on("SIGINT", () => shutdown("SIGINT"));
+}
+
+main().catch((error: unknown) => {
+  createLogger().error("fatal", {
+    error: error instanceof Error ? error.message : String(error),
+  });
+  process.exit(1);
+});

apps/runner/src/server.test.ts

@@ -0,0 +1,98 @@
+import { describe, expect, it } from "vitest";
+import { PROTOCOL_VERSION } from "@repo/core-runner-protocol";
+import { gateInbound } from "@/server";
+
+const TOKEN = "workspace-token";
+
+function frame(overrides: Record<string, unknown> = {}): string {
+  return JSON.stringify({
+    protocolVersion: PROTOCOL_VERSION,
+    token: TOKEN,
+    message: { type: "hello" },
+    ...overrides,
+  });
+}
+
+describe("gateInbound", () => {
+  it("accepts a conformant envelope and returns the inner message", () => {
+    const result = gateInbound(frame(), TOKEN);
+    expect(result).toEqual({ ok: true, message: { type: "hello" } });
+  });
+
+  it("rejects a frame that is not JSON", () => {
+    const result = gateInbound("{nope", TOKEN);
+    expect(result).toMatchObject({ ok: false, cause: "invalid-message" });
+  });
+
+  it("rejects JSON that is not an object", () => {
+    expect(gateInbound('"hello"', TOKEN)).toMatchObject({
+      ok: false,
+      cause: "invalid-message",
+    });
+    expect(gateInbound("[1,2]", TOKEN)).toMatchObject({
+      ok: false,
+      cause: "invalid-message",
+    });
+    expect(gateInbound("null", TOKEN)).toMatchObject({
+      ok: false,
+      cause: "invalid-message",
+    });
+  });
+
+  it("rejects an unsupported protocol version with the named cause", () => {
+    const result = gateInbound(frame({ protocolVersion: "999" }), TOKEN);
+    expect(result).toMatchObject({
+      ok: false,
+      cause: "unsupported-protocol-version",
+    });
+  });
+
+  it("rejects a missing protocol version as unsupported", () => {
+    const raw = JSON.stringify({ token: TOKEN, message: { type: "hello" } });
+    expect(gateInbound(raw, TOKEN)).toMatchObject({
+      ok: false,
+      cause: "unsupported-protocol-version",
+    });
+  });
+
+  it("rejects a missing token as invalid-message (schema)", () => {
+    const raw = JSON.stringify({
+      protocolVersion: PROTOCOL_VERSION,
+      message: { type: "hello" },
+    });
+    expect(gateInbound(raw, TOKEN)).toMatchObject({
+      ok: false,
+      cause: "invalid-message",
+    });
+  });
+
+  it("rejects unknown envelope fields (strict schema)", () => {
+    const result = gateInbound(frame({ extra: "field" }), TOKEN);
+    expect(result).toMatchObject({ ok: false, cause: "invalid-message" });
+  });
+
+  it("rejects an unknown message shape without echoing payload values", () => {
+    const result = gateInbound(
+      frame({
+        message: { type: "clone", gitUrl: "", pat: "super-secret-pat" },
+      }),
+      TOKEN,
+    );
+    expect(result).toMatchObject({ ok: false, cause: "invalid-message" });
+    expect(JSON.stringify(result)).not.toContain("super-secret-pat");
+  });
+
+  it("rejects a wrong token as unauthorized", () => {
+    expect(gateInbound(frame({ token: "wrong" }), TOKEN)).toMatchObject({
+      ok: false,
+      cause: "unauthorized",
+    });
+    // Same length, different bytes — exercises the constant-time compare.
+    expect(
+      gateInbound(frame({ token: "workspace-tokeX" }), TOKEN),
+    ).toMatchObject({
+      ok: false,
+      cause: "unauthorized",
+    });
+  });
+});

apps/runner/src/server.ts

@@ -0,0 +1,292 @@
+import { once } from "node:events";
+import { timingSafeEqual } from "node:crypto";
+import { WebSocketServer, type RawData, type WebSocket } from "ws";
+import {
+  envelopeSchema,
+  PROTOCOL_VERSION,
+  type RunnerErrorCause,
+  type RunnerMessage,
+  type RunnerStage,
+} from "@repo/core-runner-protocol";
+import type { Logger } from "./log";
+
+/**
+ * The runner's WS protocol server (ADR-027, protocol version "0").
+ *
+ * Session semantics:
+ *
+ * - Every inbound frame is gated: JSON → protocol version → envelope
+ *   schema (`.strict()`) → workspace-token match. Rejections emit a named
+ *   `error` event (`unsupported-protocol-version` / `invalid-message` /
+ *   `unauthorized`); pre-handshake rejections and token mismatches also
+ *   close the socket (1008 policy violation).
+ * - The first accepted message MUST be `hello`; the runner answers
+ *   `ready` ("the runner will take commands").
+ * - Commands run strictly one at a time in arrival order. While a stage
+ *   runs, the runner streams `status` events (stage + elapsedMs). A
+ *   command that succeeds ends with `ready` — the idle marker the
+ *   orchestrator (story 07) drives the session on. A command that fails
+ *   ends with a named `error` event instead; the session stays open so
+ *   the client can correct and retry.
+ * - Every outbound message is envelope-wrapped and zod-parsed before it
+ *   is sent. Outbound envelopes carry the workspace token only once the
+ *   peer has proven possession of it; rejection replies to unproven
+ *   peers carry a redacted placeholder so the token cannot leak.
+ * - Error messages summarize schema issues by path + code only — raw
+ *   payload contents (which may include a PAT) are never echoed into
+ *   errors or logs.
+ */
+export interface RunnerServerOptions {
+  host: string;
+  port: number;
+  token: string;
+  workspaceDir: string;
+  heartbeatMs: number;
+  log: Logger;
+}
+
+export interface RunnerServer {
+  host: string;
+  /** The actually-bound port (options.port `0` binds an ephemeral one). */
+  port: number;
+  /** Terminate all sessions and stop listening. Idempotent. */
+  close: () => Promise<void>;
+}
+
+/** Placeholder token on replies to peers that never proved possession. */
+const REDACTED_TOKEN = "redacted";
+
+function tokensMatch(expected: string, provided: string): boolean {
+  const expectedBuffer = Buffer.from(expected);
+  const providedBuffer = Buffer.from(provided);
+  if (expectedBuffer.length !== providedBuffer.length) return false;
+  return timingSafeEqual(expectedBuffer, providedBuffer);
+}
+
+type GateRejection = {
+  ok: false;
+  cause: Extract<
+    RunnerErrorCause,
+    "unsupported-protocol-version" | "invalid-message" | "unauthorized"
+  >;
+  detail: string;
+};
+
+type GateResult = { ok: true; message: RunnerMessage } | GateRejection;
+
+/** Issue summary that never echoes payload values (a payload can hold a PAT). */
+function summarizeIssues(error: import("zod").ZodError): string {
+  return error.issues
+    .map((issue) => `${issue.path.join(".") || "(root)"}: ${issue.code}`)
+    .join("; ");
+}
+
+/**
+ * Gate one inbound frame: JSON → version → envelope schema → token.
+ * Exported for direct unit coverage of every rejection branch.
+ */
+export function gateInbound(raw: string, expectedToken: string): GateResult {
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(raw);
+  } catch {
+    return {
+      ok: false,
+      cause: "invalid-message",
+      detail: "frame is not valid JSON",
+    };
+  }
+  if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) {
+    return {
+      ok: false,
+      cause: "invalid-message",
+      detail: "frame is not a JSON object",
+    };
+  }
+  const version = (parsed as Record<string, unknown>).protocolVersion;
+  if (version !== PROTOCOL_VERSION) {
+    return {
+      ok: false,
+      cause: "unsupported-protocol-version",
+      detail: `this runner speaks protocol version "${PROTOCOL_VERSION}"`,
+    };
+  }
+  const result = envelopeSchema.safeParse(parsed);
+  if (!result.success) {
+    return {
+      ok: false,
+      cause: "invalid-message",
+      detail: summarizeIssues(result.error),
+    };
+  }
+  if (!tokensMatch(expectedToken, result.data.token)) {
+    return {
+      ok: false,
+      cause: "unauthorized",
+      detail: "workspace token mismatch",
+    };
+  }
+  return { ok: true, message: result.data.message };
+}
+
+/** Runner-emitted event types a client must never send. */
+const RUNNER_EMITTED_TYPES = new Set(["status", "error", "ready"]);
+
+export type CommandMessage = Extract<
+  RunnerMessage,
+  { type: "clone" | "install" | "scan" | "adapter-start" | "render-frame" }
+>;
+
+export interface StageEmitter {
+  status: (stage: RunnerStage, elapsedMs: number) => void;
+  error: (
+    cause: RunnerErrorCause,
+    message: string,
+    stage?: RunnerStage,
+  ) => void;
+}
+
+/** Stages that later walking-skeleton stories wire up (05+). */
+const NOT_IMPLEMENTED: Record<
+  CommandMessage["type"],
+  { cause: RunnerErrorCause; stage?: RunnerStage }
+> = {
+  clone: { cause: "clone-failed", stage: "cloning" },
+  install: { cause: "install-failed", stage: "installing" },
+  scan: { cause: "scan-failed", stage: "scanning" },
+  "adapter-start": { cause: "adapter-start-failed", stage: "starting-preview" },
+  "render-frame": { cause: "render-failed" },
+};
+
+async function handleCommand(
+  command: CommandMessage,
+  emit: StageEmitter,
+  _options: RunnerServerOptions,
+): Promise<void> {
+  const notImplemented = NOT_IMPLEMENTED[command.type];
+  emit.error(
+    notImplemented.cause,
+    `the "${command.type}" command is not implemented by this runner build yet`,
+    notImplemented.stage,
+  );
+}
+
+function rawToString(data: RawData): string {
+  if (Array.isArray(data)) return Buffer.concat(data).toString("utf8");
+  if (data instanceof ArrayBuffer) return Buffer.from(data).toString("utf8");
+  return data.toString("utf8");
+}
+
+export async function startRunnerServer(
+  options: RunnerServerOptions,
+): Promise<RunnerServer> {
+  const { host, port, token, log } = options;
+  const wss = new WebSocketServer({ host, port });
+  await once(wss, "listening");
+  const address = wss.address();
+  if (address === null || typeof address === "string") {
+    wss.close();
+    throw new Error("runner server: could not determine bound port");
+  }
+
+  wss.on("connection", (ws: WebSocket) => {
+    let authenticated = false;
+    let commandChain: Promise<void> = Promise.resolve();
+    log.info("connection-open", {});
+
+    const send = (message: RunnerMessage): void => {
+      // Outbound conformance gate: nothing leaves unparsed.
+      const envelope = envelopeSchema.parse({
+        protocolVersion: PROTOCOL_VERSION,
+        token: authenticated ? token : REDACTED_TOKEN,
+        message,
+      });
+      ws.send(JSON.stringify(envelope));
+    };
+
+    const emit: StageEmitter = {
+      status: (stage, elapsedMs) => send({ type: "status", stage, elapsedMs }),
+      error: (cause, message, stage) =>
+        send(
+          stage
+            ? { type: "error", cause, message, stage }
+            : { type: "error", cause, message },
+        ),
+    };
+
+    ws.on("message", (data: RawData) => {
+      const gate = gateInbound(rawToString(data), token);
+      if (!gate.ok) {
+        log.info("frame-rejected", { cause: gate.cause });
+        emit.error(gate.cause, gate.detail);
+        if (gate.cause === "unauthorized" || !authenticated)
+          ws.close(1008, gate.cause);
+        return;
+      }
+
+      const message = gate.message;
+      if (!authenticated) {
+        if (message.type !== "hello") {
+          emit.error(
+            "invalid-message",
+            `handshake required: expected "hello", received "${message.type}"`,
+          );
+          ws.close(1008, "handshake-required");
+          return;
+        }
+        authenticated = true;
+        log.info("handshake-complete", {});
+        send({ type: "ready" });
+        return;
+      }
+
+      if (message.type === "hello") {
+        // Idempotent: re-confirm readiness.
+        send({ type: "ready" });
+        return;
+      }
+      if (RUNNER_EMITTED_TYPES.has(message.type)) {
+        emit.error(
+          "invalid-message",
+          `"${message.type}" is a runner-emitted event; clients must not send it`,
+        );
+        return;
+      }
+
+      const command = message as CommandMessage;
+      log.info("command-received", { command: command.type });
+      commandChain = commandChain.then(async () => {
+        try {
+          await handleCommand(command, emit, options);
+        } catch (error) {
+          // Stages map their own failures; this is the last-resort net.
+          log.error("command-crashed", {
+            command: command.type,
+            error: error instanceof Error ? error.message : String(error),
+          });
+          emit.error(
+            "invalid-message",
+            `the "${command.type}" command crashed unexpectedly`,
+          );
+        }
+      });
+    });
+
+    ws.on("close", () => {
+      log.info("connection-closed", {});
+    });
+  });
+
+  let closed: Promise<void> | null = null;
+  return {
+    host,
+    port: address.port,
+    close: () =>
+      (closed ??= (async () => {
+        for (const client of wss.clients) client.terminate();
+        await new Promise<void>((resolve, reject) => {
+          wss.close((error) => (error ? reject(error) : resolve()));
+        });
+      })()),
+  };
+}