1// @effect-diagnostics nodeBuiltinImport:off
2import * as NodeFS from "node:fs";
3import * as NodeOS from "node:os";
4import * as NodePath from "node:path";
5
6import { ThreadId } from "@t3tools/contracts";
7import * as NodeServices from "@effect/platform-node/NodeServices";
8import { assert, it } from "@effect/vitest";
9import { RotatingFileSinkError } from "@t3tools/shared/logging";
10import * as Clock from "effect/Clock";
11import * as Effect from "effect/Effect";
12import * as Deferred from "effect/Deferred";
13import * as FileSystem from "effect/FileSystem";
14import * as Fiber from "effect/Fiber";
15import * as Logger from "effect/Logger";
16import * as Schema from "effect/Schema";
17import * as TestClock from "effect/testing/TestClock";
18
19import * as ResourceAttribution from "../resourceTelemetry/ResourceAttribution.ts";
20import {
21 makeEventNdjsonLogger,
22 makeEventNdjsonLogStore,
23 type PendingRecord,
24 writeBatchedMessages,
25} from "./EventNdjsonLogger.ts";
26
27const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
28const decodeUnknownJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
29
30function ownedLogPath(basePath: string, segment: string): string {
31 const basename = NodePath.basename(basePath);
32 const extension = NodePath.extname(basename);
33 const stem = extension.length > 0 ? basename.slice(0, -extension.length) : basename;
34 return NodePath.join(NodePath.dirname(basePath), `${stem}.${segment}.log`);
35}
36
37function parseLogLine(line: string) {
38 const match = /^\[([^\]]+)\] ([A-Z]+): (.+)$/.exec(line);
39 assert.notEqual(match, null);
40 if (!match) {
41 throw new Error(`invalid log line: ${line}`);
42 }
43 const observedAt = match[1];
44 const stream = match[2];
45 const payload = match[3];
46 if (!observedAt || !stream || payload === undefined) {
47 throw new Error(`invalid log line: ${line}`);
48 }
49 return {
50 observedAt,
51 stream,
52 payload,
53 };
54}
55
56it.layer(NodeServices.layer)("EventNdjsonLogger", (it) => {
57 it.effect("summarizes circular events without exposing their contents in diagnostics", () => {
58 const messages: Array<unknown> = [];
59 const logCapture = Logger.make<unknown, void>(({ message }) => {
60 if (Array.isArray(message)) {
61 messages.push(...message);
62 } else {
63 messages.push(message);
64 }
65 });
66 const secret = "secret-circular-event-value";
67
68 return Effect.gen(function* () {
69 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
70 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
71 const circular: Record<string, unknown> = { secret };
72 circular.self = circular;
73
74 try {
75 const logger = yield* makeEventNdjsonLogger(basePath, { stream: "native" });
76 assert.exists(logger);
77 if (!logger) return;
78 yield* logger.write(circular, ThreadId.make("thread-1"));
79 yield* logger.close();
80
81 const serialized = encodeUnknownJson(messages);
82 assert.notInclude(serialized, secret);
83 const line = parseLogLine(
84 NodeFS.readFileSync(ownedLogPath(basePath, "thread-1"), "utf8").trim(),
85 );
86 assert.equal(line.payload, '{"truncated":true}');
87 } finally {
88 NodeFS.rmSync(tempDir, { recursive: true, force: true });
89 }
90 }).pipe(Effect.provide(Logger.layer([logCapture], { mergeWithExisting: false })));
91 });
92
93 it.effect("writes effect-style lines to thread-scoped files", () =>
94 Effect.gen(function* () {
95 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
96 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
97
98 try {
99 const logger = yield* makeEventNdjsonLogger(basePath, { stream: "native" });
100 assert.notEqual(logger, undefined);
101 if (!logger) {
102 return;
103 }
104
105 yield* logger.write(
106 { threadId: "provider-thread-1", id: "evt-1" },
107 ThreadId.make("thread-1"),
108 );
109 yield* logger.write(
110 { type: "turn.completed", threadId: "provider-thread-2", id: "evt-2" },
111 ThreadId.make("thread-2"),
112 );
113 yield* logger.close();
114
115 const threadOnePath = ownedLogPath(basePath, "thread-1");
116 const threadTwoPath = ownedLogPath(basePath, "thread-2");
117 assert.equal(NodeFS.existsSync(threadOnePath), true);
118 assert.equal(NodeFS.existsSync(threadTwoPath), true);
119
120 const first = parseLogLine(NodeFS.readFileSync(threadOnePath, "utf8").trim());
121 const second = parseLogLine(NodeFS.readFileSync(threadTwoPath, "utf8").trim());
122
123 assert.equal(Number.isNaN(Date.parse(first.observedAt)), false);
124 assert.equal(first.stream, "NTIVE");
125 assert.equal(first.payload, '{"threadId":"provider-thread-1","id":"evt-1"}');
126
127 assert.equal(Number.isNaN(Date.parse(second.observedAt)), false);
128 assert.equal(second.stream, "NTIVE");
129 assert.equal(
130 second.payload,
131 '{"type":"turn.completed","threadId":"provider-thread-2","id":"evt-2"}',
132 );
133 } finally {
134 NodeFS.rmSync(tempDir, { recursive: true, force: true });
135 }
136 }),
137 );
138
139 it.effect(
140 "falls back to a global segment when orchestration thread id is missing or invalid",
141 () =>
142 Effect.gen(function* () {
143 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
144 const basePath = NodePath.join(tempDir, "provider-canonical.ndjson");
145
146 try {
147 const logger = yield* makeEventNdjsonLogger(basePath, { stream: "orchestration" });
148 assert.notEqual(logger, undefined);
149 if (!logger) {
150 return;
151 }
152
153 yield* logger.write({ id: "evt-no-thread" }, null);
154 yield* logger.write({ id: "evt-invalid-thread" }, "!!!" as unknown as ThreadId);
155 yield* logger.close();
156
157 const globalPath = ownedLogPath(basePath, "_global");
158 assert.equal(NodeFS.existsSync(globalPath), true);
159 const lines = NodeFS.readFileSync(globalPath, "utf8")
160 .trim()
161 .split("\n")
162 .map((line) => parseLogLine(line));
163 assert.equal(lines.length, 2);
164 assert.equal(Number.isNaN(Date.parse(lines[0]?.observedAt ?? "")), false);
165 assert.equal(Number.isNaN(Date.parse(lines[1]?.observedAt ?? "")), false);
166 assert.equal(lines[0]?.stream, "ORCH");
167 assert.equal(lines[0]?.payload, '{"id":"evt-no-thread"}');
168 assert.equal(lines[1]?.stream, "ORCH");
169 assert.equal(lines[1]?.payload, '{"id":"evt-invalid-thread"}');
170 } finally {
171 NodeFS.rmSync(tempDir, { recursive: true, force: true });
172 }
173 }),
174 );
175
176 it.effect("shares one thread writer across native and canonical streams", () =>
177 Effect.gen(function* () {
178 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
179 const basePath = NodePath.join(tempDir, "events.log");
180
181 try {
182 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
183 const native = store.logger("native");
184 const canonical = store.logger("canonical");
185 const threadId = ThreadId.make("thread-shared");
186
187 yield* native.write({ id: "native-event" }, threadId);
188 yield* canonical.write({ type: "item.completed", id: "canonical-event" }, threadId);
189 yield* store.close();
190
191 const lines = NodeFS.readFileSync(ownedLogPath(basePath, "thread-shared"), "utf8")
192 .trim()
193 .split("\n")
194 .map(parseLogLine);
195
196 assert.deepEqual(
197 lines.map(({ stream, payload }) => ({ stream, payload })),
198 [
199 { stream: "NTIVE", payload: '{"id":"native-event"}' },
200 {
201 stream: "CANON",
202 payload: '{"type":"item.completed","id":"canonical-event"}',
203 },
204 ],
205 );
206 } finally {
207 NodeFS.rmSync(tempDir, { recursive: true, force: true });
208 }
209 }),
210 );
211
212 it.effect("keeps shared store views non-owning when one adapter closes", () =>
213 Effect.gen(function* () {
214 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
215 const basePath = NodePath.join(tempDir, "events.log");
216
217 try {
218 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
219 const native = store.logger("native");
220 const canonical = store.logger("canonical");
221 const threadId = ThreadId.make("thread-shared-close");
222
223 yield* native.write({ id: "before-close" }, threadId);
224 yield* native.close();
225 yield* canonical.write({ type: "item.completed", id: "after-close" }, threadId);
226 yield* store.close();
227
228 const lines = NodeFS.readFileSync(ownedLogPath(basePath, "thread-shared-close"), "utf8")
229 .trim()
230 .split("\n")
231 .map(parseLogLine);
232
233 assert.deepEqual(
234 lines.map(({ stream, payload }) => ({ stream, payload })),
235 [
236 { stream: "NTIVE", payload: '{"id":"before-close"}' },
237 {
238 stream: "CANON",
239 payload: '{"type":"item.completed","id":"after-close"}',
240 },
241 ],
242 );
243 } finally {
244 NodeFS.rmSync(tempDir, { recursive: true, force: true });
245 }
246 }),
247 );
248
249 it.effect("flushes an active batch without a permanent polling loop", () =>
250 Effect.gen(function* () {
251 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
252 const basePath = NodePath.join(tempDir, "events.log");
253 const threadPath = ownedLogPath(basePath, "thread-batched");
254
255 try {
256 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 1_000 });
257 yield* store
258 .logger("native")
259 .write({ id: "batched-event" }, ThreadId.make("thread-batched"));
260
261 assert.equal(NodeFS.existsSync(threadPath), false);
262 yield* TestClock.adjust(1_000);
263 yield* store.close();
264 assert.equal(NodeFS.existsSync(threadPath), true);
265 } finally {
266 NodeFS.rmSync(tempDir, { recursive: true, force: true });
267 }
268 }),
269 );
270
271 it.effect("does not strand a later batch after an interrupted write", () =>
272 Effect.gen(function* () {
273 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
274 const basePath = NodePath.join(tempDir, "events.log");
275 const threadPath = ownedLogPath(basePath, "thread-interrupted");
276
277 try {
278 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 1_000 });
279 const logger = store.logger("native");
280 const interruptedWrite = yield* logger
281 .write({ id: "possibly-interrupted" }, ThreadId.make("thread-interrupted"))
282 .pipe(Effect.forkChild);
283 yield* Effect.yieldNow;
284 yield* Fiber.interrupt(interruptedWrite);
285 yield* logger.write({ id: "accepted" }, ThreadId.make("thread-interrupted"));
286
287 yield* TestClock.adjust(1_000);
288 yield* store.close();
289
290 assert.equal(NodeFS.existsSync(threadPath), true);
291 assert.include(NodeFS.readFileSync(threadPath, "utf8"), '{"id":"accepted"}');
292 } finally {
293 NodeFS.rmSync(tempDir, { recursive: true, force: true });
294 }
295 }),
296 );
297
298 it.effect("drops transient provider events before serialization", () =>
299 Effect.gen(function* () {
300 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
301 const basePath = NodePath.join(tempDir, "events.log");
302
303 try {
304 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
305 const canonical = store.logger("canonical");
306 const native = store.logger("native");
307 const threadId = ThreadId.make("thread-filtered");
308 const circularDelta: Record<string, unknown> = { type: "content.delta" };
309 circularDelta["self"] = circularDelta;
310
311 yield* canonical.write(circularDelta, threadId);
312 yield* canonical.write({ type: "item.completed", id: "final" }, threadId);
313 yield* native.write({ type: "content.delta", id: "native-delta" }, threadId);
314 yield* native.write(
315 { method: "item/agentMessage/delta", payload: circularDelta },
316 threadId,
317 );
318 yield* native.write(
319 { method: "thread/realtime/outputAudio/delta", payload: circularDelta },
320 threadId,
321 );
322 yield* native.write(
323 { method: "thread/realtime/transcript/delta", payload: circularDelta },
324 threadId,
325 );
326 yield* native.write({ method: "turn/diff/updated", payload: circularDelta }, threadId);
327 yield* native.write(
328 {
329 event: {
330 direction: "incoming",
331 stage: "decoded",
332 payload: {
333 method: "turn/diff/updated",
334 get params() {
335 throw new Error("unused diff snapshots must not be traversed");
336 },
337 },
338 },
339 },
340 threadId,
341 );
342 yield* native.write(
343 {
344 event: {
345 method: "claude/stream_event/content_block_delta/text_delta",
346 payload: circularDelta,
347 },
348 },
349 threadId,
350 );
351 yield* native.write(
352 {
353 event: {
354 method: "session/update",
355 payload: { update: { sessionUpdate: "agent_message_chunk" } },
356 },
357 },
358 threadId,
359 );
360 yield* native.write(
361 {
362 event: {
363 direction: "incoming",
364 stage: "decoded",
365 payload: { method: "item/commandExecution/outputDelta", params: circularDelta },
366 },
367 },
368 threadId,
369 );
370 yield* native.write(
371 {
372 event: {
373 direction: "incoming",
374 stage: "decoded",
375 payload: {
376 type: "stream_event",
377 event: { type: "content_block_delta", delta: circularDelta },
378 },
379 },
380 },
381 threadId,
382 );
383 yield* native.write(
384 {
385 event: {
386 direction: "incoming",
387 stage: "raw",
388 get payload() {
389 throw new Error("raw frames must not be inspected");
390 },
391 },
392 },
393 threadId,
394 );
395 yield* native.write(
396 {
397 event: {
398 type: "message.part.updated",
399 payload: { properties: { part: { type: "text" } } },
400 },
401 },
402 threadId,
403 );
404 yield* native.write(
405 {
406 event: {
407 type: "message.part.updated",
408 payload: {
409 properties: {
410 part: { type: "tool", state: { status: "running", output: "progress" } },
411 },
412 },
413 },
414 },
415 threadId,
416 );
417 yield* native.write({ type: "turn.completed", id: "native-final" }, threadId);
418 yield* store.close();
419
420 const lines = NodeFS.readFileSync(ownedLogPath(basePath, "thread-filtered"), "utf8")
421 .trim()
422 .split("\n")
423 .map(parseLogLine);
424
425 assert.deepEqual(
426 lines.map(({ stream, payload }) => ({ stream, payload })),
427 [
428 { stream: "CANON", payload: '{"type":"item.completed","id":"final"}' },
429 { stream: "NTIVE", payload: '{"type":"turn.completed","id":"native-final"}' },
430 ],
431 );
432 } finally {
433 NodeFS.rmSync(tempDir, { recursive: true, force: true });
434 }
435 }),
436 );
437
438 it.effect("summarizes large histories without reading their items", () =>
439 Effect.gen(function* () {
440 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
441 const basePath = NodePath.join(tempDir, "events.log");
442 const turns = Array.from({ length: 10_000 });
443 Object.defineProperty(turns, 0, {
444 get: () => {
445 throw new Error("history must not be serialized");
446 },
447 });
448
449 try {
450 const store = yield* makeEventNdjsonLogStore(basePath);
451 yield* store.logger("native").write(
452 {
453 provider: "codex",
454 event: {
455 direction: "incoming",
456 stage: "decoded",
457 payload: { id: 42, result: { thread: { id: "native-thread", turns } } },
458 },
459 },
460 ThreadId.make("large-history"),
461 );
462 yield* store.close();
463
464 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-history"), "utf8");
465 assert.isBelow(Buffer.byteLength(contents), 2_048);
466 const record = decodeUnknownJson(parseLogLine(contents.trim()).payload);
467 assert.nestedPropertyVal(record, "event.payload.id", 42);
468 assert.nestedPropertyVal(record, "event.payload.result.thread.id", "native-thread");
469 assert.nestedPropertyVal(record, "event.payload.result.thread.turns.itemCount", 10_000);
470 } finally {
471 NodeFS.rmSync(tempDir, { recursive: true, force: true });
472 }
473 }),
474 );
475
476 it.effect("bounds oversized records while retaining failure details", () =>
477 Effect.gen(function* () {
478 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
479 const basePath = NodePath.join(tempDir, "events.log");
480
481 try {
482 const store = yield* makeEventNdjsonLogStore(basePath);
483 const logger = store.logger("native");
484 const threadId = ThreadId.make("large-error");
485 const failure = {
486 method: "error",
487 params: {
488 threadId: "native-thread",
489 turnId: "native-turn",
490 error: { message: "The provider is unavailable.", code: "overloaded" },
491 output: "x".repeat(128 * 1_024),
492 },
493 };
494 yield* logger.write(failure, threadId);
495 yield* logger.write({ id: "escaped", output: "\u0000".repeat(20_000) }, threadId);
496 yield* store.close();
497
498 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-error"), "utf8");
499 const records = contents
500 .trim()
501 .split("\n")
502 .map((line) => decodeUnknownJson(parseLogLine(line).payload));
503 assert.isBelow(Buffer.byteLength(contents), 64 * 1_024);
504 assert.equal(records.length, 2);
505 assert.nestedPropertyVal(records[0], "params.threadId", "native-thread");
506 assert.nestedPropertyVal(records[0], "params.turnId", "native-turn");
507 assert.nestedPropertyVal(
508 records[0],
509 "params.error.message",
510 "The provider is unavailable.",
511 );
512 assert.nestedPropertyVal(records[0], "params.error.code", "overloaded");
513 assert.propertyVal(records[1], "id", "escaped");
514 } finally {
515 NodeFS.rmSync(tempDir, { recursive: true, force: true });
516 }
517 }),
518 );
519
520 it.effect("bounds canonical diff snapshots before serializing their duplicate payloads", () =>
521 Effect.gen(function* () {
522 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
523 const basePath = NodePath.join(tempDir, "events.log");
524 const threadId = ThreadId.make("large-diff");
525 const diff = "diff-payload".repeat(128 * 1_024);
526 try {
527 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
528 yield* store.logger("canonical").write(
529 {
530 type: "turn.diff.updated",
531 threadId,
532 turnId: "native-turn",
533 raw: { method: "turn/diff/updated", payload: { diff } },
534 payload: { unifiedDiff: diff },
535 },
536 threadId,
537 );
538 yield* store.close();
539 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-diff"), "utf8");
540 assert.isBelow(Buffer.byteLength(contents), 2_048);
541 const record = decodeUnknownJson(parseLogLine(contents.trim()).payload);
542 assert.propertyVal(record, "type", "turn.diff.updated");
543 assert.propertyVal(record, "threadId", threadId);
544 assert.propertyVal(record, "turnId", "native-turn");
545 } finally {
546 NodeFS.rmSync(tempDir, { recursive: true, force: true });
547 }
548 }),
549 );
550
551 it.effect("keeps OpenCode tool input, final output, and errors in native logs", () =>
552 Effect.gen(function* () {
553 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
554 const basePath = NodePath.join(tempDir, "events.log");
555 const threadId = ThreadId.make("thread-tool-lifecycle");
556
557 try {
558 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
559 const native = store.logger("native");
560 const input = { command: "run-build" };
561 const states = [
562 { status: "pending", input },
563 { status: "completed", input, output: "Build complete", time: { start: 1, end: 2 } },
564 { status: "error", input, error: "Build failed", time: { start: 3, end: 4 } },
565 { status: "unknown", input },
566 null,
567 ];
568 const events = states.map((state) => ({
569 event: {
570 type: "message.part.updated",
571 payload: { properties: { part: { type: "tool", state } } },
572 },
573 }));
574 for (const event of events) {
575 yield* native.write(event, threadId);
576 }
577 yield* store.close();
578
579 const payloads = NodeFS.readFileSync(
580 ownedLogPath(basePath, "thread-tool-lifecycle"),
581 "utf8",
582 )
583 .trim()
584 .split("\n")
585 .map((line) => parseLogLine(line).payload);
586 assert.deepEqual(
587 payloads,
588 events.map((event) => encodeUnknownJson(event)),
589 );
590 } finally {
591 NodeFS.rmSync(tempDir, { recursive: true, force: true });
592 }
593 }),
594 );
595
596 it.effect("contains hostile event accessors inside guarded serialization", () =>
597 Effect.gen(function* () {
598 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
599 const basePath = NodePath.join(tempDir, "events.log");
600
601 try {
602 const logger = yield* makeEventNdjsonLogger(basePath, {
603 stream: "canonical",
604 batchWindowMs: 0,
605 });
606 assert.exists(logger);
607 if (!logger) return;
608 const hostile = new Proxy(
609 { id: "hostile" },
610 {
611 get(_target, property) {
612 if (property === "type") throw new Error("blocked");
613 return undefined;
614 },
615 },
616 );
617
618 yield* logger.write(hostile, ThreadId.make("thread-hostile"));
619 yield* logger.close();
620
621 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "thread-hostile"), "utf8");
622 assert.notInclude(contents, "blocked");
623 } finally {
624 NodeFS.rmSync(tempDir, { recursive: true, force: true });
625 }
626 }),
627 );
628
629 it.effect("serializes concurrent first writes for the same segment", () =>
630 Effect.gen(function* () {
631 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
632 const basePath = NodePath.join(tempDir, "provider-canonical.ndjson");
633
634 try {
635 const logger = yield* makeEventNdjsonLogger(basePath, {
636 stream: "canonical",
637 batchWindowMs: 0,
638 });
639 assert.notEqual(logger, undefined);
640 if (!logger) {
641 return;
642 }
643
644 yield* Effect.all(
645 [
646 logger.write({ id: "evt-concurrent-1" }, null),
647 logger.write({ id: "evt-concurrent-2" }, null),
648 ],
649 { concurrency: "unbounded" },
650 );
651 yield* logger.close();
652
653 const globalPath = ownedLogPath(basePath, "_global");
654 assert.equal(NodeFS.existsSync(globalPath), true);
655 const lines = NodeFS.readFileSync(globalPath, "utf8")
656 .trim()
657 .split("\n")
658 .map((line) => parseLogLine(line));
659
660 assert.equal(lines.length, 2);
661 assert.deepEqual(lines.map((line) => line.payload).toSorted(), [
662 '{"id":"evt-concurrent-1"}',
663 '{"id":"evt-concurrent-2"}',
664 ]);
665 } finally {
666 NodeFS.rmSync(tempDir, { recursive: true, force: true });
667 }
668 }),
669 );
670
671 it.effect("rotates per-thread files when max size is exceeded", () =>
672 Effect.gen(function* () {
673 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
674 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
675
676 try {
677 const store = yield* makeEventNdjsonLogStore(basePath, {
678 maxBytes: 120,
679 maxFiles: 2,
680 batchWindowMs: 0,
681 });
682 const native = store.logger("native");
683 const canonical = store.logger("canonical");
684
685 for (let index = 0; index < 10; index += 1) {
686 const logger = index % 2 === 0 ? native : canonical;
687 yield* logger.write(
688 {
689 type: "session.started",
690 threadId: "provider-thread-rotate",
691 id: `evt-${index}`,
692 payload: "x".repeat(40),
693 },
694 ThreadId.make("thread-rotate"),
695 );
696 }
697 yield* store.close();
698
699 const fileStem = NodePath.basename(ownedLogPath(basePath, "thread-rotate"));
700 const matchingFiles = NodeFS.readdirSync(tempDir)
701 .filter((entry) => entry === fileStem || entry.startsWith(`${fileStem}.`))
702 .toSorted();
703
704 assert.equal(
705 matchingFiles.some((entry) => entry === `${fileStem}.1`),
706 true,
707 );
708 assert.equal(
709 matchingFiles.some((entry) => entry === fileStem || entry === `${fileStem}.2`),
710 true,
711 );
712 assert.equal(
713 matchingFiles.some((entry) => entry === `${fileStem}.3`),
714 false,
715 );
716 } finally {
717 NodeFS.rmSync(tempDir, { recursive: true, force: true });
718 }
719 }),
720 );
721
722 it.effect("enforces aggregate age and byte retention on startup", () =>
723 Effect.gen(function* () {
724 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
725 const basePath = NodePath.join(tempDir, "events.log");
726 const expiredPath = ownedLogPath(basePath, "expired");
727 const oldPath = ownedLogPath(basePath, "old");
728 const newPath = ownedLogPath(basePath, "new");
729 const unrelatedLogPath = NodePath.join(tempDir, "unrelated.log");
730 const legacyLogPath = NodePath.join(tempDir, "legacy-thread.log");
731 const ignoredPath = NodePath.join(tempDir, "ignored.txt");
732
733 try {
734 yield* TestClock.setTime(1_800_000_000_000);
735 const now = yield* Clock.currentTimeMillis;
736 for (const filePath of [expiredPath, oldPath, newPath, unrelatedLogPath, ignoredPath]) {
737 NodeFS.writeFileSync(filePath, "x".repeat(40));
738 }
739 NodeFS.writeFileSync(
740 legacyLogPath,
741 "[2026-01-01T00:00:00.000Z] CANON: legacy provider event\n",
742 );
743 NodeFS.utimesSync(expiredPath, (now - 20_000) / 1_000, (now - 20_000) / 1_000);
744 NodeFS.utimesSync(legacyLogPath, (now - 20_000) / 1_000, (now - 20_000) / 1_000);
745 NodeFS.utimesSync(oldPath, (now - 5_000) / 1_000, (now - 5_000) / 1_000);
746 NodeFS.utimesSync(newPath, now / 1_000, now / 1_000);
747
748 const store = yield* makeEventNdjsonLogStore(basePath, {
749 maxAgeMs: 10_000,
750 maxTotalBytes: 60,
751 });
752 yield* store.close();
753
754 assert.equal(NodeFS.existsSync(expiredPath), false);
755 assert.equal(NodeFS.existsSync(legacyLogPath), false);
756 assert.equal(NodeFS.existsSync(oldPath), false);
757 assert.equal(NodeFS.existsSync(newPath), true);
758 assert.equal(NodeFS.existsSync(unrelatedLogPath), true);
759 assert.equal(NodeFS.existsSync(ignoredPath), true);
760 } finally {
761 NodeFS.rmSync(tempDir, { recursive: true, force: true });
762 }
763 }),
764 );
765
766 it.effect("does not prune an active thread sink during an unrelated flush", () =>
767 Effect.gen(function* () {
768 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
769 const basePath = NodePath.join(tempDir, "events.log");
770 const activePath = ownedLogPath(basePath, "active");
771
772 try {
773 yield* TestClock.setTime(1_800_000_000_000);
774 const store = yield* makeEventNdjsonLogStore(basePath, {
775 batchWindowMs: 0,
776 maxAgeMs: 1,
777 retentionCheckIntervalMs: 1,
778 });
779 const logger = store.logger("native");
780
781 yield* logger.write({ id: "active-before-retention" }, ThreadId.make("active"));
782 assert.equal(NodeFS.existsSync(activePath), true);
783
784 yield* TestClock.adjust("2 millis");
785 yield* logger.write({ id: "retention-trigger" }, ThreadId.make("other"));
786
787 assert.equal(NodeFS.existsSync(activePath), true);
788 yield* store.close();
789 } finally {
790 NodeFS.rmSync(tempDir, { recursive: true, force: true });
791 }
792 }),
793 );
794
795 it.effect("attributes batches that were written before a later chunk fails", () =>
796 Effect.gen(function* () {
797 const records: ReadonlyArray<PendingRecord> = [
798 {
799 stream: "native",
800 threadSegment: "thread",
801 line: "first",
802 bytes: 5,
803 },
804 {
805 stream: "canonical",
806 threadSegment: "thread",
807 line: "second",
808 bytes: 6,
809 },
810 ];
811 const attributed: Array<PendingRecord> = [];
812 let writes = 0;
813
814 const result = yield* writeBatchedMessages(
815 {
816 write: () => {
817 writes += 1;
818 return writes === 2
819 ? Effect.fail(
820 new RotatingFileSinkError({
821 operation: "write",
822 filePath: "test.log",
823 cause: "simulated disk exhaustion",
824 }),
825 )
826 : Effect.void;
827 },
828 },
829 records,
830 5,
831 (written) => attributed.push(...written),
832 ).pipe(Effect.result);
833 assert.equal(result._tag, "Failure");
834 assert.deepEqual(attributed, [records[0]]);
835 }),
836 );
837
838 it.effect(
839 "keeps unrelated work responsive and drains before close while a disk write is held",
840 () =>
841 Effect.gen(function* () {
842 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
843 const basePath = NodePath.join(tempDir, "events.log");
844 const fs = yield* FileSystem.FileSystem;
845 const started = yield* Deferred.make<void>();
846 const release = yield* Deferred.make<void>();
847 let writes = 0;
848 try {
849 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 }).pipe(
850 Effect.provideService(FileSystem.FileSystem, {
851 ...fs,
852 writeFile: (filePath, content, options) =>
853 Effect.gen(function* () {
854 if (++writes === 1) {
855 yield* Deferred.succeed(started, undefined);
856 yield* Deferred.await(release);
857 }
858 yield* fs.writeFile(filePath, content, options);
859 }),
860 }),
861 );
862 const first = yield* store
863 .logger("native")
864 .write({ id: "first" }, ThreadId.make("held"))
865 .pipe(Effect.forkChild);
866 yield* Deferred.await(started);
867 const closed = yield* Deferred.make<void>();
868 const close = yield* store
869 .close()
870 .pipe(Effect.andThen(Deferred.succeed(closed, undefined)), Effect.forkChild);
871 assert.equal(yield* Deferred.isDone(closed), false);
872 // A separate fiber runs while the file operation remains pending.
873 const heartbeat = yield* Deferred.make<void>();
874 yield* Effect.forkChild(Deferred.succeed(heartbeat, undefined));
875 yield* Deferred.await(heartbeat);
876 yield* Deferred.succeed(release, undefined);
877 yield* Fiber.join(first);
878 yield* Fiber.join(close);
879 assert.include(
880 NodeFS.readFileSync(ownedLogPath(basePath, "held"), "utf8"),
881 '{"id":"first"}',
882 );
883 } finally {
884 NodeFS.rmSync(tempDir, { recursive: true, force: true });
885 }
886 }),
887 );
888
889 it.effect("reports logical provider log writes to resource attribution", () =>
890 Effect.gen(function* () {
891 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
892 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
893
894 try {
895 const attribution = yield* ResourceAttribution.make();
896 const logger = yield* makeEventNdjsonLogger(basePath, {
897 stream: "native",
898 batchWindowMs: 0,
899 attribution,
900 });
901 assert.notEqual(logger, undefined);
902 if (!logger) {
903 return;
904 }
905
906 yield* logger.write({ id: "attributed-event" }, ThreadId.make("thread-attribution"));
907 yield* logger.close();
908
909 const snapshot = yield* attribution.snapshot;
910 assert.equal(snapshot.entries.length, 1);
911 assert.equal(snapshot.entries[0]?.component, "provider-event-log");
912 assert.equal(snapshot.entries[0]?.operation, "native.append");
913 assert.equal(snapshot.entries[0]?.count, 1);
914 assert.isAbove(snapshot.entries[0]?.logicalWriteBytes ?? 0, 0);
915 } finally {
916 NodeFS.rmSync(tempDir, { recursive: true, force: true });
917 }
918 }),
919 );
920});