Merge pull request #1484 from corbitsdev/cl-10776-drop-snapshot-debounce

refactor(session): drop the run.json snapshot debounce

Author: TheGreatAxios

src/exec/runner.ts

@@ -102,7 +102,6 @@ import {
   clearActiveDisposeHost,
 } from "../session/active-host.js";
 import {
-  createRunSnapshotDebouncer,
   finalizeRunState,
   loadState,
   saveState,
@@ -536,18 +535,13 @@ export async function runExec(config: Config): Promise<ExecResult> {
       : {}),
   };
   let stopHeartbeat: (() => void) | undefined;
-  // CL-10667: debounced per-turn boundary snapshots; terminal exposes flush.
-  let runSnapshotDebouncer:
-    | ReturnType<typeof createRunSnapshotDebouncer>
-    | undefined;
 
   const persist = async (
     status: "running" | "done" | "failed" | "cancelled",
     extra?: { error?: string },
   ): Promise<void> => {
     if (finalized && status === "running") return;
     if (status !== "running") {
-      await runSnapshotDebouncer?.flush();
       finalized = true;
       stopHeartbeat?.();
       stopHeartbeat = undefined;
@@ -601,9 +595,6 @@ export async function runExec(config: Config): Promise<ExecResult> {
   };
 
   await persist("running");
-  runSnapshotDebouncer = createRunSnapshotDebouncer({
-    write: () => persist("running"),
-  });
   stopHeartbeat = startRunHeartbeat({
     shouldTick: () => !finalized,
     tick: () => persist("running"),
@@ -1057,8 +1048,7 @@ export async function runExec(config: Config): Promise<ExecResult> {
             activeRunHandle.cacheWriteModel = `${event.data.source.provider}:${event.data.source.model}`;
           }
         }
-        // CL-10667: debounced — the write recomputes state when it fires.
-        runSnapshotDebouncer?.schedule();
+        void persist("running");
       },
       resolveContextDir: () => workdir,
     });

src/session/run-snapshot-debounce.test.ts

@@ -131,62 +131,6 @@ export async function saveState(
   return write;
 }
 
-/**
- * Debounce window for mid-run `running` snapshots (CL-10667): coalesces the
- * per-turn boundary write to ~0.2/turn at a ~1s/turn cadence, while the
- * run-liveness heartbeat keeps mtime freshness over longer gaps.
- */
-export const RUN_SNAPSHOT_DEBOUNCE_MS = 5_000;
-
-export interface RunSnapshotDebouncer {
-  /** Mark a pending write and arm the one-shot timer once. */
-  schedule: () => void;
-  /** Cancel the timer and run any pending write immediately. */
-  flush: () => Promise<void>;
-}
-
-/**
- * Lazy payoff: the deferred write recomputes its snapshot when it fires, not
- * at schedule time, so a front of rapid turns persists the state current at
- * fire time. One timer; `schedule` while armed only re-marks dirty. The
- * writer maps write failures to undefined so a stalled flush cannot wedge
- * the terminal path that awaits it.
- */
-export function createRunSnapshotDebouncer(args: {
-  write: () => Promise<void>;
-  intervalMs?: number;
-}): RunSnapshotDebouncer {
-  const intervalMs = args.intervalMs ?? RUN_SNAPSHOT_DEBOUNCE_MS;
-  let timer: ReturnType<typeof setTimeout> | null = null;
-  let dirty = false;
-  let inFlight: Promise<void> = Promise.resolve();
-
-  const run = (): void => {
-    dirty = false;
-    inFlight = args.write().catch(() => undefined);
-  };
-
-  return {
-    schedule: () => {
-      dirty = true;
-      if (timer !== null) return;
-      timer = setTimeout(() => {
-        timer = null;
-        if (dirty) run();
-      }, intervalMs);
-      timer.unref?.();
-    },
-    flush: async () => {
-      if (timer !== null) {
-        clearTimeout(timer);
-        timer = null;
-      }
-      if (dirty) run();
-      await inFlight;
-    },
-  };
-}
-
 // Terminal-write path: clears the in-memory active-run handle (active-run.ts)
 // alongside the on-disk status so the two cannot drift. Non-terminal
 // ("running") snapshots use saveState directly.

src/session/state.ts

@@ -1,6 +1,7 @@
 import { describe, expect, spyOn, test } from "bun:test";
 import { EventEmitter } from "node:events";
 import { AgentContextLockError, type Agent } from "@intx/agent";
+import { getLogger } from "@intx/log";
 import type { InferenceSource } from "@intx/types/runtime";
 
 import * as codexSession from "../../auth/codex/session.js";
@@ -17,6 +18,7 @@ import {
 } from "../../subagent/session-store.js";
 import type { ReactorInboundEvent } from "@intx/types/runtime";
 import { defined } from "../../../testkit/defined.js";
+import { LOG_NAMESPACE_ROOT } from "../../branding.js";
 import {
   stubReactorCapabilities,
   stubReactorState,
@@ -31,10 +33,12 @@ import {
   createDeliveryGeneration,
   createSessionOperationQueue,
 } from "../delivery-queue.js";
+import * as sessionState from "../../session/state.js";
 import {
   agentRebuildFailure,
   closeAgentForRebuild,
   createRunLifecycle,
+  createRunPersistence,
   finalizeTUIRun,
   observeRecoveryAttempts,
   resetSessionForRotation,
@@ -976,6 +980,62 @@ describe("rebuild close helpers", () => {
     ).toMatch(/restart/i);
   });
 
+  test("a rejecting saveState on the turn-boundary snapshot does not emit unhandledRejection", async () => {
+    const save = spyOn(sessionState, "saveState").mockRejectedValue(
+      new Error("disk full"),
+    );
+    const warn = spyOn(getLogger([LOG_NAMESPACE_ROOT, "tui"]), "warn");
+    const state = {
+      runTaskTitle: "t",
+      liveSource: { id: "p", model: "m", credentialId: "cred" },
+      startedAt: 1,
+      connectedMcpServers: [],
+      config: { cwd: "/tmp" },
+      sessionId: "sess-boundary",
+    } as unknown as RunnerState;
+    const services = {
+      runSink: { getTurnCount: () => 2 },
+      activatedToolNames: { list: () => [] },
+      activeRunHandle: {
+        sessionId: "sess-boundary",
+        cwd: "/tmp",
+        task: "t",
+        startedAt: 1,
+        turnsUsed: 0,
+      },
+      crashGuard: { isFinalized: () => false },
+    } as unknown as RunnerServices;
+    state.persistRunSnapshot = createRunPersistence(
+      state,
+      services,
+    ).persistRunSnapshot;
+
+    let unhandled: unknown = null;
+    const onUnhandledRejection = (reason: unknown): void => {
+      unhandled = reason;
+    };
+    process.on("unhandledRejection", onUnhandledRejection);
+    try {
+      // session.ts onTurnBoundarySnapshot: fire-and-forget running snapshot.
+      void state.persistRunSnapshot?.("running");
+      await new Promise((resolve) => setTimeout(resolve, 0));
+      expect(unhandled).toBeNull();
+      expect(save).toHaveBeenCalled();
+      expect(warn).toHaveBeenCalledWith(
+        "saveState failed for session {sessionId} status={status}: {error}",
+        {
+          sessionId: "sess-boundary",
+          status: "running",
+          error: "disk full",
+        },
+      );
+    } finally {
+      process.off("unhandledRejection", onUnhandledRejection);
+      save.mockRestore();
+      warn.mockRestore();
+    }
+  });
+
   // Overlay accept/decline tests stub bump() inside resolveSuspended, so
   // deleting the interrupt-site bump would not fail them. Drive the interrupt
   // helper itself.