diff --git a/.github/workflows/pr-trusted.yml b/.github/workflows/pr-trusted.yml index d870c77d9eb..682fc9762f3 100644 --- a/.github/workflows/pr-trusted.yml +++ b/.github/workflows/pr-trusted.yml @@ -354,26 +354,9 @@ jobs: fail-fast: false matrix: include: - # The server suite is pinned to maxWorkers=1 (server/vitest.config.ts), - # so it can only be parallelized across runners. Shard it to keep this - # lane off the PR critical path. The suite has grown to ~3184s of - # serial vitest wall time (mean of runs 35036001734 and 35024948947, - # 2026-09-15): at five plain general-server shards with stale - # recorded durations the worst shard ran 806s of tests while the - # best ran 417s. This now uses the same shape as release-verify.yml: - # the ~429s chat integration suite splits by collected test location - # across three dedicated runners (~143s each), and the remaining - # ~2755s levels across twelve duration-balanced shards at ~230s - # each, in line with the other ~200-290s lanes. - # - # The balance only holds while the manifest is fresh. By 2026-10-10 - # the without-chat group had grown to 871 suites and ~5031s of serial - # vitest time (mean of runs 38076844891 and 38076972444), 230 of - # them unrecorded and weighted at the median, so the twelve shards - # ran 313-554s of tests (server 2/12 at 618s was the slowest check - # in the run). Re-recorded durations level them to ~415s each; the - # shard count stays at twelve because Build and workspaces-b sit - # above that already. + # Keep twelve duration-balanced server shards. Each shard overlaps + # at most two isolated fork workers; cases within a file stay serial. + # The chat fixture keeps three independent database-backed shards. - group: general-server-without-chat group_label: server (1/12) shard_index: 0 diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 09e8038a57b..6fb6f0f6643 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -75,6 +75,16 @@ When changing the workflow path or branch, authorize the new reference before updating the caller. Retain older authorized SHA references while queued runs or supported reruns still use them. +General server shards run at most two test files at a time in separate Vitest +forks. Each file retains its own home, config and process state, and its test +cases run serially. An explicit `PAPERCLIP_TEST_DATABASE_URL` keeps those shards +serial because the database is shared. Unsharded local tests, route/authz lanes, +and native Runner lanes retain their existing concurrency. Sharded chat jobs run +two independent partitions, each with its own embedded PostgreSQL, home and +config; an explicit shared test database retains one partition per job. +The Runner transport lease-expiry and cold-admission regressions live in their +own file so those real-clock waits overlap other files in the existing pool. + For a full wall-clock profile of all main PR CI jobs, add `profile-ci` to the PR. See [CI profiling](ci-profiling.md) for the Honeycomb waterfall, coverage, operator setup, and activation requirements. diff --git a/doc/ci-profiling.md b/doc/ci-profiling.md index fe8472a7577..49aac6a4ebc 100644 --- a/doc/ci-profiling.md +++ b/doc/ci-profiling.md @@ -162,7 +162,7 @@ readiness fixtures, `profile-ci` also runs **CI work benchmark**. It uses one GitHub-hosted runner without secrets or AWS access. It compares the unchanged PR base and candidate sequentially. It records complete command times, test inventories, revisions, and failure logs in the `ci-work-benchmark` artifact. -The serialized shard must retain the same cases. The fixture comparison retains +The complete general-server shard must retain the same cases. The fixture comparison retains the database contract and splits large Git scanning from bounded staging. These command timings exclude queue time and setup; use the full Honeycomb waterfall to assess the complete CI critical path. diff --git a/packages/adapter-utils/src/workspace-snapshot-thread.test.ts b/packages/adapter-utils/src/workspace-snapshot-thread.test.ts index 5830b7039aa..53aaba4b877 100644 --- a/packages/adapter-utils/src/workspace-snapshot-thread.test.ts +++ b/packages/adapter-utils/src/workspace-snapshot-thread.test.ts @@ -81,6 +81,9 @@ describe("workspace snapshot threads", () => { await expect(fs.stat(created[0]!)).rejects.toMatchObject({ code: "ENOENT" }); }); + // Multiple real TypeScript worker cold starts can exceed the default 5s on + // CI (three queued workers use two slots). These cases verify cleanup and + // error propagation, not a worker startup deadline. Keep completion bounded. it("preserves filesystem errors, cleans failed workers and releases queued jobs", async () => { const root = await fixture(); const file = path.join(root, "file"); @@ -103,7 +106,7 @@ describe("workspace snapshot threads", () => { if (results[2].status === "fulfilled") await disposeDirectorySnapshot(results[2].value); expect(created).toHaveLength(3); for (const directory of created) await expect(fs.stat(directory)).rejects.toMatchObject({ code: "ENOENT" }); - }); + }, 15_000); it.each(["unchanged", "changed", "sparse"] as const)("isolates restore captures and cleans their manifests: %s", async mode => { const target = await fixture(), source = await fixture(); @@ -129,7 +132,7 @@ describe("workspace snapshot threads", () => { expect(await fs.readFile(path.join(target, "file"), "utf8")).toBe(mode === "unchanged" ? "before" : "after"); expect(await fs.readFile(path.join(target, "concurrent"), "utf8")).toBe("host bytes"); for (const directory of created) await expect(fs.stat(directory)).rejects.toMatchObject({ code: "ENOENT" }); - }); + }, 15_000); it("retains the in-memory snapshot contract when disk backing is not requested", async () => { const root = await fixture(); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts new file mode 100644 index 00000000000..1f0fbaef19f --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-deadlines.test.ts @@ -0,0 +1,152 @@ +import { chmod, cp, mkdir, mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises"; +import { readFileSync } from "node:fs"; +import { createHash } from "node:crypto"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, it, vi } from "vitest"; +import { QUALIFIED_ACPX_PROFILES } from "../drivers/acpx/qualified-profiles.js"; +import { codexSemanticToolSpecs } from "../drivers/codex/codex-app-server-driver.js"; +import { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js"; +import * as durableControlPlane from "../control-plane/durable-prp-control-plane.js"; +import { + createCapabilityRunnerdCodexTransport, + defaultCapabilityRunnerdBinary as qualifiedCapabilityRunnerdBinary, +} from "./runnerd-codex-transport.js"; + +// These regressions must cross real production timing boundaries. Keep their +// waits intact, but let the existing file worker pool overlap them with the +// rest of the transport suite instead of extending its serial critical path. +const defaultCapabilityRunnerdBinary = () => + process.env.PAPERCLIP_ATTACH_TRANSITION_RUNNER ?? qualifiedCapabilityRunnerdBinary(); +const fakeCodex = resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"); +function fakeCodexArgs(stateDirectory: string, ...args: string[]): string[] { + return ["--state-file", join(stateDirectory, "fake-codex-state.json"), ...args]; +} + +it("keeps a quiet active Codex turn in the same process across connection lease expiry", async () => { + const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-renew-active-")); + const callsPath = join(stateDirectory, "calls.log"); + const cores: DurablePrpControlPlane[] = []; + const OriginalCore = durableControlPlane.DurablePrpControlPlane; + const coreSpy = vi.spyOn(durableControlPlane, "DurablePrpControlPlane") + .mockImplementation(function(options: ConstructorParameters[0]) { + const core = new OriginalCore({ ...options, connectionLeaseTtlMs: 60_000 }); + cores.push(core); + return core; + } as unknown as typeof OriginalCore); + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(stateDirectory, "--hold-turn", "--record-process-start", "--call-log", callsPath), + stateDirectory, runnerReconnectGraceMs: 5_000, + }); + try { + const opened = await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Keep working quietly." }] }); + const runnerPid = bundle.evidence().runnerPid; + const codexPid = bundle.evidence().codexPid; + expect(runnerPid).toBeGreaterThan(0); + expect(codexPid).toBeGreaterThan(0); + const core = cores[0]!; + const original = structuredClone(Object.values(core.store.state.leases)[0]!); + await vi.waitFor(() => { + expect(Date.now()).toBeGreaterThan(original.expiresAtUnixMs + 1_000); + }, { timeout: 75_000, interval: 1_000 }); + const current = Object.values(core.store.state.leases)[0]!; + expect(current.expiresAtUnixMs).toBeGreaterThan(original.expiresAtUnixMs); + expect(current.leaseId).toBe(original.leaseId); + expect(core.store.state.connectionCount).toBe(1); + expect(bundle.evidence().runnerPid).toBe(runnerPid); + expect(bundle.evidence().codexPid).toBe(codexPid); + process.kill(runnerPid!, 0); + process.kill(codexPid!, 0); + const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/); + expect(calls.filter(call => call === "process-start")).toHaveLength(1); + expect(calls.filter(call => call === "thread/start")).toHaveLength(1); + expect(calls.filter(call => call === "turn/start")).toHaveLength(1); + expect(calls).not.toContain("turn/interrupt"); + expect(calls).not.toContain("thread/resume"); + const state = JSON.parse(await readFile(join(stateDirectory, "fake-codex-state.json"), "utf8")); + expect(state.threadId).toBe(opened.thread.id); + expect(state.activeTurnId).toBe("provider-turn-1"); + } finally { + await bundle.transport.close(); + coreSpy.mockRestore(); + await rm(stateDirectory, { recursive: true, force: true }); + } +}, 90_000); + +it("reopens Pi after a completed turn within its cold admission budget during warm attachment", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-pi-warm-admission-")); + const providerNode = process.platform === "linux" ? join(root, "node") : process.execPath; + if (process.platform === "linux") { + await cp(process.execPath, providerNode, { dereference: true }); + await chmod(providerNode, 0o500); + expect((await stat(providerNode)).mode & 0o777).toBe(0o500); + } + const cliRoot = join(root, "dist/cli"); + await mkdir(cliRoot, { recursive: true, mode: 0o700 }); + const sidecarPath = join(cliRoot, "acpx-runtime-sidecar.cjs"); + const journal = join(root, "commands.ndjson"); + const fixture = await readFile(fileURLToPath(new URL("./fixtures/fake-pi-warm-sidecar.cjs", import.meta.url)), "utf8"); + await writeFile(sidecarPath, fixture.replace("/* fixture-config */ null", JSON.stringify({ journal, resumeDelayMs: 32_000, commandDigest: QUALIFIED_ACPX_PROFILES.pi.commandDigest })), { mode: 0o600 }); + const digest = (path: string) => `sha256:${createHash("sha256").update(readFileSync(path)).digest("hex")}`; + const bundle = createCapabilityRunnerdCodexTransport({ + provider: "acpx", acpxAgent: "pi", piThinkingLevel: "low", acpxPermissionMode: "deny-all", + runnerBinary: defaultCapabilityRunnerdBinary(), stateDirectory: root, runnerFilesystemRoot: root, + providerNodeCommand: providerNode, providerNodeCommandSha256: digest(providerNode), + acpxSidecarPath: sidecarPath, acpxSidecarSha256: digest(sidecarPath), + providerPackAuthorityDigest: `sha256:${"d".repeat(64)}`, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, + environment: { PATH: "/usr/bin:/bin" }, + }); + bundle.transport.setServerRequestHandler(async () => ({ + success: true, + contentItems: [{ type: "inputText", text: "Completion report accepted." }], + })); + let failure: unknown; + try { + await bundle.transport.request("initialize", {}); + await bundle.transport.request("thread/start", { + cwd: root, model: "openrouter/deepseek/deepseek-v4-flash-0731", + dynamicTools: codexSemanticToolSpecs(), + completionContract: { revision: "warm-pi-contract", criterionIds: [] }, + }); + const runnerPid = bundle.evidence().runnerPid; + const notifications = bundle.transport.notifications()[Symbol.asyncIterator](); + const completeTurn = async () => { + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete the fixture turn." }] }); + for (let index = 0; index < 64; index += 1) { + const next = await Promise.race([ + notifications.next(), + new Promise((_, reject) => setTimeout(() => reject(new Error("Fixture completion timed out")), 5_000)), + ]); + if (next.value?.method === "turn/completed") return; + } + throw new Error("Fixture turn did not complete"); + }; + await completeTurn(); + const started = Date.now(); + await bundle.transport.attachRun!({ runId: "run-pi-warm-second", turnId: "turn-pi-warm-second", itemId: "item-pi-warm-second" }); + expect(Date.now() - started).toBeGreaterThan(30_000); + await completeTurn(); + expect(bundle.evidence()).toMatchObject({ runnerPid, runnerExited: false }); + const commands = (await readFile(journal, "utf8")).trim().split("\n").map((line) => JSON.parse(line)); + expect(commands.filter((entry) => entry.event === "spawn")).toHaveLength(2); + expect(commands.filter((entry) => entry.command === "session.open")).toHaveLength(2); + expect(commands.filter((entry) => entry.resumed)).toHaveLength(1); + expect(commands.filter((entry) => entry.command === "turn.start")).toHaveLength(2); + expect(commands.filter((entry) => entry.command === "session.close")).toHaveLength(1); + } catch (error) { + failure = error; + throw error; + } finally { + try { + await bundle.transport.close(); + } catch (cleanupError) { + if (failure) throw new AggregateError([failure, cleanupError], "Warm attachment and strict cleanup failed"); + throw cleanupError; + } + await rm(root, { recursive: true, force: true }); + } +}, 75_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index a67cd97afd5..4491e8b1fc3 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -1,4 +1,3 @@ -import { QUALIFIED_ACPX_PROFILES } from "../drivers/acpx/qualified-profiles.js"; import { coldAdmissionTimeoutMs, waitForRunnerCommand } from "./runnerd-codex-transport.js"; import { chmod, @@ -3341,58 +3340,6 @@ it("expands coalesced canonical items without dropping strict bindings", () => { ]); }); -it("keeps a quiet active Codex turn in the same process across connection lease expiry", async () => { - const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-renew-active-")); - const callsPath = join(stateDirectory, "calls.log"); - const cores: DurablePrpControlPlane[] = []; - const OriginalCore = durableControlPlane.DurablePrpControlPlane; - const coreSpy = vi.spyOn(durableControlPlane, "DurablePrpControlPlane") - .mockImplementation(function(options: ConstructorParameters[0]) { - const core = new OriginalCore({ ...options, connectionLeaseTtlMs: 60_000 }); - cores.push(core); - return core; - } as unknown as typeof OriginalCore); - const bundle = createCapabilityRunnerdCodexTransport({ - runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, - codexArgs: fakeCodexArgs(stateDirectory, "--hold-turn", "--record-process-start", "--call-log", callsPath), - stateDirectory, runnerReconnectGraceMs: 5_000, - }); - try { - const opened = await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }); - await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Keep working quietly." }] }); - const runnerPid = bundle.evidence().runnerPid; - const codexPid = bundle.evidence().codexPid; - expect(runnerPid).toBeGreaterThan(0); - expect(codexPid).toBeGreaterThan(0); - const core = cores[0]!; - const original = structuredClone(Object.values(core.store.state.leases)[0]!); - await vi.waitFor(() => { - expect(Date.now()).toBeGreaterThan(original.expiresAtUnixMs + 1_000); - }, { timeout: 75_000, interval: 1_000 }); - const current = Object.values(core.store.state.leases)[0]!; - expect(current.expiresAtUnixMs).toBeGreaterThan(original.expiresAtUnixMs); - expect(current.leaseId).toBe(original.leaseId); - expect(core.store.state.connectionCount).toBe(1); - expect(bundle.evidence().runnerPid).toBe(runnerPid); - expect(bundle.evidence().codexPid).toBe(codexPid); - process.kill(runnerPid!, 0); - process.kill(codexPid!, 0); - const calls = (await readFile(callsPath, "utf8")).trim().split(/\r?\n/); - expect(calls.filter(call => call === "process-start")).toHaveLength(1); - expect(calls.filter(call => call === "thread/start")).toHaveLength(1); - expect(calls.filter(call => call === "turn/start")).toHaveLength(1); - expect(calls).not.toContain("turn/interrupt"); - expect(calls).not.toContain("thread/resume"); - const state = JSON.parse(await readFile(join(stateDirectory, "fake-codex-state.json"), "utf8")); - expect(state.threadId).toBe(opened.thread.id); - expect(state.activeTurnId).toBe("provider-turn-1"); - } finally { - await bundle.transport.close(); - coreSpy.mockRestore(); - await rm(stateDirectory, { recursive: true, force: true }); - } -}, 90_000); - it("runs the lab provider boundary through authenticated durable PRP", async () => { const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-lab-provider-")); const bundle = createCapabilityRunnerdCodexTransport({ @@ -7005,81 +6952,6 @@ it("rotates PRP authority in place for a warm cross-run attachment", async () => } }, 30_000); -it("reopens Pi after a completed turn within its cold admission budget during warm attachment", async () => { - const root = await mkdtemp(join(tmpdir(), "runnerd-pi-warm-admission-")); - const providerNode = process.platform === "linux" ? join(root, "node") : process.execPath; - if (process.platform === "linux") { - await cp(process.execPath, providerNode, { dereference: true }); - await chmod(providerNode, 0o500); - expect((await stat(providerNode)).mode & 0o777).toBe(0o500); - } - const cliRoot = join(root, "dist/cli"); - await mkdir(cliRoot, { recursive: true, mode: 0o700 }); - const sidecarPath = join(cliRoot, "acpx-runtime-sidecar.cjs"); - const journal = join(root, "commands.ndjson"); - const fixture = await readFile(fileURLToPath(new URL("./fixtures/fake-pi-warm-sidecar.cjs", import.meta.url)), "utf8"); - await writeFile(sidecarPath, fixture.replace("/* fixture-config */ null", JSON.stringify({ journal, resumeDelayMs: 32_000, commandDigest: QUALIFIED_ACPX_PROFILES.pi.commandDigest })), { mode: 0o600 }); - const digest = (path: string) => `sha256:${createHash("sha256").update(readFileSync(path)).digest("hex")}`; - const bundle = createCapabilityRunnerdCodexTransport({ - provider: "acpx", acpxAgent: "pi", piThinkingLevel: "low", acpxPermissionMode: "deny-all", - runnerBinary: defaultCapabilityRunnerdBinary(), stateDirectory: root, runnerFilesystemRoot: root, - providerNodeCommand: providerNode, providerNodeCommandSha256: digest(providerNode), - acpxSidecarPath: sidecarPath, acpxSidecarSha256: digest(sidecarPath), - providerPackAuthorityDigest: `sha256:${"d".repeat(64)}`, - lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, - environment: { PATH: "/usr/bin:/bin" }, - }); - bundle.transport.setServerRequestHandler(async () => ({ - success: true, - contentItems: [{ type: "inputText", text: "Completion report accepted." }], - })); - let failure: unknown; - try { - await bundle.transport.request("initialize", {}); - await bundle.transport.request("thread/start", { - cwd: root, model: "openrouter/deepseek/deepseek-v4-flash-0731", - dynamicTools: codexSemanticToolSpecs(), - completionContract: { revision: "warm-pi-contract", criterionIds: [] }, - }); - const runnerPid = bundle.evidence().runnerPid; - const notifications = bundle.transport.notifications()[Symbol.asyncIterator](); - const completeTurn = async () => { - await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Complete the fixture turn." }] }); - for (let index = 0; index < 64; index += 1) { - const next = await Promise.race([ - notifications.next(), - new Promise((_, reject) => setTimeout(() => reject(new Error("Fixture completion timed out")), 5_000)), - ]); - if (next.value?.method === "turn/completed") return; - } - throw new Error("Fixture turn did not complete"); - }; - await completeTurn(); - const started = Date.now(); - await bundle.transport.attachRun!({ runId: "run-pi-warm-second", turnId: "turn-pi-warm-second", itemId: "item-pi-warm-second" }); - expect(Date.now() - started).toBeGreaterThan(30_000); - await completeTurn(); - expect(bundle.evidence()).toMatchObject({ runnerPid, runnerExited: false }); - const commands = (await readFile(journal, "utf8")).trim().split("\n").map((line) => JSON.parse(line)); - expect(commands.filter((entry) => entry.event === "spawn")).toHaveLength(2); - expect(commands.filter((entry) => entry.command === "session.open")).toHaveLength(2); - expect(commands.filter((entry) => entry.resumed)).toHaveLength(1); - expect(commands.filter((entry) => entry.command === "turn.start")).toHaveLength(2); - expect(commands.filter((entry) => entry.command === "session.close")).toHaveLength(1); - } catch (error) { - failure = error; - throw error; - } finally { - try { - await bundle.transport.close(); - } catch (cleanupError) { - if (failure) throw new AggregateError([failure, cleanupError], "Warm attachment and strict cleanup failed"); - throw cleanupError; - } - await rm(root, { recursive: true, force: true }); - } -}, 75_000); - it("waits for a warm runner to re-authenticate before probing attachment readiness", async () => { const stateDirectory = await mkdtemp( join(tmpdir(), "runnerd-warm-reattach-before-probe-"), diff --git a/scripts/__tests__/run-vitest-stable-shard.test.mjs b/scripts/__tests__/run-vitest-stable-shard.test.mjs index 9b92a9e174e..8c9ddedf969 100644 --- a/scripts/__tests__/run-vitest-stable-shard.test.mjs +++ b/scripts/__tests__/run-vitest-stable-shard.test.mjs @@ -1,6 +1,7 @@ import assert from "node:assert/strict"; import { spawnSync } from "node:child_process"; -import { readFileSync } from "node:fs"; +import { chmodSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import os from "node:os"; import path from "node:path"; import { fileURLToPath } from "node:url"; import test from "node:test"; @@ -519,3 +520,69 @@ test("filtered collection must match the exact assigned case identities, includi assert.throws(() => assertSelectedTests(expected, actual, lineShardFile)); } }); + + +test("parallel chat jobs form a complete disjoint partition", () => { + for (const count of [2, 3, 4]) { + const shards = Array.from({ length: count }, (_, index) => + dryRunJson(["--mode", "general", "--group", "general-chat", "--shard-index", String(index), "--shard-count", String(count)], + { PAPERCLIP_TEST_DATABASE_URL: "" }).chatVitestShards, + ); + assert.ok(shards.every((workers) => workers.length === 2)); + assert.deepEqual(shards.flat().map(({ index }) => index).sort((a, b) => a - b), + Array.from({ length: count * 2 }, (_, index) => index)); + assert.ok(shards.flat().every((shard) => shard.count === count * 2)); + } +}); + +test("shared-database and unsharded chat runs retain one serial partition", () => { + for (let index = 0; index < 3; index++) { + const result = dryRunJson(["--mode", "general", "--group", "general-chat", "--shard-index", String(index), "--shard-count", "3"], + { PAPERCLIP_TEST_DATABASE_URL: "postgresql://shared/test" }); + assert.deepEqual(result.chatVitestShards, [{ index, count: 3 }]); + } + assert.deepEqual(dryRunJson(["--mode", "general", "--group", "general-chat"], + { PAPERCLIP_TEST_DATABASE_URL: "" }).chatVitestShards, [{ index: 0, count: 1 }]); +}); + + +test("a failed chat child fails the job after its sibling finishes", { skip: process.platform === "win32" }, () => { + const directory = mkdtempSync(path.join(os.tmpdir(), "chat-launch-proof-")); + const file = path.join(repoRoot, "server/src/__tests__/chat-channels.integration.test.ts"); + const lines = readFileSync(file, "utf8").split("\n") + .flatMap((line, index) => /^\s*it\(/.test(line) ? [index + 1] : []).slice(0, 6); + assert.equal(lines.length, 6); + const tests = lines.map((line) => ({ file, projectName: "@paperclipai/server", + name: `launcher fixture ${line}`, location: { line } })); + const receipt = path.join(directory, "sibling-finished"); + const fakePnpm = path.join(directory, "pnpm"); + try { + writeFileSync(fakePnpm, `#!${process.execPath} + const fs = require('node:fs'); + const args = process.argv.slice(2); + const tests = ${JSON.stringify(tests)}; + const selected = args.flatMap(arg => { + const match = arg.match(/chat-channels\\.integration\\.test\\.ts:(\\d+)$/); + return match ? [Number(match[1])] : []; + }); + if (args.includes('list')) { + const output = args.find(arg => arg.startsWith('--json=')).slice(7); + fs.writeFileSync(output, JSON.stringify(selected.length + ? tests.filter(test => selected.includes(test.location.line)) : tests)); + } else if (selected.includes(${lines[0]})) process.exit(7); + else setTimeout(() => fs.writeFileSync(${JSON.stringify(receipt)}, 'finished'), 200); + `); + chmodSync(fakePnpm, 0o755); + const result = spawnSync(process.execPath, [script, "--mode", "general", "--group", "general-chat", + "--shard-index", "0", "--shard-count", "3"], { + cwd: repoRoot, encoding: "utf8", timeout: 30_000, + env: { ...process.env, PATH: `${directory}${path.delimiter}${process.env.PATH}`, + PAPERCLIP_TEST_DATABASE_URL: "", PAPERCLIP_CI_PROFILE: "false" }, + }); + assert.equal(result.status, 1, result.stdout + result.stderr); + assert.match(result.stderr, /failed \(7\)/); + assert.equal(readFileSync(receipt, "utf8"), "finished", "a failed worker must not abandon its sibling"); + } finally { + rmSync(directory, { recursive: true, force: true }); + } +}); diff --git a/scripts/__tests__/vitest-worker-isolation.test.mjs b/scripts/__tests__/vitest-worker-isolation.test.mjs index bd70f733df4..2c762728757 100644 --- a/scripts/__tests__/vitest-worker-isolation.test.mjs +++ b/scripts/__tests__/vitest-worker-isolation.test.mjs @@ -11,7 +11,7 @@ const vitest = path.join(root, 'node_modules/vitest/vitest.mjs'); const api = fileURLToPath(new URL('../../node_modules/vitest/dist/index.js', import.meta.url)); const setup = path.join(root, 'server/src/__tests__/setup-supertest.ts'); -test('one controller retains fresh fork, globals, environment, timers and home per serialized file', () => { +for (const parallel of [false, true]) test(`isolated forks retain fresh state with file parallelism ${parallel}`, () => { const directory = realpathSync(mkdtempSync(path.join(os.tmpdir(), 'pc-fork-proof-'))); try { const record = path.join(directory, 'workers.ndjson'); @@ -20,7 +20,7 @@ test('one controller retains fresh fork, globals, environment, timers and home p for (const file of files) writeFileSync(path.join(directory, file), ` import { expect, it, vi } from ${JSON.stringify(api)}; import { appendFileSync, existsSync, writeFileSync } from 'node:fs'; - it('starts clean and can contaminate only its own worker', () => { + it('starts clean and can contaminate only its own worker', async () => { expect(globalThis.serializedLeak).toBeUndefined(); expect(process.env.PAPERCLIP_SERIALIZED_LEAK).toBeUndefined(); expect(vi.isFakeTimers()).toBe(false); @@ -30,6 +30,16 @@ test('one controller retains fresh fork, globals, environment, timers and home p config: process.env.PAPERCLIP_CONFIG, instance: process.env.PAPERCLIP_INSTANCE_ID, }) + '\\n'); writeFileSync(process.env.PAPERCLIP_CONFIG, '{}'); + // In parallel mode both real workers must arrive before either exits. + // This fails if the controller silently falls back to serial execution. + if (${parallel}) { + const { readFileSync } = await import('node:fs'); + const deadline = Date.now() + 10_000; + while (readFileSync(${JSON.stringify(record)}, 'utf8').trim().split('\\n').length < 2) { + if (Date.now() > deadline) throw new Error('second isolated worker did not start'); + await new Promise(resolve => setTimeout(resolve, 10)); + } + } globalThis.serializedLeak = true; process.env.PAPERCLIP_SERIALIZED_LEAK = 'leaked'; vi.useFakeTimers(); @@ -40,8 +50,8 @@ test('one controller retains fresh fork, globals, environment, timers and home p writeFileSync(config, `import Sequencer from './sequencer.mjs'; export default ${JSON.stringify({ root: directory, test: { - include: order, pool: 'forks', isolate: true, fileParallelism: false, - maxWorkers: 1, setupFiles: [setup], + include: order, pool: 'forks', isolate: true, fileParallelism: parallel, + maxWorkers: parallel ? 2 : 1, setupFiles: [setup], sequence: { sequencer: '__SEQUENCER__' }, }, }).replace('"__SEQUENCER__"', 'Sequencer')};`); @@ -57,7 +67,8 @@ test('one controller retains fresh fork, globals, environment, timers and home p assert.equal(result.status, 0, `${result.error ?? ''}\n${result.stdout}\n${result.stderr}`); const workers = readFileSync(record, 'utf8').trim().split('\n').map((line) => JSON.parse(line)); assert.equal(workers.length, 2); - assert.deepEqual(workers.map((worker) => worker.name), order); + assert.deepEqual(workers.map((worker) => worker.name).sort(), [...order].sort()); + if (!parallel) assert.deepEqual(workers.map((worker) => worker.name), order); for (const key of ['pid', 'home', 'config', 'instance']) assert.equal(new Set(workers.map((worker) => worker[key])).size, 2, key); } } finally { diff --git a/scripts/benchmark-ci-work.mjs b/scripts/benchmark-ci-work.mjs index 7fa6dec7661..6bc2c8b99ec 100644 --- a/scripts/benchmark-ci-work.mjs +++ b/scripts/benchmark-ci-work.mjs @@ -11,7 +11,7 @@ const output = path.resolve(process.argv[3] ?? path.join(candidate, '.ci-benchma mkdirSync(output, { recursive: true }); const results = []; const commands = [ - ['serialized', process.execPath, ['scripts/run-vitest-stable.mjs', '--mode', 'serialized', '--shard-index', '0', '--shard-count', '9']], + ['general-server', process.execPath, ['scripts/run-vitest-stable.mjs', '--mode', 'general', '--group', 'general-server-without-chat-or-native-runner', '--shard-index', '0', '--shard-count', '12']], ['fixtures', 'pnpm', ['exec', 'vitest', 'run', '--project', '@paperclipai/server', 'server/src/services/workspace-git-snapshot-streaming.test.ts', 'server/src/__tests__/workspace-runtime.test.ts', '-t', 'streams.*all four real Git|commits fixed-port reservations']], @@ -47,7 +47,7 @@ for (const [name] of commands) { const after = results.find((result) => result.variant === 'candidate' && result.name === name); assert.ok(before.tests.length > 0, `${name}: missing test evidence`); assert.deepEqual(after.files, before.files, `${name}: file inventory changed`); - if (name === 'serialized') assert.deepEqual(after.tests, before.tests, 'serialized case inventory changed'); + if (name === 'general-server') assert.deepEqual(after.tests, before.tests, 'general server case inventory changed'); else { const contracts = (tests) => { const found = new Set(); diff --git a/scripts/run-vitest-stable.mjs b/scripts/run-vitest-stable.mjs index 847716b7067..7f283c9acab 100644 --- a/scripts/run-vitest-stable.mjs +++ b/scripts/run-vitest-stable.mjs @@ -1,5 +1,5 @@ #!/usr/bin/env node -import { spawnSync } from "node:child_process"; +import { spawn, spawnSync } from "node:child_process"; import { mkdirSync, mkdtempSync, readdirSync, readFileSync, realpathSync, statSync, writeFileSync } from "node:fs"; import os from "node:os"; import path from "node:path"; @@ -377,7 +377,7 @@ function selectSerializedSuites(routeTests, shardIndex, shardCount) { return shardFiles.map((file) => byRepoPath.get(file)); } -function runVitest(args, label, testShard = null) { +async function runVitest(args, label, testShard = null) { console.log(`\n[test:run] ${label}`); invocationIndex += 1; const tempRootParent = process.platform === "win32" ? os.tmpdir() : "/tmp"; @@ -408,7 +408,7 @@ function runVitest(args, label, testShard = null) { ...filters, "--allowOnly=false", "--includeTaskLocation", `--json=${output}`], { cwd: repoRoot, env, stdio: "inherit", }); - if (result.error || result.status !== 0) fail(`Vitest collection failed: ${result.error?.message ?? result.status}`); + if (result.error || result.status !== 0) throw new Error(`Vitest collection failed: ${result.error?.message ?? result.status}`); return JSON.parse(readFileSync(output, "utf8")); }; const allCollected = collect(args, "all"); @@ -434,27 +434,27 @@ function runVitest(args, label, testShard = null) { const runFilters = selected.lines.map((line) => `${chatSuite}:${chatSuiteRunLine(sourceLines, line)}`); args = [...args.filter((arg) => arg !== chatSuite), ...runFilters, "--allowOnly=false"]; } - const result = spawnSync("pnpm", ["exec", "vitest", "run", ...sourceOnlyVitestArgs, ...args], { - cwd: repoRoot, - env: invocationEnv("vitest", env), - stdio: "inherit", + await new Promise((resolve, reject) => { + const child = spawn("pnpm", ["exec", "vitest", "run", ...sourceOnlyVitestArgs, ...args], { + cwd: repoRoot, + env: invocationEnv("vitest", env), + stdio: "inherit", + }); + child.once("error", reject); + child.once("close", (code, signal) => { + if (code === 0) resolve(); + else reject(new Error(`${label} failed (${signal ?? code})`)); + }); }); - if (result.error) { - console.error(`[test:run] Failed to start Vitest: ${result.error.message}`); - process.exit(1); - } - if (result.status !== 0) { - process.exit(result.status ?? 1); - } } -function runGeneralSuites(routeTests) { +async function runGeneralSuites(routeTests) { for (const groupName of generalGroupNames) { - runGeneralGroup(routeTests, groupName); + await runGeneralGroup(routeTests, groupName); } } -function runProjectGroup(projects, groupName, shardIndex = null, shardCount = null) { +async function runProjectGroup(projects, groupName, shardIndex = null, shardCount = null) { // With shard args, lean on Vitest's native --shard: each matrix job runs the // same per-project invocations but only its slice of each project's test // files. Vitest's sharding is deterministic for an identical file list, so @@ -463,18 +463,32 @@ function runProjectGroup(projects, groupName, shardIndex = null, shardCount = nu shardCount !== null && shardCount > 1 ? [`--shard=${shardIndex + 1}/${shardCount}`] : []; const shardSuffix = shardArgs.length > 0 ? ` shard ${shardIndex + 1}/${shardCount}` : ""; for (const project of projects) { - runVitest(["--project", project, ...shardArgs], `${groupName} project ${project}${shardSuffix}`); + await runVitest(["--project", project, ...shardArgs], `${groupName} project ${project}${shardSuffix}`); } } -function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCount = null) { +function chatShardWorkers(index, count) { + const workers = count > 1 && !process.env.PAPERCLIP_TEST_DATABASE_URL ? 2 : 1; + return Array.from({ length: workers }, (_, offset) => ({ + index: (index ?? 0) * workers + offset, + count: (count ?? 1) * workers, + })); +} + +async function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCount = null) { if (groupName === generalChatGroupName) { - runVitest(["--project", "@paperclipai/server", ...serializedServerVitestArgs, chatSuite], - "chat integration test shard", { index: shardIndex ?? 0, count: shardCount ?? 1 }); + // Each invocation gets its own temporary home and embedded PostgreSQL. + // Wait for both children even on failure so no sibling is abandoned. + const results = await Promise.allSettled(chatShardWorkers(shardIndex, shardCount).map((shard) => + runVitest(["--project", "@paperclipai/server", ...serializedServerVitestArgs, chatSuite], + "chat integration test shard", shard), + )); + const failed = results.find((result) => result.status === "rejected"); + if (failed) throw failed.reason; return; } if (groupName === generalServerNativeRunnerGroupName) { - runVitest( + await runVitest( ["--project", "@paperclipai/server", ...serializedServerVitestArgs, ...nativeRunnerSuites], "native runner vertical-slice suite", ); @@ -497,11 +511,16 @@ function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCount = return; } - runVitest( + await runVitest( [ "--project", "@paperclipai/server", - ...serializedServerVitestArgs, + // Files have separate fork processes and isolated homes/configs. + // Keep each file's cases serial; overlap only two independent files. + // An explicit shared test database cannot be isolated this way. + ...(process.env.PAPERCLIP_TEST_DATABASE_URL + ? serializedServerVitestArgs + : ["--pool=forks", "--isolate", "--fileParallelism", "--maxWorkers=2"]), ...shardFiles, ], `${groupName} shard ${shardIndex + 1}/${shardCount}`, @@ -513,7 +532,7 @@ function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCount = for (const suite of excludedSuites) { excludeRouteArgs.push("--exclude", suite.replace(/^server\//, "")); } - runVitest( + await runVitest( [ "--project", "@paperclipai/server", @@ -530,31 +549,31 @@ function runGeneralGroup(routeTests, groupName, shardIndex = null, shardCount = // 31371439296, 2026-08-10, where workspaces-a was the slowest PR check). // Its 439 test files shard cleanly with Vitest's native --shard, so the // lane splits across runners without a duration manifest. - runProjectGroup(generalWorkspacesAProjects, groupName, shardIndex, shardCount); + await runProjectGroup(generalWorkspacesAProjects, groupName, shardIndex, shardCount); return; } if (groupName === generalWorkspacesBGroupName) { - runProjectGroup(generalWorkspacesBProjects, groupName); + await runProjectGroup(generalWorkspacesBProjects, groupName); return; } if (groupName === generalWorkspacesCGroupName) { - runProjectGroup(generalWorkspacesCProjects, groupName); + await runProjectGroup(generalWorkspacesCProjects, groupName); return; } fail(`Unknown group "${groupName}".`); } -function runSerializedSuites(routeTests, shardIndex, shardCount) { +async function runSerializedSuites(routeTests, shardIndex, shardCount) { const shardTests = selectSerializedSuites(routeTests, shardIndex, shardCount); console.log( `\n[test:run] serialized shard ${shardIndex + 1}/${shardCount} running ${shardTests.length} of ${routeTests.length} suites`, ); if (shardTests.length > 0) { - runVitest( + await runVitest( [ "--project", "@paperclipai/server", @@ -578,9 +597,9 @@ const routeTests = walk(serverTestsDir) // Every server test file that the general-server group is responsible for, // i.e. the whole server project minus the route/authz suites that run in the -// dedicated serialized shards. Sharding this list across runners is what keeps -// the general-server lane from becoming the PR critical path: the server vitest -// config pins maxWorkers to 1, so the only way to parallelize is across jobs. +// dedicated serialized shards. The default server config keeps files serial; +// the sharded entry point overlaps two isolated files per runner unless an +// explicit shared test database requires serial execution. // Suites are partitioned by recorded duration (scripts/general-server-shard.mjs) // rather than round-robin, so one slow suite cluster can't stretch a single shard. const serializedRepoPaths = new Set(routeTests.map(test => test.repoPath)); @@ -625,6 +644,8 @@ if (options.dryRun) { generalServerShardDurations, ) : null, + chatVitestShards: options.group === generalChatGroupName + ? chatShardWorkers(options.shardIndex, options.shardCount) : null, workspaceProjects: options.group === generalWorkspacesAGroupName ? generalWorkspacesAProjects @@ -647,14 +668,19 @@ if (options.dryRun) { process.exit(0); } -if (options.mode === generalModeName || options.mode === allModeName) { - if (options.group) { - runGeneralGroup(routeTests, options.group, options.shardIndex, options.shardCount); - } else { - runGeneralSuites(routeTests); +try { + if (options.mode === generalModeName || options.mode === allModeName) { + if (options.group) { + await runGeneralGroup(routeTests, options.group, options.shardIndex, options.shardCount); + } else { + await runGeneralSuites(routeTests); + } } -} -if (options.mode === serializedModeName || options.mode === allModeName) { - runSerializedSuites(routeTests, options.shardIndex ?? 0, options.shardCount ?? 1); + if (options.mode === serializedModeName || options.mode === allModeName) { + await runSerializedSuites(routeTests, options.shardIndex ?? 0, options.shardCount ?? 1); + } +} catch (error) { + console.error(`[test:run] ${error.message}`); + process.exitCode = 1; }