fix(sdk-pi): write the error log without blocking pi

packages/pi/README.md

@@ -62,6 +62,9 @@ extension errors to `telemetry-dev.log` in the pi agent directory: `~/.pi/agent`
 `PI_CODING_AGENT_DIR` when you set it. The file has mode `0600`. When the file is more than 5 MB,
 the integration moves it to `telemetry-dev.log.1`.
 
+Writes happen in the background, so a slow disk does not hold up pi. When 100 entries are already
+waiting, later ones are dropped. `session_shutdown` waits up to one second for the queued entries.
+
 A series of failed exports, for example during an ingest outage, is one entry. The next failure
 after an export reaches the ingest is a new entry.
 

packages/pi/src/error-log.ts

@@ -1,25 +1,72 @@
-import { appendFileSync, renameSync, statSync } from "node:fs";
+import { appendFile, rename, stat } from "node:fs/promises";
 import { homedir } from "node:os";
 import { join } from "node:path";
 import { inspect } from "node:util";
 
 /** pi moves its own `mcp.log` to `mcp.log.1` when the file is more than 5 MB. */
 const MAX_LOG_BYTES = 5 * 1024 * 1024;
 
+/** Entries waiting for the disk; new entries are dropped while this many are pending. */
+const MAX_PENDING_ENTRIES = 100;
+
+/** How long `flushErrorLog` waits for a stalled disk before giving up. */
+const FLUSH_DEADLINE_MS = 1000;
+
+let pending = 0;
+let queue: Promise<void> = Promise.resolve();
+
 /**
- * Appends an error to `telemetry-dev.log` in pi's agent directory. pi draws its TUI on the
- * terminal, so an error written to stderr garbles the screen and is not kept.
+ * Queues an error for `telemetry-dev.log` in pi's agent directory and returns without waiting
+ * on the disk. pi draws its TUI on the terminal, so an error written to stderr garbles the
+ * screen and is not kept.
  */
 export function writeErrorLog(error: unknown): void {
+  if (pending >= MAX_PENDING_ENTRIES) return;
+
   const path = join(agentDir(), "telemetry-dev.log");
   const entry = `${new Date().toISOString()} ${inspect(error).replaceAll("\n", "\n    ")}\n`;
 
+  pending += 1;
+  queue = queue
+    .then(() => append(path, entry))
+    .finally(() => {
+      pending -= 1;
+    });
+}
+
+/** Resolves once every queued entry has been written or has failed. Never rejects. */
+export function drainErrorLogForTesting(): Promise<void> {
+  return queue;
+}
+
+/**
+ * Resolves once every queued entry has been written or has failed, or after `FLUSH_DEADLINE_MS`
+ * if a write stalls. Never rejects.
+ */
+export async function flushErrorLog(): Promise<void> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const deadline = new Promise<void>((resolve) => {
+    timer = setTimeout(resolve, FLUSH_DEADLINE_MS);
+    timer.unref?.();
+  });
+
+  try {
+    await Promise.race([queue, deadline]);
+  } finally {
+    clearTimeout(timer);
+  }
+}
+
+async function append(path: string, entry: string): Promise<void> {
   try {
-    if ((statSync(path, { throwIfNoEntry: false })?.size ?? 0) > MAX_LOG_BYTES) {
-      renameSync(path, `${path}.1`);
-    }
+    const size = await stat(path).then(
+      (stats) => stats.size,
+      () => 0,
+    );
+
+    if (size > MAX_LOG_BYTES) await rename(path, `${path}.1`);
 
-    appendFileSync(path, entry, { mode: 0o600 });
+    await appendFile(path, entry, { mode: 0o600 });
   } catch {
     // The log is best effort; telemetry must never fail a pi session.
   }

packages/pi/src/extension.ts

@@ -10,7 +10,7 @@ import {
 } from "@telemetry-dev/sdk";
 
 import { ensureInit, type ClientOverrides, type TelemetryDevPiOptions } from "./config.ts";
-import { writeErrorLog } from "./error-log.ts";
+import { flushErrorLog, writeErrorLog } from "./error-log.ts";
 
 /** Options for {@link telemetryDevExtension}. */
 export interface TelemetryDevExtensionOptions extends TelemetryDevPiOptions {
@@ -454,6 +454,9 @@ export function telemetryDevExtension(
         } catch (error) {
           reportError(onError, error);
         }
+
+        // pi can exit right after shutdown; land the errors that flush just reported.
+        await flushErrorLog();
       },
     );
   };

packages/pi/tests/error-log.test.ts

@@ -0,0 +1,35 @@
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+import { expect, test, vi } from "vitest";
+
+import { flushErrorLog, writeErrorLog } from "../src/error-log.ts";
+
+const appendFile = vi.hoisted(() => vi.fn(() => new Promise<void>(() => {})));
+
+// A separate file, so the stalled write below cannot hold up the queue in other tests.
+vi.mock("node:fs/promises", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("node:fs/promises")>()),
+  appendFile,
+}));
+
+test("flushErrorLog waits for a stalled write until its deadline, then gives up", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "telemetry-pi-agent-"));
+  vi.stubEnv("PI_CODING_AGENT_DIR", directory);
+
+  try {
+    writeErrorLog(new Error("stalled"));
+    await vi.waitFor(() => expect(appendFile).toHaveBeenCalledOnce());
+
+    const started = performance.now();
+    await flushErrorLog();
+    const elapsed = performance.now() - started;
+
+    expect(elapsed).toBeGreaterThanOrEqual(900);
+    expect(elapsed).toBeLessThan(3000);
+  } finally {
+    vi.unstubAllEnvs();
+    await rm(directory, { recursive: true, force: true });
+  }
+});

packages/pi/tests/pi.test.ts

@@ -1,3 +1,4 @@
+import { existsSync } from "node:fs";
 import { mkdtemp, readFile, rm, stat, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
@@ -21,6 +22,7 @@ import {
 import { afterEach, expect, test, vi } from "vitest";
 
 import { resetForTesting } from "../src/config.ts";
+import { drainErrorLogForTesting, writeErrorLog } from "../src/error-log.ts";
 import { telemetryDevExtension, type TelemetryDevExtensionOptions } from "../src/extension.ts";
 
 interface Exporters extends ClientOverrides {
@@ -753,10 +755,11 @@ test("errors go to a private, rotated telemetry-dev.log in the agent directory,
     // An earlier log of more than 5 MB moves to telemetry-dev.log.1.
     await writeFile(path, "x".repeat(5 * 1024 * 1024 + 1));
     const handlers = new Map<string, (event: object, ctx: ExtensionContext) => unknown>();
+    let status = 200;
     telemetryDevExtension({
       apiKey: "td_live_test",
       exportMode: "immediate",
-      fetch: async () => new Response(null, { status: 400 }),
+      fetch: async () => new Response(null, { status }),
     })({ on: (type: string, handler: never) => handlers.set(type, handler) });
 
     await handlers.get("session_start")?.({ reason: "startup" }, makeContext());
@@ -769,6 +772,9 @@ test("errors go to a private, rotated telemetry-dev.log in the agent directory,
       },
       makeContext(),
     );
+    // The export that shutdown flushes fails; its report must still reach the log.
+    status = 400;
+    await handlers.get("session_shutdown")?.({ reason: "quit" }, makeContext());
     await resetForTesting();
 
     const log = await readFile(path, "utf8");
@@ -783,6 +789,31 @@ test("errors go to a private, rotated telemetry-dev.log in the agent directory,
   }
 });
 
+test("the error log keeps call order and drops entries past the pending limit", async () => {
+  const directory = await mkdtemp(join(tmpdir(), "telemetry-pi-agent-"));
+  vi.stubEnv("PI_CODING_AGENT_DIR", directory);
+
+  try {
+    for (let index = 0; index < 150; index++) writeErrorLog(new Error(`entry-${index}`));
+
+    // writeErrorLog returns before touching the disk.
+    expect(existsSync(join(directory, "telemetry-dev.log"))).toBe(false);
+    await drainErrorLogForTesting();
+    const entries = [
+      ...(await readFile(join(directory, "telemetry-dev.log"), "utf8")).matchAll(/entry-(\d+)/g),
+    ].map((match) => Number(match[1]));
+
+    expect(entries).toEqual(Array.from({ length: 100 }, (_, index) => index));
+
+    writeErrorLog(new Error("after drain"));
+    await drainErrorLogForTesting();
+    expect(await readFile(join(directory, "telemetry-dev.log"), "utf8")).toContain("after drain");
+  } finally {
+    vi.unstubAllEnvs();
+    await rm(directory, { recursive: true, force: true });
+  }
+});
+
 test("each event reads the current session id", async () => {
   const harness = makeHarness();
   await harness.emit({ type: "session_start", reason: "startup" }, makeContext("session-2"));