Open History to see the last commit
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 Effect from "effect/Effect";
11import * as Deferred from "effect/Deferred";
12import * as FileSystem from "effect/FileSystem";
13import * as Fiber from "effect/Fiber";
14import * as Logger from "effect/Logger";
15import * as Schema from "effect/Schema";
16import * as TestClock from "effect/testing/TestClock";
17
18import {
19 makeEventNdjsonLogStore,
20 type PendingRecord,
21 writeBatchedMessages,
22} from "./EventNdjsonLogger.ts";
23
24const encodeUnknownJson = Schema.encodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
25const decodeUnknownJson = Schema.decodeUnknownSync(Schema.fromJsonString(Schema.Unknown));
26
27function ownedLogPath(basePath: string, segment: string): string {
28 const basename = NodePath.basename(basePath);
29 const extension = NodePath.extname(basename);
30 const stem = extension.length > 0 ? basename.slice(0, -extension.length) : basename;
31 return NodePath.join(NodePath.dirname(basePath), `${stem}.${segment}.log`);
32}
33
34function parseLogLine(line: string) {
35 const match = /^\[([^\]]+)\] ([A-Z]+): (.+)$/.exec(line);
36 assert.notEqual(match, null);
37 if (!match) {
38 throw new Error(`invalid log line: ${line}`);
39 }
40 const observedAt = match[1];
41 const stream = match[2];
42 const payload = match[3];
43 if (!observedAt || !stream || payload === undefined) {
44 throw new Error(`invalid log line: ${line}`);
45 }
46 return {
47 observedAt,
48 stream,
49 payload,
50 };
51}
52
53it.layer(NodeServices.layer)("EventNdjsonLogger", (it) => {
54 it.effect("summarizes circular events without exposing their contents in diagnostics", () => {
55 const messages: Array<unknown> = [];
56 const logCapture = Logger.make<unknown, void>(({ message }) => {
57 if (Array.isArray(message)) {
58 messages.push(...message);
59 } else {
60 messages.push(message);
61 }
62 });
63 const secret = "secret-circular-event-value";
64
65 return Effect.gen(function* () {
66 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
67 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
68 const circular: Record<string, unknown> = { secret };
69 circular.self = circular;
70
71 try {
72 const store = yield* makeEventNdjsonLogStore(basePath);
73 const logger = store.logger("native");
74 yield* logger.write(circular, ThreadId.make("thread-1"));
75 yield* store.close();
76
77 const serialized = encodeUnknownJson(messages);
78 assert.notInclude(serialized, secret);
79 const line = parseLogLine(
80 NodeFS.readFileSync(ownedLogPath(basePath, "thread-1"), "utf8").trim(),
81 );
82 assert.equal(line.payload, '{"truncated":true}');
83 } finally {
84 NodeFS.rmSync(tempDir, { recursive: true, force: true });
85 }
86 }).pipe(Effect.provide(Logger.layer([logCapture], { mergeWithExisting: false })));
87 });
88
89 it.effect(
90 "falls back to a global segment when orchestration thread id is missing or invalid",
91 () =>
92 Effect.gen(function* () {
93 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
94 const basePath = NodePath.join(tempDir, "provider-canonical.ndjson");
95
96 try {
97 const store = yield* makeEventNdjsonLogStore(basePath);
98 const logger = store.logger("orchestration");
99
100 yield* logger.write({ id: "evt-no-thread" }, null);
101 yield* logger.write({ id: "evt-invalid-thread" }, "!!!" as unknown as ThreadId);
102 yield* store.close();
103
104 const globalPath = ownedLogPath(basePath, "_global");
105 assert.equal(NodeFS.existsSync(globalPath), true);
106 const lines = NodeFS.readFileSync(globalPath, "utf8")
107 .trim()
108 .split("\n")
109 .map((line) => parseLogLine(line));
110 assert.equal(lines.length, 2);
111 assert.equal(Number.isNaN(Date.parse(lines[0]?.observedAt ?? "")), false);
112 assert.equal(Number.isNaN(Date.parse(lines[1]?.observedAt ?? "")), false);
113 assert.equal(lines[0]?.stream, "ORCH");
114 assert.equal(lines[0]?.payload, '{"id":"evt-no-thread"}');
115 assert.equal(lines[1]?.stream, "ORCH");
116 assert.equal(lines[1]?.payload, '{"id":"evt-invalid-thread"}');
117 } finally {
118 NodeFS.rmSync(tempDir, { recursive: true, force: true });
119 }
120 }),
121 );
122
123 it.effect("keeps long thread ids that share a prefix in separate files", () =>
124 Effect.gen(function* () {
125 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
126 const basePath = NodePath.join(tempDir, "events.log");
127 const prefix =
128 "thread:delegated-task:command%3Amcp%3A00000000-0000-4000-8000-000000000000%3Adelegate-task%3A";
129
130 try {
131 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
132 const logger = store.logger("native");
133 yield* logger.write({ id: "first" }, ThreadId.make(`${prefix}review-the-diff`));
134 yield* logger.write({ id: "second" }, ThreadId.make(`${prefix}write-the-tests`));
135 yield* store.close();
136
137 const files = NodeFS.readdirSync(tempDir);
138 assert.equal(files.length, 2);
139 assert.deepEqual(
140 files
141 .map((name) =>
142 NodeFS.readFileSync(NodePath.join(tempDir, name), "utf8")
143 .trim()
144 .split("\n")
145 .map((line) => parseLogLine(line).payload),
146 )
147 .toSorted(),
148 [['{"id":"first"}'], ['{"id":"second"}']],
149 );
150 } finally {
151 NodeFS.rmSync(tempDir, { recursive: true, force: true });
152 }
153 }),
154 );
155
156 it.effect("keeps shared store views non-owning when one adapter closes", () =>
157 Effect.gen(function* () {
158 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
159 const basePath = NodePath.join(tempDir, "events.log");
160
161 try {
162 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
163 const native = store.logger("native");
164 const canonical = store.logger("canonical");
165 const threadId = ThreadId.make("thread-shared-close");
166
167 yield* native.write({ id: "before-close" }, threadId);
168 yield* native.close();
169 yield* canonical.write({ type: "item.completed", id: "after-close" }, threadId);
170 yield* store.close();
171
172 const lines = NodeFS.readFileSync(ownedLogPath(basePath, "thread-shared-close"), "utf8")
173 .trim()
174 .split("\n")
175 .map(parseLogLine);
176
177 assert.deepEqual(
178 lines.map(({ stream, payload }) => ({ stream, payload })),
179 [
180 { stream: "NTIVE", payload: '{"id":"before-close"}' },
181 {
182 stream: "CANON",
183 payload: '{"type":"item.completed","id":"after-close"}',
184 },
185 ],
186 );
187 } finally {
188 NodeFS.rmSync(tempDir, { recursive: true, force: true });
189 }
190 }),
191 );
192
193 it.effect("does not strand a later batch after an interrupted write", () =>
194 Effect.gen(function* () {
195 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
196 const basePath = NodePath.join(tempDir, "events.log");
197 const threadPath = ownedLogPath(basePath, "thread-interrupted");
198
199 try {
200 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 1_000 });
201 const logger = store.logger("native");
202 const interruptedWrite = yield* logger
203 .write({ id: "possibly-interrupted" }, ThreadId.make("thread-interrupted"))
204 .pipe(Effect.forkChild);
205 yield* Effect.yieldNow;
206 yield* Fiber.interrupt(interruptedWrite);
207 yield* logger.write({ id: "accepted" }, ThreadId.make("thread-interrupted"));
208
209 yield* TestClock.adjust(1_000);
210 yield* store.close();
211
212 assert.equal(NodeFS.existsSync(threadPath), true);
213 assert.include(NodeFS.readFileSync(threadPath, "utf8"), '{"id":"accepted"}');
214 } finally {
215 NodeFS.rmSync(tempDir, { recursive: true, force: true });
216 }
217 }),
218 );
219
220 it.effect("summarizes large histories without reading their items", () =>
221 Effect.gen(function* () {
222 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
223 const basePath = NodePath.join(tempDir, "events.log");
224 const turns = Array.from({ length: 10_000 });
225 Object.defineProperty(turns, 0, {
226 get: () => {
227 throw new Error("history must not be serialized");
228 },
229 });
230
231 try {
232 const store = yield* makeEventNdjsonLogStore(basePath);
233 yield* store.logger("native").write(
234 {
235 provider: "codex",
236 event: {
237 direction: "incoming",
238 stage: "decoded",
239 payload: { id: 42, result: { thread: { id: "native-thread", turns } } },
240 },
241 },
242 ThreadId.make("large-history"),
243 );
244 yield* store.close();
245
246 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-history"), "utf8");
247 assert.isBelow(Buffer.byteLength(contents), 2_048);
248 const record = decodeUnknownJson(parseLogLine(contents.trim()).payload);
249 assert.nestedPropertyVal(record, "event.payload.id", 42);
250 assert.nestedPropertyVal(record, "event.payload.result.thread.id", "native-thread");
251 assert.nestedPropertyVal(record, "event.payload.result.thread.turns.itemCount", 10_000);
252 } finally {
253 NodeFS.rmSync(tempDir, { recursive: true, force: true });
254 }
255 }),
256 );
257
258 it.effect("bounds oversized records while retaining failure details", () =>
259 Effect.gen(function* () {
260 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
261 const basePath = NodePath.join(tempDir, "events.log");
262
263 try {
264 const store = yield* makeEventNdjsonLogStore(basePath);
265 const logger = store.logger("native");
266 const threadId = ThreadId.make("large-error");
267 const failure = {
268 method: "error",
269 params: {
270 threadId: "native-thread",
271 turnId: "native-turn",
272 error: { message: "The provider is unavailable.", code: "overloaded" },
273 output: "x".repeat(128 * 1_024),
274 },
275 };
276 yield* logger.write(failure, threadId);
277 yield* logger.write({ id: "escaped", output: "\u0000".repeat(20_000) }, threadId);
278 yield* store.close();
279
280 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-error"), "utf8");
281 const records = contents
282 .trim()
283 .split("\n")
284 .map((line) => decodeUnknownJson(parseLogLine(line).payload));
285 assert.isBelow(Buffer.byteLength(contents), 64 * 1_024);
286 assert.equal(records.length, 2);
287 assert.nestedPropertyVal(records[0], "params.threadId", "native-thread");
288 assert.nestedPropertyVal(records[0], "params.turnId", "native-turn");
289 assert.nestedPropertyVal(
290 records[0],
291 "params.error.message",
292 "The provider is unavailable.",
293 );
294 assert.nestedPropertyVal(records[0], "params.error.code", "overloaded");
295 assert.propertyVal(records[1], "id", "escaped");
296 } finally {
297 NodeFS.rmSync(tempDir, { recursive: true, force: true });
298 }
299 }),
300 );
301
302 it.effect("bounds canonical diff snapshots before serializing their duplicate payloads", () =>
303 Effect.gen(function* () {
304 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
305 const basePath = NodePath.join(tempDir, "events.log");
306 const threadId = ThreadId.make("large-diff");
307 const diff = "diff-payload".repeat(128 * 1_024);
308 try {
309 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
310 yield* store.logger("canonical").write(
311 {
312 type: "turn.diff.updated",
313 threadId,
314 turnId: "native-turn",
315 raw: { method: "turn/diff/updated", payload: { diff } },
316 payload: { unifiedDiff: diff },
317 },
318 threadId,
319 );
320 yield* store.close();
321 const contents = NodeFS.readFileSync(ownedLogPath(basePath, "large-diff"), "utf8");
322 assert.isBelow(Buffer.byteLength(contents), 2_048);
323 const record = decodeUnknownJson(parseLogLine(contents.trim()).payload);
324 assert.propertyVal(record, "type", "turn.diff.updated");
325 assert.propertyVal(record, "threadId", threadId);
326 assert.propertyVal(record, "turnId", "native-turn");
327 } finally {
328 NodeFS.rmSync(tempDir, { recursive: true, force: true });
329 }
330 }),
331 );
332
333 it.effect("serializes concurrent first writes for the same segment", () =>
334 Effect.gen(function* () {
335 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
336 const basePath = NodePath.join(tempDir, "provider-canonical.ndjson");
337
338 try {
339 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 });
340 const logger = store.logger("canonical");
341
342 yield* Effect.all(
343 [
344 logger.write({ id: "evt-concurrent-1" }, null),
345 logger.write({ id: "evt-concurrent-2" }, null),
346 ],
347 { concurrency: "unbounded" },
348 );
349 yield* store.close();
350
351 const globalPath = ownedLogPath(basePath, "_global");
352 assert.equal(NodeFS.existsSync(globalPath), true);
353 const lines = NodeFS.readFileSync(globalPath, "utf8")
354 .trim()
355 .split("\n")
356 .map((line) => parseLogLine(line));
357
358 assert.equal(lines.length, 2);
359 assert.deepEqual(lines.map((line) => line.payload).toSorted(), [
360 '{"id":"evt-concurrent-1"}',
361 '{"id":"evt-concurrent-2"}',
362 ]);
363 } finally {
364 NodeFS.rmSync(tempDir, { recursive: true, force: true });
365 }
366 }),
367 );
368
369 it.effect("rotates per-thread files when max size is exceeded", () =>
370 Effect.gen(function* () {
371 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
372 const basePath = NodePath.join(tempDir, "provider-native.ndjson");
373
374 try {
375 const store = yield* makeEventNdjsonLogStore(basePath, {
376 maxBytes: 120,
377 maxFiles: 2,
378 batchWindowMs: 0,
379 });
380 const native = store.logger("native");
381 const canonical = store.logger("canonical");
382
383 for (let index = 0; index < 10; index += 1) {
384 const logger = index % 2 === 0 ? native : canonical;
385 yield* logger.write(
386 {
387 type: "session.started",
388 threadId: "provider-thread-rotate",
389 id: `evt-${index}`,
390 payload: "x".repeat(40),
391 },
392 ThreadId.make("thread-rotate"),
393 );
394 }
395 yield* store.close();
396
397 const fileStem = NodePath.basename(ownedLogPath(basePath, "thread-rotate"));
398 const matchingFiles = NodeFS.readdirSync(tempDir)
399 .filter((entry) => entry === fileStem || entry.startsWith(`${fileStem}.`))
400 .toSorted();
401
402 assert.equal(
403 matchingFiles.some((entry) => entry === `${fileStem}.1`),
404 true,
405 );
406 assert.equal(
407 matchingFiles.some((entry) => entry === fileStem || entry === `${fileStem}.2`),
408 true,
409 );
410 assert.equal(
411 matchingFiles.some((entry) => entry === `${fileStem}.3`),
412 false,
413 );
414 } finally {
415 NodeFS.rmSync(tempDir, { recursive: true, force: true });
416 }
417 }),
418 );
419
420 it.effect("retires a thread log once a retention interval passes without writes", () =>
421 Effect.gen(function* () {
422 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
423 const basePath = NodePath.join(tempDir, "events.log");
424 const finishedPath = ownedLogPath(basePath, "finished");
425 const activePath = ownedLogPath(basePath, "active");
426
427 try {
428 const store = yield* makeEventNdjsonLogStore(basePath, {
429 batchWindowMs: 0,
430 });
431 const logger = store.logger("native");
432
433 yield* logger.write({ id: "finished-first" }, ThreadId.make("finished"));
434 yield* TestClock.adjust("5 minutes");
435 yield* logger.write({ id: "active-first" }, ThreadId.make("active"));
436 assert.equal(NodeFS.existsSync(finishedPath), true);
437 NodeFS.truncateSync(finishedPath, 512 * 1024 * 1024 + 1);
438
439 yield* TestClock.adjust("5 minutes");
440 yield* logger.write({ id: "active-second" }, ThreadId.make("active"));
441 assert.equal(NodeFS.existsSync(finishedPath), false);
442 assert.equal(NodeFS.existsSync(activePath), true);
443
444 yield* logger.write({ id: "finished-again" }, ThreadId.make("finished"));
445 yield* store.close();
446 assert.deepEqual(
447 NodeFS.readFileSync(finishedPath, "utf8")
448 .trim()
449 .split("\n")
450 .map((line) => parseLogLine(line).payload),
451 ['{"id":"finished-again"}'],
452 );
453 } finally {
454 NodeFS.rmSync(tempDir, { recursive: true, force: true });
455 }
456 }),
457 );
458
459 it.effect("attributes batches that were written before a later chunk fails", () =>
460 Effect.gen(function* () {
461 const records: ReadonlyArray<PendingRecord> = [
462 {
463 stream: "native",
464 threadSegment: "thread",
465 line: "first",
466 bytes: 5,
467 },
468 {
469 stream: "canonical",
470 threadSegment: "thread",
471 line: "second",
472 bytes: 6,
473 },
474 ];
475 const attributed: Array<PendingRecord> = [];
476 let writes = 0;
477
478 const result = yield* writeBatchedMessages(
479 {
480 write: () => {
481 writes += 1;
482 return writes === 2
483 ? Effect.fail(
484 new RotatingFileSinkError({
485 operation: "write",
486 filePath: "test.log",
487 cause: "simulated disk exhaustion",
488 }),
489 )
490 : Effect.void;
491 },
492 },
493 records,
494 5,
495 (written) => attributed.push(...written),
496 ).pipe(Effect.result);
497 assert.equal(result._tag, "Failure");
498 assert.deepEqual(attributed, [records[0]]);
499 }),
500 );
501
502 it.effect(
503 "keeps unrelated work responsive and drains before close while a disk write is held",
504 () =>
505 Effect.gen(function* () {
506 const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-log-"));
507 const basePath = NodePath.join(tempDir, "events.log");
508 const fs = yield* FileSystem.FileSystem;
509 const started = yield* Deferred.make<void>();
510 const release = yield* Deferred.make<void>();
511 let writes = 0;
512 try {
513 const store = yield* makeEventNdjsonLogStore(basePath, { batchWindowMs: 0 }).pipe(
514 Effect.provideService(FileSystem.FileSystem, {
515 ...fs,
516 writeFile: (filePath, content, options) =>
517 Effect.gen(function* () {
518 if (++writes === 1) {
519 yield* Deferred.succeed(started, undefined);
520 yield* Deferred.await(release);
521 }
522 yield* fs.writeFile(filePath, content, options);
523 }),
524 }),
525 );
526 const first = yield* store
527 .logger("native")
528 .write({ id: "first" }, ThreadId.make("held"))
529 .pipe(Effect.forkChild);
530 yield* Deferred.await(started);
531 const closed = yield* Deferred.make<void>();
532 const close = yield* store
533 .close()
534 .pipe(Effect.andThen(Deferred.succeed(closed, undefined)), Effect.forkChild);
535 assert.equal(yield* Deferred.isDone(closed), false);
536
537 const heartbeat = yield* Deferred.make<void>();
538 yield* Effect.forkChild(Deferred.succeed(heartbeat, undefined));
539 yield* Deferred.await(heartbeat);
540 yield* Deferred.succeed(release, undefined);
541 yield* Fiber.join(first);
542 yield* Fiber.join(close);
543 assert.include(
544 NodeFS.readFileSync(ownedLogPath(basePath, "held"), "utf8"),
545 '{"id":"first"}',
546 );
547 } finally {
548 NodeFS.rmSync(tempDir, { recursive: true, force: true });
549 }
550 }),
551 );
552});