[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 { clampTimerMs, 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";
@@ -39,6 +34,8 @@ export const DEFAULT_IDLE_TIMEOUT_MS = 5 * 60_000; // 5 min
 export const DEFAULT_TTFB_TIMEOUT_MS = 2 * 60_000; // 2 min
 export const DEFAULT_TOOL_CALL_TIMEOUT_MS = 5 * 60_000; // 5 min
 
+/** Every field is a usable timer delay (integer ms, at least 1, within Node's timer range) —
+ *  `resolveRunLimits` guarantees it, so callers can arm any of them without re-checking. */
 export interface ResolvedRunLimits {
   totalMs: number;
   idleMs: number;
@@ -54,6 +51,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);
@@ -64,7 +63,16 @@ export function resolveRunLimits(
     );
     idleMs = Math.floor(totalMs / 2);
   }
-  return { totalMs, idleMs, ttfbMs, toolCallMs };
+  // Every field is armed directly as a timer delay, so the whole record leaves this function
+  // inside the timer domain — including the values DERIVED above rather than read from env
+  // (half of a total already sitting at its own floor rounds to 0, which fires instantly).
+  // Normalizing the result, not each derivation, keeps that true of any future adjustment here.
+  return {
+    totalMs: clampTimerMs(totalMs),
+    idleMs: clampTimerMs(idleMs),
+    ttfbMs: clampTimerMs(ttfbMs),
+    toolCallMs: clampTimerMs(toolCallMs),
+  };
 }
 
 export interface RunLimitsHandle {

services/runner/src/env.ts

@@ -0,0 +1,125 @@
+/**
+ * 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;
+  }
+
+  const clamped = clampInt(parsed, min, max);
+  if (clamped !== Math.trunc(parsed)) {
+    warnOnce(
+      name,
+      clamped === min
+        ? `${raw} below minimum ${min}; clamped`
+        : `${raw} above maximum ${max}; clamped`,
+      log,
+    );
+  }
+  return clamped;
+}
+
+/**
+ * Force an already-computed delay into the timer domain.
+ *
+ * `envInt` guards the read boundary, but a delay is often *derived* afterwards — a fraction of
+ * another limit, a backoff, a remaining budget — and arithmetic on two in-domain values can
+ * still land outside it (half of the smallest legal delay rounds to 0, which fires instantly).
+ * Run any derived delay through this so the domain holds by construction rather than by
+ * re-deriving the argument at each site.
+ */
+export function clampTimerMs(ms: number, { min = 1 }: { min?: number } = {}): number {
+  return clampInt(ms, min, TIMER_MAX_MS);
+}
+
+/** Truncate before clamping: a fractional value like 0.5 must not floor to a 0 that then
+ *  passes a naive `> 0` check. `NaN` is the only input with no magnitude to clamp toward, so
+ *  it takes the floor; ±Infinity has a direction and lands on the matching end of the range. */
+function clampInt(value: number, min: number, max: number): number {
+  if (Number.isNaN(value)) return min;
+  return Math.min(Math.max(Math.trunc(value), min), max);
+}
+
+/**
+ * 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 {
@@ -208,7 +216,7 @@ export function recordsIncomplete(sessionId: string): boolean {
  * 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
@@ -410,10 +418,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[] };