Parallelize the serialized server test suite

apps/server/src/provider/Layers/CursorAdapter.test.ts

@@ -28,6 +28,7 @@ import {
 import { ServerConfig } from "../../config.ts";
 import { ServerSettingsService } from "../../serverSettings.ts";
 import type { CursorAdapterShape } from "../Services/CursorAdapter.ts";
+import { pollUntil } from "../testUtils/pollUntil.ts";
 import { makeCursorAdapter } from "./CursorAdapter.ts";
 const decodeCursorSettings = Schema.decodeSync(CursorSettings);
 
@@ -97,36 +98,33 @@ async function readJsonLines(filePath: string) {
     .split("\n")
     .map((line) => line.trim())
     .filter((line) => line.length > 0)
-    .map((line) => JSON.parse(line) as Record<string, unknown>);
+    .flatMap((line) => {
+      // A poll can observe the mock child halfway through appending its final
+      // line. The next poll will see the complete JSON record.
+      try {
+        return [JSON.parse(line) as Record<string, unknown>];
+      } catch {
+        return [];
+      }
+    });
 }
 
-async function waitForFileContent(filePath: string, attempts = 40) {
-  for (let attempt = 0; attempt < attempts; attempt += 1) {
-    try {
-      const raw = await NodeFSP.readFile(filePath, "utf8");
-      if (raw.trim().length > 0) {
-        return raw;
-      }
-    } catch {}
-    await Effect.runPromise(Effect.yieldNow);
-  }
-  throw new Error(`Timed out waiting for file content at ${filePath}`);
+function waitForFileContent(filePath: string) {
+  return pollUntil({
+    poll: Effect.promise(() => NodeFSP.readFile(filePath, "utf8").catch(() => "")),
+    until: (raw) => raw.trim().length > 0,
+    description: `file content at ${filePath}`,
+  });
 }
 
 function waitForJsonLogMatch(
   filePath: string,
   predicate: (entry: Record<string, unknown>) => boolean,
-  attempts = 40,
 ) {
-  return Effect.gen(function* () {
-    for (let attempt = 0; attempt < attempts; attempt += 1) {
-      const requests = yield* Effect.promise(() => readJsonLines(filePath));
-      if (requests.some(predicate)) {
-        return requests;
-      }
-      yield* Effect.yieldNow;
-    }
-    return yield* Effect.promise(() => readJsonLines(filePath));
+  return pollUntil({
+    poll: Effect.promise(() => readJsonLines(filePath)),
+    until: (entries) => entries.some(predicate),
+    description: `a matching json log entry in ${filePath}`,
   });
 }
 
@@ -392,7 +390,7 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => {
 
       yield* adapter.stopSession(threadId);
 
-      const exitLog = yield* Effect.promise(() => waitForFileContent(exitLogPath));
+      const exitLog = yield* waitForFileContent(exitLogPath);
       assert.include(exitLog, "SIGTERM");
     }),
   );
@@ -444,7 +442,7 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => {
 
         yield* adapter.stopSession(threadId);
 
-        const exitLog = yield* Effect.promise(() => waitForFileContent(exitLogPath));
+        const exitLog = yield* waitForFileContent(exitLogPath);
         assert.equal(exitLog.match(/SIGTERM/g)?.length ?? 0, 2);
       }),
   );
@@ -555,7 +553,7 @@ cursorAdapterTestLayer("CursorAdapterLive", (it) => {
           modelSelection,
         });
 
-        yield* Effect.promise(() => waitForFileContent(requestLogPath));
+        yield* waitForFileContent(requestLogPath);
 
         const requestsAfterStart = yield* Effect.promise(() => readJsonLines(requestLogPath));
         const configIdsAfterStart = requestsAfterStart.flatMap((entry) =>

apps/server/src/provider/Layers/GrokAdapter.test.ts

@@ -446,9 +446,11 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
         }),
       );
       const adapter = yield* makeTestAdapter(wrapperPath);
-      const contentDelta = yield* Deferred.make<void>();
+      const trailingContentDelta = yield* Deferred.make<void>();
       const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) =>
-        event.type === "content.delta" ? Deferred.succeed(contentDelta, undefined) : Effect.void,
+        event.type === "content.delta" && event.payload.delta === "mock"
+          ? Deferred.succeed(trailingContentDelta, undefined)
+          : Effect.void,
       ).pipe(Effect.forkChild);
 
       yield* adapter.startSession({
@@ -467,10 +469,9 @@ it.layer(grokAdapterTestLayer)("GrokAdapterLive", (it) => {
         })
         .pipe(Effect.forkChild);
 
-      yield* Deferred.await(contentDelta);
-      for (let yieldAttempt = 0; yieldAttempt < 6; yieldAttempt += 1) {
-        yield* Effect.yieldNow;
-      }
+      // The mock emits this trailing chunk after the xAI prompt-complete
+      // notification, so it is a deterministic boundary for "prompt success".
+      yield* Deferred.await(trailingContentDelta).pipe(Effect.timeout("10 seconds"));
       yield* Fiber.interrupt(sendTurnFiber);
       for (let yieldAttempt = 0; yieldAttempt < 4; yieldAttempt += 1) {
         yield* Effect.yieldNow;

apps/server/src/provider/Layers/ProviderRegistry.test.ts

@@ -50,6 +50,7 @@ import type { ProviderInstance } from "../ProviderDriver.ts";
 import * as ProviderInstanceRegistry from "../Services/ProviderInstanceRegistry.ts";
 import * as ProviderRegistry from "../Services/ProviderRegistry.ts";
 import { makeManualOnlyProviderMaintenanceCapabilities } from "../providerMaintenance.ts";
+import { pollUntil } from "../testUtils/pollUntil.ts";
 const decodeServerSettings = Schema.decodeSync(ServerSettings);
 const encodeServerSettings = Schema.encodeSync(ServerSettings);
 const encodedDefaultServerSettings = encodeServerSettings(DEFAULT_SERVER_SETTINGS);
@@ -1552,18 +1553,12 @@ it.layer(TestLayer)("ProviderRegistry", (it) => {
           // Boot-time probe: the default codex instance is enabled with
           // `firstMissing`, so the real spawner yields ENOENT and the
           // snapshot should be `status: "error"`.
-          let initialProviders = yield* registry.getProviders;
-          for (
-            let attempts = 0;
-            attempts < 50 &&
-            initialProviders.find((provider) => provider.instanceId === "codex")?.status !==
-              "error";
-            attempts += 1
-          ) {
-            yield* TestClock.adjust("10 millis");
-            yield* Effect.yieldNow;
-            initialProviders = yield* registry.getProviders;
-          }
+          const initialProviders = yield* pollUntil({
+            poll: TestClock.adjust("10 millis").pipe(Effect.andThen(registry.getProviders)),
+            until: (providers) =>
+              providers.find((provider) => provider.instanceId === "codex")?.status === "error",
+            description: "the boot-time codex probe to fail against the first missing binary",
+          });
           const initialCodex = initialProviders.find((provider) => provider.instanceId === "codex");
           assert.strictEqual(initialCodex?.status, "error");
           assert.strictEqual(initialCodex?.installed, false);
@@ -1589,21 +1584,17 @@ it.layer(TestLayer)("ProviderRegistry", (it) => {
           // Poll until the injected process boundary observes the new
           // executable. This verifies the public settings-to-probe behavior
           // without depending on timestamps assigned by TestClock.
-          const refreshed = yield* Effect.gen(function* () {
-            for (let attempts = 0; attempts < 60; attempts += 1) {
-              const providers = yield* registry.getProviders;
+          const refreshed = yield* pollUntil({
+            poll: TestClock.adjust("50 millis").pipe(Effect.andThen(registry.getProviders)),
+            until: (providers) => {
               const codex = providers.find((provider) => provider.instanceId === "codex");
-              if (
+              return (
                 codex !== undefined &&
                 codex.status === "error" &&
                 spawnedCommands.includes(secondMissing)
-              ) {
-                return providers;
-              }
-              yield* TestClock.adjust("50 millis");
-              yield* Effect.yieldNow;
-            }
-            return yield* registry.getProviders;
+              );
+            },
+            description: "the codex re-probe against the second missing binary",
           });
 
           const reprobedCodex = refreshed.find((provider) => provider.instanceId === "codex");

apps/server/src/provider/testUtils/pollUntil.ts

@@ -0,0 +1,53 @@
+import * as Clock from "effect/Clock";
+import * as Duration from "effect/Duration";
+import * as Effect from "effect/Effect";
+import * as TestClock from "effect/testing/TestClock";
+
+const describeValue = (value: unknown) => {
+  try {
+    const text = JSON.stringify(value);
+    if (text === undefined) {
+      return String(value);
+    }
+    return text.length > 2_000 ? `${text.slice(0, 2_000)}…` : text;
+  } catch {
+    return String(value);
+  }
+};
+
+export interface PollUntilOptions<A, E, R> {
+  readonly poll: Effect.Effect<A, E, R>;
+  readonly until: (value: A) => boolean;
+  readonly description: string;
+  readonly timeout?: Duration.Input;
+  readonly interval?: Duration.Input;
+}
+
+/**
+ * Polls asynchronous OS work using real time even when an Effect test uses
+ * TestClock. This gives child processes and libuv callbacks time to progress.
+ */
+export const pollUntil = <A, E, R>(options: PollUntilOptions<A, E, R>) =>
+  Effect.gen(function* () {
+    const timeoutMillis = Duration.toMillis(options.timeout ?? "10 seconds");
+    const interval = options.interval ?? "25 millis";
+    const startedAt = yield* TestClock.withLive(Clock.currentTimeMillis);
+
+    for (;;) {
+      const value = yield* options.poll;
+      if (options.until(value)) {
+        return value;
+      }
+
+      const now = yield* TestClock.withLive(Clock.currentTimeMillis);
+      if (now - startedAt >= timeoutMillis) {
+        return yield* Effect.die(
+          new Error(
+            `Timed out after ${timeoutMillis}ms waiting for ${options.description}. ` +
+              `Last polled value: ${describeValue(value)}`,
+          ),
+        );
+      }
+      yield* TestClock.withLive(Effect.sleep(interval));
+    }
+  });

apps/server/vite.config.ts

@@ -64,9 +64,12 @@ export default mergeConfig(
       },
     },
     test: {
-      // The server suite exercises sqlite, git, temp worktrees, and orchestration
-      // runtimes heavily. Running files in parallel introduces load-sensitive flakes.
-      fileParallelism: false,
+      // Keep enough parallelism to avoid serializing the entire server suite while
+      // capping load from sqlite, git, temp-worktree, and orchestration tests.
+      // Isolate a demonstrably conflicting suite instead of forcing every file
+      // through one worker.
+      fileParallelism: true,
+      maxWorkers: 4,
       // Server integration tests exercise sqlite, git, and orchestration together.
       // Under package-wide runs they can exceed the default budget on loaded CI hosts.
       hookTimeout: 120_000,