feat(agent): emit rtk token-savings telemetry at end of cloud run (#3146)

Author: pauldambra

packages/agent/src/acp-extensions.ts

@@ -78,6 +78,9 @@ export const POSTHOG_NOTIFICATIONS = {
 
   /** Permission request resolved, persisted so a reconnecting client can tell it is no longer pending */
   PERMISSION_RESOLVED: "_posthog/permission_resolved",
+
+  /** RTK output-compression token savings tallied at the end of a run */
+  RTK_SAVINGS: "_posthog/rtk_savings",
 } as const;
 
 /**

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

@@ -75,6 +75,7 @@ import {
 } from "./cloud-prompt";
 import { TaskRunEventStreamSender } from "./event-stream-sender";
 import { type JwtPayload, JwtValidationError, validateJwt } from "./jwt";
+import { resolveRtkSavings } from "./rtk-savings";
 import {
   handoffLocalGitStateSchema,
   jsonRpcRequestSchema,
@@ -304,6 +305,7 @@ export class AgentServer {
   private app: Hono;
   private posthogAPI: PostHogAPIClient;
   private eventStreamSender: TaskRunEventStreamSender | null = null;
+  private rtkSavingsEmitted = false;
   private questionRelayedToSlack = false;
   private adapterEmittedTurnComplete = false;
   private detectedPrUrl: string | null = null;
@@ -2838,6 +2840,11 @@ ${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,
@@ -3365,6 +3372,7 @@ ${signedCommitInstructions}
     }
 
     if (completeEventStream) {
+      await this.emitRtkSavings();
       await this.eventStreamSender?.stop();
     }
 
@@ -3373,6 +3381,47 @@ ${signedCommitInstructions}
     this.session = null;
   }
 
+  /**
+   * Emit RTK's output-compression token savings for this run as a telemetry
+   * event, so PostHog can report how much context RTK saved. Best-effort and
+   * cloud-only (no-op when the event stream isn't configured). Fires at most
+   * once per run: it's called from both terminal seams (the error path stops
+   * the stream in signalTaskComplete, before cleanupSession runs) and must land
+   * 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;
+
+    try {
+      const savings = await resolveRtkSavings();
+      if (!savings) return;
+
+      this.eventStreamSender.enqueue({
+        type: "notification",
+        timestamp: new Date().toISOString(),
+        notification: {
+          jsonrpc: "2.0",
+          method: POSTHOG_NOTIFICATIONS.RTK_SAVINGS,
+          // snake_case: these land as PostHog event properties.
+          params: {
+            task_id: this.config.taskId,
+            run_id: this.config.runId,
+            team_id: this.config.projectId,
+            total_commands: savings.totalCommands,
+            input_tokens: savings.inputTokens,
+            output_tokens: savings.outputTokens,
+            tokens_saved: savings.tokensSaved,
+            avg_savings_pct: savings.avgSavingsPct,
+          },
+        },
+      });
+      this.logger.debug("Emitted rtk savings", { ...savings });
+    } catch (error) {
+      this.logger.debug("Failed to emit rtk savings", { error });
+    }
+  }
+
   private async captureCheckpointState(
     localGitState?: HandoffLocalGitState,
   ): Promise<void> {

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

@@ -0,0 +1,99 @@
+import { describe, expect, it, vi } from "vitest";
+import { resolveRtkSavings } from "./rtk-savings";
+
+const GAIN_JSON = JSON.stringify({
+  summary: {
+    total_commands: 2,
+    total_input: 502691,
+    total_output: 5835,
+    total_saved: 496856,
+    avg_savings_pct: 98.8392471717218,
+    total_time_ms: 3456,
+    avg_time_ms: 1728,
+  },
+  daily: [],
+});
+
+function gain(stdout: string) {
+  return vi.fn().mockResolvedValue(stdout);
+}
+
+describe("resolveRtkSavings", () => {
+  it("parses the rtk gain summary into a typed savings object", async () => {
+    const runGain = gain(GAIN_JSON);
+    const savings = await resolveRtkSavings({
+      resolveBinary: () => "/bundled/rtk",
+      runGain,
+    });
+
+    expect(runGain).toHaveBeenCalledWith("/bundled/rtk", expect.anything());
+    expect(savings).toEqual({
+      totalCommands: 2,
+      inputTokens: 502691,
+      outputTokens: 5835,
+      tokensSaved: 496856,
+      avgSavingsPct: 98.8392471717218,
+    });
+  });
+
+  it("does not run rtk when the binary is unresolved (disabled / not found)", async () => {
+    const runGain = gain(GAIN_JSON);
+    const savings = await resolveRtkSavings({
+      resolveBinary: () => undefined,
+      runGain,
+    });
+
+    expect(savings).toBeNull();
+    expect(runGain).not.toHaveBeenCalled();
+  });
+
+  it.each([
+    [
+      "nothing was tracked (zero commands)",
+      JSON.stringify({ summary: { total_commands: 0, total_saved: 0 } }),
+    ],
+    ["malformed JSON", "not json"],
+    ["missing summary block", JSON.stringify({ daily: [] })],
+    ["the payload is JSON null", "null"],
+    ["the payload is a non-object", "42"],
+  ])("returns null when %s", async (_label, stdout) => {
+    const savings = await resolveRtkSavings({
+      resolveBinary: () => "/bundled/rtk",
+      runGain: gain(stdout),
+    });
+
+    expect(savings).toBeNull();
+  });
+
+  it("returns null when rtk gain throws rather than disrupting the run", async () => {
+    const savings = await resolveRtkSavings({
+      resolveBinary: () => "/bundled/rtk",
+      runGain: vi.fn().mockRejectedValue(new Error("rtk exploded")),
+    });
+
+    expect(savings).toBeNull();
+  });
+
+  it("coerces missing or non-numeric fields to zero", async () => {
+    const savings = await resolveRtkSavings({
+      resolveBinary: () => "/bundled/rtk",
+      runGain: gain(
+        JSON.stringify({
+          summary: {
+            total_commands: 3,
+            total_saved: "lots",
+            avg_savings_pct: null,
+          },
+        }),
+      ),
+    });
+
+    expect(savings).toEqual({
+      totalCommands: 3,
+      inputTokens: 0,
+      outputTokens: 0,
+      tokensSaved: 0,
+      avgSavingsPct: 0,
+    });
+  });
+});

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

@@ -0,0 +1,86 @@
+import { execFile } from "node:child_process";
+import { promisify } from "node:util";
+import { resolveRtkPrefix } from "../adapters/claude/session/rtk";
+
+const execFileAsync = promisify(execFile);
+
+/**
+ * The `summary` block of `rtk gain --format json` — RTK's own tally of the
+ * output it compressed away before it reached the model. The counts are RTK's
+ * token estimates (its own output labels them "Input/Output tokens" and "Tokens
+ * saved"), not raw bytes.
+ */
+export interface RtkSavingsSummary {
+  totalCommands: number;
+  inputTokens: number;
+  outputTokens: number;
+  tokensSaved: number;
+  avgSavingsPct: number;
+}
+
+interface ResolveRtkSavingsOptions {
+  env?: NodeJS.ProcessEnv;
+  /** Resolves the rtk binary to invoke; undefined disables reporting. Overridable for tests. */
+  resolveBinary?: (env: NodeJS.ProcessEnv) => string | undefined;
+  /** Runs `rtk gain` and returns its stdout; overridable for tests. */
+  runGain?: (binary: string, env: NodeJS.ProcessEnv) => Promise<string>;
+}
+
+function toFiniteNumber(value: unknown): number {
+  return typeof value === "number" && Number.isFinite(value) ? value : 0;
+}
+
+function parseGainSummary(stdout: string): RtkSavingsSummary | null {
+  const parsed: unknown = JSON.parse(stdout);
+  // `JSON.parse("null")` returns null, and a non-object payload has no summary —
+  // guard before indexing rather than leaning on the caller's try/catch.
+  if (!parsed || typeof parsed !== "object") return null;
+  const summary = (parsed as { summary?: Record<string, unknown> }).summary;
+  if (!summary || typeof summary !== "object") return null;
+  return {
+    totalCommands: toFiniteNumber(summary.total_commands),
+    inputTokens: toFiniteNumber(summary.total_input),
+    outputTokens: toFiniteNumber(summary.total_output),
+    tokensSaved: toFiniteNumber(summary.total_saved),
+    avgSavingsPct: toFiniteNumber(summary.avg_savings_pct),
+  };
+}
+
+async function defaultRunGain(
+  binary: string,
+  env: NodeJS.ProcessEnv,
+): Promise<string> {
+  const { stdout } = await execFileAsync(binary, ["gain", "--format", "json"], {
+    timeout: 5_000,
+    env,
+  });
+  return stdout;
+}
+
+/**
+ * Reads RTK's own token-savings tally (`rtk gain --format json`).
+ *
+ * Best-effort: returns null when RTK is disabled or unavailable, when it has
+ * tracked nothing, or on any exec/parse failure — reporting savings must never
+ * disrupt a run. The tally is machine-global, which equals a single run's
+ * savings in the ephemeral cloud sandbox (fresh DB per run). A long-lived host
+ * would need to snapshot-and-diff instead.
+ */
+export async function resolveRtkSavings({
+  env = process.env,
+  resolveBinary = resolveRtkPrefix,
+  runGain = defaultRunGain,
+}: ResolveRtkSavingsOptions = {}): Promise<RtkSavingsSummary | null> {
+  const binary = resolveBinary(env);
+  if (!binary) return null;
+
+  try {
+    const stdout = await runGain(binary, env);
+    const summary = parseGainSummary(stdout);
+    // No rtk-wrapped commands ran this session — nothing worth reporting.
+    if (!summary || summary.totalCommands <= 0) return null;
+    return summary;
+  } catch {
+    return null;
+  }
+}