diff --git a/AGENTS.md b/AGENTS.md index 2c2638d..40c8f02 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -52,11 +52,12 @@ See `docs/guides/runbook.md` for the full workflow. Apps: -| App | Port | Purpose | -| ---------------- | ---- | ---------------------------------------------------------------------------------------------- | -| `apps/web-next` | 3000 | Next.js — the hosted editor shell + landing page; custom `server.ts` hosts Next.js + Socket.IO | -| `apps/cms` | 3001 | Payload admin | -| `apps/storybook` | 6006 | Storybook — component workshop for `core-ui` + feature UI | +| App | Port | Purpose | +| ---------------- | --------- | ---------------------------------------------------------------------------------------------- | +| `apps/web-next` | 3000 | Next.js — the hosted editor shell + landing page; custom `server.ts` hosts Next.js + Socket.IO | +| `apps/cms` | 3001 | Payload admin | +| `apps/storybook` | 6006 | Storybook — component workshop for `core-ui` + feature UI | +| `apps/runner` | ephemeral | Workspace runner (ADR-027) — WS runner-protocol server; clone/install/scan/preview stages | `auth` and `editor` are the feature packages today (`editor` is UI-only — empty `useCases`, no binders). The remaining Veect control-plane features (workspaces, projects, discovery, design-doc, …) land as sibling packages under `packages/` following the same shape. Per ADR-029, `packages/editor` is rebuilt under template conventions — the prototype codebase under `docs/product/reference/` is reference material, never vendored. @@ -66,7 +67,7 @@ Apps: ### Five tags -- **app** (3 packages) — `apps/web-next`, `apps/cms`, `apps/storybook` +- **app** (4 packages) — `apps/web-next`, `apps/cms`, `apps/storybook`, `apps/runner` - **core-composition** (3 packages) — `packages/core-api`, `core-cms` (must-have); `core-trpc` (optional, scaffolded) - **core** (9 packages) — `packages/core-shared` (must-have); `core-ui`, `core-events`, `core-realtime`, `core-audit`, `core-analytics`, `core-consent`, `core-dsr` (optional cores, all currently scaffolded — new ones via `pnpm turbo gen core-package `); `core-runner-protocol` (hand-scaffolded — outside the generator's snapshot set) - **feature** (1 package) — `packages/auth` diff --git a/apps/runner/AGENTS.md b/apps/runner/AGENTS.md new file mode 100644 index 0000000..7c00046 --- /dev/null +++ b/apps/runner/AGENTS.md @@ -0,0 +1,46 @@ +# apps/runner — workspace runner (cloud-runner process) + +The runner process of ADR-027: executes a workspace's repo code (clone, +install, scan, preview adapter) on behalf of the control plane, speaking +`@repo/core-runner-protocol` over plain WS/JSON. App-tier (`tags: +["app"]`): imperative code is expected here; there is no feature +manifest, no use-case layer, no DI container. + +## Process contract (what the provisioner — story 06 — relies on) + +- **Config is env-only** (see `src/config.ts`): `RUNNER_TOKEN` (required, + secret — never argv), `RUNNER_WORKSPACE_DIR` (required), + `RUNNER_PORT` (default `0` = ephemeral), `RUNNER_HOST` (default + `127.0.0.1`), `RUNNER_HEARTBEAT_MS` (default `1000`). +- **Port discovery:** on boot the runner writes a JSON line to stdout: + `{"level":"info","event":"listening","host":...,"port":...}`. +- Logs are JSON lines on stdout. Log event names + safe fields only — + never protocol payloads, never credentials. Integration tests grep the + entire child output for the PAT and the workspace token. +- `SIGTERM`/`SIGINT` shut down gracefully. + +## Protocol session semantics (documented in `src/server.ts`) + +- Every inbound/outbound frame is envelope-wrapped and zod-parsed + (`envelopeSchema`); rejections emit named errors + (`unsupported-protocol-version` / `invalid-message` / `unauthorized`). + Pre-handshake rejections and token mismatches close the socket (1008). +- First message must be `hello`; the runner answers `ready`. +- Commands run one at a time in arrival order. A running stage streams + `status` (stage + elapsedMs: start, heartbeats, final). Success ends + with `ready` (the idle marker the story-07 orchestrator drives on); + failure ends with a named `error` and the session stays open. +- Zod issues are summarized as path + code only — payload values (which + may include a PAT) are never echoed into errors or logs. + +## Testing + +- Unit suites live next to sources in `src/`; protocol/WS suites live in + `tests/` (`protocol-server.test.ts` runs the real server in-process so + v8 coverage sees it; `runner-process.integration.test.ts` spawns the + real child process and discovers the port from stdout). +- `tests/protocol-client.ts` envelope-parses every inbound frame, so + every test doubles as an outbound-conformance assertion. +- Coverage: app-tier vitest thresholds inherited from + `vitest.base.node`; only `src/main.ts` (bootstrap glue, exercised by + the spawn suite in a child process) is excluded. diff --git a/apps/runner/eslint.config.js b/apps/runner/eslint.config.js new file mode 100644 index 0000000..6e3e103 --- /dev/null +++ b/apps/runner/eslint.config.js @@ -0,0 +1,3 @@ +import baseConfig from "@repo/core-eslint/base"; + +export default [...baseConfig]; diff --git a/apps/runner/package.json b/apps/runner/package.json new file mode 100644 index 0000000..afea877 --- /dev/null +++ b/apps/runner/package.json @@ -0,0 +1,29 @@ +{ + "name": "@repo/runner", + "private": true, + "version": "0.0.0", + "type": "module", + "scripts": { + "build": "tsc --noEmit", + "start": "node --import tsx src/main.ts", + "lint": "eslint .", + "test": "vitest run", + "typecheck": "tsc --noEmit" + }, + "dependencies": { + "@repo/core-runner-protocol": "workspace:*", + "ws": "^8.18.0", + "zod": "^3.25.0" + }, + "devDependencies": { + "@repo/core-eslint": "workspace:*", + "@repo/core-testing": "workspace:*", + "@repo/core-typescript": "workspace:*", + "@types/node": "^22.0.0", + "@types/ws": "^8.18.0", + "@vitest/coverage-v8": "^3.2.4", + "tsx": "^4.0.0", + "typescript": "^5.8.0", + "vitest": "^3.0.0" + } +} diff --git a/apps/runner/src/config.test.ts b/apps/runner/src/config.test.ts new file mode 100644 index 0000000..a4a3b10 --- /dev/null +++ b/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(); + }); +}); diff --git a/apps/runner/src/config.ts b/apps/runner/src/config.ts new file mode 100644 index 0000000..9fbf15f --- /dev/null +++ b/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 `/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; + +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, + ), + }); +} diff --git a/apps/runner/src/log.test.ts b/apps/runner/src/log.test.ts new file mode 100644 index 0000000..3e5ceac --- /dev/null +++ b/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(); + }); +}); diff --git a/apps/runner/src/log.ts b/apps/runner/src/log.ts new file mode 100644 index 0000000..2337724 --- /dev/null +++ b/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): void; + error(event: string, fields?: Record): 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, + ) => { + sink(JSON.stringify({ level, event, ...fields })); + }; + return { + info: (event, fields) => emit("info", event, fields), + error: (event, fields) => emit("error", event, fields), + }; +} diff --git a/apps/runner/src/main.ts b/apps/runner/src/main.ts new file mode 100644 index 0000000..ef1b400 --- /dev/null +++ b/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 { + 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); +}); diff --git a/apps/runner/src/server.test.ts b/apps/runner/src/server.test.ts new file mode 100644 index 0000000..04be118 --- /dev/null +++ b/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 { + 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", + }); + }); +}); diff --git a/apps/runner/src/server.ts b/apps/runner/src/server.ts new file mode 100644 index 0000000..611b903 --- /dev/null +++ b/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; +} + +/** 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).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 { + 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 { + 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 = 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 | null = null; + return { + host, + port: address.port, + close: () => + (closed ??= (async () => { + for (const client of wss.clients) client.terminate(); + await new Promise((resolve, reject) => { + wss.close((error) => (error ? reject(error) : resolve())); + }); + })()), + }; +} diff --git a/apps/runner/tests/protocol-client.ts b/apps/runner/tests/protocol-client.ts new file mode 100644 index 0000000..6f03a37 --- /dev/null +++ b/apps/runner/tests/protocol-client.ts @@ -0,0 +1,133 @@ +import { once } from "node:events"; +import WebSocket from "ws"; +import { + envelopeSchema, + PROTOCOL_VERSION, + type Envelope, + type RunnerMessage, +} from "@repo/core-runner-protocol"; + +/** + * Test-side protocol client. Every inbound frame is parsed against the + * envelope schema — so every assertion made through this client also + * proves the runner's outbound messages are protocol-conformant. + */ +export interface ProtocolClient { + /** Envelope-wrap and send a message (token/version overridable per call). */ + send: ( + message: RunnerMessage, + overrides?: Partial>, + ) => void; + /** Send a raw frame verbatim (malformed-frame tests). */ + sendRaw: (data: string) => void; + /** All inbound messages so far, in arrival order. */ + received: () => RunnerMessage[]; + /** All inbound frames verbatim (for leak assertions on the raw wire). */ + rawReceived: () => string[]; + /** Resolve the first (possibly already-received) message matching. */ + waitFor: ( + predicate: (message: RunnerMessage) => boolean, + description: string, + timeoutMs?: number, + ) => Promise; + /** Resolves with the close code once the server closes the socket. */ + closed: Promise; + close: () => void; +} + +export async function connectProtocolClient( + port: number, + token: string, +): Promise { + const ws = new WebSocket(`ws://127.0.0.1:${port}`); + await once(ws, "open"); + + const inbound: RunnerMessage[] = []; + const rawFrames: string[] = []; + const waiters = new Set<() => void>(); + + ws.on("message", (data) => { + const raw = String(data); + rawFrames.push(raw); + const envelope = envelopeSchema.parse(JSON.parse(raw)); + inbound.push(envelope.message); + for (const notify of [...waiters]) notify(); + }); + + const closed = new Promise((resolve) => { + ws.on("close", (code) => resolve(code)); + }); + + return { + send: (message, overrides) => { + ws.send( + JSON.stringify({ + protocolVersion: PROTOCOL_VERSION, + token, + ...overrides, + message, + }), + ); + }, + sendRaw: (data) => ws.send(data), + received: () => [...inbound], + rawReceived: () => [...rawFrames], + waitFor: (predicate, description, timeoutMs = 30_000) => { + return new Promise((resolve, reject) => { + let scanned = 0; + const scan = (): void => { + while (scanned < inbound.length) { + const message = inbound[scanned]; + scanned += 1; + if (message !== undefined && predicate(message)) { + cleanup(); + resolve(message); + return; + } + } + }; + const timer = setTimeout(() => { + cleanup(); + reject( + new Error( + `timed out waiting for ${description}; received: ${JSON.stringify(inbound.map((m) => m.type))}`, + ), + ); + }, timeoutMs); + const cleanup = (): void => { + clearTimeout(timer); + waiters.delete(scan); + }; + waiters.add(scan); + scan(); + }); + }, + closed, + close: () => ws.close(), + }; +} + +/** Complete the hello/ready handshake and return the ready message. */ +export async function handshake( + client: ProtocolClient, +): Promise { + client.send({ type: "hello" }); + return client.waitFor((m) => m.type === "ready", "ready handshake reply"); +} + +/** Wait for the next `error` event and assert its named cause. */ +export async function expectNamedError( + client: ProtocolClient, + cause: string, +): Promise> { + const error = await client.waitFor( + (m) => m.type === "error", + `${cause} error`, + ); + if (error.type !== "error" || error.cause !== cause) { + throw new Error( + `expected a "${cause}" error, received: ${JSON.stringify(error)}`, + ); + } + return error; +} diff --git a/apps/runner/tests/protocol-server.test.ts b/apps/runner/tests/protocol-server.test.ts new file mode 100644 index 0000000..ef2a871 --- /dev/null +++ b/apps/runner/tests/protocol-server.test.ts @@ -0,0 +1,212 @@ +/** + * Protocol-server suite over a real WS connection (in-process server so + * v8 coverage sees `server.ts`; the spawn suite in + * `runner-process.integration.test.ts` proves the same handshake against + * a real child process). Assertions go through the protocol only — every + * inbound frame the client records is envelope-parsed, so runner + * outbound conformance is asserted implicitly on every test. + */ +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { PROTOCOL_VERSION } from "@repo/core-runner-protocol"; +import { startRunnerServer, type RunnerServer } from "@/server"; +import { createLogger } from "@/log"; +import { + connectProtocolClient, + expectNamedError, + handshake, + type ProtocolClient, +} from "./protocol-client"; + +const TOKEN = "test-workspace-token"; + +let server: RunnerServer; +let workspaceDir: string; +let logLines: string[]; +let clients: ProtocolClient[]; + +async function connect(token = TOKEN): Promise { + const client = await connectProtocolClient(server.port, token); + clients.push(client); + return client; +} + +beforeEach(async () => { + workspaceDir = await mkdtemp(path.join(os.tmpdir(), "veect-runner-test-")); + logLines = []; + clients = []; + server = await startRunnerServer({ + host: "127.0.0.1", + port: 0, + token: TOKEN, + workspaceDir, + heartbeatMs: 50, + log: createLogger((line) => logLines.push(line)), + }); +}); + +afterEach(async () => { + for (const client of clients) client.close(); + await server.close(); + await rm(workspaceDir, { recursive: true, force: true }); +}); + +describe("hello/ready handshake", () => { + it("answers a valid hello with ready", async () => { + const client = await connect(); + const ready = await handshake(client); + expect(ready).toEqual({ type: "ready" }); + }); + + it.each([ + [ + "a bad token", + "unauthorized", + (client: ProtocolClient) => client.send({ type: "hello" }), + "wrong-token", + ], + [ + "a missing token (schema)", + "invalid-message", + (client: ProtocolClient) => + client.sendRaw( + JSON.stringify({ + protocolVersion: PROTOCOL_VERSION, + message: { type: "hello" }, + }), + ), + TOKEN, + ], + [ + "an unsupported protocol version", + "unsupported-protocol-version", + (client: ProtocolClient) => + client.send({ type: "hello" }, { protocolVersion: "42" as never }), + TOKEN, + ], + [ + "a non-JSON frame", + "invalid-message", + (client: ProtocolClient) => client.sendRaw("definitely not json"), + TOKEN, + ], + ] as const)( + "rejects %s with a named %s error and closes 1008", + async (_label, cause, act, token) => { + const client = await connect(token); + act(client); + await expectNamedError(client, cause); + await expect(client.closed).resolves.toBe(1008); + }, + ); + + it("requires hello first: a command before the handshake is rejected", async () => { + const client = await connect(); + client.send({ type: "install" }); + const error = await expectNamedError(client, "invalid-message"); + expect(error.message).toContain("handshake required"); + await expect(client.closed).resolves.toBe(1008); + }); + + it("redacts the workspace token on replies to unauthenticated peers", async () => { + const client = await connect("wrong-token"); + client.send({ type: "hello" }); + await expectNamedError(client, "unauthorized"); + // The rejection envelope must not carry the real workspace token. + expect(client.rawReceived().join("")).not.toContain(TOKEN); + expect(client.rawReceived().join("")).toContain('"redacted"'); + await expect(client.closed).resolves.toBe(1008); + }); + + it("answers a repeated hello idempotently with ready", async () => { + const client = await connect(); + await handshake(client); + client.send({ type: "hello" }); + const second = await client.waitFor( + (m) => + m.type === "ready" && + client.received().filter((r) => r.type === "ready").length >= 2, + "second ready", + ); + expect(second).toEqual({ type: "ready" }); + }); +}); + +describe("post-handshake message gate", () => { + it("keeps the session open after a malformed frame post-handshake", async () => { + const client = await connect(); + await handshake(client); + client.sendRaw("{broken"); + await expectNamedError(client, "invalid-message"); + // Session survives: another hello still gets a ready. + client.send({ type: "hello" }); + const ready = await client.waitFor( + (m) => + m.type === "ready" && + client.received().filter((r) => r.type === "ready").length >= 2, + "ready after malformed frame", + ); + expect(ready).toEqual({ type: "ready" }); + }); + + it("closes the session on a bad token even post-handshake", async () => { + const client = await connect(); + await handshake(client); + client.send({ type: "install" }, { token: "tampered" }); + await expectNamedError(client, "unauthorized"); + await expect(client.closed).resolves.toBe(1008); + }); + + it("rejects runner-emitted event types sent by the client", async () => { + const client = await connect(); + await handshake(client); + client.send({ type: "status", stage: "cloning", elapsedMs: 1 }); + const error = await expectNamedError(client, "invalid-message"); + expect(error.message).toContain("runner-emitted"); + }); + + it("rejects a strict-schema violation without echoing payload values", async () => { + const client = await connect(); + await handshake(client); + client.sendRaw( + JSON.stringify({ + protocolVersion: PROTOCOL_VERSION, + token: TOKEN, + message: { + type: "clone", + gitUrl: "git://x/y.git", + pat: "sekrit-pat", + extra: true, + }, + }), + ); + const error = await expectNamedError(client, "invalid-message"); + expect(JSON.stringify(error)).not.toContain("sekrit-pat"); + expect(logLines.join("\n")).not.toContain("sekrit-pat"); + }); +}); + +describe("commands not yet wired (stories 05+)", () => { + it.each([ + ["scan", "scan-failed", "scanning"], + ["adapter-start", "adapter-start-failed", "starting-preview"], + ] as const)( + "answers %s with the named %s error", + async (type, cause, stage) => { + const client = await connect(); + await handshake(client); + client.send({ type } as never); + const error = await expectNamedError(client, cause); + expect(error).toMatchObject({ type: "error", cause, stage }); + }, + ); + + it("answers render-frame with the named render-failed error", async () => { + const client = await connect(); + await handshake(client); + client.send({ type: "render-frame", componentId: "button" }); + await expectNamedError(client, "render-failed"); + }); +}); diff --git a/apps/runner/tests/runner-process.integration.test.ts b/apps/runner/tests/runner-process.integration.test.ts new file mode 100644 index 0000000..da33eb8 --- /dev/null +++ b/apps/runner/tests/runner-process.integration.test.ts @@ -0,0 +1,48 @@ +/** + * Spawn-based integration suite: the runner as the provisioner (story 06) + * will actually run it — a real child process, config via env only, port + * discovered from stdout, real WS from the outside. + */ +import { afterEach, describe, expect, it } from "vitest"; +import { spawnRunner, type SpawnedRunner } from "./spawn-runner"; +import { + connectProtocolClient, + expectNamedError, + handshake, + type ProtocolClient, +} from "./protocol-client"; + +const TOKEN = "spawned-runner-token"; + +let runner: SpawnedRunner | undefined; +let clients: ProtocolClient[] = []; + +afterEach(async () => { + for (const client of clients) client.close(); + clients = []; + await runner?.stop(); + runner = undefined; +}); + +describe("spawned runner process", () => { + it("boots from env config, announces its port, and completes the handshake over real WS", async () => { + runner = await spawnRunner({ token: TOKEN }); + expect(runner.port).toBeGreaterThan(0); + + const client = await connectProtocolClient(runner.port, TOKEN); + clients.push(client); + const ready = await handshake(client); + expect(ready).toEqual({ type: "ready" }); + }); + + it("rejects a bad workspace token from a real client and closes the socket", async () => { + runner = await spawnRunner({ token: TOKEN }); + const client = await connectProtocolClient(runner.port, "not-the-token"); + clients.push(client); + client.send({ type: "hello" }); + await expectNamedError(client, "unauthorized"); + await expect(client.closed).resolves.toBe(1008); + // The runner never writes the workspace token to its logs. + expect(runner.output()).not.toContain(TOKEN); + }); +}); diff --git a/apps/runner/tests/spawn-runner.ts b/apps/runner/tests/spawn-runner.ts new file mode 100644 index 0000000..a58eaaf --- /dev/null +++ b/apps/runner/tests/spawn-runner.ts @@ -0,0 +1,113 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { mkdtemp, rm } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +const APP_ROOT = fileURLToPath(new URL("..", import.meta.url)); + +const START_TIMEOUT_MS = 60_000; +const KILL_TIMEOUT_MS = 5_000; + +/** + * A real runner child process, as the provisioner (story 06) will spawn + * it: config via env only, port discovered from the stdout `listening` + * line. `output()` exposes everything the process ever wrote — the + * credential-leak assertions grep it. + */ +export interface SpawnedRunner { + port: number; + workspaceDir: string; + /** Combined stdout + stderr captured so far. */ + output: () => string; + stop: () => Promise; +} + +export async function spawnRunner(options: { + token: string; +}): Promise { + const workspaceDir = await mkdtemp( + path.join(os.tmpdir(), "veect-runner-ws-"), + ); + const child: ChildProcess = spawn( + process.execPath, + ["--import", "tsx", "src/main.ts"], + { + cwd: APP_ROOT, + env: { + ...process.env, + RUNNER_TOKEN: options.token, + RUNNER_WORKSPACE_DIR: workspaceDir, + RUNNER_PORT: "0", + RUNNER_HEARTBEAT_MS: "250", + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + + let captured = ""; + child.stdout?.on("data", (chunk: Buffer) => { + captured += chunk.toString("utf8"); + }); + child.stderr?.on("data", (chunk: Buffer) => { + captured += chunk.toString("utf8"); + }); + + const stop = async (): Promise => { + if (child.exitCode === null && child.signalCode === null) { + const exited = new Promise((resolve) => + child.once("exit", () => resolve()), + ); + const killTimer = setTimeout( + () => child.kill("SIGKILL"), + KILL_TIMEOUT_MS, + ); + killTimer.unref(); + child.kill("SIGTERM"); + await exited; + clearTimeout(killTimer); + } + await rm(workspaceDir, { recursive: true, force: true }); + }; + + const port = await new Promise((resolve, reject) => { + const deadline = setTimeout(() => { + reject( + new Error( + `runner did not announce a port within ${START_TIMEOUT_MS}ms:\n${captured}`, + ), + ); + }, START_TIMEOUT_MS); + const poll = setInterval(() => { + if (child.exitCode !== null) { + clearTimeout(deadline); + clearInterval(poll); + reject( + new Error( + `runner exited before listening (code ${child.exitCode}):\n${captured}`, + ), + ); + return; + } + for (const line of captured.split("\n")) { + if (!line.includes('"listening"')) continue; + try { + const parsed = JSON.parse(line) as { event?: string; port?: number }; + if (parsed.event === "listening" && typeof parsed.port === "number") { + clearTimeout(deadline); + clearInterval(poll); + resolve(parsed.port); + return; + } + } catch { + // partial line — keep polling + } + } + }, 50); + }).catch(async (error: unknown) => { + await stop(); + throw error; + }); + + return { port, workspaceDir, output: () => captured, stop }; +} diff --git a/apps/runner/tsconfig.json b/apps/runner/tsconfig.json new file mode 100644 index 0000000..3d8fea9 --- /dev/null +++ b/apps/runner/tsconfig.json @@ -0,0 +1,12 @@ +{ + "extends": "@repo/core-typescript/base.json", + "compilerOptions": { + "rootDir": ".", + "paths": { + "@/*": ["./src/*"] + }, + "types": ["vitest/globals", "node"] + }, + "include": ["src/**/*", "tests/**/*", "vitest.config.ts"], + "exclude": ["node_modules", "dist"] +} diff --git a/apps/runner/turbo.json b/apps/runner/turbo.json new file mode 100644 index 0000000..dbcc9dd --- /dev/null +++ b/apps/runner/turbo.json @@ -0,0 +1,15 @@ +{ + "extends": ["//"], + "tags": ["app"], + "tasks": { + "test": { + "env": [ + "RUNNER_HOST", + "RUNNER_PORT", + "RUNNER_TOKEN", + "RUNNER_WORKSPACE_DIR", + "RUNNER_HEARTBEAT_MS" + ] + } + } +} diff --git a/apps/runner/vitest.config.ts b/apps/runner/vitest.config.ts new file mode 100644 index 0000000..df5ae43 --- /dev/null +++ b/apps/runner/vitest.config.ts @@ -0,0 +1,26 @@ +import path from "node:path"; +import { mergeConfig } from "vitest/config"; +import { nodeVitestConfig } from "@repo/core-typescript/vitest.base.node"; + +// Coverage excludes mirror the app-tier pattern (see apps/cms and +// apps/web-next vitest configs): bootstrap glue is excluded, thresholds +// stay inherited from the shared base — never lowered here. +export default mergeConfig(nodeVitestConfig, { + test: { + // Integration suites spawn a real runner process, run real `git + // daemon` clones, and (story task 3) a real package-manager install + // of the vite-kitchen fixture — minutes-scale on a cold cache. + testTimeout: 240_000, + hookTimeout: 60_000, + coverage: { + exclude: [ + // Process bootstrap: env → server → signal handlers. Exercised + // end-to-end by the spawn integration suite (tests/), which runs + // it in a child process where v8 in-process coverage cannot see + // it. All logic it calls (config, server) is unit-covered. + "src/main.ts", + ], + }, + }, + resolve: { alias: { "@": path.resolve(__dirname, "./src") } }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index dc87227..2f21fd1 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -117,6 +117,46 @@ importers: specifier: ^3.0.0 version: 3.2.4(@types/debug@4.1.13)(@types/node@22.19.17)(happy-dom@20.8.9)(jiti@2.7.0)(jsdom@25.0.1)(lightningcss@1.32.0)(sass@1.99.0)(terser@5.46.2)(tsx@4.21.0)(yaml@2.9.0) + apps/runner: + dependencies: + "@repo/core-runner-protocol": + specifier: workspace:* + version: link:../../packages/core-runner-protocol + ws: + specifier: ^8.18.0 + version: 8.21.0 + zod: + specifier: ^3.25.0 + version: 3.25.76 + devDependencies: + "@repo/core-eslint": + specifier: workspace:* + version: link:../../packages/core-eslint + "@repo/core-testing": + specifier: workspace:* + version: link:../../packages/core-testing + "@repo/core-typescript": + specifier: workspace:* + version: link:../../packages/core-typescript + "@types/node": + specifier: ^22.0.0 + version: 22.19.17 + "@types/ws": + specifier: ^8.18.0 + version: 8.18.1 + "@vitest/coverage-v8": + specifier: ^3.2.4 + version: 3.2.4(vitest@3.2.4(@types/debug@4.1.13)(@types/node@22.19.17)(happy-dom@20.8.9)(jiti@2.7.0)(jsdom@25.0.1)(lightningcss@1.32.0)(sass@1.99.0)(terser@5.46.2)(tsx@4.21.0)(yaml@2.9.0)) + tsx: + specifier: ^4.0.0 + version: 4.21.0 + typescript: + specifier: ^5.8.0 + version: 5.9.3 + vitest: + specifier: ^3.0.0 + version: 3.2.4(@types/debug@4.1.13)(@types/node@22.19.17)(happy-dom@20.8.9)(jiti@2.7.0)(jsdom@25.0.1)(lightningcss@1.32.0)(sass@1.99.0)(terser@5.46.2)(tsx@4.21.0)(yaml@2.9.0) + apps/storybook: devDependencies: "@playwright/test": @@ -15019,7 +15059,7 @@ snapshots: "@parcel/watcher": 2.5.6 effect: 3.21.2 multipasta: 0.2.7 - ws: 8.20.0 + ws: 8.21.0 transitivePeerDependencies: - bufferutil - utf-8-validate @@ -15034,7 +15074,7 @@ snapshots: effect: 3.21.2 mime: 3.0.0 undici: 7.24.4 - ws: 8.20.0 + ws: 8.21.0 transitivePeerDependencies: - bufferutil - utf-8-validate @@ -18187,7 +18227,7 @@ snapshots: recast: 0.23.11 semver: 7.7.4 util: 0.12.5 - ws: 8.20.0 + ws: 8.21.0 optionalDependencies: prettier: 3.8.1 transitivePeerDependencies: @@ -20595,7 +20635,7 @@ snapshots: "@types/ws": 8.18.1 entities: 7.0.1 whatwg-mimetype: 3.0.0 - ws: 8.20.0 + ws: 8.21.0 transitivePeerDependencies: - bufferutil - utf-8-validate @@ -21473,7 +21513,7 @@ snapshots: whatwg-encoding: 3.1.1 whatwg-mimetype: 4.0.0 whatwg-url: 14.2.0 - ws: 8.20.0 + ws: 8.21.0 xml-name-validator: 5.0.0 transitivePeerDependencies: - bufferutil