1// @effect-diagnostics nodeBuiltinImport:off
2/**
3 * Best-effort provider event logging with one shared writer per thread.
4 *
5 * Native and canonical views share batching, rotation, and retention state so
6 * they cannot race while appending to the same thread-scoped file.
7 */
8import * as NodeFS from "node:fs";
9import * as NodePath from "node:path";
10
11import type { ThreadId } from "@t3tools/contracts";
12import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
13import { makeAsyncRotatingFileSink } from "@t3tools/shared/logging";
14import { errorTag } from "@t3tools/shared/observability";
15import * as Clock from "effect/Clock";
16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";
18import * as FileSystem from "effect/FileSystem";
19import * as Path from "effect/Path";
20import * as Result from "effect/Result";
21import * as Exit from "effect/Exit";
22import * as Schema from "effect/Schema";
23import * as Scope from "effect/Scope";
24import * as SynchronizedRef from "effect/SynchronizedRef";
25
26import { toSafeThreadAttachmentSegment } from "../attachmentStore.ts";
27import type * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts";
28
29const MEBIBYTE = 1024 * 1024;
30const DAY_MS = 24 * 60 * 60 * 1_000;
31const DEFAULT_MAX_BYTES = 10 * MEBIBYTE;
32const DEFAULT_MAX_FILES = 10;
33const DEFAULT_BATCH_WINDOW_MS = 1_000;
34const DEFAULT_MAX_TOTAL_BYTES = 512 * MEBIBYTE;
35const DEFAULT_MAX_AGE_MS = 14 * DAY_MS;
36const DEFAULT_RETENTION_CHECK_INTERVAL_MS = 5 * 60 * 1_000;
37const DEFAULT_MAX_BUFFERED_BYTES = MEBIBYTE;
38const DEFAULT_MAX_BUFFERED_RECORDS = 512;
39const MAX_RECORD_CHARACTERS = 64 * 1024;
40const MAX_RECORD_FIELDS = 1_024;
41const MAX_RECORD_DEPTH = 16;
42const GLOBAL_THREAD_SEGMENT = "_global";
43const LOG_SCOPE = "provider-observability";
44const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
45
46const transientCanonicalEventTypes = new Set([
47 "content.delta",
48 "hook.progress",
49 "item.updated",
50 "task.progress",
51 "thread.realtime.audio.delta",
52 "tool.progress",
53 "turn.proposed.delta",
54]);
55const transientNativeMethods = new Set([
56 "item/agentMessage/delta",
57 "item/commandExecution/outputDelta",
58 "item/fileChange/outputDelta",
59 "item/plan/delta",
60 "item/reasoning/summaryTextDelta",
61 "item/reasoning/textDelta",
62 "thread/realtime/outputAudio/delta",
63 "thread/realtime/transcript/delta",
64 "turn/diff/updated",
65]);
66const transientAcpUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
67
68export type EventNdjsonStream = "native" | "canonical" | "orchestration";
69
70export interface EventNdjsonLogStore {
71 readonly filePath: string;
72 readonly logger: (stream: EventNdjsonStream) => ProviderEventLoggers.EventNdjsonLogger;
73 readonly close: () => Effect.Effect<void>;
74}
75
76export interface EventNdjsonLogStoreOptions {
77 readonly maxBytes?: number;
78 readonly maxFiles?: number;
79 readonly batchWindowMs?: number;
80 readonly maxTotalBytes?: number;
81 readonly maxAgeMs?: number;
82 readonly retentionCheckIntervalMs?: number;
83 readonly maxBufferedBytes?: number;
84 readonly maxBufferedRecords?: number;
85 readonly attribution?: ResourceAttribution.ResourceAttribution["Service"];
86}
87
88export interface EventNdjsonLoggerOptions extends EventNdjsonLogStoreOptions {
89 readonly stream: EventNdjsonStream;
90}
91
92export class EventNdjsonLogConfigurationError extends Schema.TaggedError<EventNdjsonLogConfigurationError>()(
93 "EventNdjsonLogConfigurationError",
94 {
95 filePath: Schema.String,
96 option: Schema.String,
97 value: Schema.Number,
98 minimum: Schema.Number,
99 },
100) {
101 override get message(): string {
102 return `Provider event log option '${this.option}' must be an integer >= ${this.minimum}; received ${this.value} for '${this.filePath}'`;
103 }
104}
105
106export class EventNdjsonLogDirectoryError extends Schema.TaggedError<EventNdjsonLogDirectoryError>()(
107 "EventNdjsonLogDirectoryError",
108 {
109 directory: Schema.String,
110 cause: Schema.Defect(),
111 },
112) {
113 override get message(): string {
114 return `Failed to create provider event log directory '${this.directory}'`;
115 }
116}
117
118export type EventNdjsonLogStoreError =
119 | EventNdjsonLogConfigurationError
120 | EventNdjsonLogDirectoryError;
121
122interface ResolvedOptions {
123 readonly maxBytes: number;
124 readonly maxFiles: number;
125 readonly batchWindowMs: number;
126 readonly maxTotalBytes: number;
127 readonly maxAgeMs: number;
128 readonly retentionCheckIntervalMs: number;
129 readonly maxBufferedBytes: number;
130 readonly maxBufferedRecords: number;
131 readonly attribution: ResourceAttribution.ResourceAttribution["Service"] | undefined;
132}
133
134export interface PendingRecord {
135 readonly stream: EventNdjsonStream;
136 readonly threadSegment: string;
137 readonly line: string;
138 readonly bytes: number;
139}
140
141type AsyncFileSink = Effect.Success<ReturnType<typeof makeAsyncRotatingFileSink>>;
142
143interface StoreState {
144 readonly pending: Array<PendingRecord>;
145 readonly pendingBytes: number;
146 readonly sinks: ReadonlyMap<string, AsyncFileSink>;
147 readonly flushScheduled: boolean;
148 readonly closed: boolean;
149 readonly lastRetentionAt: number;
150}
151
152interface AttributionSummary {
153 readonly stream: EventNdjsonStream;
154 readonly count: number;
155 readonly logicalWriteBytes: number;
156}
157
158interface FileOperationFailure {
159 readonly filePath: string;
160 readonly cause: unknown;
161}
162
163interface RetentionResult {
164 readonly failures: ReadonlyArray<FileOperationFailure>;
165}
166
167interface DrainResult {
168 readonly attributions: ReadonlyArray<AttributionSummary>;
169 readonly failures: ReadonlyArray<FileOperationFailure>;
170}
171
172function logWarning(message: string, context: Record<string, unknown>): Effect.Effect<void> {
173 return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE }));
174}
175
176function resolveThreadSegment(raw: string | null | undefined): string {
177 const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null;
178 return normalized ?? GLOBAL_THREAD_SEGMENT;
179}
180
181function resolveStreamLabel(stream: EventNdjsonStream): string {
182 return stream === "native" ? "NTIVE" : stream === "orchestration" ? "ORCH" : "CANON";
183}
184
185function providerLogPrefix(filePath: string): string {
186 const basename = NodePath.basename(filePath);
187 const extension = NodePath.extname(basename);
188 return `${extension.length > 0 ? basename.slice(0, -extension.length) : basename}.`;
189}
190
191function providerLogPath(directory: string, prefix: string, threadSegment: string): string {
192 return NodePath.join(directory, `${prefix}${threadSegment}.log`);
193}
194
195export function shouldPersistProviderEvent(stream: EventNdjsonStream, event: unknown): boolean {
196 if (stream === "orchestration" || typeof event !== "object" || event === null) {
197 return true;
198 }
199 try {
200 const type = Reflect.get(event, "type");
201 if (typeof type === "string" && transientCanonicalEventTypes.has(type)) {
202 return false;
203 }
204 if (stream !== "native") return true;
205
206 const nested = Reflect.get(event, "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;
218 const method = Reflect.get(nativeEvent, "method");
219 if (
220 typeof method === "string" &&
221 (transientNativeMethods.has(method) ||
222 method.startsWith("claude/stream_event/content_block_delta/"))
223 ) {
224 return false;
225 }
226
227 const nativeType = Reflect.get(nativeEvent, "type");
228 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 }
239
240 const payload = Reflect.get(nativeEvent, "payload");
241 if (typeof payload !== "object" || payload === null) return true;
242
243 if (method === "session/update") {
244 const update = Reflect.get(payload, "update");
245 if (typeof update !== "object" || update === null) return true;
246 const updateType = Reflect.get(update, "sessionUpdate");
247 return typeof updateType !== "string" || !transientAcpUpdates.has(updateType);
248 }
249
250 if (nativeType === "message.part.updated") {
251 const properties = Reflect.get(payload, "properties");
252 if (typeof properties !== "object" || properties === null) return true;
253 const part = Reflect.get(properties, "part");
254 if (typeof part !== "object" || part === null) return true;
255 const partType = Reflect.get(part, "type");
256 if (partType === "text" || partType === "reasoning") return false;
257 if (partType === "tool") {
258 // Running snapshots repeat growing output. Pending and terminal states stay in the log.
259 const state = Reflect.get(part, "state");
260 if (typeof state === "object" && state !== null) {
261 return Reflect.get(state, "status") !== "running";
262 }
263 }
264 return true;
265 }
266
267 return true;
268 } catch {
269 return true;
270 }
271}
272
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 adapters copy payloads or the logger encodes them. */
342export function 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
377export const writeBatchedMessages = Effect.fnUntraced(function* (
378 sink: AsyncFileSink,
379 records: ReadonlyArray<PendingRecord>,
380 maxBytes: number,
381 onWritten: (records: ReadonlyArray<PendingRecord>) => void,
382) {
383 let pendingRecords: Array<PendingRecord> = [];
384 let pendingBytes = 0;
385
386 const flush = Effect.gen(function* () {
387 if (pendingRecords.length === 0) return;
388 const writtenRecords = pendingRecords;
389 yield* sink.write(writtenRecords.map((record) => record.line).join(""));
390 onWritten(writtenRecords);
391 pendingRecords = [];
392 pendingBytes = 0;
393 });
394
395 for (const record of records) {
396 if (pendingBytes > 0 && pendingBytes + record.bytes > maxBytes) {
397 yield* flush;
398 }
399 pendingRecords.push(record);
400 pendingBytes += record.bytes;
401 if (pendingBytes >= maxBytes) {
402 yield* flush;
403 }
404 }
405 yield* flush;
406});
407
408async function isProviderLogFile(
409 filePath: string,
410 fileName: string,
411 filePrefix: string,
412): Promise<boolean> {
413 if (!/\.log(?:\.\d+)?$/u.test(fileName)) return false;
414 if (fileName.startsWith(filePrefix)) return true;
415
416 const descriptor = await NodeFS.promises.open(filePath, "r");
417 try {
418 const header = Buffer.alloc(256);
419 const { bytesRead } = await descriptor.read(header, 0, header.byteLength, 0);
420 return /^\[[^\]\r\n]+\] (?:NTIVE|CANON|ORCH): /u.test(header.toString("utf8", 0, bytesRead));
421 } finally {
422 await descriptor.close();
423 }
424}
425
426async function enforceRetention(input: {
427 readonly directory: string;
428 readonly maxTotalBytes: number;
429 readonly maxAgeMs: number;
430 readonly activeFilePaths: ReadonlySet<string>;
431 readonly filePrefix: string;
432 readonly now: number;
433}): Promise<RetentionResult> {
434 const failures: Array<FileOperationFailure> = [];
435 const files: Array<{ filePath: string; mtimeMs: number; size: number }> = [];
436
437 let entries: ReadonlyArray<NodeFS.Dirent>;
438 try {
439 entries = await NodeFS.promises.readdir(input.directory, { withFileTypes: true });
440 } catch (cause) {
441 return { failures: [{ filePath: input.directory, cause }] };
442 }
443
444 for (const entry of entries) {
445 if (!entry.isFile()) continue;
446 const filePath = NodePath.join(input.directory, entry.name);
447 try {
448 if (!(await isProviderLogFile(filePath, entry.name, input.filePrefix))) continue;
449 const stat = await NodeFS.promises.stat(filePath);
450 files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size });
451 } catch (cause) {
452 failures.push({ filePath, cause });
453 }
454 }
455
456 let totalBytes = files.reduce((total, file) => total + file.size, 0);
457 const remove = async (file: (typeof files)[number]) => {
458 if (input.activeFilePaths.has(file.filePath)) return false;
459 try {
460 await NodeFS.promises.rm(file.filePath, { force: true });
461 totalBytes -= file.size;
462 return true;
463 } catch (cause) {
464 failures.push({ filePath: file.filePath, cause });
465 return false;
466 }
467 };
468
469 const retained: typeof files = [];
470 for (const file of files) {
471 if (input.now - file.mtimeMs <= input.maxAgeMs || !(await remove(file))) retained.push(file);
472 }
473
474 for (const file of retained.toSorted(
475 (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath),
476 )) {
477 if (totalBytes <= input.maxTotalBytes) break;
478 await remove(file);
479 }
480
481 return { failures };
482}
483
484function validateOption(input: {
485 readonly filePath: string;
486 readonly option: string;
487 readonly value: number;
488 readonly minimum: number;
489}): EventNdjsonLogConfigurationError | undefined {
490 if (Number.isInteger(input.value) && input.value >= input.minimum) return undefined;
491 return new EventNdjsonLogConfigurationError(input);
492}
493
494function resolveOptions(
495 filePath: string,
496 options: EventNdjsonLogStoreOptions,
497): Effect.Effect<ResolvedOptions, EventNdjsonLogConfigurationError> {
498 const resolved = {
499 maxBytes: options.maxBytes ?? DEFAULT_MAX_BYTES,
500 maxFiles: options.maxFiles ?? DEFAULT_MAX_FILES,
501 batchWindowMs: options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS,
502 maxTotalBytes: options.maxTotalBytes ?? DEFAULT_MAX_TOTAL_BYTES,
503 maxAgeMs: options.maxAgeMs ?? DEFAULT_MAX_AGE_MS,
504 retentionCheckIntervalMs:
505 options.retentionCheckIntervalMs ?? DEFAULT_RETENTION_CHECK_INTERVAL_MS,
506 maxBufferedBytes: options.maxBufferedBytes ?? DEFAULT_MAX_BUFFERED_BYTES,
507 maxBufferedRecords: options.maxBufferedRecords ?? DEFAULT_MAX_BUFFERED_RECORDS,
508 attribution: options.attribution,
509 } satisfies ResolvedOptions;
510
511 const validations = [
512 ["maxBytes", resolved.maxBytes, 1],
513 ["maxFiles", resolved.maxFiles, 1],
514 ["batchWindowMs", resolved.batchWindowMs, 0],
515 ["maxTotalBytes", resolved.maxTotalBytes, 1],
516 ["maxAgeMs", resolved.maxAgeMs, 1],
517 ["retentionCheckIntervalMs", resolved.retentionCheckIntervalMs, 1],
518 ["maxBufferedBytes", resolved.maxBufferedBytes, 1],
519 ["maxBufferedRecords", resolved.maxBufferedRecords, 1],
520 ] as const;
521
522 for (const [option, value, minimum] of validations) {
523 const error = validateOption({ filePath, option, value, minimum });
524 if (error) return Effect.fail(error);
525 }
526 return Effect.succeed(resolved);
527}
528
529const drainPending = Effect.fnUntraced(function* (input: {
530 readonly directory: string;
531 readonly options: ResolvedOptions;
532 readonly state: StoreState;
533 readonly filePrefix: string;
534 readonly now: number;
535 readonly timerFired: boolean;
536 readonly close: boolean;
537 readonly makeSink: (
538 filePath: string,
539 ) => Effect.Effect<AsyncFileSink, Effect.Error<ReturnType<typeof makeAsyncRotatingFileSink>>>;
540}): Effect.fn.Return<readonly [DrainResult, StoreState]> {
541 if (input.state.closed) {
542 return [{ attributions: [], failures: [] }, input.state];
543 }
544
545 const sinks = new Map(input.state.sinks);
546 const failures: Array<FileOperationFailure> = [];
547 const attributionByStream = new Map<
548 EventNdjsonStream,
549 { count: number; logicalWriteBytes: number }
550 >();
551 const recordsBySegment = new Map<string, Array<PendingRecord>>();
552
553 for (const record of input.state.pending) {
554 const records = recordsBySegment.get(record.threadSegment) ?? [];
555 records.push(record);
556 recordsBySegment.set(record.threadSegment, records);
557 }
558
559 for (const [threadSegment, records] of recordsBySegment) {
560 const filePath = providerLogPath(input.directory, input.filePrefix, threadSegment);
561 let sink = sinks.get(threadSegment);
562 if (!sink) {
563 const created = yield* input.makeSink(filePath).pipe(Effect.result);
564 if (Result.isFailure(created)) {
565 failures.push({ filePath, cause: created.failure });
566 continue;
567 }
568 sink = created.success;
569 sinks.set(threadSegment, sink);
570 }
571
572 const written = yield* writeBatchedMessages(
573 sink,
574 records,
575 input.options.maxBytes,
576 (writtenRecords) => {
577 for (const record of writtenRecords) {
578 const current = attributionByStream.get(record.stream) ?? {
579 count: 0,
580 logicalWriteBytes: 0,
581 };
582 attributionByStream.set(record.stream, {
583 count: current.count + 1,
584 logicalWriteBytes: current.logicalWriteBytes + record.bytes,
585 });
586 }
587 },
588 ).pipe(Effect.result);
589 if (Result.isFailure(written)) {
590 sinks.delete(threadSegment);
591 failures.push({ filePath, cause: written.failure });
592 }
593 }
594
595 const retentionDue =
596 input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs;
597 const retention = retentionDue
598 ? yield* Effect.promise(() =>
599 enforceRetention({
600 directory: input.directory,
601 maxTotalBytes: input.options.maxTotalBytes,
602 maxAgeMs: input.options.maxAgeMs,
603 activeFilePaths: new Set(
604 Array.from(sinks.keys(), (threadSegment) =>
605 providerLogPath(input.directory, input.filePrefix, threadSegment),
606 ),
607 ),
608 filePrefix: input.filePrefix,
609 now: input.now,
610 }),
611 )
612 : { failures: [] };
613
614 return [
615 {
616 attributions: Array.from(attributionByStream, ([stream, value]) => ({
617 stream,
618 ...value,
619 })),
620 failures: [...failures, ...retention.failures],
621 },
622 {
623 pending: [],
624 pendingBytes: 0,
625 sinks,
626 flushScheduled: input.timerFired ? false : input.state.flushScheduled,
627 closed: input.close,
628 lastRetentionAt: retentionDue ? input.now : input.state.lastRetentionAt,
629 },
630 ];
631});
632
633const serializeEvent = Effect.fnUntraced(function* (event: unknown) {
634 return yield* encodeUnknownJsonString(event).pipe(
635 Effect.catch((error) =>
636 logWarning("failed to serialize provider event log record", {
637 errorTag: errorTag(error),
638 }).pipe(Effect.as(undefined)),
639 ),
640 );
641});
642
643export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
644 filePath: string,
645 options: EventNdjsonLogStoreOptions = {},
646): Effect.fn.Return<
647 EventNdjsonLogStore,
648 EventNdjsonLogStoreError,
649 FileSystem.FileSystem | Path.Path
650> {
651 const fs = yield* FileSystem.FileSystem;
652 const path = yield* Path.Path;
653 const resolved = yield* resolveOptions(filePath, options);
654 const directory = NodePath.dirname(filePath);
655 const filePrefix = providerLogPrefix(filePath);
656
657 yield* fs
658 .makeDirectory(directory, { recursive: true })
659 .pipe(Effect.mapError((cause) => new EventNdjsonLogDirectoryError({ directory, cause })));
660 const makeSink = (filePath: string) =>
661 makeAsyncRotatingFileSink({
662 filePath,
663 maxBytes: resolved.maxBytes,
664 maxFiles: resolved.maxFiles,
665 }).pipe(
666 Effect.provideService(FileSystem.FileSystem, fs),
667 Effect.provideService(Path.Path, path),
668 );
669
670 const initializedAt = yield* Clock.currentTimeMillis;
671 const initialRetention = yield* Effect.promise(() =>
672 enforceRetention({
673 directory,
674 maxTotalBytes: resolved.maxTotalBytes,
675 maxAgeMs: resolved.maxAgeMs,
676 activeFilePaths: new Set(),
677 filePrefix,
678 now: initializedAt,
679 }),
680 );
681 for (const failure of initialRetention.failures) {
682 yield* logWarning("provider event log retention failed", {
683 filePath: failure.filePath,
684 errorTag: errorTag(failure.cause),
685 });
686 }
687
688 const stateRef = yield* SynchronizedRef.make<StoreState>({
689 pending: [],
690 pendingBytes: 0,
691 sinks: new Map(),
692 flushScheduled: false,
693 closed: false,
694 lastRetentionAt: initializedAt,
695 });
696 const timerScope = yield* Scope.make();
697
698 const flush = Effect.fnUntraced(function* (timerFired: boolean, close: boolean) {
699 const startedAt = yield* Clock.currentTimeMillis;
700 const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) =>
701 drainPending({
702 directory,
703 options: resolved,
704 state,
705 filePrefix,
706 now: startedAt,
707 timerFired,
708 close,
709 makeSink,
710 }),
711 );
712
713 for (const failure of result.failures) {
714 yield* logWarning("provider event log write or retention failed", {
715 filePath: failure.filePath,
716 errorTag: errorTag(failure.cause),
717 });
718 }
719
720 if (resolved.attribution && result.attributions.length > 0) {
721 const completedAt = yield* Clock.currentTimeMillis;
722 const durationMs = Math.max(0, completedAt - startedAt);
723 const totalBytes = result.attributions.reduce(
724 (total, entry) => total + entry.logicalWriteBytes,
725 0,
726 );
727 yield* Effect.forEach(
728 result.attributions,
729 (entry) =>
730 resolved.attribution?.record({
731 component: "provider-event-log",
732 operation: `${entry.stream}.append`,
733 logicalWriteBytes: entry.logicalWriteBytes,
734 count: entry.count,
735 durationMs:
736 totalBytes === 0
737 ? 0
738 : Math.round(durationMs * (entry.logicalWriteBytes / totalBytes)),
739 }) ?? Effect.void,
740 { discard: true },
741 );
742 }
743 }, Effect.uninterruptible);
744
745 const scheduleFlush = Effect.fnUntraced(function* () {
746 yield* Effect.forkIn(
747 Effect.sleep(resolved.batchWindowMs).pipe(Effect.andThen(flush(true, false))),
748 timerScope,
749 { startImmediately: true },
750 );
751 });
752
753 const close = Effect.fnUntraced(function* () {
754 yield* flush(false, true);
755 yield* Scope.close(timerScope, Exit.void);
756 });
757
758 const loggerViews = new Map<EventNdjsonStream, ProviderEventLoggers.EventNdjsonLogger>();
759 const logger = (stream: EventNdjsonStream): ProviderEventLoggers.EventNdjsonLogger => {
760 const existing = loggerViews.get(stream);
761 if (existing) return existing;
762
763 const write = Effect.fnUntraced(function* (event: unknown, threadId: ThreadId | null) {
764 if (!shouldPersistProviderEvent(stream, event)) return;
765 let payload = yield* serializeEvent(boundProviderEventForLogging(event));
766 if (payload === undefined) return;
767 // Escaping can expand strings beyond their input size. Keep that bounded
768 // serialization out of the file too, while retaining routing/error fields.
769 if (Buffer.byteLength(payload) > MAX_RECORD_CHARACTERS) {
770 payload = yield* serializeEvent(summarizeProviderEvent(event));
771 if (payload === undefined) return;
772 }
773
774 const observedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso));
775 const line = `[${observedAt}] ${resolveStreamLabel(stream)}: ${payload}\n`;
776 const bytes = Buffer.byteLength(line);
777 const action = yield* SynchronizedRef.modifyEffect(stateRef, (state) => {
778 if (state.closed) {
779 return Effect.succeed([{ flush: false }, state] as const);
780 }
781 const pending = state.pending;
782 pending.push({ stream, threadSegment: resolveThreadSegment(threadId), line, bytes });
783 const pendingBytes = state.pendingBytes + bytes;
784 const flush =
785 resolved.batchWindowMs === 0 ||
786 pending.length >= resolved.maxBufferedRecords ||
787 pendingBytes >= resolved.maxBufferedBytes;
788 const schedule = !flush && !state.flushScheduled;
789 const nextState = {
790 ...state,
791 pending,
792 pendingBytes,
793 flushScheduled: state.flushScheduled || schedule,
794 };
795 return (schedule ? scheduleFlush() : Effect.void).pipe(
796 Effect.as([{ flush }, nextState] as const),
797 );
798 }).pipe(Effect.uninterruptible);
799
800 if (action.flush) {
801 yield* flush(false, false);
802 }
803 });
804
805 const view = {
806 filePath,
807 write,
808 close: () => Effect.void,
809 } satisfies ProviderEventLoggers.EventNdjsonLogger;
810 loggerViews.set(stream, view);
811 return view;
812 };
813
814 return { filePath, logger, close } satisfies EventNdjsonLogStore;
815});
816
817export const makeEventNdjsonLogger = Effect.fnUntraced(function* (
818 filePath: string,
819 options: EventNdjsonLoggerOptions,
820): Effect.fn.Return<
821 ProviderEventLoggers.EventNdjsonLogger | undefined,
822 never,
823 FileSystem.FileSystem | Path.Path
824> {
825 const store = yield* makeEventNdjsonLogStore(filePath, options).pipe(
826 Effect.catch((error) =>
827 logWarning(error.message, { error }).pipe(
828 Effect.as<EventNdjsonLogStore | undefined>(undefined),
829 ),
830 ),
831 );
832 if (!store) return undefined;
833 return { ...store.logger(options.stream), close: store.close };
834});