authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-03 18:14:45-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-03 18:22:29-08:00
log86e87a8789196663e5e919466676566dadb32f20
tree6874093cfabd71d06a220bf53837718f9ffbee8c
parent404e78a506ad1a6b0bd55eaa703fb0563e7dea7d

better lables

5 files changed, 390 insertions(+), 274 deletions(-)

apps/server/src/provider/Layers/CodexAdapter.ts+31-31
......@@ -436,7 +436,9 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
436436 const nativeEventLogger =
437437 options?.nativeEventLogger ??
438438 (options?.nativeEventLogPath !== undefined
439 ? makeEventNdjsonLogger(options.nativeEventLogPath)
439 ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {
440 stream: "native",
441 })
440442 : undefined);
441443
442444 const manager = yield* Effect.acquireRelease(
......@@ -499,9 +501,11 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
499501 new Error(`Invalid attachment id '${attachment.id}'.`),
500502 );
501503 }
502 const bytes = yield* fileSystem.readFile(attachmentPath).pipe(
503 Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause)),
504 );
504 const bytes = yield* fileSystem
505 .readFile(attachmentPath)
506 .pipe(
507 Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause)),
508 );
505509 return {
506510 type: "image" as const,
507511 url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`,
......@@ -583,33 +587,29 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) =>
583587 const runtimeEventQueue = yield* Queue.unbounded<ProviderRuntimeEvent>();
584588
585589 yield* Effect.acquireRelease(
586 Effect.sync(() => {
587 const writeNativeEvent = (event: ProviderEvent) => {
588 if (!nativeEventLogger) {
589 return;
590 }
591 const loggedAt = new Date().toISOString();
592 const resolveOrchestrationThreadId = Effect.catch(
593 directory.getThreadId(event.sessionId).pipe(Effect.map(Option.getOrUndefined)),
594 () => Effect.void,
595 );
596 const writeLogEffect = Effect.tap(resolveOrchestrationThreadId, (orchestrationThreadId) =>
597 Effect.sync(() => {
598 nativeEventLogger.write({
599 observedAt: loggedAt,
600 stream: "native",
601 orchestrationThreadId: orchestrationThreadId ?? null,
602 event,
603 });
604 }),
605 );
606 void Effect.runPromise(writeLogEffect);
607 };
608
609 const listener = (event: ProviderEvent) => {
610 writeNativeEvent(event);
611 Queue.offerAllUnsafe(runtimeEventQueue, mapToRuntimeEvents(event));
612 };
590 Effect.gen(function* () {
591 const writeNativeEvent = (event: ProviderEvent) =>
592 Effect.gen(function* () {
593 if (!nativeEventLogger) {
594 return;
595 }
596 const orchestrationThreadId = yield* Effect.catch(
597 directory.getThreadId(event.sessionId).pipe(
598 Effect.map((threadIdOption) =>
599 Option.isSome(threadIdOption) ? threadIdOption.value : null,
600 ),
601 ),
602 () => Effect.succeed(null),
603 );
604 yield* nativeEventLogger.write(event, orchestrationThreadId);
605 });
606
607 const services = yield* Effect.services<never>();
608 const listener = (event: ProviderEvent) =>
609 Effect.gen(function* () {
610 yield* writeNativeEvent(event);
611 yield* Queue.offerAll(runtimeEventQueue, mapToRuntimeEvents(event));
612 }).pipe(Effect.runPromiseWith(services));
613613 manager.on("event", listener);
614614 return listener;
615615 }),
apps/server/src/provider/Layers/EventNdjsonLogger.test.ts+142-101
......@@ -2,116 +2,157 @@ import fs from "node:fs";
22import os from "node:os";
33import path from "node:path";
44
5import { describe, expect, it } from "vitest";
5import { ThreadId } from "@t3tools/contracts";
6import { assert, describe, it } from "@effect/vitest";
7import { Effect } from "effect";
68
79import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
810
11function parseLogLine(line: string) {
12 const match = /^\[([^\]]+)\] ([A-Z]+): (.+)$/.exec(line);
13 assert.notEqual(match, null);
14 if (!match) {
15 throw new Error(`invalid log line: ${line}`);
16 }
17 const observedAt = match[1];
18 const stream = match[2];
19 const payload = match[3];
20 if (!observedAt || !stream || payload === undefined) {
21 throw new Error(`invalid log line: ${line}`);
22 }
23 return {
24 observedAt,
25 stream,
26 payload,
27 };
28}
29
930describe("EventNdjsonLogger", () => {
10 it("writes effect-style lines to thread-scoped files", () => {
11 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
12 const basePath = path.join(tempDir, "provider-native.ndjson");
13
14 try {
15 const logger = makeEventNdjsonLogger(basePath);
16 expect(logger).toBeDefined();
17 if (!logger) {
18 return;
31 it.effect("writes effect-style lines to thread-scoped files", () =>
32 Effect.gen(function* () {
33 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
34 const basePath = path.join(tempDir, "provider-native.ndjson");
35
36 try {
37 const logger = yield* makeEventNdjsonLogger(basePath, { stream: "native" });
38 assert.notEqual(logger, undefined);
39 if (!logger) {
40 return;
41 }
42
43 yield* logger.write(
44 { threadId: "provider-thread-1", id: "evt-1" },
45 ThreadId.makeUnsafe("thread-1"),
46 );
47 yield* logger.write(
48 { type: "turn.completed", threadId: "provider-thread-2", id: "evt-2" },
49 ThreadId.makeUnsafe("thread-2"),
50 );
51 yield* logger.close();
52
53 const threadOnePath = path.join(tempDir, "thread-1.log");
54 const threadTwoPath = path.join(tempDir, "thread-2.log");
55 assert.equal(fs.existsSync(threadOnePath), true);
56 assert.equal(fs.existsSync(threadTwoPath), true);
57
58 const first = parseLogLine(fs.readFileSync(threadOnePath, "utf8").trim());
59 const second = parseLogLine(fs.readFileSync(threadTwoPath, "utf8").trim());
60
61 assert.equal(Number.isNaN(Date.parse(first.observedAt)), false);
62 assert.equal(first.stream, "NATIVE");
63 assert.equal(first.payload, '{"threadId":"provider-thread-1","id":"evt-1"}');
64
65 assert.equal(Number.isNaN(Date.parse(second.observedAt)), false);
66 assert.equal(second.stream, "NATIVE");
67 assert.equal(
68 second.payload,
69 '{"type":"turn.completed","threadId":"provider-thread-2","id":"evt-2"}',
70 );
71 } finally {
72 fs.rmSync(tempDir, { recursive: true, force: true });
1973 }
20
21 logger.write({
22 observedAt: "2026-03-03T12:00:00.000Z",
23 stream: "native",
24 orchestrationThreadId: "thread-1",
25 event: { threadId: "provider-thread-1", id: "evt-1" },
26 });
27 logger.write({
28 observedAt: "2026-03-03T12:00:01.000Z",
29 stream: "canonical",
30 orchestrationThreadId: "thread-2",
31 event: { type: "turn.completed", threadId: "provider-thread-2", id: "evt-2" },
32 });
33 logger.close();
34
35 const threadOnePath = path.join(tempDir, "thread-1.log");
36 const threadTwoPath = path.join(tempDir, "thread-2.log");
37 expect(fs.existsSync(threadOnePath)).toBe(true);
38 expect(fs.existsSync(threadTwoPath)).toBe(true);
39
40 const first = fs.readFileSync(threadOnePath, "utf8").trim();
41 const second = fs.readFileSync(threadTwoPath, "utf8").trim();
42 expect(first).toBe(
43 '[2026-03-03T12:00:00.000Z] NATIVE: {"threadId":"provider-thread-1","id":"evt-1"}',
44 );
45 expect(second).toBe(
46 '[2026-03-03T12:00:01.000Z] ORCHESTRATION: {"type":"turn.completed","threadId":"provider-thread-2","id":"evt-2"}',
47 );
48 } finally {
49 fs.rmSync(tempDir, { recursive: true, force: true });
50 }
51 });
52
53 it("drops records without orchestration thread ids", () => {
54 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
55 const basePath = path.join(tempDir, "provider-canonical.ndjson");
56
57 try {
58 const logger = makeEventNdjsonLogger(basePath);
59 expect(logger).toBeDefined();
60 if (!logger) {
61 return;
74 }),
75 );
76
77 it.effect("falls back to a global segment when orchestration thread id is missing or invalid", () =>
78 Effect.gen(function* () {
79 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
80 const basePath = path.join(tempDir, "provider-canonical.ndjson");
81
82 try {
83 const logger = yield* makeEventNdjsonLogger(basePath, { stream: "orchestration" });
84 assert.notEqual(logger, undefined);
85 if (!logger) {
86 return;
87 }
88
89 yield* logger.write({ id: "evt-no-thread" }, null);
90 yield* logger.write({ id: "evt-invalid-thread" }, "!!!" as unknown as ThreadId);
91 yield* logger.close();
92
93 const globalPath = path.join(tempDir, "_global.log");
94 assert.equal(fs.existsSync(globalPath), true);
95 const lines = fs
96 .readFileSync(globalPath, "utf8")
97 .trim()
98 .split("\n")
99 .map((line) => parseLogLine(line));
100 assert.equal(lines.length, 2);
101 assert.equal(Number.isNaN(Date.parse(lines[0]?.observedAt ?? "")), false);
102 assert.equal(Number.isNaN(Date.parse(lines[1]?.observedAt ?? "")), false);
103 assert.equal(lines[0]?.stream, "ORCHESTRATION");
104 assert.equal(lines[0]?.payload, '{"id":"evt-no-thread"}');
105 assert.equal(lines[1]?.stream, "ORCHESTRATION");
106 assert.equal(lines[1]?.payload, '{"id":"evt-invalid-thread"}');
107 } finally {
108 fs.rmSync(tempDir, { recursive: true, force: true });
62109 }
110 }),
111 );
63112
64 logger.write({ stream: "orchestration", event: { id: "evt-no-thread" } });
65 logger.close();
66
67 const logFiles = fs.readdirSync(tempDir).filter((entry) => entry.endsWith(".log"));
68 expect(logFiles).toEqual([]);
69 } finally {
70 fs.rmSync(tempDir, { recursive: true, force: true });
71 }
72 });
73
74 it("rotates per-thread files when max size is exceeded", () => {
75 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
76 const basePath = path.join(tempDir, "provider-native.ndjson");
77
78 try {
79 const logger = makeEventNdjsonLogger(basePath, {
80 maxBytes: 120,
81 maxFiles: 2,
82 });
83 expect(logger).toBeDefined();
84 if (!logger) {
85 return;
86 }
113 it.effect("rotates per-thread files when max size is exceeded", () =>
114 Effect.gen(function* () {
115 const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-"));
116 const basePath = path.join(tempDir, "provider-native.ndjson");
87117
88 for (let index = 0; index < 10; index += 1) {
89 logger.write({
118 try {
119 const logger = yield* makeEventNdjsonLogger(basePath, {
90120 stream: "native",
91 orchestrationThreadId: "thread-rotate",
92 event: {
93 threadId: "provider-thread-rotate",
94 id: `evt-${index}`,
95 payload: "x".repeat(40),
96 },
121 maxBytes: 120,
122 maxFiles: 2,
97123 });
124 assert.notEqual(logger, undefined);
125 if (!logger) {
126 return;
127 }
128
129 for (let index = 0; index < 10; index += 1) {
130 yield* logger.write(
131 {
132 threadId: "provider-thread-rotate",
133 id: `evt-${index}`,
134 payload: "x".repeat(40),
135 },
136 ThreadId.makeUnsafe("thread-rotate"),
137 );
138 }
139 yield* logger.close();
140
141 const fileStem = "thread-rotate.log";
142 const matchingFiles = fs
143 .readdirSync(tempDir)
144 .filter((entry) => entry === fileStem || entry.startsWith(`${fileStem}.`))
145 .toSorted();
146
147 assert.equal(matchingFiles.some((entry) => entry === `${fileStem}.1`), true);
148 assert.equal(
149 matchingFiles.some((entry) => entry === fileStem || entry === `${fileStem}.2`),
150 true,
151 );
152 assert.equal(matchingFiles.some((entry) => entry === `${fileStem}.3`), false);
153 } finally {
154 fs.rmSync(tempDir, { recursive: true, force: true });
98155 }
99 logger.close();
100
101 const fileStem = "thread-rotate.log";
102 const matchingFiles = fs
103 .readdirSync(tempDir)
104 .filter((entry) => entry === fileStem || entry.startsWith(`${fileStem}.`))
105 .toSorted();
106
107 expect(matchingFiles.some((entry) => entry === `${fileStem}.1`)).toBe(true);
108 expect(matchingFiles.some((entry) => entry === fileStem || entry === `${fileStem}.2`)).toBe(
109 true,
110 );
111 expect(matchingFiles.some((entry) => entry === `${fileStem}.3`)).toBe(false);
112 } finally {
113 fs.rmSync(tempDir, { recursive: true, force: true });
114 }
115 });
116
156 }),
157 );
117158});
apps/server/src/provider/Layers/EventNdjsonLogger.ts+199-127
......@@ -8,178 +8,250 @@
88import fs from "node:fs";
99import path from "node:path";
1010
11import type { ThreadId } from "@t3tools/contracts";
1112import { RotatingFileSink } from "@t3tools/shared/logging";
13import { Effect, Exit, Logger, Scope } from "effect";
14
1215import { toSafeThreadAttachmentSegment } from "../../attachmentStore.ts";
13import { createLogger } from "../../logger.ts";
1416
15const logger = createLogger("provider-observability");
1617const DEFAULT_MAX_BYTES = 10 * 1024 * 1024;
1718const DEFAULT_MAX_FILES = 10;
19const DEFAULT_BATCH_WINDOW_MS = 200;
20const GLOBAL_THREAD_SEGMENT = "_global";
21const LOG_SCOPE = "provider-observability";
22
23export type EventNdjsonStream = "native" | "canonical" | "orchestration";
1824
1925export interface EventNdjsonLogger {
2026 readonly filePath: string;
21 write: (record: unknown) => void;
22 close: () => void;
27 write: (event: unknown, threadId: ThreadId | null) => Effect.Effect<void>;
28 close: () => Effect.Effect<void>;
2329}
2430
2531export interface EventNdjsonLoggerOptions {
32 readonly stream: EventNdjsonStream;
2633 readonly maxBytes?: number;
2734 readonly maxFiles?: number;
35 readonly batchWindowMs?: number;
2836}
2937
30function stringifyRecord(record: unknown): string | undefined {
31 try {
32 return JSON.stringify(record);
33 } catch (error) {
34 logger.warn("failed to serialize provider event log record", { error });
35 return undefined;
36 }
38interface ThreadWriter {
39 writeMessage: (message: string) => Effect.Effect<void>;
40 close: () => Effect.Effect<void>;
3741}
3842
39function normalizeThreadSegment(raw: unknown): string | null {
40 if (typeof raw !== "string") {
41 return null;
42 }
43 return toSafeThreadAttachmentSegment(raw);
43function logWarning(message: string, context: Record<string, unknown>): Effect.Effect<void> {
44 return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE }));
4445}
4546
46function asObject(value: unknown): Record<string, unknown> | undefined {
47 if (!value || typeof value !== "object" || Array.isArray(value)) {
48 return undefined;
49 }
50 return value as Record<string, unknown>;
47function resolveThreadSegment(raw: string | null | undefined): string {
48 const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null;
49 return normalized ?? GLOBAL_THREAD_SEGMENT;
5150}
5251
53function resolveThreadSegment(record: unknown): string | null {
54 const root = asObject(record);
55 return normalizeThreadSegment(root?.orchestrationThreadId);
52function formatLoggerMessage(message: unknown): string {
53 if (Array.isArray(message)) {
54 return message.map((part) => (typeof part === "string" ? part : String(part))).join(" ");
55 }
56 return typeof message === "string" ? message : String(message);
5657}
5758
58function resolveThreadFilePath(baseFilePath: string, threadSegment: string): string {
59 const dir = path.dirname(baseFilePath);
60 return path.join(dir, `${threadSegment}.log`);
59function makeLineLogger(streamLabel: string): Logger.Logger<unknown, string> {
60 return Logger.make(
61 ({ date, message }) =>
62 `[${date.toISOString()}] ${streamLabel}: ${formatLoggerMessage(message)}\n`,
63 );
6164}
6265
63function resolveObservedAt(raw: unknown): string {
64 if (typeof raw !== "string") {
65 return new Date().toISOString();
66 }
67 const trimmed = raw.trim();
68 if (trimmed.length === 0) {
69 return new Date().toISOString();
66function resolveStreamLabel(stream: EventNdjsonStream): string {
67 switch (stream) {
68 case "native":
69 return "NTIVE";
70 case "canonical":
71 case "orchestration":
72 default:
73 return "CANON";
7074 }
71 return trimmed;
7275}
7376
74function resolveStreamLabel(raw: unknown): string {
75 if (typeof raw !== "string") {
76 return "ORCHESTRATION";
77 }
78 const normalized = raw.trim().toLowerCase();
79 if (normalized === "native") {
80 return "NATIVE";
81 }
82 if (normalized === "canonical" || normalized === "orchestration") {
83 return "ORCHESTRATION";
84 }
85 if (normalized.length === 0) {
86 return "ORCHESTRATION";
87 }
88 return normalized.toUpperCase();
77function toLogMessage(event: unknown): Effect.Effect<string | undefined> {
78 return Effect.gen(function* () {
79 const serialized = yield* Effect.sync(() => {
80 try {
81 return { ok: true as const, value: JSON.stringify(event) };
82 } catch (error) {
83 return { ok: false as const, error };
84 }
85 });
86
87 if (!serialized.ok) {
88 yield* logWarning("failed to serialize provider event log record", {
89 error: serialized.error,
90 });
91 return undefined;
92 }
93
94 if (typeof serialized.value !== "string") {
95 return undefined;
96 }
97
98 return serialized.value;
99 });
89100}
90101
91function toLogLine(record: unknown): string | undefined {
92 const root = asObject(record);
93 const payload = root && "event" in root ? root.event : record;
94 const serializedPayload = stringifyRecord(payload);
95 if (!serializedPayload) {
96 return undefined;
97 }
98 const observedAt = resolveObservedAt(root?.observedAt);
99 const streamLabel = resolveStreamLabel(root?.stream);
100 return `[${observedAt}] ${streamLabel}: ${serializedPayload}\n`;
102function makeThreadWriter(input: {
103 readonly filePath: string;
104 readonly maxBytes: number;
105 readonly maxFiles: number;
106 readonly batchWindowMs: number;
107 readonly streamLabel: string;
108}): Effect.Effect<ThreadWriter | undefined> {
109 return Effect.gen(function* () {
110 const sinkResult = yield* Effect.sync(() => {
111 try {
112 return {
113 ok: true as const,
114 sink: new RotatingFileSink({
115 filePath: input.filePath,
116 maxBytes: input.maxBytes,
117 maxFiles: input.maxFiles,
118 throwOnError: true,
119 }),
120 };
121 } catch (error) {
122 return { ok: false as const, error };
123 }
124 });
125
126 if (!sinkResult.ok) {
127 yield* logWarning("failed to initialize provider thread log file", {
128 filePath: input.filePath,
129 error: sinkResult.error,
130 });
131 return undefined;
132 }
133
134 const sink = sinkResult.sink;
135 const scope = yield* Scope.make();
136 const lineLogger = makeLineLogger(input.streamLabel);
137 const batchedLogger = yield* Logger.batched(lineLogger, {
138 window: input.batchWindowMs,
139 flush: (messages) =>
140 Effect.gen(function* () {
141 const flushResult = yield* Effect.sync(() => {
142 try {
143 for (const message of messages) {
144 sink.write(message);
145 }
146 return { ok: true as const };
147 } catch (error) {
148 return { ok: false as const, error };
149 }
150 });
151
152 if (!flushResult.ok) {
153 yield* logWarning("provider event log batch flush failed", {
154 filePath: input.filePath,
155 error: flushResult.error,
156 });
157 }
158 }),
159 }).pipe(Effect.provideService(Scope.Scope, scope));
160
161 const loggerLayer = Logger.layer([batchedLogger], { mergeWithExisting: false });
162
163 return {
164 writeMessage(message: string) {
165 return Effect.log(message).pipe(Effect.provide(loggerLayer));
166 },
167 close() {
168 return Scope.close(scope, Exit.void);
169 },
170 } satisfies ThreadWriter;
171 });
101172}
102173
103174export function makeEventNdjsonLogger(
104175 filePath: string,
105 options?: EventNdjsonLoggerOptions,
106): EventNdjsonLogger | undefined {
107 const maxBytes = options?.maxBytes ?? DEFAULT_MAX_BYTES;
108 const maxFiles = options?.maxFiles ?? DEFAULT_MAX_FILES;
109
110 try {
111 fs.mkdirSync(path.dirname(filePath), { recursive: true });
112 } catch (error) {
113 logger.warn("failed to create provider event log directory", { filePath, error });
114 return undefined;
115 }
176 options: EventNdjsonLoggerOptions,
177): Effect.Effect<EventNdjsonLogger | undefined> {
178 return Effect.gen(function* () {
179 const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES;
180 const maxFiles = options.maxFiles ?? DEFAULT_MAX_FILES;
181 const batchWindowMs = options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS;
182 const streamLabel = resolveStreamLabel(options.stream);
183
184 const directoryReady = yield* Effect.sync(() => {
185 try {
186 fs.mkdirSync(path.dirname(filePath), { recursive: true });
187 return true;
188 } catch (error) {
189 return { ok: false as const, error };
190 }
191 });
192 if (directoryReady !== true) {
193 yield* logWarning("failed to create provider event log directory", {
194 filePath,
195 error: directoryReady.error,
196 });
197 return undefined;
198 }
116199
117 try {
118 let broken = false;
119 const sinks = new Map<string, RotatingFileSink>();
200 const threadWriters = new Map<string, ThreadWriter>();
120201 const failedSegments = new Set<string>();
121202
122 const resolveSink = (threadSegment: string): RotatingFileSink | undefined => {
123 if (failedSegments.has(threadSegment)) {
124 return undefined;
125 }
126 const existingSink = sinks.get(threadSegment);
127 if (existingSink) {
128 return existingSink;
129 }
203 const resolveThreadWriter = (threadSegment: string): Effect.Effect<ThreadWriter | undefined> =>
204 Effect.gen(function* () {
205 if (failedSegments.has(threadSegment)) {
206 return undefined;
207 }
208 const existing = threadWriters.get(threadSegment);
209 if (existing) {
210 return existing;
211 }
130212
131 const threadFilePath = resolveThreadFilePath(filePath, threadSegment);
132 try {
133 const sink = new RotatingFileSink({
134 filePath: threadFilePath,
213 const writer = yield* makeThreadWriter({
214 filePath: path.join(path.dirname(filePath), `${threadSegment}.log`),
135215 maxBytes,
136216 maxFiles,
137 throwOnError: true,
138 });
139 sinks.set(threadSegment, sink);
140 return sink;
141 } catch (error) {
142 failedSegments.add(threadSegment);
143 logger.warn("failed to initialize provider thread log file", {
144 filePath: threadFilePath,
145 threadSegment,
146 error,
217 batchWindowMs,
218 streamLabel,
147219 });
148 return undefined;
149 }
150 };
220 if (!writer) {
221 failedSegments.add(threadSegment);
222 return undefined;
223 }
224
225 threadWriters.set(threadSegment, writer);
226 return writer;
227 });
151228
152229 return {
153230 filePath,
154 write(record: unknown) {
155 if (broken) {
156 return;
157 }
158 const threadSegment = resolveThreadSegment(record);
159 if (!threadSegment) {
160 return;
161 }
162 const line = toLogLine(record);
163 if (!line) {
164 return;
165 }
166 const sink = resolveSink(threadSegment);
167 if (!sink) {
168 return;
169 }
170 try {
171 sink.write(line);
172 } catch (error) {
173 broken = true;
174 logger.warn("provider event log write failed", { filePath, threadSegment, error });
175 }
231 write(event: unknown, threadId: ThreadId | null) {
232 return Effect.gen(function* () {
233 const threadSegment = resolveThreadSegment(threadId);
234 const message = yield* toLogMessage(event);
235 if (!message) {
236 return;
237 }
238
239 const writer = yield* resolveThreadWriter(threadSegment);
240 if (!writer) {
241 return;
242 }
243
244 yield* writer.writeMessage(message);
245 });
176246 },
177247 close() {
178 sinks.clear();
248 return Effect.gen(function* () {
249 for (const writer of threadWriters.values()) {
250 yield* writer.close();
251 }
252 threadWriters.clear();
253 });
179254 },
180 };
181 } catch (error) {
182 logger.warn("failed to create provider event log file", { filePath, error });
183 return undefined;
184 }
255 } satisfies EventNdjsonLogger;
256 });
185257}
apps/server/src/provider/Layers/ProviderService.ts+9-12
......@@ -111,7 +111,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
111111 const canonicalEventLogger =
112112 options?.canonicalEventLogger ??
113113 (options?.canonicalEventLogPath !== undefined
114 ? makeEventNdjsonLogger(options.canonicalEventLogPath)
114 ? yield* makeEventNdjsonLogger(options.canonicalEventLogPath, {
115 stream: "canonical",
116 })
115117 : undefined);
116118
117119 const registry = yield* ProviderAdapterRegistry;
......@@ -143,22 +145,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
143145 canonicalizeRuntimeEventSession(event).pipe(
144146 Effect.tap((canonicalEvent) =>
145147 canonicalEventLogger
146 ? Effect.tap(
148 ? Effect.flatMap(
147149 Effect.catch(
148150 directory.getThreadId(canonicalEvent.sessionId).pipe(
149 Effect.map(Option.getOrUndefined),
151 Effect.map((threadIdOption) =>
152 Option.isSome(threadIdOption) ? threadIdOption.value : null,
153 ),
150154 ),
151 () => Effect.void,
155 () => Effect.succeed<ThreadId | null>(null),
152156 ),
153157 (orchestrationThreadId) =>
154 Effect.sync(() => {
155 canonicalEventLogger.write({
156 observedAt: new Date().toISOString(),
157 stream: "canonical",
158 orchestrationThreadId: orchestrationThreadId ?? null,
159 event: canonicalEvent,
160 });
161 }),
158 canonicalEventLogger.write(canonicalEvent, orchestrationThreadId),
162159 )
163160 : Effect.void,
164161 ),
apps/server/src/serverLayers.ts+9-3
......@@ -43,19 +43,25 @@ export function makeServerProviderLayer(): Layer.Layer<
4343 return Effect.gen(function* () {
4444 const { stateDir } = yield* ServerConfig;
4545 const providerLogsDir = path.join(stateDir, "logs", "provider");
46 const providerEventLogger = makeEventNdjsonLogger(path.join(providerLogsDir, "events.log"));
46 const providerEventLogPath = path.join(providerLogsDir, "events.log");
47 const nativeEventLogger = yield* makeEventNdjsonLogger(providerEventLogPath, {
48 stream: "native",
49 });
50 const canonicalEventLogger = yield* makeEventNdjsonLogger(providerEventLogPath, {
51 stream: "canonical",
52 });
4753 const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(
4854 Layer.provide(ProviderSessionRuntimeRepositoryLive),
4955 );
5056 const codexAdapterLayer = makeCodexAdapterLive(
51 providerEventLogger ? { nativeEventLogger: providerEventLogger } : undefined,
57 nativeEventLogger ? { nativeEventLogger } : undefined,
5258 );
5359 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(
5460 Layer.provide(codexAdapterLayer),
5561 Layer.provideMerge(providerSessionDirectoryLayer),
5662 );
5763 return makeProviderServiceLive(
58 providerEventLogger ? { canonicalEventLogger: providerEventLogger } : undefined,
64 canonicalEventLogger ? { canonicalEventLogger } : undefined,
5965 ).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer));
6066 }).pipe(Layer.unwrap);
6167}