fix(cloud-agent-next): skip serialization of oversized ingest events

services/cloud-agent-next/src/shared/ingest-frame.test.ts

@@ -3,6 +3,7 @@ import {
   MAX_INGEST_BUFFERED_BYTES,
   MAX_INGEST_EVENT_BYTES,
   IngestEventBuffer,
+  estimateSerializedBytes,
   isLifecycleIngestEvent,
   prepareIngestFrame,
 } from './ingest-frame.js';
@@ -231,6 +232,125 @@ describe('prepareIngestFrame', () => {
     expect(sent.streamEventType).toBe('wrapper_event_truncated');
     expect(sent.data.kiloEventName).toBe('message.part.updated');
   });
+
+  it('compacts a very large message.updated without serializing the full payload', () => {
+    // A 10 MB event would block the event loop for seconds if JSON.stringify
+    // were called on the full payload. The estimate must short-circuit and
+    // route directly to compaction.
+    const event: IngestEvent = {
+      streamEventType: 'kilocode',
+      timestamp: '2026-04-14T08:00:00.000Z',
+      data: {
+        event: 'message.updated',
+        type: 'message.updated',
+        properties: {
+          info: {
+            id: 'msg_big',
+            sessionID: 'sess_1',
+            role: 'assistant',
+            time: { started: 1, completed: 2 },
+            parts: [{ type: 'text', text: 'z'.repeat(10_000_000) }],
+          },
+        },
+      },
+    };
+    const frame = prepareIngestFrame(event);
+    expect(frame.kind).toBe('send');
+    if (frame.kind !== 'send') return;
+    expect(frame.compacted).toBe(true);
+    expect(frame.bytes).toBeLessThanOrEqual(MAX_INGEST_EVENT_BYTES);
+    // The huge text must not appear in the compacted output.
+    expect(frame.serialized).not.toContain('zzzzz');
+    const sent = JSON.parse(frame.serialized);
+    expect(sent.data.properties.info.id).toBe('msg_big');
+    expect(sent.data.properties.info.parts).toBeUndefined();
+    // originalBytes reflects the estimate (a lower bound), still over budget.
+    expect(frame.originalBytes).toBeGreaterThan(MAX_INGEST_EVENT_BYTES);
+  });
+
+  it('preserves the exact byte count when the estimate is under budget but serialization exceeds it', () => {
+    // Each emoji is 2 UTF-16 code units but 4 UTF-8 bytes. A string of 400K
+    // emoji has .length = 800K (under the 1 MiB estimate budget) but UTF-8
+    // byte length ≈ 1.6 MB (over the send budget). The estimate passes, so
+    // tryStringify runs and produces an over-budget result — originalBytes
+    // must reflect the exact measurement, not the lower estimate.
+    const event: IngestEvent = {
+      streamEventType: 'status',
+      timestamp: '2026-04-14T08:00:00.000Z',
+      data: { message: '😀'.repeat(400_000) },
+    };
+    const frame = prepareIngestFrame(event);
+    expect(frame.originalBytes).toBeGreaterThan(MAX_INGEST_EVENT_BYTES);
+    if (frame.kind === 'send') {
+      expect(frame.compacted).toBe(true);
+    }
+  });
+});
+
+describe('estimateSerializedBytes', () => {
+  it('returns a lower bound on the actual serialized size', () => {
+    const value = { a: 'hello', b: 42, c: true, d: null, e: [1, 2, 3] };
+    const actual = JSON.stringify(value).length;
+    const estimate = estimateSerializedBytes(value, 1_000_000);
+    expect(estimate).toBeLessThanOrEqual(actual);
+    expect(estimate).toBeGreaterThan(0);
+  });
+
+  it('short-circuits when the estimate exceeds the budget', () => {
+    const value = { big: 'x'.repeat(2_000_000) };
+    const estimate = estimateSerializedBytes(value, 1_000_000);
+    expect(estimate).toBeGreaterThan(1_000_000);
+    // Should not walk the entire object — just enough to exceed the budget.
+    expect(estimate).toBeLessThan(2_100_000);
+  });
+
+  it('handles circular references without infinite looping', () => {
+    const circular: Record<string, unknown> = { a: 'hello' };
+    circular.self = circular;
+    const estimate = estimateSerializedBytes(circular, 1_000_000);
+    expect(estimate).toBeLessThan(100);
+  });
+
+  it('counts shared (non-cyclic) references at each occurrence', () => {
+    // A shared sub-object referenced from two keys must be counted twice,
+    // matching JSON.stringify (which serializes it at each occurrence).
+    const shared = { text: 'x'.repeat(100_000) };
+    const value = { a: shared, b: shared };
+    const actual = JSON.stringify(value).length;
+    const estimate = estimateSerializedBytes(value, 1_000_000);
+    expect(estimate).toBeLessThanOrEqual(actual);
+    // Counting shared only once would underestimate by ~100K bytes.
+    expect(estimate).toBeGreaterThan(200_000);
+  });
+
+  it('short-circuits on wide arrays via structural overhead', () => {
+    // 2M-element sparse array: comma overhead (2M-1) alone exceeds the 1M
+    // budget, so the estimate returns without queuing or walking elements.
+    const value = { arr: new Array(2_000_000) };
+    const estimate = estimateSerializedBytes(value, 1_000_000);
+    expect(estimate).toBeGreaterThan(1_000_000);
+    expect(estimate).toBeLessThan(2_010_000);
+  });
+
+  it('returns 2 for an empty object', () => {
+    expect(estimateSerializedBytes({}, 1_000_000)).toBe(2);
+  });
+
+  it('returns 2 for an empty array', () => {
+    expect(estimateSerializedBytes([], 1_000_000)).toBe(2);
+  });
+
+  it('treats errors as oversized so compaction is used', () => {
+    const obj: Record<string, unknown> = {};
+    Object.defineProperty(obj, 'throwingGetter', {
+      get: () => {
+        throw new Error('boom');
+      },
+      enumerable: true,
+    });
+    const estimate = estimateSerializedBytes(obj, 1_000_000);
+    expect(estimate).toBeGreaterThan(1_000_000);
+  });
 });
 
 describe('IngestEventBuffer', () => {

services/cloud-agent-next/src/shared/ingest-frame.ts

@@ -317,6 +317,75 @@ function tryStringify(event: IngestEvent): string | undefined {
   }
 }
 
+/**
+ * Cheap lower-bound estimate of the JSON-serialized UTF-8 byte length of a
+ * value. Walks the object tree summing string lengths and structural
+ * overhead, short-circuiting as soon as the running total exceeds `budget`.
+ *
+ * This avoids calling `JSON.stringify` on oversized events (which blocks the
+ * event loop for seconds on multi-MB payloads) just to discover they need
+ * compaction. The estimate is a conservative lower bound — if it says
+ * "under budget", the full serialization is attempted (which is fast for
+ * genuinely small events). If it says "over budget", compaction runs
+ * directly without ever materializing the full serialized string.
+ *
+ * Ancestor cycles (true circular references) are detected via a `WeakSet`
+ * tracking the current path and skipped. Shared (non-cyclic) references are
+ * counted at each occurrence, matching `JSON.stringify`.
+ */
+export function estimateSerializedBytes(value: unknown, budget: number): number {
+  try {
+    let total = 0;
+    const stack: Array<{ v: unknown } | { exit: object }> = [{ v: value }];
+    const path = new WeakSet<object>();
+    while (stack.length > 0) {
+      const entry = stack.pop();
+      if (entry === undefined) break;
+      if ('exit' in entry) {
+        path.delete(entry.exit);
+        continue;
+      }
+      const item = entry.v;
+      if (typeof item === 'string') {
+        total += item.length + 2; // surrounding quotes
+      } else if (typeof item === 'number') {
+        total += 1; // lower bound (actual: 1–21 chars)
+      } else if (typeof item === 'boolean') {
+        total += 4; // "true" (4) or "false" (5)
+      } else if (item === null) {
+        total += 4; // "null"
+      } else if (typeof item === 'object') {
+        if (path.has(item)) continue; // ancestor cycle — skip
+        path.add(item);
+        stack.push({ exit: item }); // remove from path after children
+        if (Array.isArray(item)) {
+          total += 2; // brackets
+          total += Math.max(0, item.length - 1); // commas
+          if (total > budget) return total;
+          for (let i = item.length - 1; i >= 0; i--) {
+            stack.push({ v: item[i] });
+          }
+        } else {
+          const keys = Object.keys(item);
+          total += 2; // braces
+          for (const key of keys) {
+            total += key.length + 3; // "key":
+          }
+          total += Math.max(0, keys.length - 1); // commas
+          if (total > budget) return total;
+          for (let i = keys.length - 1; i >= 0; i--) {
+            stack.push({ v: (item as Record<string, unknown>)[keys[i]] });
+          }
+        }
+      }
+      if (total > budget) return total;
+    }
+    return total;
+  } catch {
+    return budget + 1; // treat as oversized on error
+  }
+}
+
 /**
  * Trim, serialize, measure, and (if needed) compact an ingest event so its
  * serialized UTF-8 byte length stays under `MAX_INGEST_EVENT_BYTES`.
@@ -326,18 +395,38 @@ export function prepareIngestFrame(event: IngestEvent): PreparedIngestFrame {
     ...event,
     data: trimPayload(event.streamEventType, event.data),
   };
-  const originalSerialized = tryStringify(trimmed);
-  const originalBytes = originalSerialized === undefined ? 0 : byteLength(originalSerialized);
-  if (originalSerialized !== undefined && originalBytes <= MAX_INGEST_EVENT_BYTES) {
-    return {
-      kind: 'send',
-      serialized: originalSerialized,
-      bytes: originalBytes,
-      compacted: false,
-      originalBytes,
-    };
+
+  // Pre-serialization estimate: if the event is clearly over budget, skip
+  // the full JSON.stringify (which blocks the event loop for seconds on
+  // multi-MB payloads) and go straight to compaction. The estimate is a
+  // conservative lower bound, so events that pass are still measured
+  // exactly via tryStringify below.
+  const estimate = estimateSerializedBytes(trimmed, MAX_INGEST_EVENT_BYTES);
+
+  let originalSerialized: string | undefined;
+  let exactBytes: number | undefined;
+  if (estimate <= MAX_INGEST_EVENT_BYTES) {
+    originalSerialized = tryStringify(trimmed);
+    if (originalSerialized !== undefined) {
+      const originalBytes = byteLength(originalSerialized);
+      if (originalBytes <= MAX_INGEST_EVENT_BYTES) {
+        return {
+          kind: 'send',
+          serialized: originalSerialized,
+          bytes: originalBytes,
+          compacted: false,
+          originalBytes,
+        };
+      }
+      exactBytes = originalBytes;
+    }
   }
 
+  // Oversized or unserializable — compact without full serialization.
+  // Prefer the exact measurement when serialization succeeded; otherwise
+  // fall back to the estimate (a lower bound sufficient for observability).
+  const originalBytes = exactBytes ?? estimate;
+
   const compactedEvent = compactIngestEvent(trimmed, originalBytes);
   const compactedSerialized = tryStringify(compactedEvent);
   if (compactedSerialized !== undefined) {
@@ -372,7 +461,9 @@ export function prepareIngestFrame(event: IngestEvent): PreparedIngestFrame {
     kind: 'dropped',
     originalBytes,
     reason:
-      originalSerialized === undefined ? 'unserializable_ingest_event' : 'oversized_ingest_event',
+      originalSerialized === undefined && estimate <= MAX_INGEST_EVENT_BYTES
+        ? 'unserializable_ingest_event'
+        : 'oversized_ingest_event',
   };
 }
 

services/cloud-agent-next/wrapper/src/global-feed.ts

@@ -246,6 +246,22 @@ export function openKiloGlobalFeed(options: OpenKiloGlobalFeedOptions): KiloGlob
     let backpressureSince: number | undefined;
     for await (const data of parseSseDataStream(response.body)) {
       if (closed || abortController.signal.aborted) break;
+
+      // Pre-parse size guard: skip JSON.parse + JSON.stringify on oversized
+      // SSE frames, which block the event loop for seconds on multi-MB
+      // payloads. data.length is a UTF-16 code-unit count (≤ UTF-8 byte
+      // length), so this is a safe lower-bound over-limit check.
+      if (data.length > MAX_GLOBAL_FEED_WEBSOCKET_BUFFERED_BYTES) {
+        logToFile(
+          JSON.stringify({
+            message: 'kilo_global_feed_event_dropped_oversized',
+            eventBytes: data.length,
+            limitBytes: MAX_GLOBAL_FEED_WEBSOCKET_BUFFERED_BYTES,
+          })
+        );
+        continue;
+      }
+
       let parsed: unknown;
       try {
         parsed = JSON.parse(data);