fix(agent): keep rtk cache rename on one filesystem and pin error-path emit ordering

- mkdtempSync beside cacheDir (not os.tmpdir()) so renameSync stays
  same-filesystem and never throws EXDEV on Linux/Docker containers
- wrap renameSync in try/catch: swallow Windows destination-exists error
  when the winner already placed the file (concurrent cold-cache builds)
- fix error message to interpolate extractedBinary (the path checked)
  not cachedBinary (the destination)
- signalTaskComplete ordering test now passes a savings-returning stub so
  the rtk_savings enqueue is asserted to precede stop() on that seam
- fix ENOBUFS comment -> ERR_CHILD_PROCESS_STDIO_MAXBUFFER (real error)
- fix double space in emitRtkSavings comment

Generated-By: PostHog Code
Task-Id: b86391cc-4634-4a2b-ae34-fe1f9c0626a0

Author: pauldambra

packages/agent/build/rtk-binary.mjs

@@ -188,17 +188,27 @@ export async function ensureRtkBinary(destDir) {
     }
     // Extract into a temp dir first so a killed build cannot leave a partial
     // binary in cacheDir that existsSync accepts on the next run.
-    const tmpDir = mkdtempSync(join(tmpdir(), "posthog-rtk-"));
+    // Use dirname(cacheDir) so the temp dir and cacheDir are on the same
+    // filesystem: rename(2) fails with EXDEV across filesystems (e.g. tmpfs /tmp
+    // vs overlay node_modules/.cache in Docker / CI containers).
+    const tmpDir = mkdtempSync(join(dirname(cacheDir), ".tmp-"));
     try {
       await extractArchive(archivePath, tmpDir, target);
       rmSync(archivePath, { force: true });
       const extractedBinary = join(tmpDir, binName);
       if (!existsSync(extractedBinary)) {
         throw new Error(
-          `rtk binary missing after extraction: ${cachedBinary} — the release archive layout for ${target} may have changed`,
+          `rtk binary missing after extraction: ${extractedBinary} — the release archive layout for ${target} may have changed`,
         );
       }
-      renameSync(extractedBinary, cachedBinary);
+      try {
+        renameSync(extractedBinary, cachedBinary);
+      } catch (renameError) {
+        // Two concurrent cold-cache builds can race to the same destination.
+        // On Windows renameSync throws when the destination already exists;
+        // if the winner already wrote the file, the loser can proceed safely.
+        if (!existsSync(cachedBinary)) throw renameError;
+      }
     } finally {
       rmSync(tmpDir, { force: true, recursive: true });
     }

packages/agent/src/server/agent-server.test.ts

@@ -549,17 +549,24 @@ describe("AgentServer HTTP Mode", () => {
 
     it("writes terminal failure status before completing event ingest", async () => {
       const order: string[] = [];
-      const testServer = createServer() as unknown as {
+      // Use a savings-returning resolver so the rtk_savings enqueue fires on this
+      // path — the null default would skip emitRtkSavings and leave the ordering
+      // of rtk enqueue < stop() untested on the signalTaskComplete seam.
+      const testServer = createServer({
+        resolveRtkSavings: async () => ({
+          totalCommands: 1,
+          inputTokens: 100,
+          outputTokens: 50,
+          tokensSaved: 30,
+          avgSavingsPct: 30,
+        }),
+      }) as unknown as {
         eventStreamSender: {
-          enqueue: (event: Record<string, unknown>) => void;
-          stop: () => Promise<void>;
+          enqueue: ReturnType<typeof vi.fn>;
+          stop: ReturnType<typeof vi.fn>;
         };
         posthogAPI: {
-          updateTaskRun: (
-            taskId: string,
-            runId: string,
-            payload: Record<string, unknown>,
-          ) => Promise<unknown>;
+          updateTaskRun: ReturnType<typeof vi.fn>;
         };
         signalTaskComplete(
           payload: JwtPayload,
@@ -595,7 +602,8 @@ describe("AgentServer HTTP Mode", () => {
         "boom",
       );
 
-      expect(order).toEqual(["enqueue", "update", "stop"]);
+      // error enqueue -> status update -> rtk_savings enqueue -> stop
+      expect(order).toEqual(["enqueue", "update", "enqueue", "stop"]);
       expect(testServer.eventStreamSender.enqueue).toHaveBeenCalledWith(
         expect.objectContaining({
           type: "notification",
@@ -613,6 +621,13 @@ describe("AgentServer HTTP Mode", () => {
           error_message: "boom",
         },
       );
+      // Assert rtk_savings enqueue precedes stop() on this seam.
+      const enqueueOrders =
+        testServer.eventStreamSender.enqueue.mock.invocationCallOrder;
+      const stopOrder =
+        testServer.eventStreamSender.stop.mock.invocationCallOrder[0];
+      // The last enqueue (rtk_savings) must land before stop()
+      expect(enqueueOrders[enqueueOrders.length - 1]).toBeLessThan(stopOrder);
     });
 
     it("still stops event ingest when terminal failure status update fails", async () => {

packages/agent/src/server/agent-server.ts

@@ -3392,7 +3392,7 @@ ${signedCommitInstructions}
   private async emitRtkSavings(): Promise<void> {
     if (!this.eventStreamSender || this.rtkSavingsAttempted) return;
     // Set before the await: a failed attempt still counts — prevents retry
-    // storms on the terminal path.  The flag name reflects this intent.
+    // storms on the terminal path. The flag name reflects this intent.
     this.rtkSavingsAttempted = true;
 
     try {

packages/agent/src/server/rtk-savings.ts

@@ -53,7 +53,7 @@ async function defaultRunGain(
   const { stdout } = await execFileAsync(binary, ["gain", "--format", "json"], {
     timeout: 5_000,
     // The daily array grows on long-lived hosts; cap the buffer explicitly so
-    // ENOBUFS never silently swallows savings on desktop reuse.
+    // ERR_CHILD_PROCESS_STDIO_MAXBUFFER never silently swallows savings on desktop reuse.
     maxBuffer: 10 * 1024 * 1024,
     env,
   });