fix(agent): clarify at-most-once emit semantics and harden rtk binary extraction

Rename rtkSavingsEmitted -> rtkSavingsAttempted (the flag is set before the
await, so a failed attempt still prevents retries; the name now says so).
Move emitRtkSavings after updateTaskRun on the error path so the "failed"
status lands before the up-to-5s rtk gain probe. Extract the cached rtk binary
atomically via a temp dir + rename so a killed build cannot leave a partial
binary that existsSync accepts on subsequent runs. Add explicit maxBuffer to
the execFileAsync call to make the 10 MB cap intentional. Add readability
comments to the double-cleanup test.

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

Author: pauldambra

packages/agent/build/rtk-binary.mjs

@@ -6,10 +6,13 @@ import {
   createWriteStream,
   existsSync,
   mkdirSync,
+  mkdtempSync,
   readFileSync,
+  renameSync,
   rmSync,
   writeFileSync,
 } from "node:fs";
+import { tmpdir } from "node:os";
 import { dirname, join, resolve } from "node:path";
 import { pipeline } from "node:stream/promises";
 import { setTimeout as sleep } from "node:timers/promises";
@@ -183,12 +186,21 @@ export async function ensureRtkBinary(destDir) {
       rmSync(archivePath, { force: true });
       throw error;
     }
-    await extractArchive(archivePath, cacheDir, target);
-    rmSync(archivePath, { force: true });
-    if (!existsSync(cachedBinary)) {
-      throw new Error(
-        `rtk binary missing after extraction: ${cachedBinary} — the release archive layout for ${target} may have changed`,
-      );
+    // 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-"));
+    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`,
+        );
+      }
+      renameSync(extractedBinary, cachedBinary);
+    } finally {
+      rmSync(tmpDir, { force: true, recursive: true });
     }
   }
 

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

@@ -517,9 +517,13 @@ describe("AgentServer HTTP Mode", () => {
       const session = testServer.session;
       await testServer.cleanupSession({ completeEventStream: true });
       // A failed run hits both terminal seams; the savings must only count once.
+      // Re-arm session: the first cleanup nulled this.session, so without this
+      // the second cleanup exits early and never reaches emitRtkSavings.
       testServer.session = session;
       await testServer.cleanupSession({ completeEventStream: true });
 
+      // The once-only assertion is meaningful because the stubbed stop() does
+      // not gate enqueue — rtkSavingsAttempted is the actual guard under test.
       expect(testServer.eventStreamSender.enqueue).toHaveBeenCalledOnce();
       expect(testServer.eventStreamSender.enqueue).toHaveBeenCalledWith(
         expect.objectContaining({

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

@@ -305,7 +305,7 @@ export class AgentServer {
   private app: Hono;
   private posthogAPI: PostHogAPIClient;
   private eventStreamSender: TaskRunEventStreamSender | null = null;
-  private rtkSavingsEmitted = false;
+  private rtkSavingsAttempted = false;
   private questionRelayedToSlack = false;
   private adapterEmittedTurnComplete = false;
   private detectedPrUrl: string | null = null;
@@ -2840,11 +2840,6 @@ ${signedCommitInstructions}
       error: errorMessage ?? "Agent error",
     });
 
-    // Emit before the finally below stops the stream — failed runs still used
-    // RTK-compressed commands and must be counted. cleanupSession's emit runs
-    // after this stop on the error path, so it can't rely on that one.
-    await this.emitRtkSavings();
-
     try {
       await this.posthogAPI.updateTaskRun(payload.task_id, payload.run_id, {
         status,
@@ -2854,6 +2849,11 @@ ${signedCommitInstructions}
     } catch (error) {
       this.logger.error("Failed to signal task completion", error);
     } finally {
+      // Emit before stop() — failed runs still used RTK-compressed commands
+      // and must be counted. cleanupSession's emit runs after this stop on
+      // the error path, so it can't rely on that one. Placed after
+      // updateTaskRun so the "failed" status lands promptly.
+      await this.emitRtkSavings();
       await this.eventStreamSender?.stop();
     }
   }
@@ -3390,8 +3390,10 @@ ${signedCommitInstructions}
    * before the stream is stopped, since enqueue is a no-op once stopped.
    */
   private async emitRtkSavings(): Promise<void> {
-    if (!this.eventStreamSender || this.rtkSavingsEmitted) return;
-    this.rtkSavingsEmitted = true;
+    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.
+    this.rtkSavingsAttempted = true;
 
     try {
       const savings = await (

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

@@ -52,6 +52,9 @@ async function defaultRunGain(
 ): Promise<string> {
   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.
+    maxBuffer: 10 * 1024 * 1024,
     env,
   });
   return stdout;