authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-09-17 16:41:05-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-17 16:41:05-07:00
logb17cc2ab5d5029f898121798ca19db3288cba616
treeb8e219c7550f8b65b9fd95598948442fc63b3586
parent4cc984fedc33184c0a0fb67ea17fb02902ae2224
signature Signed by PGP key B5690EEEBB952194

fix(server): bound provider event log records before serialization (#12305)

2 files changed, 309 insertions(+), 6 deletions(-)

apps/server/src/provider/Layers/EventNdjsonLogger.test.ts+171-2
......@@ -21,6 +21,7 @@ import {
2121} from "./EventNdjsonLogger.ts";
2222
2323const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
24const decodeUnknownJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
2425
2526function ownedLogPath(basePath: string, segment: string): string {
2627 const basename = NodePath.basename(basePath);
......@@ -49,7 +50,7 @@ function parseLogLine(line: string) {
4950}
5051
5152describe("EventNdjsonLogger", () => {
52 it.effect("logs bounded diagnostics when an event cannot be serialized", () => {
53 it.effect("summarizes circular events without exposing their contents in diagnostics", () => {
5354 const messages: Array<unknown> = [];
5455 const logCapture = Logger.make<unknown, void>(({ message }) => {
5556 if (Array.isArray(message)) {
......@@ -71,10 +72,14 @@ describe("EventNdjsonLogger", () => {
7172 assert.exists(logger);
7273 if (!logger) return;
7374 yield* logger.write(circular, ThreadId.make("thread-1"));
75 yield* logger.close();
7476
7577 const serialized = encodeUnknownJson(messages);
7678 assert.notInclude(serialized, secret);
77 assert.include(serialized, '"errorTag":"SchemaError"');
79 const line = parseLogLine(
80 NodeFS.readFileSync(ownedLogPath(basePath, "thread-1"), "utf8").trim(),
81 );
82 assert.equal(line.payload, '{"truncated":true}');
7883 } finally {
7984 NodeFS.rmSync(tempDir, { recursive: true, force: true });
8085 }
......@@ -314,6 +319,22 @@ describe("EventNdjsonLogger", () => {
314319 { method: "thread/realtime/transcript/delta", payload: circularDelta },
315320 threadId,
316321 );
322 yield* native.write({ method: "turn/diff/updated", payload: circularDelta }, threadId);
323 yield* native.write(
324 {
325 event: {
326 direction: "incoming",
327 stage: "decoded",
328 payload: {
329 method: "turn/diff/updated",
330 get params() {
331 throw new Error("unused diff snapshots must not be traversed");
332 },
333 },
334 },
335 },
336 threadId,
337 );
317338 yield* native.write(
318339 {
319340 event: {
......@@ -332,6 +353,41 @@ describe("EventNdjsonLogger", () => {
332353 },
333354 threadId,
334355 );
356 yield* native.write(
357 {
358 event: {
359 direction: "incoming",
360 stage: "decoded",
361 payload: { method: "item/commandExecution/outputDelta", params: circularDelta },
362 },
363 },
364 threadId,
365 );
366 yield* native.write(
367 {
368 event: {
369 direction: "incoming",
370 stage: "decoded",
371 payload: {
372 type: "stream_event",
373 event: { type: "content_block_delta", delta: circularDelta },
374 },
375 },
376 },
377 threadId,
378 );
379 yield* native.write(
380 {
381 event: {
382 direction: "incoming",
383 stage: "raw",
384 get payload() {
385 throw new Error("raw frames must not be inspected");
386 },
387 },
388 },
389 threadId,
390 );
335391 yield* native.write(
336392 {
337393 event: {
......@@ -375,6 +431,119 @@ describe("EventNdjsonLogger", () => {
375431 }),
376432 );
377433
434 it.effect("summarizes large histories without reading their items", () =>
435 Effect.gen(function* () {
436 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
437 const basePath = NodePath.join(tempDir, "events.log");
438 const turns = Array.from({ length: 10_000 });
439 Object.defineProperty(turns, 0, {
440 get: () => {
441 throw new Error("history must not be serialized");
442 },
443 });
444
445 try {
446 const store = yield* makeEventNdjsonLogStore(basePath);
447 yield* store.logger("native").write(
448 {
449 provider: "codex",
450 event: {
451 direction: "incoming",
452 stage: "decoded",
453 payload: { id: 42, result: { thread: { id: "native-thread", turns } } },
454 },
455 },
456 ThreadId.make("large-history"),
457 );
458 yield* store.close();
459
460 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-history"), "utf8");
461 assert.isBelow(Buffer.byteLength(contents), 2_048);
462 const record = decodeUnknownJson(parseLogLine(contents.trim()).payload);
463 assert.nestedPropertyVal(record, "event.payload.id", 42);
464 assert.nestedPropertyVal(record, "event.payload.result.thread.id", "native-thread");
465 assert.nestedPropertyVal(record, "event.payload.result.thread.turns.itemCount", 10_000);
466 } finally {
467 NodeFS.rmSync(tempDir, { recursive: true, force: true });
468 }
469 }),
470 );
471
472 it.effect("bounds oversized records while retaining failure details", () =>
473 Effect.gen(function* () {
474 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
475 const basePath = NodePath.join(tempDir, "events.log");
476
477 try {
478 const store = yield* makeEventNdjsonLogStore(basePath);
479 const logger = store.logger("native");
480 const threadId = ThreadId.make("large-error");
481 const failure = {
482 method: "error",
483 params: {
484 threadId: "native-thread",
485 turnId: "native-turn",
486 error: { message: "The provider is unavailable.", code: "overloaded" },
487 output: "x".repeat(128 * 1_024),
488 },
489 };
490 yield* logger.write(failure, threadId);
491 yield* logger.write({ id: "escaped", output: "\u0000".repeat(20_000) }, threadId);
492 yield* store.close();
493
494 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-error"), "utf8");
495 const records = contents
496 .trim()
497 .split("\n")
498 .map((line) => decodeUnknownJson(parseLogLine(line).payload));
499 assert.isBelow(Buffer.byteLength(contents), 64 * 1_024);
500 assert.equal(records.length, 2);
501 assert.nestedPropertyVal(records[0], "params.threadId", "native-thread");
502 assert.nestedPropertyVal(records[0], "params.turnId", "native-turn");
503 assert.nestedPropertyVal(
504 records[0],
505 "params.error.message",
506 "The provider is unavailable.",
507 );
508 assert.nestedPropertyVal(records[0], "params.error.code", "overloaded");
509 assert.propertyVal(records[1], "id", "escaped");
510 } finally {
511 NodeFS.rmSync(tempDir, { recursive: true, force: true });
512 }
513 }),
514 );
515
516 it.effect("bounds canonical diff snapshots before serializing their duplicate payloads", () =>
517 Effect.gen(function* () {
518 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
519 const basePath = NodePath.join(tempDir, "events.log");
520 const threadId = ThreadId.make("large-diff");
521 const diff = "diff-payload".repeat(128 * 1_024);
522 try {
523 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
524 yield* store.logger("canonical").write(
525 {
526 type: "turn.diff.updated",
527 threadId,
528 turnId: "native-turn",
529 raw: { method: "turn/diff/updated", payload: { diff } },
530 payload: { unifiedDiff: diff },
531 },
532 threadId,
533 );
534 yield* store.close();
535 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-diff"), "utf8");
536 assert.isBelow(Buffer.byteLength(contents), 2_048);
537 const record = decodeUnknownJson(parseLogLine(contents.trim()).payload);
538 assert.propertyVal(record, "type", "turn.diff.updated");
539 assert.propertyVal(record, "threadId", threadId);
540 assert.propertyVal(record, "turnId", "native-turn");
541 } finally {
542 NodeFS.rmSync(tempDir, { recursive: true, force: true });
543 }
544 }),
545 );
546
378547 it.effect("keeps OpenCode tool input, final output, and errors in native logs", () =>
379548 Effect.gen(function* () {
380549 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
apps/server/src/provider/Layers/EventNdjsonLogger.ts+138-4
......@@ -32,6 +32,9 @@ const DEFAULT_MAX_AGE_MS = 14 * DAY_MS;
3232const DEFAULT_RETENTION_CHECK_INTERVAL_MS = 5 * 60 * 1_000;
3333const DEFAULT_MAX_BUFFERED_BYTES = MEBIBYTE;
3434const DEFAULT_MAX_BUFFERED_RECORDS = 512;
35const MAX_RECORD_CHARACTERS = 64 * 1024;
36const MAX_RECORD_FIELDS = 1_024;
37const MAX_RECORD_DEPTH = 16;
3538const GLOBAL_THREAD_SEGMENT = "_global";
3639const LOG_SCOPE = "provider-observability";
3740const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
......@@ -54,6 +57,7 @@ const transientNativeMethods = new Set([
5457 "item/reasoning/textDelta",
5558 "thread/realtime/outputAudio/delta",
5659 "thread/realtime/transcript/delta",
60 "turn/diff/updated",
5761]);
5862const transientAcpUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
5963
......@@ -188,7 +192,7 @@ function providerLogPath(directory: string, prefix: string, threadSegment: strin
188192 return NodePath.join(directory, `${prefix}${threadSegment}.log`);
189193}
190194
191function shouldPersist(stream: EventNdjsonStream, event: unknown): boolean {
195function shouldPersistProviderEvent(stream: EventNdjsonStream, event: unknown): boolean {
192196 if (stream === "orchestration" || typeof event !== "object" || event === null) {
193197 return true;
194198 }
......@@ -200,7 +204,17 @@ function shouldPersist(stream: EventNdjsonStream, event: unknown): boolean {
200204 if (stream !== "native") return true;
201205
202206 const nested = Reflect.get(event, "event");
203 const nativeEvent = typeof nested === "object" && nested !== null ? nested : event;
207 const envelope = typeof nested === "object" && nested !== null ? nested : event;
208 // Decoded frames carry the same information as raw frames without another
209 // copy of every token delta. Decode failures have their own diagnostic frame.
210 if (Reflect.get(envelope, "stage") === "raw") return false;
211 const decodedPayload = Reflect.get(envelope, "payload");
212 const nativeEvent =
213 Reflect.get(envelope, "stage") === "decoded" &&
214 typeof decodedPayload === "object" &&
215 decodedPayload !== null
216 ? decodedPayload
217 : envelope;
204218 const method = Reflect.get(nativeEvent, "method");
205219 if (
206220 typeof method === "string" &&
......@@ -212,6 +226,16 @@ function shouldPersist(stream: EventNdjsonStream, event: unknown): boolean {
212226
213227 const nativeType = Reflect.get(nativeEvent, "type");
214228 if (nativeType === "message.part.delta") return false;
229 if (nativeType === "stream_event") {
230 const streamEvent = Reflect.get(nativeEvent, "event");
231 if (
232 typeof streamEvent === "object" &&
233 streamEvent !== null &&
234 Reflect.get(streamEvent, "type") === "content_block_delta"
235 ) {
236 return false;
237 }
238 }
215239
216240 const payload = Reflect.get(nativeEvent, "payload");
217241 if (typeof payload !== "object" || payload === null) return true;
......@@ -246,6 +270,110 @@ function shouldPersist(stream: EventNdjsonStream, event: unknown): boolean {
246270 }
247271}
248272
273const summaryFields = [
274 "provider",
275 "protocol",
276 "kind",
277 "providerSessionId",
278 "direction",
279 "stage",
280 "type",
281 "subtype",
282 "method",
283 "id",
284 "threadId",
285 "turnId",
286 "requestId",
287 "session_id",
288 "status",
289 "is_error",
290 "api_error_status",
291 "terminal_reason",
292 "stop_reason",
293 "operation",
294 "code",
295 "willRetry",
296 "message",
297 "event",
298 "payload",
299 "params",
300 "result",
301 "thread",
302 "turn",
303 "error",
304 "turns",
305 "items",
306 "content",
307] as const;
308
309function summarizeProviderEvent(event: unknown): unknown {
310 let remainingFields = 128;
311 let remainingCharacters = 8 * 1024;
312 const summarize = (value: unknown, depth: number): unknown => {
313 if (typeof value === "string") {
314 if (value.length > Math.min(1_024, remainingCharacters)) {
315 return { omittedCharacters: value.length };
316 }
317 remainingCharacters -= value.length;
318 return value;
319 }
320 if (value === null || typeof value === "number" || typeof value === "boolean") return value;
321 if (typeof value !== "object") return undefined;
322 if (Array.isArray(value)) return { itemCount: value.length };
323 const summary: Record<string, unknown> = { truncated: true };
324 if (depth >= 6) return summary;
325 for (const key of summaryFields) {
326 if (remainingFields <= 0) break;
327 const nested = Reflect.get(value, key);
328 if (nested === undefined) continue;
329 remainingFields -= 1;
330 summary[key] = summarize(nested, depth + 1);
331 }
332 return summary;
333 };
334 try {
335 return summarize(event, 0);
336 } catch {
337 return { truncated: true };
338 }
339}
340
341/** Bounds traversal before the logger encodes payloads. */
342function boundProviderEventForLogging(event: unknown): unknown {
343 let remainingCharacters = MAX_RECORD_CHARACTERS;
344 let remainingFields = MAX_RECORD_FIELDS;
345 const ancestors = new WeakSet<object>();
346 const fits = (value: unknown, depth: number): boolean => {
347 if (typeof value === "string") {
348 remainingCharacters -= value.length;
349 return remainingCharacters >= 0;
350 }
351 if (typeof value !== "object" || value === null) return true;
352 if (depth > MAX_RECORD_DEPTH || ancestors.has(value)) return false;
353 if (Array.isArray(value) && value.length > remainingFields) return false;
354 ancestors.add(value);
355 for (const key in value) {
356 if (!Object.hasOwn(value, key)) continue;
357 remainingFields -= 1;
358 remainingCharacters -= key.length;
359 if (
360 remainingFields < 0 ||
361 remainingCharacters < 0 ||
362 !fits(Reflect.get(value, key), depth + 1)
363 )
364 return false;
365 }
366 ancestors.delete(value);
367 return true;
368 };
369 try {
370 if (fits(event, 0)) return event;
371 } catch {
372 // A failing accessor must not escape into provider processing.
373 }
374 return summarizeProviderEvent(event);
375}
376
249377export function writeBatchedMessages(
250378 sink: Pick<RotatingFileSink, "write">,
251379 records: ReadonlyArray<PendingRecord>,
......@@ -612,9 +740,15 @@ export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
612740 if (existing) return existing;
613741
614742 const write = Effect.fnUntraced(function* (event: unknown, threadId: ThreadId | null) {
615 if (!shouldPersist(stream, event)) return;
616 const payload = yield* serializeEvent(event);
743 if (!shouldPersistProviderEvent(stream, event)) return;
744 let payload = yield* serializeEvent(boundProviderEventForLogging(event));
617745 if (payload === undefined) return;
746 // Escaping can expand strings beyond their input size. Keep that bounded
747 // serialization out of the file too, while retaining routing/error fields.
748 if (Buffer.byteLength(payload) > MAX_RECORD_CHARACTERS) {
749 payload = yield* serializeEvent(summarizeProviderEvent(event));
750 if (payload === undefined) return;
751 }
618752
619753 const observedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso));
620754 const line = `[${observedAt}] ${resolveStreamLabel(stream)}: ${payload}\n`;