Open History to see the last commit
1// @effect-diagnostics nodeBuiltinImport:off
2/** Native and canonical views share write state to avoid races on the same thread file. */
3import * as NodeCrypto from "node:crypto";
4import * as NodeFS from "node:fs";
5import * as NodePath from "node:path";
6
7import type { ThreadId } from "@t3tools/contracts";
8import type * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers";
9import { makeAsyncRotatingFileSink } from "@t3tools/shared/logging";
10import { errorTag } from "@t3tools/shared/observability";
11import * as Clock from "effect/Clock";
12import * as DateTime from "effect/DateTime";
13import * as Effect from "effect/Effect";
14import * as FileSystem from "effect/FileSystem";
15import * as Path from "effect/Path";
16import * as Result from "effect/Result";
17import * as Exit from "effect/Exit";
18import * as Schema from "effect/Schema";
19import * as Scope from "effect/Scope";
20import * as SynchronizedRef from "effect/SynchronizedRef";
21
22import {
23 ATTACHMENT_ID_THREAD_SEGMENT_MAX_CHARS,
24 toSafeThreadAttachmentSegment,
25} from "../attachmentStore.ts";
26import type * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts";
27
28const MEBIBYTE = 1024 * 1024;
29const DAY_MS = 24 * 60 * 60 * 1_000;
30const DEFAULT_MAX_BYTES = 10 * MEBIBYTE;
31const DEFAULT_MAX_FILES = 10;
32const DEFAULT_BATCH_WINDOW_MS = 1_000;
33const DEFAULT_MAX_TOTAL_BYTES = 512 * MEBIBYTE;
34const DEFAULT_MAX_AGE_MS = 14 * DAY_MS;
35const DEFAULT_RETENTION_CHECK_INTERVAL_MS = 5 * 60 * 1_000;
36const DEFAULT_MAX_BUFFERED_BYTES = MEBIBYTE;
37const DEFAULT_MAX_BUFFERED_RECORDS = 512;
38const MAX_RECORD_CHARACTERS = 64 * 1024;
39const MAX_RECORD_FIELDS = 1_024;
40const MAX_RECORD_DEPTH = 16;
41const GLOBAL_THREAD_SEGMENT = "_global";
42const LOG_SCOPE = "provider-observability";
43const encodeUnknownJsonString = Schema.encodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
44
45const transientCanonicalEventTypes = new Set([
46 "content.delta",
47 "hook.progress",
48 "item.updated",
49 "task.progress",
50 "thread.realtime.audio.delta",
51 "tool.progress",
52 "turn.proposed.delta",
53]);
54const transientNativeMethods = new Set([
55 "item/agentMessage/delta",
56 "item/commandExecution/outputDelta",
57 "item/fileChange/outputDelta",
58 "item/plan/delta",
59 "item/reasoning/summaryTextDelta",
60 "item/reasoning/textDelta",
61 "thread/realtime/outputAudio/delta",
62 "thread/realtime/transcript/delta",
63 "turn/diff/updated",
64]);
65const transientAcpUpdates = new Set(["agent_message_chunk", "agent_thought_chunk"]);
66
67export type EventNdjsonStream = "native" | "canonical" | "orchestration";
68
69export interface EventNdjsonLogStore {
70 readonly filePath: string;
71 readonly logger: (stream: EventNdjsonStream) => ProviderEventLoggers.EventNdjsonLogger;
72 readonly close: () => Effect.Effect<void>;
73}
74
75export interface EventNdjsonLogStoreOptions {
76 readonly maxBytes?: number;
77 readonly maxFiles?: number;
78 readonly batchWindowMs?: number;
79 readonly attribution?: ResourceAttribution.ResourceAttribution["Service"];
80}
81
82export class EventNdjsonLogDirectoryError extends Schema.TaggedError<EventNdjsonLogDirectoryError>()(
83 "EventNdjsonLogDirectoryError",
84 {
85 directory: Schema.String,
86 cause: Schema.Defect(),
87 },
88) {
89 override get message(): string {
90 return `Failed to create provider event log directory '${this.directory}'`;
91 }
92}
93
94export type EventNdjsonLogStoreError = EventNdjsonLogDirectoryError;
95
96interface ResolvedOptions {
97 readonly maxBytes: number;
98 readonly maxFiles: number;
99 readonly batchWindowMs: number;
100 readonly maxTotalBytes: number;
101 readonly maxAgeMs: number;
102 readonly retentionCheckIntervalMs: number;
103 readonly maxBufferedBytes: number;
104 readonly maxBufferedRecords: number;
105 readonly attribution: ResourceAttribution.ResourceAttribution["Service"] | undefined;
106}
107
108export interface PendingRecord {
109 readonly stream: EventNdjsonStream;
110 readonly threadSegment: string;
111 readonly line: string;
112 readonly bytes: number;
113}
114
115type AsyncFileSink = Effect.Success<ReturnType<typeof makeAsyncRotatingFileSink>>;
116
117interface StoreState {
118 readonly pending: Array<PendingRecord>;
119 readonly pendingBytes: number;
120 readonly sinks: ReadonlyMap<string, AsyncFileSink>;
121 readonly writtenSinceRetention: ReadonlySet<string>;
122 readonly flushScheduled: boolean;
123 readonly closed: boolean;
124 readonly lastRetentionAt: number;
125}
126
127interface AttributionSummary {
128 readonly stream: EventNdjsonStream;
129 readonly count: number;
130 readonly logicalWriteBytes: number;
131}
132
133interface FileOperationFailure {
134 readonly filePath: string;
135 readonly cause: unknown;
136}
137
138interface RetentionResult {
139 readonly failures: ReadonlyArray<FileOperationFailure>;
140}
141
142interface DrainResult {
143 readonly attributions: ReadonlyArray<AttributionSummary>;
144 readonly failures: ReadonlyArray<FileOperationFailure>;
145}
146
147function logWarning(message: string, context: Record<string, unknown>): Effect.Effect<void> {
148 return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE }));
149}
150
151function resolveThreadSegment(raw: string | null | undefined): string {
152 if (typeof raw !== "string") return GLOBAL_THREAD_SEGMENT;
153 const normalized = toSafeThreadAttachmentSegment(raw);
154 if (normalized === null) return GLOBAL_THREAD_SEGMENT;
155 // Delegated thread ids share a prefix longer than the segment, so ids that may be cut get a hash.
156 if (raw.length <= ATTACHMENT_ID_THREAD_SEGMENT_MAX_CHARS) return normalized;
157 const hash = NodeCrypto.createHash("sha256").update(raw).digest("hex").slice(0, 16);
158 return `${normalized}-${hash}`;
159}
160
161function resolveStreamLabel(stream: EventNdjsonStream): string {
162 return stream === "native" ? "NTIVE" : stream === "orchestration" ? "ORCH" : "CANON";
163}
164
165function providerLogPrefix(filePath: string): string {
166 const basename = NodePath.basename(filePath);
167 const extension = NodePath.extname(basename);
168 return `${extension.length > 0 ? basename.slice(0, -extension.length) : basename}.`;
169}
170
171function providerLogPath(directory: string, prefix: string, threadSegment: string): string {
172 return NodePath.join(directory, `${prefix}${threadSegment}.log`);
173}
174
175export function shouldPersistProviderEvent(stream: EventNdjsonStream, event: unknown): boolean {
176 if (stream === "orchestration" || typeof event !== "object" || event === null) {
177 return true;
178 }
179 try {
180 const type = Reflect.get(event, "type");
181 if (typeof type === "string" && transientCanonicalEventTypes.has(type)) {
182 return false;
183 }
184 if (stream !== "native") return true;
185
186 const nested = Reflect.get(event, "event");
187 const envelope = typeof nested === "object" && nested !== null ? nested : event;
188 // Decoded frames preserve raw-frame information without duplicating every token delta.
189 if (Reflect.get(envelope, "stage") === "raw") return false;
190 const decodedPayload = Reflect.get(envelope, "payload");
191 const nativeEvent =
192 Reflect.get(envelope, "stage") === "decoded" &&
193 typeof decodedPayload === "object" &&
194 decodedPayload !== null
195 ? decodedPayload
196 : envelope;
197 const method = Reflect.get(nativeEvent, "method");
198 if (
199 typeof method === "string" &&
200 (transientNativeMethods.has(method) ||
201 method.startsWith("claude/stream_event/content_block_delta/"))
202 ) {
203 return false;
204 }
205
206 const nativeType = Reflect.get(nativeEvent, "type");
207 if (nativeType === "system" && Reflect.get(nativeEvent, "subtype") === "thinking_tokens") {
208 return false;
209 }
210 if (nativeType === "message.part.delta") return false;
211 if (nativeType === "stream_event") {
212 const streamEvent = Reflect.get(nativeEvent, "event");
213 if (
214 typeof streamEvent === "object" &&
215 streamEvent !== null &&
216 Reflect.get(streamEvent, "type") === "content_block_delta"
217 ) {
218 return false;
219 }
220 }
221
222 const payload = Reflect.get(nativeEvent, "payload");
223 if (typeof payload !== "object" || payload === null) return true;
224
225 if (method === "session/update") {
226 const update = Reflect.get(payload, "update");
227 if (typeof update !== "object" || update === null) return true;
228 const updateType = Reflect.get(update, "sessionUpdate");
229 return typeof updateType !== "string" || !transientAcpUpdates.has(updateType);
230 }
231
232 if (nativeType === "message.part.updated") {
233 const properties = Reflect.get(payload, "properties");
234 if (typeof properties !== "object" || properties === null) return true;
235 const part = Reflect.get(properties, "part");
236 if (typeof part !== "object" || part === null) return true;
237 const partType = Reflect.get(part, "type");
238 if (partType === "text" || partType === "reasoning") return false;
239 if (partType === "tool") {
240 // Running snapshots repeat growing output.
241 const state = Reflect.get(part, "state");
242 if (typeof state === "object" && state !== null) {
243 return Reflect.get(state, "status") !== "running";
244 }
245 }
246 return true;
247 }
248
249 return true;
250 } catch {
251 return true;
252 }
253}
254
255const summaryFields = [
256 "provider",
257 "protocol",
258 "kind",
259 "providerSessionId",
260 "direction",
261 "stage",
262 "type",
263 "subtype",
264 "method",
265 "id",
266 "threadId",
267 "turnId",
268 "requestId",
269 "session_id",
270 "status",
271 "is_error",
272 "api_error_status",
273 "terminal_reason",
274 "stop_reason",
275 "operation",
276 "code",
277 "willRetry",
278 "message",
279 "event",
280 "payload",
281 "params",
282 "result",
283 "thread",
284 "turn",
285 "error",
286 "turns",
287 "items",
288 "content",
289] as const;
290
291function summarizeProviderEvent(event: unknown): unknown {
292 let remainingFields = 128;
293 let remainingCharacters = 8 * 1024;
294 const summarize = (value: unknown, depth: number): unknown => {
295 if (typeof value === "string") {
296 if (value.length > Math.min(1_024, remainingCharacters)) {
297 return { omittedCharacters: value.length };
298 }
299 remainingCharacters -= value.length;
300 return value;
301 }
302 if (value === null || typeof value === "number" || typeof value === "boolean") return value;
303 if (typeof value !== "object") return undefined;
304 if (Array.isArray(value)) return { itemCount: value.length };
305 const summary: Record<string, unknown> = { truncated: true };
306 if (depth >= 6) return summary;
307 for (const key of summaryFields) {
308 if (remainingFields <= 0) break;
309 const nested = Reflect.get(value, key);
310 if (nested === undefined) continue;
311 remainingFields -= 1;
312 summary[key] = summarize(nested, depth + 1);
313 }
314 return summary;
315 };
316 try {
317 return summarize(event, 0);
318 } catch {
319 return { truncated: true };
320 }
321}
322
323export function boundProviderEventForLogging(event: unknown): unknown {
324 let remainingCharacters = MAX_RECORD_CHARACTERS;
325 let remainingFields = MAX_RECORD_FIELDS;
326 const ancestors = new WeakSet<object>();
327 const fits = (value: unknown, depth: number): boolean => {
328 if (typeof value === "string") {
329 remainingCharacters -= value.length;
330 return remainingCharacters >= 0;
331 }
332 if (typeof value !== "object" || value === null) return true;
333 if (depth > MAX_RECORD_DEPTH || ancestors.has(value)) return false;
334 if (Array.isArray(value) && value.length > remainingFields) return false;
335 ancestors.add(value);
336 for (const key in value) {
337 if (!Object.hasOwn(value, key)) continue;
338 remainingFields -= 1;
339 remainingCharacters -= key.length;
340 if (
341 remainingFields < 0 ||
342 remainingCharacters < 0 ||
343 !fits(Reflect.get(value, key), depth + 1)
344 )
345 return false;
346 }
347 ancestors.delete(value);
348 return true;
349 };
350 try {
351 if (fits(event, 0)) return event;
352 } catch {}
353 return summarizeProviderEvent(event);
354}
355
356export const writeBatchedMessages = Effect.fnUntraced(function* (
357 sink: AsyncFileSink,
358 records: ReadonlyArray<PendingRecord>,
359 maxBytes: number,
360 onWritten: (records: ReadonlyArray<PendingRecord>) => void,
361) {
362 let pendingRecords: Array<PendingRecord> = [];
363 let pendingBytes = 0;
364
365 const flush = Effect.gen(function* () {
366 if (pendingRecords.length === 0) return;
367 const writtenRecords = pendingRecords;
368 yield* sink.write(writtenRecords.map((record) => record.line).join(""));
369 onWritten(writtenRecords);
370 pendingRecords = [];
371 pendingBytes = 0;
372 });
373
374 for (const record of records) {
375 if (pendingBytes > 0 && pendingBytes + record.bytes > maxBytes) {
376 yield* flush;
377 }
378 pendingRecords.push(record);
379 pendingBytes += record.bytes;
380 if (pendingBytes >= maxBytes) {
381 yield* flush;
382 }
383 }
384 yield* flush;
385});
386
387async function isProviderLogFile(
388 filePath: string,
389 fileName: string,
390 filePrefix: string,
391): Promise<boolean> {
392 if (!/\.log(?:\.\d+)?$/u.test(fileName)) return false;
393 if (fileName.startsWith(filePrefix)) return true;
394
395 const descriptor = await NodeFS.promises.open(filePath, "r");
396 try {
397 const header = Buffer.alloc(256);
398 const { bytesRead } = await descriptor.read(header, 0, header.byteLength, 0);
399 return /^\[[^\]\r\n]+\] (?:NTIVE|CANON|ORCH): /u.test(header.toString("utf8", 0, bytesRead));
400 } finally {
401 await descriptor.close();
402 }
403}
404
405async function enforceRetention(input: {
406 readonly directory: string;
407 readonly maxTotalBytes: number;
408 readonly maxAgeMs: number;
409 readonly activeFilePaths: ReadonlySet<string>;
410 readonly filePrefix: string;
411 readonly now: number;
412}): Promise<RetentionResult> {
413 const failures: Array<FileOperationFailure> = [];
414 const files: Array<{ filePath: string; mtimeMs: number; size: number }> = [];
415
416 let entries: ReadonlyArray<NodeFS.Dirent>;
417 try {
418 entries = await NodeFS.promises.readdir(input.directory, { withFileTypes: true });
419 } catch (cause) {
420 return { failures: [{ filePath: input.directory, cause }] };
421 }
422
423 for (const entry of entries) {
424 if (!entry.isFile()) continue;
425 const filePath = NodePath.join(input.directory, entry.name);
426 try {
427 if (!(await isProviderLogFile(filePath, entry.name, input.filePrefix))) continue;
428 const stat = await NodeFS.promises.stat(filePath);
429 files.push({ filePath, mtimeMs: stat.mtimeMs, size: stat.size });
430 } catch (cause) {
431 failures.push({ filePath, cause });
432 }
433 }
434
435 let totalBytes = files.reduce((total, file) => total + file.size, 0);
436 const remove = async (file: (typeof files)[number]) => {
437 if (input.activeFilePaths.has(file.filePath)) return false;
438 try {
439 await NodeFS.promises.rm(file.filePath, { force: true });
440 totalBytes -= file.size;
441 return true;
442 } catch (cause) {
443 failures.push({ filePath: file.filePath, cause });
444 return false;
445 }
446 };
447
448 const retained: typeof files = [];
449 for (const file of files) {
450 if (input.now - file.mtimeMs <= input.maxAgeMs || !(await remove(file))) retained.push(file);
451 }
452
453 for (const file of retained.toSorted(
454 (left, right) => left.mtimeMs - right.mtimeMs || left.filePath.localeCompare(right.filePath),
455 )) {
456 if (totalBytes <= input.maxTotalBytes) break;
457 await remove(file);
458 }
459
460 return { failures };
461}
462
463function resolveOptions(options: EventNdjsonLogStoreOptions): Effect.Effect<ResolvedOptions> {
464 const resolved = {
465 maxBytes: options.maxBytes ?? DEFAULT_MAX_BYTES,
466 maxFiles: options.maxFiles ?? DEFAULT_MAX_FILES,
467 batchWindowMs: options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS,
468 maxTotalBytes: DEFAULT_MAX_TOTAL_BYTES,
469 maxAgeMs: DEFAULT_MAX_AGE_MS,
470 retentionCheckIntervalMs: DEFAULT_RETENTION_CHECK_INTERVAL_MS,
471 maxBufferedBytes: DEFAULT_MAX_BUFFERED_BYTES,
472 maxBufferedRecords: DEFAULT_MAX_BUFFERED_RECORDS,
473 attribution: options.attribution,
474 } satisfies ResolvedOptions;
475
476 return Effect.succeed(resolved);
477}
478
479const drainPending = Effect.fnUntraced(function* (input: {
480 readonly directory: string;
481 readonly options: ResolvedOptions;
482 readonly state: StoreState;
483 readonly filePrefix: string;
484 readonly now: number;
485 readonly timerFired: boolean;
486 readonly close: boolean;
487 readonly makeSink: (
488 filePath: string,
489 ) => Effect.Effect<AsyncFileSink, Effect.Error<ReturnType<typeof makeAsyncRotatingFileSink>>>;
490}): Effect.fn.Return<readonly [DrainResult, StoreState]> {
491 if (input.state.closed) {
492 return [{ attributions: [], failures: [] }, input.state];
493 }
494
495 const sinks = new Map(input.state.sinks);
496 const writtenSegments = new Set(input.state.writtenSinceRetention);
497 const failures: Array<FileOperationFailure> = [];
498 const attributionByStream = new Map<
499 EventNdjsonStream,
500 { count: number; logicalWriteBytes: number }
501 >();
502 const recordsBySegment = new Map<string, Array<PendingRecord>>();
503
504 for (const record of input.state.pending) {
505 const records = recordsBySegment.get(record.threadSegment) ?? [];
506 records.push(record);
507 recordsBySegment.set(record.threadSegment, records);
508 }
509
510 for (const [threadSegment, records] of recordsBySegment) {
511 const filePath = providerLogPath(input.directory, input.filePrefix, threadSegment);
512 let sink = sinks.get(threadSegment);
513 if (!sink) {
514 const created = yield* input.makeSink(filePath).pipe(Effect.result);
515 if (Result.isFailure(created)) {
516 failures.push({ filePath, cause: created.failure });
517 continue;
518 }
519 sink = created.success;
520 sinks.set(threadSegment, sink);
521 }
522 writtenSegments.add(threadSegment);
523
524 const written = yield* writeBatchedMessages(
525 sink,
526 records,
527 input.options.maxBytes,
528 (writtenRecords) => {
529 for (const record of writtenRecords) {
530 const current = attributionByStream.get(record.stream) ?? {
531 count: 0,
532 logicalWriteBytes: 0,
533 };
534 attributionByStream.set(record.stream, {
535 count: current.count + 1,
536 logicalWriteBytes: current.logicalWriteBytes + record.bytes,
537 });
538 }
539 },
540 ).pipe(Effect.result);
541 if (Result.isFailure(written)) {
542 sinks.delete(threadSegment);
543 failures.push({ filePath, cause: written.failure });
544 }
545 }
546
547 const retentionDue =
548 input.now - input.state.lastRetentionAt >= input.options.retentionCheckIntervalMs;
549 if (retentionDue) {
550 // A sink idle for a whole retention interval reopens from disk on its next
551 // write, so retention may remove its files without leaving a stale size.
552 for (const threadSegment of sinks.keys()) {
553 if (!writtenSegments.has(threadSegment)) sinks.delete(threadSegment);
554 }
555 }
556 const retention = retentionDue
557 ? yield* Effect.promise(() =>
558 enforceRetention({
559 directory: input.directory,
560 maxTotalBytes: input.options.maxTotalBytes,
561 maxAgeMs: input.options.maxAgeMs,
562 activeFilePaths: new Set(
563 Array.from(sinks.keys(), (threadSegment) =>
564 providerLogPath(input.directory, input.filePrefix, threadSegment),
565 ),
566 ),
567 filePrefix: input.filePrefix,
568 now: input.now,
569 }),
570 )
571 : { failures: [] };
572
573 return [
574 {
575 attributions: Array.from(attributionByStream, ([stream, value]) => ({
576 stream,
577 ...value,
578 })),
579 failures: [...failures, ...retention.failures],
580 },
581 {
582 pending: [],
583 pendingBytes: 0,
584 sinks,
585 writtenSinceRetention: retentionDue ? new Set() : writtenSegments,
586 flushScheduled: input.timerFired ? false : input.state.flushScheduled,
587 closed: input.close,
588 lastRetentionAt: retentionDue ? input.now : input.state.lastRetentionAt,
589 },
590 ];
591});
592
593const serializeEvent = Effect.fnUntraced(function* (event: unknown) {
594 return yield* encodeUnknownJsonString(event).pipe(
595 Effect.catch((error) =>
596 logWarning("failed to serialize provider event log record", {
597 errorTag: errorTag(error),
598 }).pipe(Effect.as(undefined)),
599 ),
600 );
601});
602
603export const makeEventNdjsonLogStore = Effect.fnUntraced(function* (
604 filePath: string,
605 options: EventNdjsonLogStoreOptions = {},
606): Effect.fn.Return<
607 EventNdjsonLogStore,
608 EventNdjsonLogStoreError,
609 FileSystem.FileSystem | Path.Path
610> {
611 const fs = yield* FileSystem.FileSystem;
612 const path = yield* Path.Path;
613 const resolved = yield* resolveOptions(options);
614 const directory = NodePath.dirname(filePath);
615 const filePrefix = providerLogPrefix(filePath);
616
617 yield* fs
618 .makeDirectory(directory, { recursive: true })
619 .pipe(Effect.mapError((cause) => new EventNdjsonLogDirectoryError({ directory, cause })));
620 const makeSink = (filePath: string) =>
621 makeAsyncRotatingFileSink({
622 filePath,
623 maxBytes: resolved.maxBytes,
624 maxFiles: resolved.maxFiles,
625 }).pipe(
626 Effect.provideService(FileSystem.FileSystem, fs),
627 Effect.provideService(Path.Path, path),
628 );
629
630 const initializedAt = yield* Clock.currentTimeMillis;
631 const initialRetention = yield* Effect.promise(() =>
632 enforceRetention({
633 directory,
634 maxTotalBytes: resolved.maxTotalBytes,
635 maxAgeMs: resolved.maxAgeMs,
636 activeFilePaths: new Set(),
637 filePrefix,
638 now: initializedAt,
639 }),
640 );
641 for (const failure of initialRetention.failures) {
642 yield* logWarning("provider event log retention failed", {
643 filePath: failure.filePath,
644 errorTag: errorTag(failure.cause),
645 });
646 }
647
648 const stateRef = yield* SynchronizedRef.make<StoreState>({
649 pending: [],
650 pendingBytes: 0,
651 sinks: new Map(),
652 writtenSinceRetention: new Set(),
653 flushScheduled: false,
654 closed: false,
655 lastRetentionAt: initializedAt,
656 });
657 const timerScope = yield* Scope.make();
658
659 const flush = Effect.fnUntraced(function* (timerFired: boolean, close: boolean) {
660 const startedAt = yield* Clock.currentTimeMillis;
661 const result = yield* SynchronizedRef.modifyEffect(stateRef, (state) =>
662 drainPending({
663 directory,
664 options: resolved,
665 state,
666 filePrefix,
667 now: startedAt,
668 timerFired,
669 close,
670 makeSink,
671 }),
672 );
673
674 for (const failure of result.failures) {
675 yield* logWarning("provider event log write or retention failed", {
676 filePath: failure.filePath,
677 errorTag: errorTag(failure.cause),
678 });
679 }
680
681 if (resolved.attribution && result.attributions.length > 0) {
682 const completedAt = yield* Clock.currentTimeMillis;
683 const durationMs = Math.max(0, completedAt - startedAt);
684 const totalBytes = result.attributions.reduce(
685 (total, entry) => total + entry.logicalWriteBytes,
686 0,
687 );
688 yield* Effect.forEach(
689 result.attributions,
690 (entry) =>
691 resolved.attribution?.record({
692 component: "provider-event-log",
693 operation: `${entry.stream}.append`,
694 logicalWriteBytes: entry.logicalWriteBytes,
695 count: entry.count,
696 durationMs:
697 totalBytes === 0
698 ? 0
699 : Math.round(durationMs * (entry.logicalWriteBytes / totalBytes)),
700 }) ?? Effect.void,
701 { discard: true },
702 );
703 }
704 }, Effect.uninterruptible);
705
706 const scheduleFlush = Effect.fnUntraced(function* () {
707 yield* Effect.forkIn(
708 Effect.sleep(resolved.batchWindowMs).pipe(Effect.andThen(flush(true, false))),
709 timerScope,
710 { startImmediately: true },
711 );
712 });
713
714 const close = Effect.fnUntraced(function* () {
715 yield* flush(false, true);
716 yield* Scope.close(timerScope, Exit.void);
717 });
718
719 const loggerViews = new Map<EventNdjsonStream, ProviderEventLoggers.EventNdjsonLogger>();
720 const logger = (stream: EventNdjsonStream): ProviderEventLoggers.EventNdjsonLogger => {
721 const existing = loggerViews.get(stream);
722 if (existing) return existing;
723
724 const write = Effect.fnUntraced(function* (event: unknown, threadId: ThreadId | null) {
725 if (!shouldPersistProviderEvent(stream, event)) return;
726 let payload = yield* serializeEvent(boundProviderEventForLogging(event));
727 if (payload === undefined) return;
728 // Escaping can expand strings beyond their input size.
729 if (Buffer.byteLength(payload) > MAX_RECORD_CHARACTERS) {
730 payload = yield* serializeEvent(summarizeProviderEvent(event));
731 if (payload === undefined) return;
732 }
733
734 const observedAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso));
735 const line = `[${observedAt}] ${resolveStreamLabel(stream)}: ${payload}\n`;
736 const bytes = Buffer.byteLength(line);
737 const threadSegment = resolveThreadSegment(threadId);
738 const action = yield* SynchronizedRef.modifyEffect(stateRef, (state) => {
739 if (state.closed) {
740 return Effect.succeed([{ flush: false }, state] as const);
741 }
742 const pending = state.pending;
743 pending.push({ stream, threadSegment, line, bytes });
744 const pendingBytes = state.pendingBytes + bytes;
745 const flush =
746 resolved.batchWindowMs === 0 ||
747 pending.length >= resolved.maxBufferedRecords ||
748 pendingBytes >= resolved.maxBufferedBytes;
749 const schedule = !flush && !state.flushScheduled;
750 const nextState = {
751 ...state,
752 pending,
753 pendingBytes,
754 flushScheduled: state.flushScheduled || schedule,
755 };
756 return (schedule ? scheduleFlush() : Effect.void).pipe(
757 Effect.as([{ flush }, nextState] as const),
758 );
759 }).pipe(Effect.uninterruptible);
760
761 if (action.flush) {
762 yield* flush(false, false);
763 }
764 });
765
766 const view = {
767 filePath,
768 write,
769 close: () => Effect.void,
770 } satisfies ProviderEventLoggers.EventNdjsonLogger;
771 loggerViews.set(stream, view);
772 return view;
773 };
774
775 return { filePath, logger, close } satisfies EventNdjsonLogStore;
776});