[fix] Bound the session-records fetch and consume the persist-drop signal

services/runner/src/engines/sandbox_agent/run-limits.ts

@@ -10,6 +10,8 @@
  * legitimate, human-timescale wait, not a wedge, and must never be reaped by these deadlines.
  */
 
+import { envTimerMs } from "../../env.ts";
+
 export interface Clock {
   now(): number;
   setTimeout(fn: () => void, ms: number): NodeJS.Timeout;
@@ -22,13 +24,6 @@ const realClock: Clock = {
   clearTimeout: (handle) => clearTimeout(handle),
 };
 
-function envMs(name: string, defaultMs: number): number {
-  const raw = process.env[name];
-  if (raw === undefined || raw === "") return defaultMs;
-  const parsed = Number(raw);
-  return Number.isFinite(parsed) && parsed >= 0 ? parsed : defaultMs;
-}
-
 export const TOTAL_DEADLINE_ENV = "AGENTA_RUNNER_RUN_TOTAL_TIMEOUT_MS";
 export const IDLE_TIMEOUT_ENV = "AGENTA_RUNNER_RUN_IDLE_TIMEOUT_MS";
 export const TTFB_TIMEOUT_ENV = "AGENTA_RUNNER_RUN_TTFB_TIMEOUT_MS";
@@ -54,6 +49,8 @@ export interface ResolvedRunLimits {
 export function resolveRunLimits(
   log: (message: string) => void = () => {},
 ): ResolvedRunLimits {
+  const envMs = (name: string, defaultMs: number): number =>
+    envTimerMs(name, defaultMs, { log });
   const totalMs = envMs(TOTAL_DEADLINE_ENV, DEFAULT_TOTAL_DEADLINE_MS);
   let idleMs = envMs(IDLE_TIMEOUT_ENV, DEFAULT_IDLE_TIMEOUT_MS);
   const ttfbMs = envMs(TTFB_TIMEOUT_ENV, DEFAULT_TTFB_TIMEOUT_MS);

services/runner/src/env.ts

@@ -0,0 +1,105 @@
+/**
+ * Numeric environment config read against an explicit domain.
+ *
+ * Every ops-tunable number in the runner is a timer delay, a byte budget, or an attempt count —
+ * each with a domain the consuming API actually accepts. Hand-rolled `Number(process.env.X ?? d)`
+ * reads do not encode that domain, and every way of getting it wrong fails *silently in the
+ * direction of "no protection at all"*:
+ *
+ *  - junk ("30s", "5 min") parses to `NaN`; `setTimeout(fn, NaN)` fires on the next tick, so a
+ *    coalescing window or poll cadence collapses into a busy path instead of falling back.
+ *  - a sub-millisecond value floors to `0`; `AbortSignal.timeout(0)` aborts the request
+ *    immediately, so every call fails and the feature degrades permanently and quietly.
+ *  - a value above `TIMER_MAX_MS` overflows Node's 32-bit timer: the delay silently becomes 1 ms
+ *    (a "very long timeout" turns into an instant one), and past 2^32-1
+ *    `AbortSignal.timeout` throws `ERR_OUT_OF_RANGE` outright.
+ *
+ * So the domain is stated once, here, and clamped into rather than trusted. A bad override is
+ * never fatal — it falls back or clamps and warns once (a hot path must not spam stderr) —
+ * because a typo in one env var must not break every run in the process.
+ */
+
+/**
+ * Node timers (`setTimeout`, `AbortSignal.timeout`) take a delay that fits in a 32-bit signed
+ * int. Above this the delay silently clamps to 1 ms; `AbortSignal.timeout` throws past 2^32-1.
+ * ~24.8 days — no runner timeout is legitimately longer.
+ */
+export const TIMER_MAX_MS = 2_147_483_647;
+
+function defaultLog(msg: string): void {
+  process.stderr.write(`${msg}\n`);
+}
+
+/** Names already warned about, so a per-turn or per-poll read warns once, not every call. */
+const warnedNames = new Set<string>();
+
+/** Test-only: forget which names have warned, so a case can assert on the warning it triggers. */
+export function resetEnvWarnings(): void {
+  warnedNames.clear();
+}
+
+export interface EnvIntOptions {
+  /** Lowest accepted value; anything below is clamped up to it. Defaults to 1 — zero is a
+   *  degenerate delay/count everywhere in the runner, never a meaningful "disabled". */
+  min?: number;
+  /** Highest accepted value; anything above is clamped down to it. */
+  max?: number;
+  /** Where a bad-override warning goes. Defaults to stderr. */
+  log?: (msg: string) => void;
+}
+
+/**
+ * Read `name` as an integer inside `[min, max]`, falling back to `fallback` (a trusted code
+ * constant, used as-is) when unset or unparseable.
+ *
+ * Unset or empty is the normal case and stays silent; a value that was *set but unusable* is
+ * reported once, since it means the operator's intent is not in effect.
+ */
+export function envInt(
+  name: string,
+  fallback: number,
+  { min = 1, max = Number.MAX_SAFE_INTEGER, log = defaultLog }: EnvIntOptions = {},
+): number {
+  const raw = process.env[name];
+  if (raw === undefined || raw.trim() === "") return fallback;
+
+  const parsed = Number(raw);
+  if (!Number.isFinite(parsed)) {
+    warnOnce(name, `unparseable value '${raw}'; using ${fallback}`, log);
+    return fallback;
+  }
+
+  // Truncate first, then clamp: a fractional value like 0.5 would otherwise floor to a 0 that
+  // passes a naive `> 0` check.
+  const truncated = Math.trunc(parsed);
+  if (truncated < min) {
+    warnOnce(name, `${raw} below minimum ${min}; clamped`, log);
+    return min;
+  }
+  if (truncated > max) {
+    warnOnce(name, `${raw} above maximum ${max}; clamped`, log);
+    return max;
+  }
+  return truncated;
+}
+
+/**
+ * Read `name` as a timer delay in ms. Same as `envInt` with the ceiling Node's timers impose,
+ * so an override can shorten or lengthen a timeout but can never turn it into "fires instantly".
+ */
+export function envTimerMs(
+  name: string,
+  fallbackMs: number,
+  options: EnvIntOptions = {},
+): number {
+  return envInt(name, fallbackMs, {
+    ...options,
+    max: Math.min(options.max ?? TIMER_MAX_MS, TIMER_MAX_MS),
+  });
+}
+
+function warnOnce(name: string, detail: string, log: (msg: string) => void): void {
+  if (warnedNames.has(name)) return;
+  warnedNames.add(name);
+  log(`[env] ${name}: ${detail}`);
+}

services/runner/src/sessions/persist.ts

@@ -23,6 +23,7 @@
  */
 
 import { apiBase } from "../apiBase.ts";
+import { envInt, envTimerMs } from "../env.ts";
 import type { AgentEvent } from "../protocol.ts";
 import type { Redactor } from "../redaction.ts";
 import { stableRecordId } from "./record-id.ts";
@@ -33,6 +34,10 @@ const INGEST_RETRY_BASE_MS = 100;
 // 6 attempts ≈ 100+200+400+800+1600ms of backoff (~3.1s) — bounded per event so a real outage
 // can't hang the turn-end drain indefinitely.
 const DURABLE_INGEST_MAX_RETRIES = 6;
+// Ceiling on the override: the backoff doubles per attempt, so each extra attempt doubles the
+// worst-case drain wait. 12 attempts ≈ 3.4 min of backoff — the point past which "retry harder"
+// stops being a tuning knob and becomes a hung turn.
+const DURABLE_INGEST_MAX_RETRIES_CAP = 12;
 
 /** Durable-records upgrades (stronger retry + drop counting) are opt-in and read at call time
  * so the flag can be toggled per test. Off → the fire-and-forget legacy path, unchanged. */
@@ -42,8 +47,11 @@ function durableRecordsEnabled(): boolean {
 
 /** Attempts before a durable-mode drop; env-overridable for ops tuning (and fast tests). */
 function durableMaxRetries(): number {
-  const n = Number(process.env.AGENTA_RECORDS_INGEST_MAX_RETRIES);
-  return Number.isFinite(n) && n > 0 ? Math.floor(n) : DURABLE_INGEST_MAX_RETRIES;
+  return envInt("AGENTA_RECORDS_INGEST_MAX_RETRIES", DURABLE_INGEST_MAX_RETRIES, {
+    min: 1,
+    max: DURABLE_INGEST_MAX_RETRIES_CAP,
+    log,
+  });
 }
 
 function log(msg: string): void {
@@ -190,7 +198,7 @@ export function takePersistFailures(sessionId: string): number {
  * substitute for a close signal the harness may never send (a call that streams then
  * stalls without a `tool_result`).
  */
-const OPEN_TOOL_TTL_MS = Number(process.env.AGENTA_RECORD_TOOL_TTL_MS ?? 3000);
+const OPEN_TOOL_TTL_MS = envTimerMs("AGENTA_RECORD_TOOL_TTL_MS", 3_000, { log });
 
 /**
  * Build an emitter that persists every event via the ingest chain AND calls the
@@ -392,10 +400,20 @@ export function buildPersistingEmitter(
     );
   };
 
-  const flush = (): Promise<void> => {
+  const flush = async (): Promise<void> => {
     // A paused call ends the turn with its slot still open — persist it before draining.
     flushOpenTool();
-    return drainPersist(sessionId);
+    await drainPersist(sessionId);
+    // Consume the drop signal at the turn-end drain: records that exhausted retries mean the durable
+    // log is incomplete, so next turn's reconstruction may be missing context. Reading here also
+    // clears the per-session counter so it can't accumulate unread. (Only ever non-zero in durable
+    // mode.)
+    const dropped = takePersistFailures(sessionId);
+    if (dropped > 0) {
+      log(
+        `WARN session=${sessionId} durable log incomplete: ${dropped} record(s) dropped this turn; reconstruction may lack context`,
+      );
+    }
   };
 
   return { emit, persist, flush };

services/runner/src/sessions/records-query.ts

@@ -5,12 +5,29 @@
  */
 
 import { apiBase } from "../apiBase.ts";
+import { envTimerMs } from "../env.ts";
 import type { SessionRecordRow } from "./reconstruct.ts";
 
 function log(msg: string): void {
   process.stderr.write(`[sessions/records-query] ${msg}\n`);
 }
 
+/** Bound the reconstruction fetch: it sits on the turn's critical path (the prompt waits on it),
+ * so a stalled query must fail fast to the inbound-history fallback rather than hang on Undici's
+ * long request-level timeout. */
+const DEFAULT_QUERY_TIMEOUT_MS = 5_000;
+
+/** Env-overridable for ops tuning; read at call time so it can be tuned per test. The floor
+ * matters here: a timeout that resolved to 0 would abort every query on the spot, silently
+ * pinning reconstruction to the inbound-history fallback. */
+function queryTimeoutMs(): number {
+  return envTimerMs(
+    "AGENTA_SESSIONS_RECORDS_QUERY_TIMEOUT_MS",
+    DEFAULT_QUERY_TIMEOUT_MS,
+    { min: 1, log },
+  );
+}
+
 /**
  * Fetch a session's durable record log, ordered for reconstruction (the endpoint returns records
  * by ingest time, then per-turn `record_index`). Returns `null` on failure so the caller can fall
@@ -29,6 +46,7 @@ export async function fetchSessionRecords(
         authorization: auth(),
       },
       body: JSON.stringify({ session_id: sessionId }),
+      signal: AbortSignal.timeout(queryTimeoutMs()),
     });
     if (!res.ok) throw new Error(`HTTP ${res.status}`);
     const body = (await res.json()) as { records?: SessionRecordRow[] };

services/runner/tests/unit/env-num.test.ts

@@ -0,0 +1,96 @@
+/**
+ * The numeric-env reader (src/env.ts). Each case here is a way a hand-rolled
+ * `Number(process.env.X ?? d)` read used to fail open — silently producing a value that
+ * disables the protection the env var exists to tune.
+ */
+import { describe, it, beforeEach, assert, vi } from "vitest";
+
+import { envInt, envTimerMs, resetEnvWarnings, TIMER_MAX_MS } from "../../src/env.ts";
+
+const NAME = "AGENTA_TEST_ENV_NUM";
+
+describe("envInt", () => {
+  beforeEach(() => {
+    resetEnvWarnings();
+  });
+
+  it("unset or blank falls back silently (the normal case)", () => {
+    const warns: string[] = [];
+    assert.equal(envInt(NAME, 500, { log: (m) => warns.push(m) }), 500);
+    vi.stubEnv(NAME, "");
+    assert.equal(envInt(NAME, 500, { log: (m) => warns.push(m) }), 500);
+    vi.stubEnv(NAME, "   ");
+    assert.equal(envInt(NAME, 500, { log: (m) => warns.push(m) }), 500);
+    assert.deepEqual(warns, []);
+  });
+
+  it("a usable value wins", () => {
+    vi.stubEnv(NAME, "250");
+    assert.equal(envInt(NAME, 500), 250);
+  });
+
+  it("unparseable falls back and says so once", () => {
+    const warns: string[] = [];
+    vi.stubEnv(NAME, "30s");
+    assert.equal(envInt(NAME, 500, { log: (m) => warns.push(m) }), 500);
+    assert.equal(envInt(NAME, 500, { log: (m) => warns.push(m) }), 500);
+    assert.equal(warns.length, 1, "a per-turn read must warn once, not every call");
+    assert.match(warns[0], /unparseable value '30s'/);
+  });
+
+  // The CodeRabbit finding on #5501, at the root: 0.5 truncates to 0, and a 0 ms
+  // AbortSignal.timeout aborts the request on the spot.
+  it("a sub-unit value clamps to the minimum instead of truncating to zero", () => {
+    vi.stubEnv(NAME, "0.5");
+    assert.equal(envInt(NAME, 500, { min: 1 }), 1);
+  });
+
+  it("zero and negatives clamp to the minimum", () => {
+    vi.stubEnv(NAME, "0");
+    assert.equal(envInt(NAME, 500, { min: 1 }), 1);
+    vi.stubEnv(NAME, "-7");
+    assert.equal(envInt(NAME, 500, { min: 1 }), 1);
+  });
+
+  it("fractions above the minimum truncate", () => {
+    vi.stubEnv(NAME, "42.9");
+    assert.equal(envInt(NAME, 500), 42);
+  });
+
+  it("clamps down to an explicit maximum", () => {
+    const warns: string[] = [];
+    vi.stubEnv(NAME, "99");
+    assert.equal(envInt(NAME, 6, { min: 1, max: 12, log: (m) => warns.push(m) }), 12);
+    assert.match(warns[0], /above maximum 12/);
+  });
+});
+
+describe("envTimerMs", () => {
+  beforeEach(() => {
+    resetEnvWarnings();
+  });
+
+  it("clamps to the 32-bit timer ceiling rather than overflowing to 1ms", () => {
+    vi.stubEnv(NAME, String(TIMER_MAX_MS + 1));
+    assert.equal(envTimerMs(NAME, 5_000), TIMER_MAX_MS);
+    // Past 2^32-1 AbortSignal.timeout throws outright; the clamp keeps it constructible.
+    vi.stubEnv(NAME, "1e12");
+    resetEnvWarnings();
+    const ms = envTimerMs(NAME, 5_000);
+    assert.equal(ms, TIMER_MAX_MS);
+    assert.doesNotThrow(() => AbortSignal.timeout(ms));
+  });
+
+  it("keeps a caller's own maximum when it is tighter than the timer ceiling", () => {
+    vi.stubEnv(NAME, "60000");
+    assert.equal(envTimerMs(NAME, 5_000, { max: 10_000 }), 10_000);
+  });
+
+  it("never yields a delay that aborts immediately", () => {
+    for (const raw of ["0", "0.9", "-1", "abc", ""]) {
+      vi.stubEnv(NAME, raw);
+      resetEnvWarnings();
+      assert.isAtLeast(envTimerMs(NAME, 5_000), 1, `raw=${raw}`);
+    }
+  });
+});

services/runner/tests/unit/session-persist.test.ts

@@ -482,17 +482,48 @@ describe("durable records (AGENTA_RECORDS_DURABLE)", () => {
     vi.stubEnv("AGENTA_RECORDS_DURABLE", "true");
     vi.stubEnv("AGENTA_RECORDS_INGEST_MAX_RETRIES", "2"); // keep the test fast
     fetchFailCount = 99;
-    const { emit, flush } = buildPersistingEmitter("sess-durable", () => "t");
+    const { emit } = buildPersistingEmitter("sess-durable", () => "t");
     emit({ type: "message", text: "x" });
     emit({ type: "done" });
-    await flush();
+    // Drain directly (not flush): flush's turn-end consumer would read + clear the count.
+    await drainPersist("sess-durable");
 
     assert.equal(postedBodies.length, 0); // both records dropped
     assert.equal(takePersistFailures("sess-durable"), 2);
     // take clears the count.
     assert.equal(takePersistFailures("sess-durable"), 0);
   });
 
+  it("durable: the turn-end flush consumes the drop count (warns + clears)", async () => {
+    vi.stubEnv("AGENTA_RECORDS_DURABLE", "true");
+    vi.stubEnv("AGENTA_RECORDS_INGEST_MAX_RETRIES", "2");
+    fetchFailCount = 99;
+    const warns: string[] = [];
+    const writeSpy = vi
+      .spyOn(process.stderr, "write")
+      .mockImplementation((chunk: string | Uint8Array) => {
+        warns.push(String(chunk));
+        return true;
+      });
+    try {
+      const { emit, flush } = buildPersistingEmitter("sess-flush", () => "t");
+      emit({ type: "message", text: "x" });
+      await flush();
+
+      // The drain surfaced the incomplete durable log at turn end...
+      assert.equal(
+        warns.some(
+          (w) => w.includes("sess-flush") && w.includes("durable log incomplete"),
+        ),
+        true,
+      );
+    } finally {
+      writeSpy.mockRestore();
+    }
+    // ...and cleared the counter, so nothing accumulates unread.
+    assert.equal(takePersistFailures("sess-flush"), 0);
+  });
+
   it("durable: a transient failure recovers within the retry budget (no drop counted)", async () => {
     vi.stubEnv("AGENTA_RECORDS_DURABLE", "true");
     vi.stubEnv("AGENTA_RECORDS_INGEST_MAX_RETRIES", "5");