feat: scale-out phase A — instance ids, the schema, leases, the run protocol, a multi-server e2e harness

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- **Scale-out, phase A: the foundation for API pods that share one database and runs that outlive them** (`docs/plans/scale-out.md`). Every API process has an id (`OPTIO_INSTANCE_ID`, the pod's name from the chart, plus a suffix new at every boot; `GET /api/health` says which instance answered, to a signed-in request only). One migration adds every table and column the plan's phases build on: the attachment columns on runs (`exec_state`, `exec_pid`, `consumed_bytes`, `attached_by`, `attach_lease_until`, on tasks, Job runs, PR-review runs and agent turns), `leases`, `ws_upgrade_tokens`, `inbound_webhook_deliveries`, `glance_state`, `installed_skill_files`, `ticket_sync_claims`, and one active PR review per PR URL. `services/lease-service.ts` is the one way an instance claims something for a while. The **run protocol** (`packages/container-runtime/src/run-protocol.ts`) is on every container runtime — `startRun`, `attachRun`, `deliverStdin`, `killRun`: a start script launches the agent under a supervisor in its own session in the run's home, with stdin fed from a file through a FIFO until the `__OPTIO_STDIN_EOF__` line, its output appended to `output.ndjson` and its exit code to `exit`; the supervisor writes its own pid file (`<pid> <starttime>`) once its TERM trap is in place, and attach, kill and the pooled start guard trust a pid only when the live process's start time matches it; stdin bytes travel over the exec's stdin (`head -c <n>`), never in a script or its environment; an attach is `tail --pid` from a byte offset; the exec that started a run can go away and any instance can attach later. `tini` is the pods' pid 1 (`images/base.Dockerfile`, every init script), so finished runs are reaped — **the agent images must be rebuilt for the protocol**. The repo-cleanup worker removes the run homes of finished runs from every pod (`run-home-sweep-service.ts`), the backstop for the attached worker that removes them once a run is terminal and consumed (phase B). `buildTaskStartScript` / `buildPooledStartScript` build the start scripts beside the exec scripts the workers still use (they switch in phase B). The fake runtime keeps its pods on disk (`OPTIO_FAKE_RUNTIME_DIR`) and plays a protocol run as a detached process (`fake-agent.mjs`, every event with a `seq`), and `test-utils/e2e/api-cluster.ts` boots N real API servers on one database for the e2e tier: two boot together, one is SIGKILLed, a third joins.
+
 - **Android: every trigger type in the automations and agent trigger sheets.** A Local automation's Add trigger sheet and a persistent agent's New trigger sheet take all fourteen trigger types — GitLab, Jira, Pylon, PagerDuty, Sentry, Alertmanager and Datadog events included, with their event kinds, identity and filters — through the same rows as the New work form (one trigger editor in `core:ui`, `triggers/`). A Pylon / Alertmanager / Datadog trigger's own URL and shared secret are shown once, with copy buttons, right after it is created, in the sheets and in the New work form.
 - **Every agent runtime gets the work's MCP servers.** Connections' tools and MCP servers reached only Claude Code (`.mcp.json`) and Codex (`config.toml`); a Job on Gemini, a persistent agent on Copilot, or a Task on OpenCode or Cursor had the connections' credentials and notes but no tools. Each runtime now gets the same servers in the file it reads, established from the versions the agent image installs: Gemini CLI in the user `settings.json` of a `GEMINI_CLI_HOME` of the run's own (the adapter's settings merged in, each server `trust: true`), OpenCode in an `OPENCODE_CONFIG` file merged before the project's own, GitHub Copilot CLI through `--additional-mcp-config @<file>` (`type: "local"`, every tool enabled), and Cursor in the project's `.cursor/mcp.json` with `--approve-mcps`. See `docs/connections.md` → "Which runtime reads which file".
 - The per-run files live in one **run home**, `/home/agent/optio/runs/<run id>` (`OPTIO_RUN_HOME`), named after the task, Job run, or agent turn — so a retry lands in the same place — and removed when the run's script exits and again when the task's worktree is cleaned up. Codex's per-run `CODEX_HOME` moves there too; it used to be a random directory that was never removed from the repo pod's home volume.

apps/api/e2e/api-cluster.e2e.test.ts

@@ -0,0 +1,117 @@
+/**
+ * E2E: two real API servers on one database (the scale-out harness,
+ * src/test-utils/e2e/api-cluster.ts). Both boot (migrations under the
+ * advisory lock, workers registered twice), work made through one is read
+ * through the other, a run completes with both taking BullMQ jobs, and one
+ * server's SIGKILL leaves the other serving and starting new runs.
+ *
+ * The runs here still go through the classic exec path (the workers move to
+ * the run protocol in phase B); the fake runtime keeps its pods on disk
+ * under the cluster's directory, which both servers read.
+ */
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { existsSync, readdirSync } from "node:fs";
+import { join } from "node:path";
+import { startApiCluster, type ApiCluster } from "../src/test-utils/e2e/api-cluster.js";
+import { waitFor } from "../src/test-utils/e2e/api-server.js";
+
+let cluster: ApiCluster;
+
+beforeAll(async () => {
+  cluster = await startApiCluster({ size: 2 });
+}, 240_000);
+
+afterAll(async () => {
+  await cluster?.stop();
+});
+
+interface RunBody {
+  run: { id: string; state: string; costUsd: string | null; errorMessage: string | null };
+}
+
+async function createJob(i: number, name: string): Promise<string> {
+  const { status, body } = await cluster.client(i)<{ workflow: { id: string } }>("/api/jobs", {
+    method: "POST",
+    body: JSON.stringify({
+      name,
+      promptTemplate: "Say hello from the cluster",
+      agentRuntime: "claude-code",
+    }),
+  });
+  expect(status).toBe(201);
+  return body.workflow.id;
+}
+
+async function startRun(i: number, workflowId: string): Promise<string> {
+  const { status, body } = await cluster.client(i)<RunBody>(`/api/jobs/${workflowId}/runs`, {
+    method: "POST",
+    body: JSON.stringify({}),
+  });
+  expect(status).toBe(201);
+  return body.run.id;
+}
+
+async function waitForRun(i: number, runId: string, states: string[]): Promise<RunBody["run"]> {
+  return waitFor(
+    async () => {
+      const { body } = await cluster.client(i)<RunBody>(`/api/workflow-runs/${runId}`);
+      return states.includes(body.run.state) ? body.run : null;
+    },
+    { timeoutMs: 90_000, label: `run ${runId} → ${states.join("|")}` },
+  );
+}
+
+describe("api cluster e2e", () => {
+  it("both servers are healthy; the public probe names no instance", async () => {
+    const [a, b] = await Promise.all([
+      cluster.client(0)<{ healthy: boolean; instanceId?: string }>("/api/health"),
+      cluster.client(1)<{ healthy: boolean; instanceId?: string }>("/api/health"),
+    ]);
+    expect(a.body.healthy).toBe(true);
+    expect(b.body.healthy).toBe(true);
+    // Auth is disabled here, so no request is signed in: nothing identifying.
+    expect(a.body).not.toHaveProperty("instanceId");
+    expect(b.body).not.toHaveProperty("instanceId");
+    expect(cluster.servers.map((s) => s.name)).toEqual(["api-0", "api-1"]);
+  });
+
+  it("a Job created through A is read through B, and its run completes", async () => {
+    const workflowId = await createJob(0, "cluster job");
+    const { status, body } = await cluster.client(1)<{ workflow: { id: string; name: string } }>(
+      `/api/jobs/${workflowId}`,
+    );
+    expect(status).toBe(200);
+    expect(body.workflow.name).toBe("cluster job");
+
+    const runId = await startRun(0, workflowId);
+    const run = await waitForRun(1, runId, ["completed", "failed"]);
+    expect(run.state).toBe("completed");
+    expect(run.costUsd).toBe("0.0123");
+    // The fake kept its pod on disk, where both servers look.
+    expect(existsSync(join(cluster.fakeRuntimeDir, "containers"))).toBe(true);
+    expect(readdirSync(join(cluster.fakeRuntimeDir, "containers")).length).toBeGreaterThan(0);
+  });
+
+  it("after A is SIGKILLed, B keeps serving and runs new work", async () => {
+    await cluster.kill(0, "SIGKILL");
+    expect(cluster.live().map((s) => s.name)).toEqual(["api-1"]);
+    await expect(cluster.client(0)("/api/health")).rejects.toThrow(/killed/);
+
+    const workflowId = await createJob(1, "after the kill");
+    const runId = await startRun(1, workflowId);
+    const run = await waitForRun(1, runId, ["completed", "failed"]);
+    expect(run.state).toBe("completed");
+
+    const { body } = await cluster.any()<{ healthy: boolean }>("/api/health");
+    expect(body.healthy).toBe(true);
+  });
+
+  it("a third server joins the same database and work", async () => {
+    const c = await cluster.add();
+    expect(c.name).toBe("api-2");
+    const { body } = await cluster.client(2)<{ workflows: Array<{ name: string }> }>("/api/jobs");
+    expect(body.workflows.map((w) => w.name)).toEqual(
+      expect.arrayContaining(["cluster job", "after the kill"]),
+    );
+  }, 150_000);
+});

apps/api/src/db/migrate-safe.test.ts

@@ -0,0 +1,59 @@
+import { describe, expect, it } from "vitest";
+import { viewSourceColumns, withoutColumns } from "./migrate-safe.js";
+import { RUN_VIEWS_SQL } from "./run-views.js";
+
+const VIEW = `CREATE VIEW "workflow_runs" AS
+     SELECT "id", "kind", "work_id" AS "workflow_id", "trigger_id",
+       "recovery_required", "exec_state", "consumed_bytes",
+       "created_at", "updated_at"
+     FROM "tasks" WHERE "kind" = 'standalone'
+     WITH CASCADED CHECK OPTION`;
+
+describe("run views between historical migrations", () => {
+  it("knows the source columns a view selects, aliases included as their source", () => {
+    expect(viewSourceColumns(VIEW)).toEqual([
+      "id",
+      "kind",
+      "work_id",
+      "trigger_id",
+      "recovery_required",
+      "exec_state",
+      "consumed_bytes",
+      "created_at",
+      "updated_at",
+    ]);
+    expect(
+      viewSourceColumns(`CREATE VIEW "repo_tasks" AS SELECT * FROM "tasks" WHERE "kind" = 'repo'`),
+    ).toEqual([]);
+    expect(
+      viewSourceColumns(`ALTER VIEW "workflow_runs" ALTER COLUMN "kind" SET DEFAULT 'standalone'`),
+    ).toEqual([]);
+  });
+
+  it("leaves out the columns a later migration adds, keeping every other item intact", () => {
+    const out = withoutColumns(VIEW, ["exec_state", "consumed_bytes", "recovery_required"]);
+    expect(out).toContain(
+      `SELECT "id", "kind", "work_id" AS "workflow_id", "trigger_id", "created_at", "updated_at" FROM "tasks" WHERE "kind" = 'standalone'`,
+    );
+    expect(out).toContain("WITH CASCADED CHECK OPTION");
+    expect(withoutColumns(VIEW, [])).toBe(VIEW);
+    expect(withoutColumns(`SELECT * FROM "tasks"`, ["exec_state"])).toBe(`SELECT * FROM "tasks"`);
+  });
+
+  it("the real views' later columns need no list: each is derived from the SQL", () => {
+    const columns = RUN_VIEWS_SQL.flatMap(viewSourceColumns);
+    for (const c of [
+      "exec_state",
+      "exec_pid",
+      "consumed_bytes",
+      "attached_by",
+      "attach_lease_until",
+      "recovery_required",
+    ]) {
+      expect(columns).toContain(c);
+    }
+    const stripped = RUN_VIEWS_SQL.map((s) => withoutColumns(s, ["attach_lease_until"]));
+    expect(stripped.join("\n")).not.toContain("attach_lease_until");
+    expect(stripped.join("\n")).toContain('"attached_by"');
+  });
+});

apps/api/src/db/migrate-safe.ts

@@ -33,6 +33,67 @@ interface MigrationEntry {
 
 const ADVISORY_LOCK_ID = 8_675_309; // arbitrary, unique to optio migrations
 
+/** The items of a view statement's select list, trimmed. */
+function selectList(stmt: string): { list: string[]; from: string } | null {
+  const m = stmt.match(/SELECT([\s\S]*?)(\sFROM\s)/i);
+  if (!m) return null;
+  return {
+    list: m[1]
+      .split(",")
+      .map((s) => s.trim())
+      .filter(Boolean),
+    from: m[2],
+  };
+}
+
+/** The source column of a select item (`"col"` or `"col" AS "alias"`), or null (`*`, expressions). */
+function sourceColumn(item: string): string | null {
+  const m = item.match(/^"([^"]+)"(?:\s+AS\s+"[^"]+")?$/i);
+  return m ? m[1] : null;
+}
+
+/** The `tasks` columns a view statement selects (aliases are not columns). */
+export function viewSourceColumns(stmt: string): string[] {
+  const parsed = selectList(stmt);
+  if (!parsed) return [];
+  const names = new Set<string>();
+  for (const item of parsed.list) {
+    const col = sourceColumn(item);
+    if (col) names.add(col);
+  }
+  return [...names];
+}
+
+/**
+ * The view statement without the given source columns. A fresh install
+ * remakes the views between every historical migration, before a later
+ * migration has added a column the views select: the column is left out
+ * until it exists, and nothing has to list which columns those are.
+ */
+export function withoutColumns(stmt: string, missing: string[]): string {
+  if (missing.length === 0) return stmt;
+  const parsed = selectList(stmt);
+  if (!parsed) return stmt;
+  const kept = parsed.list.filter((item) => {
+    const col = sourceColumn(item);
+    return !col || !missing.includes(col);
+  });
+  return stmt.replace(
+    /SELECT([\s\S]*?)(\sFROM\s)/i,
+    () => `SELECT ${kept.join(", ")}${parsed.from}`,
+  );
+}
+
+/** The view columns the `tasks` table does not have yet. */
+async function missingRunViewColumns(tx: { execute: Database["execute"] }): Promise<string[]> {
+  const rows = await tx.execute<{ column_name: string }>(sql`
+    SELECT column_name FROM information_schema.columns
+    WHERE table_schema = current_schema() AND table_name = 'tasks'`);
+  const present = new Set(rows.map((r) => r.column_name));
+  const wanted = new Set(RUN_VIEWS_SQL.flatMap(viewSourceColumns));
+  return [...wanted].filter((c) => !present.has(c));
+}
+
 /** Drop the run views if they are views (before the runs table, `workflow_runs` was a table). */
 const DROP_RUN_VIEWS = sql.raw(`DO $$ BEGIN
   IF EXISTS (SELECT 1 FROM pg_views WHERE schemaname = current_schema() AND viewname = 'workflow_runs') THEN
@@ -100,13 +161,10 @@ export async function migrateSafe(db: Database, migrationsFolder: string): Promi
           await tx.execute(DROP_RUN_VIEWS);
           // A fresh install passes through historical migrations before new
           // run columns exist. Keep the intermediate views compatible until
-          // the migration adding the column has run.
-          const [{ present }] = await tx.execute<{ present: boolean }>(sql`
-            SELECT EXISTS (SELECT 1 FROM information_schema.columns
-              WHERE table_schema = current_schema() AND table_name = 'tasks'
-              AND column_name = 'recovery_required') AS present`);
+          // the migration adding each column has run.
+          const missing = await missingRunViewColumns(tx);
           for (const stmt of RUN_VIEWS_SQL) {
-            await tx.execute(sql.raw(present ? stmt : stmt.replace('"recovery_required", ', "")));
+            await tx.execute(sql.raw(withoutColumns(stmt, missing)));
           }
         }
         await tx.execute(