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..b6146985cd 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,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 = { 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 = {}; + 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..5e1c6cfbc8 100644 --- a/services/cloud-agent-next/src/shared/ingest-frame.ts +++ b/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(); + 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)[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', }; } 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);