fix(observe): cap ingest batches by bytes and drop batches the daemon refuses

.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"
   ],

docs/observability.md

@@ -413,10 +413,25 @@ 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
+2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A
+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) 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 —
+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
+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

llms-full.txt

@@ -14079,10 +14079,25 @@ 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
+2 s, 4 s ... capped at 30 s), so a daemon restart mid-run loses nothing. A
+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) 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 —
+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
+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

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([

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[] {

src/observe/__fixtures__/crashAfterRun.mts

@@ -0,0 +1,60 @@
+/**
+ * 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
+ * - 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)
+ *
+ * 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 } : mode === 'backlog' || mode === 'bigbacklog' ? { flushSize: 10_000 } : {}),
+});
+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') {
+  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' || 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);
+} else {
+  throw new Error('fixture: uncaught exception after run');
+}

src/observe/blockingPost.ts

@@ -0,0 +1,67 @@
+/**
+ * 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 requests
+ * run in a throwaway worker thread with its own event loop while this
+ * 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
+ * about to exit, which terminates it.
+ *
+ * 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 };
+}
+
+/** 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 FIRST_BODY = 1;
+
+/**
+ * The worker's whole program (CommonJS, `eval: true`): the bodies in order,
+ * 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 (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: 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, 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): boolean[] {
+  const none = bodies.map(() => false);
+  if (bodies.length === 0) return none;
+  try {
+    const threads = getBuiltinModule<WorkerThreadsLike>('node:worker_threads');
+    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 bodies.map((_, i) => Atomics.load(flag, FIRST_BODY + i) === 1);
+  } catch {
+    return none;
+  }
+}

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;
+}

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;