From afe37d427d2ce059bbd4c7791c679b038e13e108 Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Tue, 14 Jul 2026 12:06:17 +0200 Subject: [PATCH 1/2] fix(cloud-agent-next): skip serialization of oversized ingest events prepareIngestFrame called JSON.stringify on the full event payload to measure its size before deciding whether to compact. For a 49 MB event this blocks the event loop for seconds, starving liveness pings and leading to wrapper_failure. The global feed had the same problem: JSON.parse + JSON.stringify ran on every SSE frame before the size check could drop it. Add estimateSerializedBytes: a cheap, short-circuiting tree walk that returns a conservative lower bound on the serialized byte length without materializing the full string. prepareIngestFrame now skips JSON.stringify when the estimate already exceeds the 1 MiB budget and routes directly to compaction. Add a pre-parse data.length guard in the global feed so oversized SSE frames are dropped before JSON.parse runs. Root cause of the incident in ses_0a57aabf3fffFvDB0le1HDtL8V: a 49 MB message.updated event blocked the wrapper event loop during size measurement, missing the liveness ping deadline. --- .../src/shared/ingest-frame.test.ts | 81 ++++++++++++++++ .../src/shared/ingest-frame.ts | 97 ++++++++++++++++--- .../wrapper/src/global-feed.ts | 16 +++ 3 files changed, 183 insertions(+), 11 deletions(-) diff --git a/services/cloud-agent-next/src/shared/ingest-frame.test.ts b/services/cloud-agent-next/src/shared/ingest-frame.test.ts index 19ebbffd46..881a794ebd 100644 --- a/services/cloud-agent-next/src/shared/ingest-frame.test.ts +++ b/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,86 @@ 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); + }); +}); + +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 = { a: 'hello' }; + circular.self = circular; + const estimate = estimateSerializedBytes(circular, 1_000_000); + expect(estimate).toBeLessThan(100); + }); + + 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 = {}; + 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', () => { diff --git a/services/cloud-agent-next/src/shared/ingest-frame.ts b/services/cloud-agent-next/src/shared/ingest-frame.ts index ad0a65c3b2..cf3add0d66 100644 --- a/services/cloud-agent-next/src/shared/ingest-frame.ts +++ b/services/cloud-agent-next/src/shared/ingest-frame.ts @@ -317,6 +317,61 @@ 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. + * + * Circular references are detected via a `WeakSet` and skipped. + */ +export function estimateSerializedBytes(value: unknown, budget: number): number { + try { + let total = 0; + const stack: unknown[] = [value]; + const visited = new WeakSet(); + while (stack.length > 0) { + const item = stack.pop(); + 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 (visited.has(item)) continue; // circular reference — skip + visited.add(item); + if (Array.isArray(item)) { + total += 2; // brackets + for (let i = item.length - 1; i >= 0; i--) { + stack.push(item[i]); + } + } else { + total += 2; // braces + const keys = Object.keys(item); + for (let i = keys.length - 1; i >= 0; i--) { + const key = keys[i]; + total += key.length + 3; // "key": + stack.push((item as Record)[key]); + } + } + } + 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 +381,36 @@ 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; + 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, + }; + } + } } + // Oversized or unserializable — compact without full serialization. + // The estimate serves as originalBytes (a lower bound sufficient for + // observability; the exact size is not actionable post-compaction). + const originalBytes = estimate; + const compactedEvent = compactIngestEvent(trimmed, originalBytes); const compactedSerialized = tryStringify(compactedEvent); if (compactedSerialized !== undefined) { @@ -372,7 +445,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', }; } diff --git a/services/cloud-agent-next/wrapper/src/global-feed.ts b/services/cloud-agent-next/wrapper/src/global-feed.ts index 3c1ecdff77..a69ff74f63 100644 --- a/services/cloud-agent-next/wrapper/src/global-feed.ts +++ b/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); From 845b2749f462ff337f019bb046e302a705a3ad6e Mon Sep 17 00:00:00 2001 From: Evgeny Shurakov Date: Tue, 14 Jul 2026 12:57:18 +0200 Subject: [PATCH 2/2] fix(cloud-agent-next): address ingest frame estimate review feedback - Track ancestor path only (not all visited objects) so shared non-cyclic references are counted at each occurrence, matching JSON.stringify - Account for array/object structural overhead (commas) and short-circuit before queuing children, preventing O(n) stack growth on wide arrays - Preserve exact byte count when tryStringify succeeds but exceeds budget, instead of discarding it in favor of the lower estimate --- .../src/shared/ingest-frame.test.ts | 39 ++++++++++++++++ .../src/shared/ingest-frame.ts | 44 +++++++++++++------ 2 files changed, 69 insertions(+), 14 deletions(-) diff --git a/services/cloud-agent-next/src/shared/ingest-frame.test.ts b/services/cloud-agent-next/src/shared/ingest-frame.test.ts index 881a794ebd..b6146985cd 100644 --- a/services/cloud-agent-next/src/shared/ingest-frame.test.ts +++ b/services/cloud-agent-next/src/shared/ingest-frame.test.ts @@ -267,6 +267,24 @@ describe('prepareIngestFrame', () => { // 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', () => { @@ -293,6 +311,27 @@ describe('estimateSerializedBytes', () => { 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); }); diff --git a/services/cloud-agent-next/src/shared/ingest-frame.ts b/services/cloud-agent-next/src/shared/ingest-frame.ts index cf3add0d66..5e1c6cfbc8 100644 --- a/services/cloud-agent-next/src/shared/ingest-frame.ts +++ b/services/cloud-agent-next/src/shared/ingest-frame.ts @@ -329,15 +329,23 @@ function tryStringify(event: IngestEvent): string | undefined { * genuinely small events). If it says "over budget", compaction runs * directly without ever materializing the full serialized string. * - * Circular references are detected via a `WeakSet` and skipped. + * 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: unknown[] = [value]; - const visited = new WeakSet(); + const stack: Array<{ v: unknown } | { exit: object }> = [{ v: value }]; + const path = new WeakSet(); while (stack.length > 0) { - const item = stack.pop(); + 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') { @@ -347,20 +355,26 @@ export function estimateSerializedBytes(value: unknown, budget: number): number } else if (item === null) { total += 4; // "null" } else if (typeof item === 'object') { - if (visited.has(item)) continue; // circular reference — skip - visited.add(item); + 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(item[i]); + stack.push({ v: item[i] }); } } else { - total += 2; // braces const keys = Object.keys(item); - for (let i = keys.length - 1; i >= 0; i--) { - const key = keys[i]; + total += 2; // braces + for (const key of keys) { total += key.length + 3; // "key": - stack.push((item as Record)[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)[keys[i]] }); } } } @@ -390,6 +404,7 @@ export function prepareIngestFrame(event: IngestEvent): PreparedIngestFrame { 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) { @@ -403,13 +418,14 @@ export function prepareIngestFrame(event: IngestEvent): PreparedIngestFrame { originalBytes, }; } + exactBytes = originalBytes; } } // Oversized or unserializable — compact without full serialization. - // The estimate serves as originalBytes (a lower bound sufficient for - // observability; the exact size is not actionable post-compaction). - const originalBytes = estimate; + // 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);