From c506ba3d172db317be682d22ea3a79d4a5decffa Mon Sep 17 00:00:00 2001
From: Ali Mohammad <freelancealimohammad@gmail.com>
Date: Sat, 24 Oct 2026 17:43:33 +0300
Subject: [PATCH 1/3] fix(observe): flush telemetry when an agent process
 crashes

An uncaught exception or unhandled rejection killed the process with the
reporter's buffer unsent. The reporter now watches uncaughtExceptionMonitor
(which also fires, origin 'unhandledRejection', for a rejection that
crashes) and sends what it holds - the batch an in-flight POST carried
included - in one request that blocks the dying process for at most 1s.

An async fetch from the monitor never opens its socket (Node exits as soon
as the monitor returns; measured 0 connections), so the POST runs in a
throwaway worker thread while the main thread sleeps on Atomics.wait.
No uncaughtException or unhandledRejection listener is added, so Node's
crash output, exit code and --unhandled-rejections mode are untouched; a
script with its own uncaughtException listener survives and keeps
reporting. close() removes the hook.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .fallowrc.json                             |   1 +
 docs/observability.md                      |   9 +-
 llms-full.txt                              |   9 +-
 src/deploy/adapters/cloudflare.test.ts     |   2 +-
 src/deploy/adapters/cloudflare.ts          |   4 +-
 src/observe/__fixtures__/crashAfterRun.mts |  53 +++++++++
 src/observe/blockingPost.ts                |  59 ++++++++++
 src/observe/reporter.test.ts               | 119 +++++++++++++++++++++
 src/observe/reporter.ts                    |  66 ++++++++++--
 9 files changed, 307 insertions(+), 15 deletions(-)
 create mode 100644 src/observe/__fixtures__/crashAfterRun.mts
 create mode 100644 src/observe/blockingPost.ts

diff --git a/.fallowrc.json b/.fallowrc.json
index 439f702d..eeb4c009 100644
--- a/.fallowrc.json
+++ b/.fallowrc.json
@@ -22,6 +22,7 @@
     "src/deploy/shims/pi.worker.ts",
     "src/deploy/shims/isolatedVm.worker.ts",
     "src/agentDir/__fixtures__/**/*.{ts,js}",
+    "src/observe/__fixtures__/*.mts",
     "examples/agent-dir/tools/*.ts",
     "registry/**/*.ts"
   ],
diff --git a/docs/observability.md b/docs/observability.md
index 86e71cb7..dbdc262a 100644
--- a/docs/observability.md
+++ b/docs/observability.md
@@ -416,7 +416,14 @@ down, the failed batch is requeued and retried with exponential backoff (1 s,
 2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. Only
 events evicted past the buffer cap (`maxBuffered`, default 5000, oldest
 first) or still unsent after the exit flush's final attempt are dropped and
-counted in the reporter's `dropped`. Eval scores flow through the same pipe —
+counted in the reporter's `dropped`. A process that dies of an uncaught
+exception or an unhandled rejection still reports its last run: the reporter
+watches `uncaughtExceptionMonitor` and sends what it holds in one request that
+holds up the crash for at most 1 s. Node's error output and exit code stay
+as they were, and a script with its own `uncaughtException` handler keeps
+reporting. An explicit `process.exit()` or a SIGKILL still loses what is
+buffered. Eval
+scores flow through the same pipe —
 `lousho eval --observe` emits them even when `LOUSHO_OBSERVE` is unset.
 
 ## The built server
diff --git a/llms-full.txt b/llms-full.txt
index 9123744c..ad332643 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -14082,7 +14082,14 @@ down, the failed batch is requeued and retried with exponential backoff (1 s,
 2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. Only
 events evicted past the buffer cap (`maxBuffered`, default 5000, oldest
 first) or still unsent after the exit flush's final attempt are dropped and
-counted in the reporter's `dropped`. Eval scores flow through the same pipe —
+counted in the reporter's `dropped`. A process that dies of an uncaught
+exception or an unhandled rejection still reports its last run: the reporter
+watches `uncaughtExceptionMonitor` and sends what it holds in one request that
+holds up the crash for at most 1 s. Node's error output and exit code stay
+as they were, and a script with its own `uncaughtException` handler keeps
+reporting. An explicit `process.exit()` or a SIGKILL still loses what is
+buffered. Eval
+scores flow through the same pipe —
 `lousho eval --observe` emits them even when `LOUSHO_OBSERVE` is unset.
 
 ## The built server
diff --git a/src/deploy/adapters/cloudflare.test.ts b/src/deploy/adapters/cloudflare.test.ts
index 6af8e859..a74a781c 100644
--- a/src/deploy/adapters/cloudflare.test.ts
+++ b/src/deploy/adapters/cloudflare.test.ts
@@ -162,7 +162,7 @@ describe('CloudflareWorkerAdapter', () => {
 
   it('findNodeBuiltinReferences accepts only the allowlisted getBuiltinModule probes', () => {
     const probes =
-      'loadBuiltinModule2("node:diagnostics_channel"); loadBuiltinModule4("node:dns"); process.getBuiltinModule?.("node:async_hooks"); loadBuiltinModule("node:module"); await loadNodeModule("node:dns"); getBuiltinModule("node:fs");';
+      'loadBuiltinModule2("node:diagnostics_channel"); loadBuiltinModule4("node:dns"); process.getBuiltinModule?.("node:async_hooks"); loadBuiltinModule("node:module"); await loadNodeModule("node:dns"); getBuiltinModule("node:fs"); getBuiltinModule(\'node:worker_threads\');';
     expect(findNodeBuiltinReferences(probes)).toEqual([]);
     // Any other builtin, or an allowlisted id outside a probe call, is still a leak.
     expect(findNodeBuiltinReferences('loadBuiltinModule("node:net"); const id = "node:dns"; import("node:async_hooks");')).toEqual([
diff --git a/src/deploy/adapters/cloudflare.ts b/src/deploy/adapters/cloudflare.ts
index 63134471..7d009e58 100644
--- a/src/deploy/adapters/cloudflare.ts
+++ b/src/deploy/adapters/cloudflare.ts
@@ -325,10 +325,12 @@ export default { fetch, scheduled };
  *    the worker graph via createAgent's LOUSHO_OBSERVE wiring; every access is
  *    guarded by `typeof process === 'undefined'` and degrades to an
  *    in-memory reporter identity on runtimes without builtins.
+ *  - node:worker_threads: src/observe/blockingPost.ts (the reporter's crash
+ *    flush), same guard; a runtime without it reports the send as failed.
  * Only these ids, and only as the argument of a `getBuiltinModule` /
  * `loadBuiltinModule` / `loadNodeModule` call, are exempt; any other `node:` string still fails.
  */
-const WORKER_SAFE_BUILTIN_PROBE = /\b(?:get|load)(?:Node|Builtin)Module\d*(?:\?\.)?\(\s*(["'`])node:(?:module|dns|diagnostics_channel|async_hooks|fs|crypto|os|path)\1\s*\)/g;
+const WORKER_SAFE_BUILTIN_PROBE = /\b(?:get|load)(?:Node|Builtin)Module\d*(?:\?\.)?\(\s*(["'`])node:(?:module|dns|diagnostics_channel|async_hooks|fs|crypto|os|path|worker_threads)\1\s*\)/g;
 
 /** Returns every `node:`-prefixed module specifier referenced in `source`. */
 export function findNodeBuiltinReferences(source: string): string[] {
diff --git a/src/observe/__fixtures__/crashAfterRun.mts b/src/observe/__fixtures__/crashAfterRun.mts
new file mode 100644
index 00000000..42eb5b08
--- /dev/null
+++ b/src/observe/__fixtures__/crashAfterRun.mts
@@ -0,0 +1,53 @@
+/**
+ * Spawned by reporter.test.ts ('crash flush'): build a reporter with its
+ * default exit hooks, finish one span (buffered: the interval flush is far
+ * off and the batch is below the size threshold), then die without closing
+ * it - the way an agent script dies of a bug after a run.
+ *
+ *   node --import tsx crashAfterRun.mts <mode>
+ *
+ * - throw:    a top-level throw (Node reports it with origin 'unhandledRejection')
+ * - reject:   an unhandled promise rejection
+ * - inflight: flushSize 1, so the span's batch is already in a POST the
+ *             collector holds open when a timer callback throws
+ * - handled:  the script has its own 'uncaughtException' listener, so it
+ *             survives the throw, finishes a second span and closes normally
+ *
+ * Env: LOUSHO_OBSERVE (collector URL), LOUSHO_FIXTURE_CWD (identity dir).
+ */
+import { observeExporter } from '../reporter';
+
+const mode = process.argv[2] ?? 'throw';
+const reporter = observeExporter({
+  url: process.env.LOUSHO_OBSERVE,
+  cwd: process.env.LOUSHO_FIXTURE_CWD,
+  project: 'crash-fixture',
+  agentName: 'crash-fixture',
+  heartbeatIntervalMs: 0,
+  ...(mode === 'inflight' ? { flushSize: 1 } : {}),
+});
+const finish = (id: string): void => {
+  const span = { id, name: 'invoke_agent crash-fixture', attributes: {}, startTime: Date.now() };
+  reporter.onSpanStart(span);
+  reporter.onSpanEnd({ ...span, endTime: Date.now() });
+};
+finish('crash-span');
+
+if (mode === 'reject') {
+  void Promise.reject(new Error('fixture: unhandled rejection after run'));
+} else if (mode === 'inflight') {
+  setTimeout(() => {
+    throw new Error('fixture: uncaught exception after run');
+  }, 300);
+} else if (mode === 'handled') {
+  process.on('uncaughtException', (error) => {
+    console.error(`handled: ${error.message}`);
+    finish('after-span');
+    void reporter.close();
+  });
+  setTimeout(() => {
+    throw new Error('fixture: uncaught exception after run');
+  }, 10);
+} else {
+  throw new Error('fixture: uncaught exception after run');
+}
diff --git a/src/observe/blockingPost.ts b/src/observe/blockingPost.ts
new file mode 100644
index 00000000..f5cfffb7
--- /dev/null
+++ b/src/observe/blockingPost.ts
@@ -0,0 +1,59 @@
+/**
+ * A POST that blocks the calling thread until it settles - the reporter's
+ * crash flush. `uncaughtExceptionMonitor` listeners run synchronously and
+ * Node exits the moment they return, so an ordinary `fetch` started there
+ * never even opens its socket (measured: 0 connections). Here the request
+ * runs in a throwaway worker thread with its own event loop while this
+ * thread sleeps on `Atomics.wait`; the worker flips a shared flag when the
+ * response (or an error) arrives.
+ *
+ * The worker is given no `execArgv` (no loaders such as tsx: it only needs
+ * the global `fetch`) and is left running when this returns: the process is
+ * about to exit, which terminates it.
+ *
+ * Builtins load through `process.getBuiltinModule`: this module ships in the
+ * Cloudflare Worker bundle (see WORKER_SAFE_BUILTIN_PROBE), where it reports
+ * `false` instead of breaking the bundle.
+ */
+
+interface WorkerThreadsLike {
+  Worker: new (code: string, options: { eval: true; execArgv: string[]; workerData: unknown }) => { unref(): void };
+}
+
+// Named to match the worker leak check's exempted call pattern (see
+// WORKER_SAFE_BUILTIN_PROBE in src/deploy/adapters/cloudflare.ts). Do not rename.
+function getBuiltinModule<T>(name: string): T | undefined {
+  if (typeof process === 'undefined') return undefined;
+  return (process.getBuiltinModule?.(name) as T | undefined) ?? undefined;
+}
+
+const PENDING = 0;
+const OK = 1;
+
+/** The worker's whole program (CommonJS, `eval: true`): one fetch, then publish the outcome. */
+const WORKER_SOURCE = `
+const { workerData: d } = process.getBuiltinModule("node:worker_threads");
+const flag = new Int32Array(d.flag);
+const done = (state) => { Atomics.store(flag, 0, state); Atomics.notify(flag, 0); };
+fetch(d.url, { method: 'POST', headers: d.headers, body: d.body, signal: AbortSignal.timeout(d.timeoutMs) })
+  .then((res) => done(res.ok ? ${OK} : 2), () => done(2));
+`;
+
+/**
+ * POST `body` to `url`, blocking at most `timeoutMs` (worker start-up
+ * included). True only when the server answered 2xx in time; false on any
+ * failure, a timeout, or a runtime without worker threads.
+ */
+export function blockingPost(url: string, headers: Record<string, string>, body: string, timeoutMs: number): boolean {
+  const threads = getBuiltinModule<WorkerThreadsLike>('node:worker_threads');
+  if (!threads || typeof SharedArrayBuffer === 'undefined') return false;
+  try {
+    const flag = new SharedArrayBuffer(4);
+    const worker = new threads.Worker(WORKER_SOURCE, { eval: true, execArgv: [], workerData: { flag, url, headers, body, timeoutMs } });
+    worker.unref(); // never keeps a process alive past a timed-out wait
+    Atomics.wait(new Int32Array(flag), 0, PENDING, timeoutMs);
+    return Atomics.load(new Int32Array(flag), 0) === OK;
+  } catch {
+    return false;
+  }
+}
diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts
index 281633d6..ecf9881d 100644
--- a/src/observe/reporter.test.ts
+++ b/src/observe/reporter.test.ts
@@ -5,12 +5,16 @@
  * drops-and-counts without ever throwing into the run, and the env
  * (`LOUSHO_OBSERVE`) wiring composes with an explicit exporter.
  */
+import { spawn } from 'node:child_process';
 import * as fs from 'node:fs';
+import * as http from 'node:http';
+import type { AddressInfo } from 'node:net';
 import * as os from 'node:os';
 import * as path from 'node:path';
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 import { GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/semconv';
 import type { Span, TraceExporter } from '../execution/tracing';
+import { blockingPost } from './blockingPost';
 import type { IngestEvent } from './envelope';
 import { fanOutExporters, observeExporter, observeExporterFromEnv, withObserveExporter, type ObserveReporter } from './reporter';
 
@@ -445,6 +449,121 @@ describe('observeExporter', () => {
   });
 });
 
+describe('crash flush', () => {
+  const fixture = path.join(__dirname, '__fixtures__', 'crashAfterRun.mts');
+  const repoRoot = path.resolve(__dirname, '..', '..');
+  type Mode = 'throw' | 'reject' | 'inflight' | 'handled';
+
+  /**
+   * Run the fixture against an in-test collector; resolves with its exit and
+   * the batch of every ingest request received. `holdFirst` never answers the
+   * first request (a POST still in flight when the process dies).
+   */
+  async function crash(mode: Mode, holdFirst = false): Promise<{ code: number | null; stderr: string; batches: IngestEvent[][] }> {
+    const batches: IngestEvent[][] = [];
+    const server = http.createServer((req, res) => {
+      let body = '';
+      req.on('data', (chunk: Buffer) => (body += chunk.toString()));
+      req.on('end', () => {
+        if (req.url === '/api/v1/ingest') batches.push((JSON.parse(body) as { batch: IngestEvent[] }).batch);
+        if (holdFirst && batches.length === 1) return;
+        res.writeHead(202, { 'content-type': 'application/json' }).end('{}');
+      });
+    });
+    await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+    const { port } = server.address() as AddressInfo;
+    try {
+      const child = spawn(process.execPath, ['--import', 'tsx', fixture, mode], {
+        cwd: repoRoot,
+        env: { ...process.env, LOUSHO_OBSERVE: `http://127.0.0.1:${port}`, LOUSHO_FIXTURE_CWD: cwd },
+        stdio: ['ignore', 'ignore', 'pipe'],
+      });
+      let stderr = '';
+      child.stderr.on('data', (chunk: Buffer) => (stderr += chunk.toString()));
+      const code = await new Promise<number | null>((resolve) => child.on('close', resolve));
+      return { code, stderr, batches };
+    } finally {
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  }
+  const obsIds = (batch: IngestEvent[]): string[] => batch.flatMap((e) => (e.type === 'observation' ? [e.data.obs_id] : []));
+
+  it('flushes pending events when the process dies of an uncaught exception', async () => {
+    const { code, stderr, batches } = await crash('throw');
+    expect(code).toBe(1); // Node's own crash exit code, unchanged
+    expect(stderr).toContain('fixture: uncaught exception after run');
+    expect(batches.flatMap(obsIds)).toEqual(['crash-span']);
+    expect(batches.flat().map((e) => e.type)).toContain('agent.deregister');
+  }, 30_000);
+
+  it('flushes pending events when the process dies of an unhandled rejection', async () => {
+    const { code, stderr, batches } = await crash('reject');
+    expect(code).toBe(1);
+    expect(stderr).toContain('fixture: unhandled rejection after run');
+    expect(batches.flatMap(obsIds)).toEqual(['crash-span']);
+  }, 30_000);
+
+  it('resends the batch a POST still in flight was carrying when the process died', async () => {
+    const { code, batches } = await crash('inflight', true);
+    expect(code).toBe(1);
+    expect(batches.map(obsIds)).toEqual([['crash-span'], ['crash-span']]);
+  }, 30_000);
+
+  it('keeps reporting when the script handles uncaughtException itself', async () => {
+    const { code, stderr, batches } = await crash('handled');
+    expect(code).toBe(0); // the user's listener kept the process alive; the reporter did not exit it
+    expect(stderr).toContain('handled: fixture: uncaught exception after run');
+    expect(batches.flatMap(obsIds)).toEqual(['crash-span', 'after-span']);
+  }, 30_000);
+
+  it('only flushes, and stays hooked, when an uncaughtException listener keeps the process alive', async () => {
+    // vitest itself listens for 'uncaughtException', so this is the survivable-throw path.
+    const before = process.listenerCount('uncaughtExceptionMonitor');
+    const { r, posted } = reporter({ registerExitHooks: true });
+    for (const id of ['one', 'two']) {
+      const span = makeSpan({ id });
+      r.onSpanStart(span);
+      r.onSpanEnd(span);
+      // Call the listeners directly: emitting the event for real would also reach vitest's own monitors.
+      const monitor = process.listeners('uncaughtExceptionMonitor').at(-1);
+      monitor?.(new Error('survived'), 'uncaughtException');
+      await vi.waitFor(() => expect(observations(posted).map((e) => e.data.obs_id)).toContain(id));
+    }
+    expect(process.listenerCount('uncaughtExceptionMonitor')).toBe(before + 1);
+    expect(allEvents(posted).map((e) => e.type)).not.toContain('agent.deregister'); // not closed
+  });
+
+  it('removes the crash hook on close()', async () => {
+    const before = process.listenerCount('uncaughtExceptionMonitor');
+    const { r } = reporter({ registerExitHooks: true });
+    expect(process.listenerCount('uncaughtExceptionMonitor')).toBe(before + 1);
+    await r.close();
+    expect(process.listenerCount('uncaughtExceptionMonitor')).toBe(before);
+  });
+});
+
+describe('blockingPost', () => {
+  it('reports false for an unreachable collector', () => {
+    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, '{}', 1000)).toBe(false);
+  });
+
+  it('gives up after its timeout', async () => {
+    // Blocking this thread means an in-process server can never answer: the wait must end on its own.
+    const server = http.createServer(() => undefined);
+    await new Promise<void>((resolve) => server.listen(0, '127.0.0.1', resolve));
+    const { port } = server.address() as AddressInfo;
+    try {
+      const t0 = Date.now();
+      expect(blockingPost(`http://127.0.0.1:${port}/`, {}, '{}', 200)).toBe(false);
+      expect(Date.now() - t0).toBeLessThan(1000);
+    } finally {
+      server.closeAllConnections();
+      await new Promise<void>((resolve) => server.close(() => resolve()));
+    }
+  });
+});
+
 describe('env wiring', () => {
   it('returns undefined when LOUSHO_OBSERVE is unset or off', () => {
     expect(observeExporterFromEnv({ env: {} })).toBeUndefined();
diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts
index eca5d4b2..65db3fab 100644
--- a/src/observe/reporter.ts
+++ b/src/observe/reporter.ts
@@ -14,8 +14,10 @@
  *   id (the `TraceLine.id`), so a resent batch upserts by key.
  * - Flush on size (50), on interval (5s), and on process exit (~2s cap:
  *   close() aborts a regular flush still in flight and makes one final
- *   attempt with the 2s timeout).
- *   Crash loss is accepted and bounded: nothing persists mid run.
+ *   attempt with the 2s timeout), and on a crash (an uncaught exception or
+ *   unhandled rejection: one POST that blocks the dying process for at most
+ *   1s, see blockingPost.ts; Node's crash output and exit code are untouched).
+ *   SIGKILL loss is accepted and bounded: nothing persists mid run.
  * - The daemon being down requeues the failed batch and retries with
  *   exponential backoff (1s base, 30s cap); only events evicted past the
  *   buffer cap, or still unsent after close()'s final attempt, drop into a
@@ -33,6 +35,7 @@ import { GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/sem
 import { SDKError } from '../execution/errors';
 import type { Span, TraceExporter } from '../execution/tracing';
 import { VERSION } from '../version';
+import { blockingPost } from './blockingPost';
 import { OBSERVE_DEFAULT_PORT, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope';
 import { observeTargetFromEnv } from './emit';
 import { hostName, resolveIdentity, resolveProject } from './identity';
@@ -108,6 +111,8 @@ const DEFAULT_HEARTBEAT_MS = 60_000;
 const DEFAULT_MAX_BUFFERED = 5000;
 const DEFAULT_REQUEST_TIMEOUT_MS = 5000;
 const EXIT_FLUSH_TIMEOUT_MS = 2000;
+/** How long a crash may block the dying process to deliver what is pending. */
+const CRASH_FLUSH_TIMEOUT_MS = 1000;
 const RETRY_BASE_MS = 1000;
 const RETRY_CAP_MS = 30_000;
 
@@ -337,9 +342,12 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
   // its own errors: a rejected link would skip every later flush and surface
   // as an unhandled rejection at the `void flush()` call sites.
   let chain: Promise<void> = Promise.resolve();
+  /** The batch a POST is carrying right now: a crash resends it, since that POST can never finish. */
+  let inFlight: IngestEvent[] = [];
   const send = async (timeoutMs: number, final: boolean): Promise<void> => {
     const batch = outbox.take(final);
     if (batch.length === 0) return;
+    inFlight = batch;
     try {
       await post(batch, timeoutMs, final);
       outbox.delivered();
@@ -349,6 +357,8 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
       if (final || !closing.signal.aborted) warnOnce(`could not reach the observe daemon at ${url}`, error);
       if (final) droppedCount += batch.length;
       else outbox.failed(batch);
+    } finally {
+      inFlight = [];
     }
   };
   const flush = (timeoutMs: number, final = false): Promise<void> => {
@@ -390,8 +400,9 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
     timers.push(t);
   }
 
-  const close = async (): Promise<void> => {
-    if (closed) return;
+  /** Stop timers and hooks and queue the goodbye; false when already closed. Shared by close() and the crash flush. */
+  const shutdown = (): boolean => {
+    if (closed) return false;
     closed = true;
     for (const t of timers) clearInterval(t);
     for (const [signal, handler] of exitHooks) {
@@ -401,16 +412,35 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
         // no process.off in this runtime
       }
     }
-    // Graceful exit: say goodbye, then one final flush with the ~2s cap. Straight
-    // into the outbox (enqueue drops after close, push could start a regular size
-    // flush); a full buffer evicts its oldest event rather than the deregister.
-    // Aborting `closing` fails a regular flush still in flight at once - its
-    // batch is requeued and rides along in the final attempt.
+    // Straight into the outbox (enqueue drops after close, push could start a
+    // regular size flush); a full buffer evicts its oldest event rather than the deregister.
     outbox.push({ type: 'agent.deregister', data: { agent_id: identity.agentId, ts: now() } });
+    return true;
+  };
+
+  const close = async (): Promise<void> => {
+    if (!shutdown()) return;
+    // Graceful exit: one final flush with the ~2s cap. Aborting `closing` fails
+    // a regular flush still in flight at once - its batch is requeued and
+    // rides along in the final attempt.
     closing.abort();
     await flush(EXIT_FLUSH_TIMEOUT_MS, true);
   };
 
+  /**
+   * The process is dying of an uncaught exception or unhandled rejection:
+   * Node exits as soon as the monitor returns, so no async send can finish.
+   * Send everything pending - the batch a regular POST was carrying included
+   * (re-sent events upsert by key) - in one POST that blocks for at most 1s.
+   */
+  const crashFlush = (): void => {
+    if (!shutdown()) return;
+    const batch = inFlight.concat(outbox.take(true));
+    if (!blockingPost(`${url}/api/v1/ingest`, headers, JSON.stringify({ batch }), CRASH_FLUSH_TIMEOUT_MS)) {
+      droppedCount += batch.length;
+    }
+  };
+
   const exitHooks: Array<[string, () => void]> = [];
   if (options.registerExitHooks !== false && typeof process !== 'undefined' && typeof process.once === 'function') {
     // 'beforeExit' fires while async work still runs; the in-flight POST keeps the loop
@@ -420,13 +450,27 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
     const onSignal = (code: number) => () => {
       void close().finally(() => process.exit(code));
     };
-    const hooks: Array<[NodeJS.Signals | 'beforeExit', () => void]> = [
+    // 'uncaughtExceptionMonitor' observes a crash without handling it: Node still
+    // prints the error and exits with its own code. It also sees an unhandled
+    // rejection that crashes the process (origin 'unhandledRejection'), and stays
+    // silent when one does not (--unhandled-rejections=warn, or a user listener),
+    // so no 'unhandledRejection' listener is added - adding one would itself stop
+    // the crash. With an 'uncaughtException' listener the process survives: keep
+    // reporting, just flush now.
+    const onCrash = () => {
+      if (process.listenerCount('uncaughtException') > 0) void flush(requestTimeoutMs);
+      else crashFlush();
+    };
+    const hooks: Array<[string, () => void]> = [
       ['beforeExit', onBeforeExit],
       ['SIGINT', onSignal(130)],
       ['SIGTERM', onSignal(143)],
+      ['uncaughtExceptionMonitor', onCrash],
     ];
     for (const [signal, handler] of hooks) {
-      process.once(signal as NodeJS.Signals, handler);
+      // The monitor stays on: a process that survived one throw can still die of the next.
+      if (signal === 'uncaughtExceptionMonitor') process.on(signal, handler);
+      else process.once(signal as NodeJS.Signals, handler);
       exitHooks.push([signal, handler]);
     }
   }

From b8a67d735eb2a3bc0de667435f1b8a0d33908a39 Mon Sep 17 00:00:00 2001
From: Ali Mohammad <freelancealimohammad@gmail.com>
Date: Sat, 24 Oct 2026 18:03:50 +0300
Subject: [PATCH 2/3] fix(observe): cap ingest batches at the daemon's
 1000-event limit

The daemon rejects a batch over OBSERVE_MAX_BATCH_EVENTS (1000), but the
outbox handed out everything pending (up to 5000): after an outage with a
larger backlog every retry got a 400 and was requeued forever, and a crash
with more than 1000 pending lost all of it.

- outbox.take() returns at most 1000 events, oldest first; a flush keeps
  sending while its last batch was full, so a backlog drains at once
  instead of one batch per interval. close()'s final attempt shares one
  ~2s deadline across its batches; a failure drops it and the rest.
- The crash flush sends successive <=1000-event POSTs from the one worker
  inside the same 1s budget; blockingPost now returns how many bodies were
  delivered and the undelivered rest is counted in dropped.
- A capture callback (process.setUncaughtExceptionCaptureCallback: REPL,
  domains) counts as surviving the throw, like an uncaughtException listener.
- The builtin lookup sits inside blockingPost's try; getBuiltinModule is
  shared from builtins.ts by identity.ts and blockingPost.ts.
- docs: "up to 1 s per reporter", batch cap, paragraph fix.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/observability.md                      | 26 ++++----
 llms-full.txt                              | 26 ++++----
 src/observe/__fixtures__/crashAfterRun.mts | 13 ++--
 src/observe/blockingPost.ts                | 72 ++++++++++++----------
 src/observe/builtins.ts                    | 14 +++++
 src/observe/identity.ts                    | 13 ++--
 src/observe/reporter.test.ts               | 61 ++++++++++++------
 src/observe/reporter.ts                    | 57 ++++++++++++-----
 8 files changed, 180 insertions(+), 102 deletions(-)
 create mode 100644 src/observe/builtins.ts

diff --git a/docs/observability.md b/docs/observability.md
index dbdc262a..5a35984b 100644
--- a/docs/observability.md
+++ b/docs/observability.md
@@ -413,17 +413,21 @@ The daemon also serves two stores agents can use at runtime:
 
 Reporting is fire-and-forget and never blocks a run: when the daemon is
 down, the failed batch is requeued and retried with exponential backoff (1 s,
-2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. Only
-events evicted past the buffer cap (`maxBuffered`, default 5000, oldest
-first) or still unsent after the exit flush's final attempt are dropped and
-counted in the reporter's `dropped`. A process that dies of an uncaught
-exception or an unhandled rejection still reports its last run: the reporter
-watches `uncaughtExceptionMonitor` and sends what it holds in one request that
-holds up the crash for at most 1 s. Node's error output and exit code stay
-as they were, and a script with its own `uncaughtException` handler keeps
-reporting. An explicit `process.exit()` or a SIGKILL still loses what is
-buffered. Eval
-scores flow through the same pipe —
+2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A
+backlog drains in batches of at most 1000 events (the daemon's per-request
+cap), one right after another. Only events evicted past the buffer cap
+(`maxBuffered`, default 5000, oldest first) or still unsent after the exit
+flush's final attempt are dropped and counted in the reporter's `dropped`.
+
+A process that dies of an uncaught exception or an unhandled rejection still
+reports its last run: the reporter watches `uncaughtExceptionMonitor` and
+sends what it holds, oldest first, holding up the crash for up to 1 s per
+reporter. Node's error output and exit code stay as they were, and a script
+that survives the throw (its own `uncaughtException` handler, or a capture
+callback such as the REPL's) keeps reporting. An explicit `process.exit()` or
+a SIGKILL still loses what is buffered.
+
+Eval scores flow through the same pipe —
 `lousho eval --observe` emits them even when `LOUSHO_OBSERVE` is unset.
 
 ## The built server
diff --git a/llms-full.txt b/llms-full.txt
index ad332643..7487dc1c 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -14079,17 +14079,21 @@ The daemon also serves two stores agents can use at runtime:
 
 Reporting is fire-and-forget and never blocks a run: when the daemon is
 down, the failed batch is requeued and retried with exponential backoff (1 s,
-2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. Only
-events evicted past the buffer cap (`maxBuffered`, default 5000, oldest
-first) or still unsent after the exit flush's final attempt are dropped and
-counted in the reporter's `dropped`. A process that dies of an uncaught
-exception or an unhandled rejection still reports its last run: the reporter
-watches `uncaughtExceptionMonitor` and sends what it holds in one request that
-holds up the crash for at most 1 s. Node's error output and exit code stay
-as they were, and a script with its own `uncaughtException` handler keeps
-reporting. An explicit `process.exit()` or a SIGKILL still loses what is
-buffered. Eval
-scores flow through the same pipe —
+2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A
+backlog drains in batches of at most 1000 events (the daemon's per-request
+cap), one right after another. Only events evicted past the buffer cap
+(`maxBuffered`, default 5000, oldest first) or still unsent after the exit
+flush's final attempt are dropped and counted in the reporter's `dropped`.
+
+A process that dies of an uncaught exception or an unhandled rejection still
+reports its last run: the reporter watches `uncaughtExceptionMonitor` and
+sends what it holds, oldest first, holding up the crash for up to 1 s per
+reporter. Node's error output and exit code stay as they were, and a script
+that survives the throw (its own `uncaughtException` handler, or a capture
+callback such as the REPL's) keeps reporting. An explicit `process.exit()` or
+a SIGKILL still loses what is buffered.
+
+Eval scores flow through the same pipe —
 `lousho eval --observe` emits them even when `LOUSHO_OBSERVE` is unset.
 
 ## The built server
diff --git a/src/observe/__fixtures__/crashAfterRun.mts b/src/observe/__fixtures__/crashAfterRun.mts
index 42eb5b08..3e3d951b 100644
--- a/src/observe/__fixtures__/crashAfterRun.mts
+++ b/src/observe/__fixtures__/crashAfterRun.mts
@@ -10,8 +10,10 @@
  * - reject:   an unhandled promise rejection
  * - inflight: flushSize 1, so the span's batch is already in a POST the
  *             collector holds open when a timer callback throws
+ * - backlog:  1500 spans pending (past the daemon's 1000-event batch cap), then a top-level throw
  * - handled:  the script has its own 'uncaughtException' listener, so it
  *             survives the throw, finishes a second span and closes normally
+ * - capture:  the same through process.setUncaughtExceptionCaptureCallback (REPL, domains)
  *
  * Env: LOUSHO_OBSERVE (collector URL), LOUSHO_FIXTURE_CWD (identity dir).
  */
@@ -24,7 +26,7 @@ const reporter = observeExporter({
   project: 'crash-fixture',
   agentName: 'crash-fixture',
   heartbeatIntervalMs: 0,
-  ...(mode === 'inflight' ? { flushSize: 1 } : {}),
+  ...(mode === 'inflight' ? { flushSize: 1 } : mode === 'backlog' ? { flushSize: 10_000 } : {}),
 });
 const finish = (id: string): void => {
   const span = { id, name: 'invoke_agent crash-fixture', attributes: {}, startTime: Date.now() };
@@ -32,6 +34,7 @@ const finish = (id: string): void => {
   reporter.onSpanEnd({ ...span, endTime: Date.now() });
 };
 finish('crash-span');
+if (mode === 'backlog') for (let i = 1; i < 1500; i++) finish(`crash-span-${i}`);
 
 if (mode === 'reject') {
   void Promise.reject(new Error('fixture: unhandled rejection after run'));
@@ -39,12 +42,14 @@ if (mode === 'reject') {
   setTimeout(() => {
     throw new Error('fixture: uncaught exception after run');
   }, 300);
-} else if (mode === 'handled') {
-  process.on('uncaughtException', (error) => {
+} else if (mode === 'handled' || mode === 'capture') {
+  const survive = (error: Error): void => {
     console.error(`handled: ${error.message}`);
     finish('after-span');
     void reporter.close();
-  });
+  };
+  if (mode === 'capture') process.setUncaughtExceptionCaptureCallback(survive);
+  else process.on('uncaughtException', survive);
   setTimeout(() => {
     throw new Error('fixture: uncaught exception after run');
   }, 10);
diff --git a/src/observe/blockingPost.ts b/src/observe/blockingPost.ts
index f5cfffb7..84484346 100644
--- a/src/observe/blockingPost.ts
+++ b/src/observe/blockingPost.ts
@@ -1,59 +1,67 @@
 /**
- * A POST that blocks the calling thread until it settles - the reporter's
+ * POSTs that block the calling thread until they settle - the reporter's
  * crash flush. `uncaughtExceptionMonitor` listeners run synchronously and
  * Node exits the moment they return, so an ordinary `fetch` started there
- * never even opens its socket (measured: 0 connections). Here the request
- * runs in a throwaway worker thread with its own event loop while this
- * thread sleeps on `Atomics.wait`; the worker flips a shared flag when the
- * response (or an error) arrives.
+ * never even opens its socket (measured: 0 connections). Here the requests
+ * run in a throwaway worker thread with its own event loop while this
+ * thread sleeps on `Atomics.wait`; the worker counts each delivered body
+ * into a shared flag and wakes this thread when it is done.
  *
  * The worker is given no `execArgv` (no loaders such as tsx: it only needs
  * the global `fetch`) and is left running when this returns: the process is
  * about to exit, which terminates it.
  *
- * Builtins load through `process.getBuiltinModule`: this module ships in the
- * Cloudflare Worker bundle (see WORKER_SAFE_BUILTIN_PROBE), where it reports
- * `false` instead of breaking the bundle.
+ * Builtins load through `getBuiltinModule` (builtins.ts): this module ships
+ * in the Cloudflare Worker bundle, where it reports nothing delivered
+ * instead of breaking the bundle.
  */
+import { getBuiltinModule } from './builtins';
 
 interface WorkerThreadsLike {
   Worker: new (code: string, options: { eval: true; execArgv: string[]; workerData: unknown }) => { unref(): void };
 }
 
-// Named to match the worker leak check's exempted call pattern (see
-// WORKER_SAFE_BUILTIN_PROBE in src/deploy/adapters/cloudflare.ts). Do not rename.
-function getBuiltinModule<T>(name: string): T | undefined {
-  if (typeof process === 'undefined') return undefined;
-  return (process.getBuiltinModule?.(name) as T | undefined) ?? undefined;
-}
-
-const PENDING = 0;
-const OK = 1;
+/** Shared flag slots: [0] 0 while the worker runs, 1 once it stopped; [1] bodies delivered so far. */
+const DONE = 0;
+const DELIVERED = 1;
 
-/** The worker's whole program (CommonJS, `eval: true`): one fetch, then publish the outcome. */
+/**
+ * The worker's whole program (CommonJS, `eval: true`): the bodies in order,
+ * one at a time, all inside the one deadline; the first failure stops it.
+ * The id literal is double-quoted so the bundle check's probe matcher sees it.
+ */
 const WORKER_SOURCE = `
 const { workerData: d } = process.getBuiltinModule("node:worker_threads");
 const flag = new Int32Array(d.flag);
-const done = (state) => { Atomics.store(flag, 0, state); Atomics.notify(flag, 0); };
-fetch(d.url, { method: 'POST', headers: d.headers, body: d.body, signal: AbortSignal.timeout(d.timeoutMs) })
-  .then((res) => done(res.ok ? ${OK} : 2), () => done(2));
+(async () => {
+  for (const body of d.bodies) {
+    const left = d.deadline - Date.now();
+    if (left <= 0) break;
+    const res = await fetch(d.url, { method: 'POST', headers: d.headers, body, signal: AbortSignal.timeout(left) });
+    if (!res.ok) break;
+    Atomics.add(flag, ${DELIVERED}, 1);
+  }
+})().catch(() => undefined).finally(() => { Atomics.store(flag, ${DONE}, 1); Atomics.notify(flag, ${DONE}); });
 `;
 
 /**
- * POST `body` to `url`, blocking at most `timeoutMs` (worker start-up
- * included). True only when the server answered 2xx in time; false on any
- * failure, a timeout, or a runtime without worker threads.
+ * POST each of `bodies` to `url` in order, blocking at most `timeoutMs` in
+ * all (worker start-up included). Returns how many leading bodies the server
+ * answered 2xx in time: the rest were not delivered (a failure, the
+ * deadline, or a runtime without worker threads, which delivers 0).
  */
-export function blockingPost(url: string, headers: Record<string, string>, body: string, timeoutMs: number): boolean {
-  const threads = getBuiltinModule<WorkerThreadsLike>('node:worker_threads');
-  if (!threads || typeof SharedArrayBuffer === 'undefined') return false;
+export function blockingPost(url: string, headers: Record<string, string>, bodies: readonly string[], timeoutMs: number): number {
+  if (bodies.length === 0) return 0;
   try {
-    const flag = new SharedArrayBuffer(4);
-    const worker = new threads.Worker(WORKER_SOURCE, { eval: true, execArgv: [], workerData: { flag, url, headers, body, timeoutMs } });
+    const threads = getBuiltinModule<WorkerThreadsLike>('node:worker_threads');
+    if (!threads) return 0;
+    const flag = new Int32Array(new SharedArrayBuffer(8));
+    const deadline = Date.now() + timeoutMs;
+    const worker = new threads.Worker(WORKER_SOURCE, { eval: true, execArgv: [], workerData: { flag: flag.buffer, url, headers, bodies, deadline } });
     worker.unref(); // never keeps a process alive past a timed-out wait
-    Atomics.wait(new Int32Array(flag), 0, PENDING, timeoutMs);
-    return Atomics.load(new Int32Array(flag), 0) === OK;
+    Atomics.wait(flag, DONE, 0, timeoutMs);
+    return Atomics.load(flag, DELIVERED);
   } catch {
-    return false;
+    return 0;
   }
 }
diff --git a/src/observe/builtins.ts b/src/observe/builtins.ts
new file mode 100644
index 00000000..3e135c0f
--- /dev/null
+++ b/src/observe/builtins.ts
@@ -0,0 +1,14 @@
+/**
+ * Lazy Node builtins for the observe modules that also ship in the Worker
+ * bundle (identity.ts, blockingPost.ts): `undefined` on runtimes without
+ * them, never an import for the bundler to resolve.
+ *
+ * Named to match the worker leak check's exempted call pattern (see
+ * WORKER_SAFE_BUILTIN_PROBE in src/deploy/adapters/cloudflare.ts): a builtin
+ * id literal stays out of the flagged set only while it appears as this
+ * function's argument at the call site. Do not rename.
+ */
+export function getBuiltinModule<T>(name: string): T | undefined {
+  if (typeof process === 'undefined') return undefined;
+  return (process.getBuiltinModule?.(name) as T | undefined) ?? undefined;
+}
diff --git a/src/observe/identity.ts b/src/observe/identity.ts
index 9f6c5e33..2f041246 100644
--- a/src/observe/identity.ts
+++ b/src/observe/identity.ts
@@ -18,6 +18,8 @@
  * breaking non-Node runtimes.
  */
 
+import { getBuiltinModule } from './builtins';
+
 /** The slices of the builtins used here, typed so the lazy loads stay honest. */
 interface FsLike {
   existsSync(path: string): boolean;
@@ -36,15 +38,8 @@ interface PathLike {
   resolve(...parts: string[]): string;
 }
 
-// Named to match the worker leak check's exempted call pattern (see
-// WORKER_SAFE_BUILTIN_PROBE in src/deploy/adapters/cloudflare.ts): the
-// the builtin id literals below stay out of the flagged set only while they
-// appear as this function's argument. Do not rename.
-function getBuiltinModule<T>(name: string): T | undefined {
-  // Reporters bundled for non-Node runtimes have no fs; identity falls back to memory.
-  if (typeof process === 'undefined') return undefined;
-  return (process.getBuiltinModule?.(name) as T | undefined) ?? undefined;
-}
+// Reporters bundled for non-Node runtimes have no fs; identity falls back to memory.
+// Keep each id literal as getBuiltinModule's direct argument (see builtins.ts).
 
 const lazyFs = () => getBuiltinModule('node:fs') as FsLike | undefined;
 const lazyCrypto = () => getBuiltinModule('node:crypto') as CryptoLike | undefined;
diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts
index ecf9881d..dd6a19a3 100644
--- a/src/observe/reporter.test.ts
+++ b/src/observe/reporter.test.ts
@@ -330,33 +330,50 @@ describe('observeExporter', () => {
       expect(r.dropped).toBe(3);
     });
 
-    it('requeues a batch too large to spread into call arguments, and the flush chain survives', async () => {
+    it('drains a backlog past the daemon batch cap in successive capped batches, and the flush chain survives', async () => {
       const unhandled: unknown[] = [];
       const onUnhandled = (reason: unknown) => unhandled.push(reason);
       process.on('unhandledRejection', onUnhandled);
       try {
-        // Count events without JSON.parse-ing a ~50MB body twice.
         const sizes: number[] = [];
+        const ids: string[] = [];
         const fetchFn = vi.fn(async (_input: string | URL | Request, init?: RequestInit): Promise<Response> => {
-          sizes.push(String(init?.body).split('"type":').length - 1);
-          if (sizes.length === 1) throw new TypeError('fetch failed');
+          const batch = (JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch;
+          sizes.push(batch.length);
+          if (sizes.length === 1) throw new TypeError('fetch failed'); // the daemon is down
+          for (const e of batch) if (e.type === 'observation') ids.push(e.data.obs_id);
           return new Response('{}', { status: 202 });
         }) as unknown as typeof fetch;
-        const total = 500_000; // past the spread limit even on a worker thread's larger stack (200k fits there)
-        const { r } = reporter({ fetchFn, maxBuffered: total, flushSize: total + 1, flushIntervalMs: 1000 });
+        const total = 2500;
+        const { r } = reporter({ fetchFn, flushSize: total + 1, flushIntervalMs: 1000 });
         for (let i = 0; i < total; i++) r.onSpanEnd({ id: `s${i}`, name: 'n', attributes: {}, startTime: 1, endTime: 2 });
-        await vi.advanceTimersByTimeAsync(1000); // rejected: the whole batch is requeued
-        await vi.advanceTimersByTimeAsync(1000); // the retry still runs and carries all of it
-        expect(sizes).toEqual([total, total]);
+        await vi.advanceTimersByTimeAsync(1000); // rejected: the oldest 1000 are requeued
+        await vi.advanceTimersByTimeAsync(1000); // the daemon is back: one flush drains it all, batch after batch
+        expect(sizes).toEqual([1000, 1000, 1000, 500]);
+        expect(ids).toEqual(Array.from({ length: total }, (_, i) => `s${i}`)); // every event, oldest first
         expect(r.dropped).toBe(0);
         await r.close();
-        expect(sizes).toHaveLength(3); // the chain still runs close()'s flush (the deregister)
+        expect(sizes).toHaveLength(5); // the chain still runs close()'s flush (the deregister)
         await vi.advanceTimersByTimeAsync(10); // let any stray rejection surface
         expect(unhandled).toEqual([]);
       } finally {
         process.off('unhandledRejection', onUnhandled);
       }
-    }, 60_000);
+    });
+
+    it("close()'s final attempt sends a backlog in capped batches and drops the rest when one fails", async () => {
+      const sizes: number[] = [];
+      const fetchFn = vi.fn(async (_input: string | URL | Request, init?: RequestInit): Promise<Response> => {
+        sizes.push((JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch.length);
+        return new Response('{}', { status: sizes.length === 2 ? 500 : 202 });
+      }) as unknown as typeof fetch;
+      vi.spyOn(console, 'warn').mockImplementation(() => undefined);
+      const { r } = reporter({ fetchFn, flushSize: 10_000 });
+      for (let i = 0; i < 2500; i++) r.onSpanEnd({ id: `s${i}`, name: 'n', attributes: {}, startTime: 1, endTime: 2 });
+      await r.close();
+      expect(sizes).toEqual([1000, 1000]); // the second failed: it and the remaining 501 (500 + the deregister) are dropped
+      expect(r.dropped).toBe(1501);
+    });
   });
 
   it('clamps maxBuffered to 1 so the deregister survives', async () => {
@@ -452,7 +469,7 @@ describe('observeExporter', () => {
 describe('crash flush', () => {
   const fixture = path.join(__dirname, '__fixtures__', 'crashAfterRun.mts');
   const repoRoot = path.resolve(__dirname, '..', '..');
-  type Mode = 'throw' | 'reject' | 'inflight' | 'handled';
+  type Mode = 'throw' | 'reject' | 'inflight' | 'handled' | 'capture' | 'backlog';
 
   /**
    * Run the fixture against an in-test collector; resolves with its exit and
@@ -504,15 +521,22 @@ describe('crash flush', () => {
     expect(batches.flatMap(obsIds)).toEqual(['crash-span']);
   }, 30_000);
 
+  it('sends a crash backlog past the daemon batch cap as successive capped POSTs', async () => {
+    const { code, batches } = await crash('backlog');
+    expect(code).toBe(1);
+    expect(batches.map((b) => b.length)).toEqual([1000, 501]); // 1500 observations + the deregister, oldest first
+    expect(batches.flatMap(obsIds)).toEqual(['crash-span', ...Array.from({ length: 1499 }, (_, i) => `crash-span-${i + 1}`)]);
+  }, 30_000);
+
   it('resends the batch a POST still in flight was carrying when the process died', async () => {
     const { code, batches } = await crash('inflight', true);
     expect(code).toBe(1);
     expect(batches.map(obsIds)).toEqual([['crash-span'], ['crash-span']]);
   }, 30_000);
 
-  it('keeps reporting when the script handles uncaughtException itself', async () => {
-    const { code, stderr, batches } = await crash('handled');
-    expect(code).toBe(0); // the user's listener kept the process alive; the reporter did not exit it
+  it.each([['handled'], ['capture']] as const)('keeps reporting when the script survives the throw itself (%s)', async (mode) => {
+    const { code, stderr, batches } = await crash(mode);
+    expect(code).toBe(0); // the user's listener / capture callback kept the process alive; the reporter did not exit it
     expect(stderr).toContain('handled: fixture: uncaught exception after run');
     expect(batches.flatMap(obsIds)).toEqual(['crash-span', 'after-span']);
   }, 30_000);
@@ -544,8 +568,9 @@ describe('crash flush', () => {
 });
 
 describe('blockingPost', () => {
-  it('reports false for an unreachable collector', () => {
-    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, '{}', 1000)).toBe(false);
+  it('delivers nothing to an unreachable collector', () => {
+    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, ['{}', '{}'], 1000)).toBe(0);
+    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, [], 1000)).toBe(0);
   });
 
   it('gives up after its timeout', async () => {
@@ -555,7 +580,7 @@ describe('blockingPost', () => {
     const { port } = server.address() as AddressInfo;
     try {
       const t0 = Date.now();
-      expect(blockingPost(`http://127.0.0.1:${port}/`, {}, '{}', 200)).toBe(false);
+      expect(blockingPost(`http://127.0.0.1:${port}/`, {}, ['{}'], 200)).toBe(0);
       expect(Date.now() - t0).toBeLessThan(1000);
     } finally {
       server.closeAllConnections();
diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts
index 65db3fab..954eff03 100644
--- a/src/observe/reporter.ts
+++ b/src/observe/reporter.ts
@@ -36,7 +36,7 @@ import { SDKError } from '../execution/errors';
 import type { Span, TraceExporter } from '../execution/tracing';
 import { VERSION } from '../version';
 import { blockingPost } from './blockingPost';
-import { OBSERVE_DEFAULT_PORT, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope';
+import { OBSERVE_DEFAULT_PORT, OBSERVE_MAX_BATCH_EVENTS, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope';
 import { observeTargetFromEnv } from './emit';
 import { hostName, resolveIdentity, resolveProject } from './identity';
 
@@ -124,7 +124,8 @@ const retryDelayMs = (failures: number): number => Math.min(RETRY_BASE_MS * 2 **
  * evicted first, reported through `onEvict`), and gated by a retry backoff
  * after a failed send. Every send path - size, interval, exit - goes through
  * `take` / `delivered` / `failed`, so a failed batch is never lost while
- * there is room to keep it.
+ * there is room to keep it. A batch is at most OBSERVE_MAX_BATCH_EVENTS
+ * events (the daemon rejects larger ones), oldest first.
  */
 function createOutbox(cap: number, now: () => number, onEvict: (count: number) => void) {
   let items: IngestEvent[] = [];
@@ -148,9 +149,13 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) =
       items.push(event);
       trim();
     },
-    /** Everything pending, for one send - or nothing while backing off, unless `force` (the final attempt). */
+    /** The oldest batch's worth, for one send - or nothing while backing off, unless `force` (the final attempt). */
     take(force = false): IngestEvent[] {
-      return force || now() >= retryAt ? items.splice(0, items.length) : [];
+      return force || now() >= retryAt ? items.splice(0, OBSERVE_MAX_BATCH_EVENTS) : [];
+    },
+    /** Everything pending, uncapped (the crash flush; counting what a failed final attempt leaves). */
+    drain(): IngestEvent[] {
+      return items.splice(0, items.length);
     },
     delivered(): void {
       failures = 0;
@@ -344,23 +349,37 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
   let chain: Promise<void> = Promise.resolve();
   /** The batch a POST is carrying right now: a crash resends it, since that POST can never finish. */
   let inFlight: IngestEvent[] = [];
-  const send = async (timeoutMs: number, final: boolean): Promise<void> => {
-    const batch = outbox.take(final);
-    if (batch.length === 0) return;
+  /** One POST; false when it failed (the batch is then requeued, or - final - dropped with everything left). */
+  const sendBatch = async (batch: IngestEvent[], timeoutMs: number, final: boolean): Promise<boolean> => {
     inFlight = batch;
     try {
       await post(batch, timeoutMs, final);
       outbox.delivered();
+      return true;
     } catch (error) {
       // Never slow or fail a run: requeue for a later retry, drop-and-count only as the last resort.
       // close() aborting a regular flush is not an outage: its batch rides along in the final attempt.
       if (final || !closing.signal.aborted) warnOnce(`could not reach the observe daemon at ${url}`, error);
-      if (final) droppedCount += batch.length;
+      if (final) droppedCount += batch.length + outbox.drain().length;
       else outbox.failed(batch);
+      return false;
     } finally {
       inFlight = [];
     }
   };
+  /**
+   * Send batch after batch while the last one was full: a backlog (after an
+   * outage) drains at once instead of one batch per interval. Regular flushes
+   * give each POST its own timeout; the final attempt shares one deadline.
+   */
+  const send = async (timeoutMs: number, final: boolean): Promise<void> => {
+    const deadline = Date.now() + timeoutMs;
+    let batch = outbox.take(final);
+    while (batch.length > 0) {
+      if (!(await sendBatch(batch, final ? Math.max(1, deadline - Date.now()) : timeoutMs, final))) return;
+      batch = batch.length < OBSERVE_MAX_BATCH_EVENTS ? [] : outbox.take(final);
+    }
+  };
   const flush = (timeoutMs: number, final = false): Promise<void> => {
     const run = chain.then(() => send(timeoutMs, final)).catch(() => undefined);
     chain = run;
@@ -430,15 +449,18 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
   /**
    * The process is dying of an uncaught exception or unhandled rejection:
    * Node exits as soon as the monitor returns, so no async send can finish.
-   * Send everything pending - the batch a regular POST was carrying included
-   * (re-sent events upsert by key) - in one POST that blocks for at most 1s.
+   * Send everything pending, oldest first - the batch a regular POST was
+   * carrying included (re-sent events upsert by key) - as successive
+   * batch-sized POSTs that together block for at most 1s.
    */
   const crashFlush = (): void => {
     if (!shutdown()) return;
-    const batch = inFlight.concat(outbox.take(true));
-    if (!blockingPost(`${url}/api/v1/ingest`, headers, JSON.stringify({ batch }), CRASH_FLUSH_TIMEOUT_MS)) {
-      droppedCount += batch.length;
-    }
+    const pending = inFlight.concat(outbox.drain());
+    const batches: IngestEvent[][] = [];
+    for (let i = 0; i < pending.length; i += OBSERVE_MAX_BATCH_EVENTS) batches.push(pending.slice(i, i + OBSERVE_MAX_BATCH_EVENTS));
+    const bodies = batches.map((batch) => JSON.stringify({ batch }));
+    const delivered = blockingPost(`${url}/api/v1/ingest`, headers, bodies, CRASH_FLUSH_TIMEOUT_MS);
+    for (const batch of batches.slice(delivered)) droppedCount += batch.length;
   };
 
   const exitHooks: Array<[string, () => void]> = [];
@@ -455,10 +477,11 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
     // rejection that crashes the process (origin 'unhandledRejection'), and stays
     // silent when one does not (--unhandled-rejections=warn, or a user listener),
     // so no 'unhandledRejection' listener is added - adding one would itself stop
-    // the crash. With an 'uncaughtException' listener the process survives: keep
-    // reporting, just flush now.
+    // the crash. With an 'uncaughtException' listener or a capture callback
+    // (REPL, domains) the process survives: keep reporting, just flush now.
+    const survives = () => process.listenerCount('uncaughtException') > 0 || process.hasUncaughtExceptionCaptureCallback?.() === true;
     const onCrash = () => {
-      if (process.listenerCount('uncaughtException') > 0) void flush(requestTimeoutMs);
+      if (survives()) void flush(requestTimeoutMs);
       else crashFlush();
     };
     const hooks: Array<[string, () => void]> = [

From 14f6e17fd25a7bb7b7dc52dce0ef080f604d19a8 Mon Sep 17 00:00:00 2001
From: Ali Mohammad <freelancealimohammad@gmail.com>
Date: Sat, 24 Oct 2026 18:11:11 +0300
Subject: [PATCH 3/3] fix(observe): cap ingest batches by bytes and drop
 batches the daemon refuses

With content capture on by default (O6), 1000 events can pass the daemon's
5MB body cap: every retry got a 413 and was requeued forever.

- Each event is serialized once when queued; a batch stops growing before
  its body would pass BATCH_BUDGET_BYTES (4MB), measured incrementally
  alongside the 1000-event cap. One event over the budget but under the
  hard cap goes out alone; one over the hard cap is dropped and counted
  on arrival (warnOnce). The POST body is joined from those strings.
- A 400/413 answer drops and counts that batch instead of requeueing it.
- The crash flush cuts its batches the same way; blockingPost reports per
  body, and the worker skips a body the server refuses instead of stopping.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/observability.md                      |  12 +-
 llms-full.txt                              |  12 +-
 src/observe/__fixtures__/crashAfterRun.mts |   8 +-
 src/observe/blockingPost.ts                |  36 ++---
 src/observe/reporter.test.ts               |  99 +++++++++++++-
 src/observe/reporter.ts                    | 145 ++++++++++++++++-----
 6 files changed, 243 insertions(+), 69 deletions(-)

diff --git a/docs/observability.md b/docs/observability.md
index 5a35984b..c6efd471 100644
--- a/docs/observability.md
+++ b/docs/observability.md
@@ -414,10 +414,14 @@ The daemon also serves two stores agents can use at runtime:
 Reporting is fire-and-forget and never blocks a run: when the daemon is
 down, the failed batch is requeued and retried with exponential backoff (1 s,
 2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A
-backlog drains in batches of at most 1000 events (the daemon's per-request
-cap), one right after another. Only events evicted past the buffer cap
-(`maxBuffered`, default 5000, oldest first) or still unsent after the exit
-flush's final attempt are dropped and counted in the reporter's `dropped`.
+backlog drains in batches one right after another, each at most 1000 events
+and 4 MB of JSON (under the daemon's per-request caps of 1000 events and
+5 MB). Only these are dropped and counted in the reporter's `dropped`:
+events evicted past the buffer cap (`maxBuffered`, default 5000, oldest
+first); events still unsent after the exit flush's final attempt; a single
+event too large for any request (over 5 MB serialized, for example with
+content capture on); and a batch the daemon refuses outright (`400`/`413`),
+which is never retried.
 
 A process that dies of an uncaught exception or an unhandled rejection still
 reports its last run: the reporter watches `uncaughtExceptionMonitor` and
diff --git a/llms-full.txt b/llms-full.txt
index 7487dc1c..1af0e749 100644
--- a/llms-full.txt
+++ b/llms-full.txt
@@ -14080,10 +14080,14 @@ The daemon also serves two stores agents can use at runtime:
 Reporting is fire-and-forget and never blocks a run: when the daemon is
 down, the failed batch is requeued and retried with exponential backoff (1 s,
 2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A
-backlog drains in batches of at most 1000 events (the daemon's per-request
-cap), one right after another. Only events evicted past the buffer cap
-(`maxBuffered`, default 5000, oldest first) or still unsent after the exit
-flush's final attempt are dropped and counted in the reporter's `dropped`.
+backlog drains in batches one right after another, each at most 1000 events
+and 4 MB of JSON (under the daemon's per-request caps of 1000 events and
+5 MB). Only these are dropped and counted in the reporter's `dropped`:
+events evicted past the buffer cap (`maxBuffered`, default 5000, oldest
+first); events still unsent after the exit flush's final attempt; a single
+event too large for any request (over 5 MB serialized, for example with
+content capture on); and a batch the daemon refuses outright (`400`/`413`),
+which is never retried.
 
 A process that dies of an uncaught exception or an unhandled rejection still
 reports its last run: the reporter watches `uncaughtExceptionMonitor` and
diff --git a/src/observe/__fixtures__/crashAfterRun.mts b/src/observe/__fixtures__/crashAfterRun.mts
index 3e3d951b..ae553b3a 100644
--- a/src/observe/__fixtures__/crashAfterRun.mts
+++ b/src/observe/__fixtures__/crashAfterRun.mts
@@ -11,6 +11,7 @@
  * - inflight: flushSize 1, so the span's batch is already in a POST the
  *             collector holds open when a timer callback throws
  * - backlog:  1500 spans pending (past the daemon's 1000-event batch cap), then a top-level throw
+ * - bigbacklog: 30 spans of ~200KB input (~6MB, past the byte budget), then a top-level throw
  * - handled:  the script has its own 'uncaughtException' listener, so it
  *             survives the throw, finishes a second span and closes normally
  * - capture:  the same through process.setUncaughtExceptionCaptureCallback (REPL, domains)
@@ -26,14 +27,15 @@ const reporter = observeExporter({
   project: 'crash-fixture',
   agentName: 'crash-fixture',
   heartbeatIntervalMs: 0,
-  ...(mode === 'inflight' ? { flushSize: 1 } : mode === 'backlog' ? { flushSize: 10_000 } : {}),
+  ...(mode === 'inflight' ? { flushSize: 1 } : mode === 'backlog' || mode === 'bigbacklog' ? { flushSize: 10_000 } : {}),
 });
-const finish = (id: string): void => {
-  const span = { id, name: 'invoke_agent crash-fixture', attributes: {}, startTime: Date.now() };
+const finish = (id: string, attributes: Record<string, unknown> = {}): void => {
+  const span = { id, name: 'invoke_agent crash-fixture', attributes, startTime: Date.now() };
   reporter.onSpanStart(span);
   reporter.onSpanEnd({ ...span, endTime: Date.now() });
 };
 finish('crash-span');
+if (mode === 'bigbacklog') for (let i = 0; i < 30; i++) finish(`big-${i}`, { 'gen_ai.input.messages': 'x'.repeat(200 * 1024) });
 if (mode === 'backlog') for (let i = 1; i < 1500; i++) finish(`crash-span-${i}`);
 
 if (mode === 'reject') {
diff --git a/src/observe/blockingPost.ts b/src/observe/blockingPost.ts
index 84484346..f988c60b 100644
--- a/src/observe/blockingPost.ts
+++ b/src/observe/blockingPost.ts
@@ -4,8 +4,8 @@
  * Node exits the moment they return, so an ordinary `fetch` started there
  * never even opens its socket (measured: 0 connections). Here the requests
  * run in a throwaway worker thread with its own event loop while this
- * thread sleeps on `Atomics.wait`; the worker counts each delivered body
- * into a shared flag and wakes this thread when it is done.
+ * thread sleeps on `Atomics.wait`; the worker marks each delivered body
+ * in a shared array and wakes this thread when it is done.
  *
  * The worker is given no `execArgv` (no loaders such as tsx: it only needs
  * the global `fetch`) and is left running when this returns: the process is
@@ -21,47 +21,47 @@ interface WorkerThreadsLike {
   Worker: new (code: string, options: { eval: true; execArgv: string[]; workerData: unknown }) => { unref(): void };
 }
 
-/** Shared flag slots: [0] 0 while the worker runs, 1 once it stopped; [1] bodies delivered so far. */
+/** Shared slots: [0] 0 while the worker runs, 1 once it stopped; [1 + i] 1 once body i got a 2xx. */
 const DONE = 0;
-const DELIVERED = 1;
+const FIRST_BODY = 1;
 
 /**
  * The worker's whole program (CommonJS, `eval: true`): the bodies in order,
- * one at a time, all inside the one deadline; the first failure stops it.
+ * one at a time, all inside the one deadline. A body the server answers with
+ * an error is skipped; no answer at all (down, deadline) stops it.
  * The id literal is double-quoted so the bundle check's probe matcher sees it.
  */
 const WORKER_SOURCE = `
 const { workerData: d } = process.getBuiltinModule("node:worker_threads");
 const flag = new Int32Array(d.flag);
 (async () => {
-  for (const body of d.bodies) {
+  for (let i = 0; i < d.bodies.length; i++) {
     const left = d.deadline - Date.now();
     if (left <= 0) break;
-    const res = await fetch(d.url, { method: 'POST', headers: d.headers, body, signal: AbortSignal.timeout(left) });
-    if (!res.ok) break;
-    Atomics.add(flag, ${DELIVERED}, 1);
+    const res = await fetch(d.url, { method: 'POST', headers: d.headers, body: d.bodies[i], signal: AbortSignal.timeout(left) });
+    if (res.ok) Atomics.store(flag, ${FIRST_BODY} + i, 1);
   }
 })().catch(() => undefined).finally(() => { Atomics.store(flag, ${DONE}, 1); Atomics.notify(flag, ${DONE}); });
 `;
 
 /**
  * POST each of `bodies` to `url` in order, blocking at most `timeoutMs` in
- * all (worker start-up included). Returns how many leading bodies the server
- * answered 2xx in time: the rest were not delivered (a failure, the
- * deadline, or a runtime without worker threads, which delivers 0).
+ * all (worker start-up included). Returns, per body, whether the server
+ * answered it 2xx in time; all false on a runtime without worker threads.
  */
-export function blockingPost(url: string, headers: Record<string, string>, bodies: readonly string[], timeoutMs: number): number {
-  if (bodies.length === 0) return 0;
+export function blockingPost(url: string, headers: Record<string, string>, bodies: readonly string[], timeoutMs: number): boolean[] {
+  const none = bodies.map(() => false);
+  if (bodies.length === 0) return none;
   try {
     const threads = getBuiltinModule<WorkerThreadsLike>('node:worker_threads');
-    if (!threads) return 0;
-    const flag = new Int32Array(new SharedArrayBuffer(8));
+    if (!threads) return none;
+    const flag = new Int32Array(new SharedArrayBuffer(4 * (FIRST_BODY + bodies.length)));
     const deadline = Date.now() + timeoutMs;
     const worker = new threads.Worker(WORKER_SOURCE, { eval: true, execArgv: [], workerData: { flag: flag.buffer, url, headers, bodies, deadline } });
     worker.unref(); // never keeps a process alive past a timed-out wait
     Atomics.wait(flag, DONE, 0, timeoutMs);
-    return Atomics.load(flag, DELIVERED);
+    return bodies.map((_, i) => Atomics.load(flag, FIRST_BODY + i) === 1);
   } catch {
-    return 0;
+    return none;
   }
 }
diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts
index dd6a19a3..28c5fce6 100644
--- a/src/observe/reporter.test.ts
+++ b/src/observe/reporter.test.ts
@@ -376,6 +376,89 @@ describe('observeExporter', () => {
     });
   });
 
+  describe('byte cap', () => {
+    const BUDGET = 4 * 1024 * 1024; // BATCH_BUDGET_BYTES
+    /** A finished span whose input is `bytes` of ASCII. */
+    const bigSpan = (id: string, bytes: number): Span =>
+      makeSpan({ id, attributes: { [GenAiAttr.OPERATION_NAME]: GenAiOperation.CHAT, [GenAiAttr.INPUT_MESSAGES]: 'x'.repeat(bytes) } });
+    /** A fetch that records each body's size and ids, answering `status(n)` to the n-th call (1-based). */
+    function sizedFetch(status: (n: number) => number = () => 202) {
+      const calls: { bytes: number; ids: string[] }[] = [];
+      const fetchFn = vi.fn(async (_input: string | URL | Request, init?: RequestInit): Promise<Response> => {
+        const body = String(init?.body);
+        const batch = (JSON.parse(body) as { batch: IngestEvent[] }).batch;
+        calls.push({ bytes: Buffer.byteLength(body), ids: batch.flatMap((e) => (e.type === 'observation' ? [e.data.obs_id] : [])) });
+        return new Response('{}', { status: status(calls.length) });
+      }) as unknown as typeof fetch;
+      return { fetchFn, calls };
+    }
+
+    it('cuts batches of large events at the byte budget, all delivered in order', async () => {
+      const { fetchFn, calls } = sizedFetch();
+      const { r } = reporter({ fetchFn, flushSize: 1000 });
+      for (let i = 0; i < 50; i++) {
+        const span = bigSpan(`big${i}`, 200 * 1024);
+        r.onSpanStart(span);
+        r.onSpanEnd(span);
+      }
+      await r.close();
+      expect(calls.length).toBeGreaterThanOrEqual(3); // ~10MB of events
+      for (const call of calls) expect(call.bytes).toBeLessThanOrEqual(BUDGET);
+      expect(calls.flatMap((c) => c.ids)).toEqual(Array.from({ length: 50 }, (_, i) => `big${i}`));
+      expect(r.dropped).toBe(0);
+    });
+
+    it('drops and counts one event too large for any request, and keeps delivering', async () => {
+      const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
+      const { fetchFn, calls } = sizedFetch();
+      const { r } = reporter({ fetchFn, flushSize: 1000 });
+      for (const span of [bigSpan('huge', 6 * 1024 * 1024), makeSpan({ id: 'after' })]) {
+        r.onSpanStart(span);
+        r.onSpanEnd(span);
+      }
+      expect(r.dropped).toBe(1);
+      expect(warn).toHaveBeenCalledOnce();
+      expect(String(warn.mock.calls[0][0])).toMatch(/larger than the daemon's 5242880-byte request cap/);
+      await r.close();
+      expect(calls.flatMap((c) => c.ids)).toEqual(['after']);
+      expect(r.dropped).toBe(1);
+    });
+
+    it('sends one event over the budget but under the hard cap on its own', async () => {
+      // The input is carried twice (input + attrs): ~4.4MB of event, over the 4MB budget, under the 5MB cap.
+      const { fetchFn, calls } = sizedFetch();
+      const { r } = reporter({ fetchFn, flushSize: 1000 });
+      for (const span of [makeSpan({ id: 'before' }), bigSpan('large', 2.2 * 1024 * 1024), makeSpan({ id: 'after' })]) {
+        r.onSpanStart(span);
+        r.onSpanEnd(span);
+      }
+      await r.close();
+      expect(calls.map((c) => c.ids)).toEqual([['before'], ['large'], ['after']]);
+      expect(r.dropped).toBe(0);
+    });
+
+    it('drops and counts a batch the daemon answers 413, and keeps delivering', async () => {
+      const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
+      const { fetchFn, calls } = sizedFetch((n) => (n === 1 ? 413 : 202));
+      const { r } = reporter({ fetchFn, flushSize: 2 });
+      const finish = (id: string) => {
+        const span = makeSpan({ id });
+        r.onSpanStart(span);
+        r.onSpanEnd(span);
+      };
+      finish('a');
+      finish('b'); // a size flush: answered 413
+      await vi.waitFor(() => expect(r.dropped).toBe(2));
+      finish('c');
+      finish('d');
+      await r.close();
+      expect(calls[0].ids).toEqual(['a', 'b']);
+      expect(calls.flatMap((c) => c.ids)).toEqual(['a', 'b', 'c', 'd']); // the refused batch is not resent
+      expect(r.dropped).toBe(2); // the refused batch, not requeued
+      expect(String(warn.mock.calls[0][0])).toContain('refused a batch of 2 events (413)');
+    });
+  });
+
   it('clamps maxBuffered to 1 so the deregister survives', async () => {
     const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined);
     try {
@@ -469,7 +552,7 @@ describe('observeExporter', () => {
 describe('crash flush', () => {
   const fixture = path.join(__dirname, '__fixtures__', 'crashAfterRun.mts');
   const repoRoot = path.resolve(__dirname, '..', '..');
-  type Mode = 'throw' | 'reject' | 'inflight' | 'handled' | 'capture' | 'backlog';
+  type Mode = 'throw' | 'reject' | 'inflight' | 'handled' | 'capture' | 'backlog' | 'bigbacklog';
 
   /**
    * Run the fixture against an in-test collector; resolves with its exit and
@@ -528,6 +611,14 @@ describe('crash flush', () => {
     expect(batches.flatMap(obsIds)).toEqual(['crash-span', ...Array.from({ length: 1499 }, (_, i) => `crash-span-${i + 1}`)]);
   }, 30_000);
 
+  it('cuts a crash backlog of large events at the byte budget', async () => {
+    const { code, batches } = await crash('bigbacklog');
+    expect(code).toBe(1);
+    expect(batches.length).toBeGreaterThanOrEqual(2); // 30 x 200KB = ~6MB
+    for (const batch of batches) expect(Buffer.byteLength(JSON.stringify({ batch }))).toBeLessThanOrEqual(4 * 1024 * 1024);
+    expect(batches.flatMap(obsIds)).toEqual(['crash-span', ...Array.from({ length: 30 }, (_, i) => `big-${i}`)]);
+  }, 30_000);
+
   it('resends the batch a POST still in flight was carrying when the process died', async () => {
     const { code, batches } = await crash('inflight', true);
     expect(code).toBe(1);
@@ -569,8 +660,8 @@ describe('crash flush', () => {
 
 describe('blockingPost', () => {
   it('delivers nothing to an unreachable collector', () => {
-    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, ['{}', '{}'], 1000)).toBe(0);
-    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, [], 1000)).toBe(0);
+    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, ['{}', '{}'], 1000)).toEqual([false, false]);
+    expect(blockingPost('http://127.0.0.1:1/api/v1/ingest', {}, [], 1000)).toEqual([]);
   });
 
   it('gives up after its timeout', async () => {
@@ -580,7 +671,7 @@ describe('blockingPost', () => {
     const { port } = server.address() as AddressInfo;
     try {
       const t0 = Date.now();
-      expect(blockingPost(`http://127.0.0.1:${port}/`, {}, ['{}'], 200)).toBe(0);
+      expect(blockingPost(`http://127.0.0.1:${port}/`, {}, ['{}'], 200)).toEqual([false]);
       expect(Date.now() - t0).toBeLessThan(1000);
     } finally {
       server.closeAllConnections();
diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts
index 954eff03..e607ef79 100644
--- a/src/observe/reporter.ts
+++ b/src/observe/reporter.ts
@@ -36,7 +36,7 @@ import { SDKError } from '../execution/errors';
 import type { Span, TraceExporter } from '../execution/tracing';
 import { VERSION } from '../version';
 import { blockingPost } from './blockingPost';
-import { OBSERVE_DEFAULT_PORT, OBSERVE_MAX_BATCH_EVENTS, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope';
+import { OBSERVE_DEFAULT_PORT, OBSERVE_MAX_BATCH_EVENTS, OBSERVE_MAX_BODY_BYTES, type IngestEvent, type ObservationData, type ObservationKind, type PromptUseData } from './envelope';
 import { observeTargetFromEnv } from './emit';
 import { hostName, resolveIdentity, resolveProject } from './identity';
 
@@ -119,16 +119,73 @@ const RETRY_CAP_MS = 30_000;
 /** Wait before the next send after `failures` consecutive failed ones (0-based): 1s, 2s, 4s ... capped at 30s. */
 const retryDelayMs = (failures: number): number => Math.min(RETRY_BASE_MS * 2 ** failures, RETRY_CAP_MS);
 
+/**
+ * The JSON a POST carries for `entries`: byte for byte what
+ * `JSON.stringify({ batch: events })` would produce, built from each event's
+ * JSON serialized once on arrival.
+ */
+const bodyOf = (entries: readonly OutboxEntry[]): string => `{"batch":[${entries.map((e) => e.json).join(',')}]}`;
+
+/** `{"batch":[]}`: the envelope's bytes around the events. */
+const ENVELOPE_BYTES = 12;
+/**
+ * A batch stops growing before its body would pass this - kept under the
+ * daemon's OBSERVE_MAX_BODY_BYTES (5MB, a `413` past it) with room to spare.
+ */
+const BATCH_BUDGET_BYTES = 4 * 1024 * 1024;
+
+/** One pending event, serialized once when queued so batching can measure it. */
+interface OutboxEntry {
+  json: string;
+  bytes: number;
+}
+
+const utf8 = new TextEncoder();
+
+/** An event as an outbox entry; `undefined` when its body alone would pass the daemon's hard cap. */
+function entryOf(event: IngestEvent): OutboxEntry | undefined {
+  const json = JSON.stringify(event);
+  const bytes = utf8.encode(json).length;
+  return ENVELOPE_BYTES + bytes > OBSERVE_MAX_BODY_BYTES ? undefined : { json, bytes };
+}
+
+/**
+ * Where the batch starting at `start` ends: at most OBSERVE_MAX_BATCH_EVENTS
+ * events (the daemon rejects more) and BATCH_BUDGET_BYTES of body, measured
+ * incrementally. The first event is always taken, so one event over the
+ * budget (but under the hard cap) goes out alone instead of wedging the queue.
+ */
+function batchEnd(entries: readonly OutboxEntry[], start: number): number {
+  let end = start;
+  let bytes = ENVELOPE_BYTES;
+  while (end < entries.length && end - start < OBSERVE_MAX_BATCH_EVENTS) {
+    bytes += entries[end].bytes + (end > start ? 1 : 0); // the comma
+    if (end > start && bytes > BATCH_BUDGET_BYTES) break;
+    end++;
+  }
+  return end;
+}
+
+/** `entries` cut into consecutive batches, oldest first. */
+function batchesOf(entries: readonly OutboxEntry[]): OutboxEntry[][] {
+  const batches: OutboxEntry[][] = [];
+  for (let start = 0; start < entries.length; ) {
+    const end = batchEnd(entries, start);
+    batches.push(entries.slice(start, end));
+    start = end;
+  }
+  return batches;
+}
+
 /**
  * The reporter's pending events: FIFO, capped at `cap` (the oldest are
  * evicted first, reported through `onEvict`), and gated by a retry backoff
  * after a failed send. Every send path - size, interval, exit - goes through
  * `take` / `delivered` / `failed`, so a failed batch is never lost while
- * there is room to keep it. A batch is at most OBSERVE_MAX_BATCH_EVENTS
- * events (the daemon rejects larger ones), oldest first.
+ * there is room to keep it. A batch is cut by {@link batchEnd}, oldest first.
  */
 function createOutbox(cap: number, now: () => number, onEvict: (count: number) => void) {
-  let items: IngestEvent[] = [];
+  let items: OutboxEntry[] = [];
   let failures = 0;
   let retryAt = 0;
   const trim = (): void => {
@@ -145,16 +202,20 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) =
     ready(): boolean {
       return now() >= retryAt;
     },
-    push(event: IngestEvent): void {
-      items.push(event);
+    /** Queue `event`; false (nothing queued) when it is too large for any request. */
+    push(event: IngestEvent): boolean {
+      const entry = entryOf(event);
+      if (!entry) return false;
+      items.push(entry);
       trim();
+      return true;
     },
-    /** The oldest batch's worth, for one send - or nothing while backing off, unless `force` (the final attempt). */
-    take(force = false): IngestEvent[] {
-      return force || now() >= retryAt ? items.splice(0, OBSERVE_MAX_BATCH_EVENTS) : [];
+    /** The oldest batch, for one send - or nothing while backing off, unless `force` (the final attempt). */
+    take(force = false): OutboxEntry[] {
+      return force || now() >= retryAt ? items.splice(0, batchEnd(items, 0)) : [];
     },
-    /** Everything pending, uncapped (the crash flush; counting what a failed final attempt leaves). */
-    drain(): IngestEvent[] {
+    /** Everything pending, uncut (the crash flush; counting what a failed final attempt leaves). */
+    drain(): OutboxEntry[] {
       return items.splice(0, items.length);
     },
     delivered(): void {
@@ -162,8 +223,8 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) =
       retryAt = 0;
     },
     /** Put a failed batch back in front of anything queued since, trim to the cap, and back off. */
-    failed(batch: IngestEvent[]): void {
-      items = batch.concat(items); // no spread: a batch can hold up to `cap` events, too many for call arguments
+    failed(batch: OutboxEntry[]): void {
+      items = batch.concat(items); // no spread: keeps call arguments bounded whatever the batch size
       trim();
       retryAt = now() + retryDelayMs(failures++);
     },
@@ -310,8 +371,14 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
     );
   });
 
+  /** Count events the daemon will never take - too large, or a batch it refused - and warn once. */
+  const discard = (count: number, why: string): void => {
+    droppedCount += count;
+    warnOnce(why, new SDKError('observe event rejected', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }));
+  };
+
   const push = (event: IngestEvent) => {
-    outbox.push(event);
+    if (!outbox.push(event)) discard(1, `an event is larger than the daemon's ${OBSERVE_MAX_BODY_BYTES}-byte request cap; dropping it`);
     if (outbox.size >= flushSize && outbox.ready()) void flush(requestTimeoutMs);
   };
 
@@ -327,17 +394,16 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
   /** Aborted by close() so a regular flush in flight can't hold the exit flush past its ~2s cap. */
   const closing = new AbortController();
 
-  const post = async (events: IngestEvent[], timeoutMs: number, final: boolean): Promise<void> => {
+  /** One ingest POST; resolves with the status, rejects only when no answer arrived. */
+  const post = async (entries: OutboxEntry[], timeoutMs: number, final: boolean): Promise<number> => {
     const timeout = AbortSignal.timeout(timeoutMs);
     const res = await fetchFn(`${url}/api/v1/ingest`, {
       method: 'POST',
       headers,
-      body: JSON.stringify({ batch: events }),
+      body: bodyOf(entries),
       signal: final ? timeout : AbortSignal.any([timeout, closing.signal]),
     });
-    if (!res.ok) {
-      throw new SDKError(`ingest answered ${res.status}`, 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false });
-    }
+    return res.status;
   };
 
   // Flushes are serialized on a chain: a fire-and-forget size flush and the
@@ -348,12 +414,17 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
   // as an unhandled rejection at the `void flush()` call sites.
   let chain: Promise<void> = Promise.resolve();
   /** The batch a POST is carrying right now: a crash resends it, since that POST can never finish. */
-  let inFlight: IngestEvent[] = [];
+  let inFlight: OutboxEntry[] = [];
   /** One POST; false when it failed (the batch is then requeued, or - final - dropped with everything left). */
-  const sendBatch = async (batch: IngestEvent[], timeoutMs: number, final: boolean): Promise<boolean> => {
+  const sendBatch = async (batch: OutboxEntry[], timeoutMs: number, final: boolean): Promise<boolean> => {
     inFlight = batch;
     try {
-      await post(batch, timeoutMs, final);
+      const status = await post(batch, timeoutMs, final);
+      // A batch the daemon refuses as such (too large, malformed) would be refused again on every retry.
+      if (status === 400 || status === 413) discard(batch.length, `the observe daemon refused a batch of ${batch.length} events (${status}); dropping it`);
+      else if (status < 200 || status >= 300) {
+        throw new SDKError(`ingest answered ${status}`, 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false });
+      }
       outbox.delivered();
       return true;
     } catch (error) {
@@ -368,16 +439,19 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
     }
   };
   /**
-   * Send batch after batch while the last one was full: a backlog (after an
-   * outage) drains at once instead of one batch per interval. Regular flushes
-   * give each POST its own timeout; the final attempt shares one deadline.
+   * Send batch after batch while a cap cut the last one short: a backlog
+   * (after an outage) drains at once instead of one batch per interval.
+   * Regular flushes give each POST its own timeout; the final attempt shares
+   * one deadline.
    */
   const send = async (timeoutMs: number, final: boolean): Promise<void> => {
     const deadline = Date.now() + timeoutMs;
-    let batch = outbox.take(final);
-    while (batch.length > 0) {
+    for (;;) {
+      const batch = outbox.take(final);
+      const cut = outbox.size > 0; // read before the await: events queued meanwhile wait for their own trigger
+      if (batch.length === 0) return;
       if (!(await sendBatch(batch, final ? Math.max(1, deadline - Date.now()) : timeoutMs, final))) return;
-      batch = batch.length < OBSERVE_MAX_BATCH_EVENTS ? [] : outbox.take(final);
+      if (!cut) return;
     }
   };
   const flush = (timeoutMs: number, final = false): Promise<void> => {
@@ -450,17 +524,16 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe
    * The process is dying of an uncaught exception or unhandled rejection:
    * Node exits as soon as the monitor returns, so no async send can finish.
    * Send everything pending, oldest first - the batch a regular POST was
-   * carrying included (re-sent events upsert by key) - as successive
-   * batch-sized POSTs that together block for at most 1s.
+   * carrying included (re-sent events upsert by key) - as successive POSTs,
+   * cut like any other batch, that together block for at most 1s.
    */
   const crashFlush = (): void => {
     if (!shutdown()) return;
-    const pending = inFlight.concat(outbox.drain());
-    const batches: IngestEvent[][] = [];
-    for (let i = 0; i < pending.length; i += OBSERVE_MAX_BATCH_EVENTS) batches.push(pending.slice(i, i + OBSERVE_MAX_BATCH_EVENTS));
-    const bodies = batches.map((batch) => JSON.stringify({ batch }));
-    const delivered = blockingPost(`${url}/api/v1/ingest`, headers, bodies, CRASH_FLUSH_TIMEOUT_MS);
-    for (const batch of batches.slice(delivered)) droppedCount += batch.length;
+    const batches = batchesOf(inFlight.concat(outbox.drain()));
+    const delivered = blockingPost(`${url}/api/v1/ingest`, headers, batches.map(bodyOf), CRASH_FLUSH_TIMEOUT_MS);
+    batches.forEach((batch, i) => {
+      if (!delivered[i]) droppedCount += batch.length;
+    });
   };
 
   const exitHooks: Array<[string, () => void]> = [];