test(server): keep responsiveness gates independent of runner scheduling (#8002)

Raw timer gaps and elapsed callback or GC spans include Windows runner preemption. Share callback CPU accounting across the daemon hot-path probes and retain deterministic async admission, real yields, and sync-IO tracing at filesystem boundaries.
2026-10-27 19:52:49 +00:00 · 2026-10-27 07:35:44 +00:00
parent 07007a485d
commit 6b7d49f0d8
12 changed files with 270 additions and 432 deletions
@@ -1,3 +1,4 @@
+import { measureMainThreadWork } from "../../test-utils/main-thread-work.js";
 import { describe, expect, it } from "vitest";
 import { selectReferenceProjectedTimelinePage } from "./test-utils/timeline-page-reference.js";
 import type { AgentTimelineItem } from "./agent-sdk-types.js";
@@ -258,25 +259,19 @@ it.each([false, true])(
    for (let index = 0; index < 100_000; index++) {
      store.append("large", { type: "user_message", text: String(index) });
    }
-    // Generation and its GC are outside the measured request window.
-    await new Promise<void>((resolve) => setImmediate(resolve));
-    const started = performance.now();
-    const probe = new Promise<number>((resolve) =>
-      setImmediate(() => resolve(performance.now() - started)),
+    const { result: tail, blockingMs } = await measureMainThreadWork(
+      `timeline tail main-thread work (wholeTurns=${wholeTurns})`,
+      () => {
+        let page = store.fetch("large", { limit: 200, wholeTurns });
+        for (let repeat = 1; repeat < 8; repeat++)
+          page = store.fetch("large", { limit: 200, wholeTurns });
+        return page;
+      },
    );
-    let page = store.fetch("large", { limit: 200, wholeTurns });
-    for (let repeat = 1; repeat < 8; repeat++)
-      page = store.fetch("large", { limit: 200, wholeTurns });
-    const duration = performance.now() - started;
-    const loopGap = await probe;
-    console.info({ duration, loopGap });
-    expect(page.rows).toHaveLength(200);
-    expect(page.startSeq).toBe(99_801);
-    expect(page.endSeq).toBe(100_000);
-    // CI runners can pause for scheduling/GC; the local per-request 2 ms gate
-    // is measured separately. Eight requests amplify the old full-history work.
-    expect(duration).toBeLessThan(50);
-    expect(loopGap).toBeLessThan(50);
+    expect(tail.rows).toHaveLength(200);
+    expect(tail.startSeq).toBe(99_801);
+    expect(tail.endSeq).toBe(100_000);
+    expect(blockingMs).toBeLessThan(50);
  },
 );

@@ -1,3 +1,4 @@
+import { measureMainThreadWork } from "../../../../test-utils/main-thread-work.js";
 import fs, { chmodSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { open } from "node:fs/promises";
 import { performance } from "node:perf_hooks";
@@ -117,13 +118,6 @@ describe("ClaudeAgentSession persisted history load", () => {
      await file.close();
    }
    const daemon = await createTestPaseoDaemon();
-    let lastTick = performance.now();
-    let longestGap = 0;
-    const probe = setInterval(() => {
-      const now = performance.now();
-      longestGap = Math.max(longestGap, now - lastTick);
-      lastTick = now;
-    }, 2);
    let loading = true;
    let answeredDuringLoad = false;
    const health = fetch(`http://127.0.0.1:${daemon.port}/api/health`).then(async (response) => {
@@ -134,33 +128,39 @@ describe("ClaudeAgentSession persisted history load", () => {
    });
    let session: AgentSession | undefined;
    try {
-      session = await resume(replyFactory("live after large resume"));
-      let index = 0;
-      for await (const event of session.streamHistory()) {
-        expect(event).toEqual({
-          type: "timeline",
-          provider: "claude",
-          timestamp: undefined,
-          item: { type: "assistant_message", text: `${index}:${text}` },
-        });
-        index++;
-      }
-      loading = false;
-      await health;
-      await new Promise((resolve) => setTimeout(resolve, 5));
+      const {
+        result: { resumed: loadedSession, historyCount },
+        blockingMs,
+      } = await measureMainThreadWork("Claude history main-thread work", async () => {
+        const resumed = await resume(replyFactory("live after large resume"));
+        session = resumed;
+        let index = 0;
+        for await (const event of resumed.streamHistory()) {
+          expect(event).toEqual({
+            type: "timeline",
+            provider: "claude",
+            timestamp: undefined,
+            item: { type: "assistant_message", text: `${index}:${text}` },
+          });
+          index++;
+        }
+        loading = false;
+        await health;
+        return { resumed, historyCount: index };
+      });
      console.info(
        JSON.stringify({
          bytes,
          count,
-          longestGap,
+          blockingMs,
          answeredDuringLoad,
          elapsedMs: performance.now() - startedAt,
        }),
      );
-      expect(index).toBe(count);
-      expect(longestGap).toBeLessThan(50);
+      expect(historyCount).toBe(count);
+      expect(blockingMs).toBeLessThan(50);
      expect(answeredDuringLoad).toBe(true);
-      const live = await collectSessionTurnEvents(session, "follow up");
+      const live = await collectSessionTurnEvents(loadedSession, "follow up");
      expect(
        live
          .filter((event) => event.type === "timeline" && event.item.type === "assistant_message")
@@ -169,7 +169,6 @@ describe("ClaudeAgentSession persisted history load", () => {
        { type: "assistant_message", messageId: "live-reply", text: "live after large resume" },
      ]);
    } finally {
-      clearInterval(probe);
      await session?.close();
      await daemon.close();
    }
@@ -1,5 +1,4 @@
-import { performance } from "node:perf_hooks";
-import { setTimeout as delay } from "node:timers/promises";
+import { measureMainThreadWork } from "../../../test-utils/main-thread-work.js";
 import { describe, expect, it } from "vitest";
 import { JsonlFrameDecoder, type JsonlFrameProblem } from "./jsonl-frame-decoder.js";

@@ -72,26 +71,15 @@ describe("provider line admission", () => {
  it("keeps the loop responsive while accumulating a fragmented provider response", async () => {
    const { decoder, frames, problems } = decoderHarness();
    const part = "x".repeat(512);
-    let last = performance.now();
-    let longestGap = 0;
-    const probe = setInterval(() => {
-      const now = performance.now();
-      longestGap = Math.max(longestGap, now - last);
-      last = now;
-    }, 1);
-    try {
-      await delay(5);
+    const { blockingMs } = await measureMainThreadWork("fragmented JSONL main-thread work", () => {
      decoder.write('{"type":"notice","text":"');
-      // 1 MiB is enough to expose the former rescan of every accumulated prefix.
+      // 1 MiB exposes the former rescan of every accumulated prefix.
      for (let index = 0; index < 2048; index++) decoder.write(part);
      decoder.write('"}\n');
-      await delay(5);
-      expect(frames).toEqual([{ type: "notice", text: part.repeat(2048) }]);
-      expect(problems).toEqual([]);
-      expect(longestGap).toBeLessThan(50);
-    } finally {
-      clearInterval(probe);
-    }
+    });
+    expect(frames).toEqual([{ type: "notice", text: part.repeat(2048) }]);
+    expect(problems).toEqual([]);
+    expect(blockingMs).toBeLessThan(50);
  });

  it("rejects an unterminated over-cap UTF-8 line and stops accepting frames", () => {
@@ -1,3 +1,4 @@
+import { measureMainThreadWork } from "../../../../test-utils/main-thread-work.js";
 import { BUILTIN_PROVIDER_IDS } from "@getpaseo/protocol/provider-manifest";
 import { mkdtemp, open, rm } from "node:fs/promises";
 import { tmpdir } from "node:os";
@@ -50,51 +51,45 @@ test("replays selected OMP history without blocking the event loop", async () =>
      } finally {
        await file.close();
      }
-      let last = performance.now();
-      let longestGap = 0;
-      const probe = setInterval(() => {
-        const now = performance.now();
-        longestGap = Math.max(longestGap, now - last);
-        last = now;
-      }, 5);
-      try {
-        const replayState = { finished: false };
-        let repliesDuringReplay = 0;
-        const replay = readActiveOmpEntryChain(sessionFile, "row-0").finally(() => {
-          replayState.finished = true;
-        });
-        // A real WebSocket RPC served by the isolated daemon on the same event loop.
-        const requests = (async () => {
-          while (!replayState.finished) {
-            await client.getDaemonStatus();
-            if (!replayState.finished) repliesDuringReplay++;
-            await delay(10);
-          }
-        })();
-        const [entries] = await Promise.all([replay, requests]);
-        await delay(10);
-        console.log(
-          JSON.stringify({
-            bytes,
-            count,
-            longestGap,
-            repliesDuringReplay,
-            generationAndLoadMs: performance.now() - started,
-          }),
-        );
-        expect(entries).toEqual([
-          {
-            type: "message",
-            id: "row-0",
-            parentId: null,
-            message: { role: "user", content: text },
-          },
-        ]);
-        expect(longestGap).toBeLessThan(50);
-        expect(repliesDuringReplay).toBeGreaterThan(0);
-      } finally {
-        clearInterval(probe);
-      }
+      let repliesDuringReplay = 0;
+      const { result: replayedEntries, blockingMs } = await measureMainThreadWork(
+        "OMP history main-thread work",
+        async () => {
+          const replayState = { finished: false };
+          const replay = readActiveOmpEntryChain(sessionFile, "row-0").finally(() => {
+            replayState.finished = true;
+          });
+          // A real WebSocket RPC served by the isolated daemon on the same event loop.
+          const requests = (async () => {
+            while (!replayState.finished) {
+              await client.getDaemonStatus();
+              if (!replayState.finished) repliesDuringReplay++;
+              await delay(10);
+            }
+          })();
+          const [entries] = await Promise.all([replay, requests]);
+          return entries;
+        },
+      );
+      console.log(
+        JSON.stringify({
+          bytes,
+          count,
+          blockingMs,
+          repliesDuringReplay,
+          generationAndLoadMs: performance.now() - started,
+        }),
+      );
+      expect(replayedEntries).toEqual([
+        {
+          type: "message",
+          id: "row-0",
+          parentId: null,
+          message: { role: "user", content: text },
+        },
+      ]);
+      expect(blockingMs).toBeLessThan(50);
+      expect(repliesDuringReplay).toBeGreaterThan(0);
    } finally {
      await client.close().finally(() => daemon.close());
    }
@@ -1,3 +1,4 @@
+import { measureMainThreadWork } from "../test-utils/main-thread-work.js";
 import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
@@ -25,22 +26,18 @@ describe("server config", () => {
    const env = { PASEO_PASSWORD: "synthetic-audit-password" };
    const persisted = loadPersistedConfig(paseoHome);
    const startup = resolveConfigFromPersisted(paseoHome, persisted, { env });
-    const started = performance.now();
-    const timer = new Promise<number>((resolve) =>
-      setTimeout(() => resolve(performance.now() - started), 0),
+    const { result: reloaded, blockingMs } = await measureMainThreadWork(
+      "config resolve main-thread work",
+      () =>
+        resolveReloadableConfig(
+          { ...persisted, daemon: { ...persisted.daemon, browserTools: { enabled: true } } },
+          { env },
+        ),
    );
-    const reloaded = resolveReloadableConfig(
-      {
-        ...persisted,
-        daemon: { ...persisted.daemon, browserTools: { enabled: true } },
-      },
-      { env },
-    );
-    const gap = await timer;
    expect(reloaded).not.toHaveProperty("auth");
    expect(reloaded.browserToolsEnabled).toBe(true);
    expect(startup.auth?.password).toBeTypeOf("string");
-    expect(gap).toBeLessThan(50);
+    expect(blockingMs).toBeLessThan(50);
  });

  test("records when the daemon is managed by Paseo Desktop", async () => {
@@ -1,6 +1,6 @@
+import { measureMainThreadWork } from "../../test-utils/main-thread-work.js";
 import { execFile } from "node:child_process";
 import { promisify } from "node:util";
-import { performance } from "node:perf_hooks";
 import pino from "pino";
 import { createPaseoDaemon } from "../bootstrap.js";
 import { loadConfig } from "../config.js";
@@ -231,21 +231,12 @@ test("password-protected config reload preserves startup auth and session constr
    await client.fetchAgents({ subscribe: {} });
    const configPath = path.join(home, "config.json");
    await writeFile(configPath, JSON.stringify({ daemon: { browserTools: { enabled: true } } }));
-    let longestGap = 0;
-    let previous = performance.now();
-    const probe = setInterval(() => {
-      const now = performance.now();
-      longestGap = Math.max(longestGap, now - previous);
-      previous = now;
-    }, 1);
-    try {
-      const reloaded = await client.reloadDaemonConfig();
-      longestGap = Math.max(longestGap, performance.now() - previous);
-      expect(reloaded.appliedPaths).toContain("daemon.browserTools.enabled");
-      expect(longestGap).toBeLessThan(50);
-    } finally {
-      clearInterval(probe);
-    }
+    const { result: reloaded, blockingMs } = await measureMainThreadWork(
+      "daemon config reload main-thread work",
+      () => client.reloadDaemonConfig(),
+    );
+    expect(reloaded.appliedPaths).toContain("daemon.browserTools.enabled");
+    expect(blockingMs).toBeLessThan(50);
    expect((await client.getDaemonConfig()).config.browserTools.enabled).toBe(true);
    // A new physical session uses ready daemon state; corrupt config is reported only by an explicit read.
    await writeFile(configPath, "invalid JSON");
@@ -1,9 +1,7 @@
+import { measureMainThreadWork } from "../../../test-utils/main-thread-work.js";
 import assert from "node:assert/strict";
-import { createHook, executionAsyncResource } from "node:async_hooks";
 import { open as openFile, writeFile } from "node:fs/promises";
-import { performance, PerformanceObserver } from "node:perf_hooks";
 import path from "node:path";
-import { MessagePort } from "node:worker_threads";
 import pino from "pino";
 import { DaemonConfigStore } from "../../daemon-config-store.js";
 import { PluginService } from "../index.js";
@@ -53,125 +51,9 @@ try {
    () => service.enablePlugin("responsive-build"),
    () => service.reloadPlugin("responsive-build"),
  ]) {
-    // Node 22.19+ exposes per-thread CPU accounting; the repository's older
-    // Node declarations do not include it yet. .tool-versions pins Node 22.20.
-    const threadCpuUsage: (previous?: NodeJS.CpuUsage) => NodeJS.CpuUsage = Reflect.get(
-      process,
-      "threadCpuUsage",
-    );
-    const startedAt = performance.now();
-    let previous = startedAt;
-    let cpu = threadCpuUsage();
-    let longestWall = 0;
-    let longestBlocking = 0;
-    let longestWallCpu = 0;
-    let workerStarts = 0;
-    let portCallbacks = 0;
-    let longestWallEvents = { workerStarts: 0, portCallbacks: 0 };
-    let longestBlockingWindow = { wallMs: 0, cpuMs: 0, resource: "" };
-    let probe: ReturnType<typeof setInterval>;
-    const callbacks = new Map<number, { wall: number; cpu: NodeJS.CpuUsage; resource: string }>();
-    const recordBlocking = (
-      blockingMs: number,
-      window: { wallMs: number; cpuMs: number; resource: string },
-    ) => {
-      if (blockingMs > longestBlocking) {
-        longestBlocking = blockingMs;
-        longestBlockingWindow = window;
-      }
-    };
-    let longestGc = 0;
-    const observeGc = (entries: readonly import("node:perf_hooks").PerformanceEntry[]) => {
-      for (const entry of entries) {
-        if (entry.startTime >= startedAt) longestGc = Math.max(longestGc, entry.duration);
-      }
-    };
-    const gc = new PerformanceObserver((list) => observeGc(list.getEntries()));
-    gc.observe({ entryTypes: ["gc"] });
-    // Measure work within callbacks, not CPU-counter increments attributed to
-    // the idle interval between them. Exclude this probe's own timer callback.
-    const activity = createHook({
-      init(_id, type) {
-        if (type === "WORKER") workerStarts++;
-      },
-      before(id) {
-        const resource = executionAsyncResource();
-        if (resource instanceof MessagePort) portCallbacks++;
-        if (resource === probe) return;
-        callbacks.set(id, {
-          wall: performance.now(),
-          cpu: threadCpuUsage(),
-          resource: Object.prototype.toString.call(resource),
-        });
-      },
-      after(id) {
-        const start = callbacks.get(id);
-        if (!start) return;
-        callbacks.delete(id);
-        const wallMs = performance.now() - start.wall;
-        const { user, system } = threadCpuUsage(start.cpu);
-        const cpuMs = (user + system) / 1_000;
-        // A callback prevents the loop from serving IO for its entire span.
-        // Use the high-resolution clock, not Windows' coarse CPU increments.
-        recordBlocking(wallMs, { wallMs, cpuMs, resource: start.resource });
-      },
-    }).enable();
-    const sample = () => {
-      const now = performance.now();
-      const currentCpu = threadCpuUsage();
-      const wallMs = now - previous;
-      const cpuMs = (currentCpu.user - cpu.user + currentCpu.system - cpu.system) / 1_000;
-      // Native worker construction and message deserialization can precede
-      // async_hooks.before. Include their interval when those events occurred.
-      if (workerStarts || portCallbacks) {
-        recordBlocking(Math.min(wallMs, cpuMs), {
-          wallMs,
-          cpuMs,
-          resource: "worker/port interval",
-        });
-      }
-      if (wallMs > longestWall) {
-        longestWall = wallMs;
-        longestWallCpu = cpuMs;
-        longestWallEvents = { workerStarts, portCallbacks };
-      }
-      workerStarts = 0;
-      portCallbacks = 0;
-      previous = now;
-      cpu = currentCpu;
-    };
-    // Sleep between probes: a continuously runnable setImmediate loop consumes
-    // a core while the worker compiles. Raw gaps remain diagnostics; the
-    // assertion measures elapsed daemon-thread callback spans and GC pauses.
-    probe = setInterval(sample, 1);
-    try {
-      // The caller's current callback began before the hook was enabled.
-      // Include synchronous work before the action's first await explicitly.
-      const actionStartedAt = performance.now();
-      const actionCpu = threadCpuUsage();
-      const pending = action();
-      const actionWallMs = performance.now() - actionStartedAt;
-      const actionCost = threadCpuUsage(actionCpu);
-      recordBlocking(actionWallMs, {
-        wallMs: actionWallMs,
-        cpuMs: (actionCost.user + actionCost.system) / 1_000,
-        resource: "action()",
-      });
-      await pending;
-      await new Promise<void>((resolve) => setImmediate(resolve));
-      sample();
-      observeGc(gc.takeRecords());
-    } finally {
-      clearInterval(probe);
-      activity.disable();
-      gc.disconnect();
-    }
-    longestBlocking = Math.max(longestBlocking, longestGc);
-    process.stderr.write(
-      `plugin compile main-thread blocking: ${longestBlocking.toFixed(2)} ms ${JSON.stringify({ longestWall, longestWallCpu, ...longestWallEvents, longestBlockingWindow, longestGc })}\n`,
-    );
+    const { blockingMs } = await measureMainThreadWork("plugin compile main-thread work", action);
    assert.ok(service.catalog()[0]?.clientBundle.includes("compiled fixture"));
-    results.push({ blockingMs: longestBlocking });
+    results.push({ blockingMs });
  }
  // Evaluate after both probes so parsing the returned code is not charged to compilation.
  const factory = globalThis.eval(service.catalog()[0]!.clientBundle) as (
@@ -3,7 +3,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { describe, expect, it } from "vitest";
 import { setImmediate } from "node:timers/promises";
-import { performance } from "node:perf_hooks";
+import { measureMainThreadWork } from "../../test-utils/main-thread-work.js";
 import { SessionDelivery } from "../session/owned-subscriptions/index.js";
 import { highlightCode, highlightCodeWithBudget } from "@getpaseo/highlight";
 import {
@@ -840,26 +840,19 @@ describe("diff highlighting budget", () => {
@@ -1 +1 @@
 -const value = Math.max(42, 7);
 +const value = Math.max(43, 7);`)[0];
-      let lastTick = performance.now();
-      let longestGap = 0;
-      const timer = setInterval(() => {
-        const now = performance.now();
-        longestGap = Math.max(longestGap, now - lastTick);
-        lastTick = now;
-      }, 1);
      try {
-        await setImmediate();
-        const highlighted = await highlightDiffWithFileContent(
-          file,
-          cwd,
-          source === "supplied" ? { oldFileContent } : {},
+        const { result: highlighted, blockingMs } = await measureMainThreadWork(
+          `diff highlight 5MiB ${source} main-thread work`,
+          () =>
+            highlightDiffWithFileContent(
+              file,
+              cwd,
+              source === "supplied" ? { oldFileContent } : {},
+            ),
        );
-        await new Promise((resolve) => setTimeout(resolve, 5));
-        console.info({ sourceBytes: Buffer.byteLength(oldFileContent), longestGap });
-        expect(longestGap).toBeLessThan(50);
+        expect(blockingMs).toBeLessThan(50);
        expect(highlighted).toEqual(highlightDiffFromHunks(file));
      } finally {
-        clearInterval(timer);
        rmSync(cwd, { recursive: true, force: true });
      }
    },
@@ -952,23 +945,17 @@ it("highlights a one-line hunk from an over-budget JSON source without blocking
@@ -2 +2 @@
 -  "value": 42,
 +  "value": 43,`);
-  let last = performance.now(),
-    longestGap = 0;
-  const timer = setInterval(() => {
-    const now = performance.now();
-    longestGap = Math.max(longestGap, now - last);
-    last = now;
-  }, 1);
  try {
-    const result = await highlightDiffWithFileContent(file, cwd, { oldFileContent });
-    await new Promise((resolve) => setTimeout(resolve, 5));
-    expect(longestGap).toBeLessThan(50);
+    const { result, blockingMs } = await measureMainThreadWork(
+      "diff highlight large JSON main-thread work",
+      () => highlightDiffWithFileContent(file, cwd, { oldFileContent }),
+    );
+    expect(blockingMs).toBeLessThan(50);
    expect(result.hunks[0].lines.find((line) => line.type === "add")?.tokens).toContainEqual({
      text: "43",
      style: "number",
    });
  } finally {
-    clearInterval(timer);
    rmSync(cwd, { recursive: true, force: true });
  }
 });
@@ -534,6 +534,7 @@ test("Git refresh, command tracing and 10,000 ignored alias events do no synchro
  const repositoryRoot = fileURLToPath(new URL("../../../../", import.meta.url));
  const source = `
    import assert from 'node:assert/strict';
+    import { measureMainThreadWork } from './packages/server/src/test-utils/main-thread-work.ts';
    import { write } from 'node:fs';
    import { mkdtemp, mkdir, writeFile, symlink, rm } from 'node:fs/promises';
    import { tmpdir } from 'node:os';
@@ -575,29 +576,23 @@ test("Git refresh, command tracing and 10,000 ignored alias events do no synchro
      await symlink(worktree, alias, process.platform === 'win32' ? 'junction' : 'dir');
      const subscription = await service.requestWorkingTreeWatch(alias, () => { notifications++; });
      const events = Array.from({ length: process.env.GIT_IO_TRACE ? 2 : 10000 }, (_, i) => ({ type: 'update', path: join(i % 2 ? alias : worktree, 'ignored', 'noise') }));
-      // A shared CI runner can deschedule the child during a wall-clock sample.
-      // Keep the 20ms gate, but take three independent batches and require one
-      // responsive callback/timer pair. All batches must remain filesystem-free.
-      const samples = [];
-      for (let trial = 0; trial < (process.env.GIT_IO_TRACE ? 1 : 3); trial++) {
-        await tick();
-        let fsCalls = 0;
-        const hook = createHook({ init(_id, type) { if (type.startsWith('FSREQ')) fsCalls++; } });
+      let fsCalls = 0;
+      const hook = createHook({ init(_id, type) { if (type.startsWith('FSREQ')) fsCalls++; } });
+      await tick();
+      const { blockingMs } = await measureMainThreadWork('Git filter main-thread work', async () => {
        await marker('FILTER_START');
        hook.enable();
-        const started = performance.now();
-        const pulse = new Promise((resolve) => setTimeout(() => resolve(performance.now() - started), 0));
-        callback(null, events);
-        const filterMs = performance.now() - started;
-        hook.disable();
-        const loopGapMs = await pulse;
-        await marker('FILTER_END');
-        assert.equal(fsCalls, 0, "filesystem calls");
-        assert.equal(notifications, 0, "ignored event notification");
-        samples.push({ filterMs, loopGapMs, fsCalls });
-      }
-      console.log(JSON.stringify({ ...samples[0], samples }));
-      assert.ok(process.env.GIT_IO_TRACE || samples.some(({ filterMs, loopGapMs }) => filterMs < 20 && loopGapMs < 20), 'no responsive filter sample: ' + JSON.stringify(samples));
+        try {
+          callback(null, events);
+        } finally {
+          hook.disable();
+          await marker('FILTER_END');
+        }
+      });
+      assert.equal(fsCalls, 0, "filesystem calls");
+      assert.equal(notifications, 0, "ignored event notification");
+      assert.ok(blockingMs < 20, 'blocking filter work: ' + blockingMs);
+      console.log(JSON.stringify({ blockingMs, fsCalls }));
      // Alias identity remains useful after deletion: no per-event realpath can recover this file.
      callback(null, [{ type: 'delete', path: join(worktree, 'tracked.txt') }]);
      assert.equal(notifications, 1);
@@ -1,5 +1,4 @@
-import { createHook, executionAsyncResource } from "node:async_hooks";
-import { performance, PerformanceObserver } from "node:perf_hooks";
+import { measureMainThreadWork } from "../test-utils/main-thread-work.js";
 import { execFileSync } from "node:child_process";
 import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
@@ -660,97 +659,20 @@ describe("WorkspaceReconciliationService", () => {
      workspaceRegistry,
      logger: createTestLogger(),
    });
-    // Windows can deschedule Node inside a callback too. Budget charged thread
-    // CPU within callback spans, capped by wall time because Windows CPU counters
-    // advance in coarse increments. Also require real asynchronous inspections:
-    // synchronous filesystem waits can block the loop without consuming CPU.
-    // Node 22.20 provides this API; our older Node declarations omit it.
-    const threadCpuUsage: (previous?: NodeJS.CpuUsage) => NodeJS.CpuUsage = Reflect.get(
-      process,
-      "threadCpuUsage",
-    );
-    const startedAt = performance.now();
-    let last = startedAt;
-    let longestGap = 0;
-    let longestCallback = 0;
-    let longestCallbackCpu = 0;
-    let longestBlocking = 0;
-    let asyncInspections = 0;
-    let longestGc = 0;
-    const callbacks = new Map<number, { wall: number; cpu: NodeJS.CpuUsage }>();
-    const observeGc = (entries: readonly import("node:perf_hooks").PerformanceEntry[]) => {
-      for (const entry of entries) {
-        if (entry.startTime >= startedAt) longestGc = Math.max(longestGc, entry.duration);
-      }
-    };
-    const gc = new PerformanceObserver((list) => observeGc(list.getEntries()));
-    gc.observe({ entryTypes: ["gc"] });
-    let probe: ReturnType<typeof setInterval>;
-    const activity = createHook({
-      init(_id, type) {
-        if (type === "FSREQPROMISE") asyncInspections++;
-      },
-      before(id) {
-        if (executionAsyncResource() === probe) return;
-        callbacks.set(id, { wall: performance.now(), cpu: threadCpuUsage() });
-      },
-      after(id) {
-        const start = callbacks.get(id);
-        if (!start) return;
-        callbacks.delete(id);
-        const duration = performance.now() - start.wall;
-        const { user, system } = threadCpuUsage(start.cpu);
-        const cpuMs = (user + system) / 1_000;
-        longestBlocking = Math.max(longestBlocking, Math.min(duration, cpuMs));
-        if (duration > longestCallback) {
-          longestCallback = duration;
-          longestCallbackCpu = cpuMs;
-        }
-      },
-    }).enable();
-    const sample = () => {
-      const now = performance.now();
-      longestGap = Math.max(longestGap, now - last);
-      last = now;
-    };
-    probe = setInterval(sample, 1);
-    let result;
    try {
-      // Include the initial synchronous span: its callback predates the hook.
-      const actionStartedAt = performance.now();
-      const actionCpu = threadCpuUsage();
-      const pending = service.runOnce();
-      const actionWallMs = performance.now() - actionStartedAt;
-      const { user, system } = threadCpuUsage(actionCpu);
-      longestCallback = Math.max(longestCallback, actionWallMs);
-      longestBlocking = Math.max(longestBlocking, Math.min(actionWallMs, (user + system) / 1_000));
-      result = await pending;
-      await new Promise<void>((resolve) => setImmediate(resolve));
-      sample();
-      observeGc(gc.takeRecords());
+      const { result, blockingMs, filesystemRequests } = await measureMainThreadWork(
+        "workspace reconciliation main-thread work",
+        () => service.runOnce(),
+      );
+      expect(blockingMs).toBeLessThan(20);
+      expect(filesystemRequests).toBe(20001);
+      expect(result.changesApplied).toHaveLength(20000);
+      expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe(
+        true,
+      );
    } finally {
-      clearInterval(probe);
-      activity.disable();
-      gc.disconnect();
      service.dispose();
    }
-    // GC CPU during callbacks is included above. GC elapsed durations, like
-    // callback elapsed durations, include runner preemption: log them separately.
-    const blockingMs = longestBlocking;
-    process.stderr.write(
-      `workspace reconciliation main-thread blocking: ${JSON.stringify({
-        blockingMs,
-        asyncInspections,
-        longestCallback,
-        longestCallbackCpu,
-        longestGc,
-        longestGap,
-      })}\n`,
-    );
-    expect(blockingMs).toBeLessThan(20);
-    expect(asyncInspections).toBe(20001);
-    expect(result.changesApplied).toHaveLength(20000);
-    expect([...workspaces.values()].every((workspace) => workspace.archivedAt !== null)).toBe(true);
  }, 30000);

  test("preserves workspace archival that lands during boot reconciliation", async () => {
@@ -0,0 +1,108 @@
+import { createHook, executionAsyncResource } from "node:async_hooks";
+import { performance, PerformanceObserver, type PerformanceEntry } from "node:perf_hooks";
+import { MessagePort } from "node:worker_threads";
+
+/** Measure main-thread work without charging idle IO or runner descheduling. */
+export async function measureMainThreadWork<T>(label: string, action: () => T | Promise<T>) {
+  // Node 22.20 supplies this API; the repository's older declarations omit it.
+  const threadCpuUsage: (previous?: NodeJS.CpuUsage) => NodeJS.CpuUsage = Reflect.get(
+    process,
+    "threadCpuUsage",
+  );
+  let blockingMs = 0;
+  let filesystemRequests = 0;
+  let yields = 0;
+  let longestCallbackMs = 0;
+  let longestCallbackCpuMs = 0;
+  let longestGapMs = 0;
+  let longestGcMs = 0;
+  let gcCount = 0;
+  let workerEvents = 0;
+  let draining = false;
+  const startedAt = performance.now();
+  let previous = startedAt;
+  let previousCpu = threadCpuUsage();
+  let probe: ReturnType<typeof setInterval>;
+  let drain: ReturnType<typeof setImmediate>;
+  const callbacks = new Map<number, { wall: number; cpu: NodeJS.CpuUsage }>();
+  const record = (wallMs: number, cpu: NodeJS.CpuUsage) => {
+    const cpuMs = (cpu.user + cpu.system) / 1_000;
+    // Windows CPU accounting is coarse. Cap charged CPU by the real span so a
+    // counter increment cannot inflate a short callback. Zero-CPU waits drop out.
+    blockingMs = Math.max(blockingMs, Math.min(wallMs, cpuMs));
+    if (wallMs > longestCallbackMs) {
+      longestCallbackMs = wallMs;
+      longestCallbackCpuMs = cpuMs;
+    }
+  };
+  const observeGc = (entries: readonly PerformanceEntry[]) => {
+    for (const entry of entries) {
+      if (entry.startTime < startedAt) continue;
+      gcCount++;
+      longestGcMs = Math.max(longestGcMs, entry.duration);
+    }
+  };
+  const gc = new PerformanceObserver((list) => observeGc(list.getEntries()));
+  gc.observe({ entryTypes: ["gc"] });
+  const activity = createHook({
+    init(_id, type) {
+      if (type.startsWith("FSREQ")) filesystemRequests++;
+      if (!draining && type === "Immediate") yields++;
+      if (type === "WORKER") workerEvents++;
+    },
+    before(id) {
+      const resource = executionAsyncResource();
+      if (resource === probe || resource === drain) return;
+      if (resource instanceof MessagePort) workerEvents++;
+      callbacks.set(id, { wall: performance.now(), cpu: threadCpuUsage() });
+    },
+    after(id) {
+      const start = callbacks.get(id);
+      if (!start) return;
+      callbacks.delete(id);
+      record(performance.now() - start.wall, threadCpuUsage(start.cpu));
+    },
+  }).enable();
+  const sample = () => {
+    const now = performance.now();
+    const cpu = threadCpuUsage();
+    const wallMs = now - previous;
+    longestGapMs = Math.max(longestGapMs, wallMs);
+    // Worker construction/message cloning can precede async_hooks.before.
+    if (workerEvents) {
+      blockingMs = Math.max(
+        blockingMs,
+        Math.min(wallMs, (cpu.user - previousCpu.user + cpu.system - previousCpu.system) / 1_000),
+      );
+    }
+    workerEvents = 0;
+    previous = now;
+    previousCpu = cpu;
+  };
+  probe = setInterval(sample, 1);
+  let result: T;
+  try {
+    // The caller's current callback began before the hook was installed.
+    const wall = performance.now();
+    const cpu = threadCpuUsage();
+    const pending = action();
+    record(performance.now() - wall, threadCpuUsage(cpu));
+    result = await pending;
+    draining = true;
+    await new Promise<void>((resolve) => {
+      drain = setImmediate(resolve);
+    });
+    sample();
+    observeGc(gc.takeRecords());
+  } finally {
+    clearInterval(probe);
+    activity.disable();
+    gc.disconnect();
+  }
+  // GC CPU inside callbacks is charged above. Elapsed GC spans, like elapsed
+  // callback spans, include preemption and remain diagnostics rather than gates.
+  process.stderr.write(
+    `${label}: ${JSON.stringify({ blockingMs, filesystemRequests, yields, longestCallbackMs, longestCallbackCpuMs, longestGcMs, gcCount, longestGapMs })}\n`,
+  );
+  return { result, blockingMs, filesystemRequests, yields };
+}
@@ -3,7 +3,7 @@ import { execFile } from "node:child_process";
 import { promisify } from "node:util";
 import { fileURLToPath } from "node:url";
 import { writeFile, truncate, rm } from "node:fs/promises";
-import { setImmediate as nextTurn } from "node:timers/promises";
+import { measureMainThreadWork } from "../test-utils/main-thread-work.js";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { afterEach, beforeEach, describe, expect, it } from "vitest";
@@ -67,7 +67,7 @@ describe("paseo config file substrate", () => {
    );
  });

-  it("keeps the longest loop stall below 20ms while admitting large configs", async () => {
+  it("keeps main-thread work below 20ms while admitting large configs", async () => {
    const config = {
      scripts: Object.fromEntries(
        Array.from({ length: 7_000 }, (_, i) => [`script${i}`, { command: "echo hi" }]),
@@ -75,48 +75,27 @@ describe("paseo config file substrate", () => {
      worktree: { setup: "npm ci" },
    };
    await writeFile(join(tempDir, "paseo.json"), JSON.stringify(config));
-    let previous = performance.now();
-    let longestGap = 0;
-    const probe = setInterval(() => {
-      const now = performance.now();
-      longestGap = Math.max(longestGap, now - previous);
-      previous = now;
-    }, 1);
-    let result: Awaited<ReturnType<typeof readPaseoConfigForEdit>>;
-    try {
-      result = await readPaseoConfigForEdit(tempDir);
-      await nextTurn();
-      longestGap = Math.max(longestGap, performance.now() - previous);
-    } finally {
-      clearInterval(probe);
-    }
+    const { result, blockingMs, filesystemRequests, yields } = await measureMainThreadWork(
+      "project config 7000 scripts main-thread work",
+      () => readPaseoConfigForEdit(tempDir),
+    );
    expect(result).toEqual({ ok: true, config, revision: await statPaseoConfigPath(tempDir) });
-    process.stdout.write(`project config 7,000 scripts max loop gap: ${longestGap.toFixed(2)}ms\n`);
-    expect(longestGap).toBeLessThan(20);
+    expect(blockingMs).toBeLessThan(20);
+    expect(filesystemRequests).toBeGreaterThan(0);
+    expect(yields).toBeGreaterThan(0);
  });

-  it("rejects a 32MiB config with a loop stall below 20ms", async () => {
+  it("rejects a 32MiB config with main-thread work below 20ms", async () => {
    await writeFile(
      join(tempDir, "paseo.json"),
      JSON.stringify({ extra: "x".repeat(32 * 1024 * 1024) }),
    );
-    let previous = performance.now();
-    let longestGap = 0;
-    const probe = setInterval(() => {
-      const now = performance.now();
-      longestGap = Math.max(longestGap, now - previous);
-      previous = now;
-    }, 1);
-    let result: Awaited<ReturnType<typeof readPaseoConfigForEdit>>;
-    try {
-      result = await readPaseoConfigForEdit(tempDir);
-      await nextTurn();
-      longestGap = Math.max(longestGap, performance.now() - previous);
-    } finally {
-      clearInterval(probe);
-    }
-    process.stdout.write(`project config 32MiB max loop gap: ${longestGap.toFixed(2)}ms\n`);
-    expect(longestGap).toBeLessThan(20);
+    const { result, blockingMs, filesystemRequests } = await measureMainThreadWork(
+      "project config 32MiB main-thread work",
+      () => readPaseoConfigForEdit(tempDir),
+    );
+    expect(blockingMs).toBeLessThan(20);
+    expect(filesystemRequests).toBeGreaterThan(0);
    expect(result).toEqual({ ok: false, error: { code: "invalid_project_config" } });
  });