From 86e87a8789196663e5e919466676566dadb32f20 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Mar 2026 18:14:45 -0800 Subject: [PATCH] better lables --- .../src/provider/Layers/CodexAdapter.ts | 60 ++-- .../provider/Layers/EventNdjsonLogger.test.ts | 225 +++++++----- .../src/provider/Layers/EventNdjsonLogger.ts | 324 +++++++++++------- .../src/provider/Layers/ProviderService.ts | 21 +- apps/server/src/serverLayers.ts | 12 +- 5 files changed, 379 insertions(+), 263 deletions(-) diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 89f65e3b210a8aa3258f55847b4714b97d287056..8bec7388932d6ba7fa31869304926a7277e39f69 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -436,7 +436,9 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => const nativeEventLogger = options?.nativeEventLogger ?? (options?.nativeEventLogPath !== undefined - ? makeEventNdjsonLogger(options.nativeEventLogPath) + ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { + stream: "native", + }) : undefined); const manager = yield* Effect.acquireRelease( @@ -499,9 +501,11 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => new Error(`Invalid attachment id '${attachment.id}'.`), ); } - const bytes = yield* fileSystem.readFile(attachmentPath).pipe( - Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause)), - ); + const bytes = yield* fileSystem + .readFile(attachmentPath) + .pipe( + Effect.mapError((cause) => toRequestError(input.sessionId, "turn/start", cause)), + ); return { type: "image" as const, url: `data:${attachment.mimeType};base64,${Buffer.from(bytes).toString("base64")}`, @@ -583,33 +587,29 @@ const makeCodexAdapter = (options?: CodexAdapterLiveOptions) => const runtimeEventQueue = yield* Queue.unbounded(); yield* Effect.acquireRelease( - Effect.sync(() => { - const writeNativeEvent = (event: ProviderEvent) => { - if (!nativeEventLogger) { - return; - } - const loggedAt = new Date().toISOString(); - const resolveOrchestrationThreadId = Effect.catch( - directory.getThreadId(event.sessionId).pipe(Effect.map(Option.getOrUndefined)), - () => Effect.void, - ); - const writeLogEffect = Effect.tap(resolveOrchestrationThreadId, (orchestrationThreadId) => - Effect.sync(() => { - nativeEventLogger.write({ - observedAt: loggedAt, - stream: "native", - orchestrationThreadId: orchestrationThreadId ?? null, - event, - }); - }), - ); - void Effect.runPromise(writeLogEffect); - }; + Effect.gen(function* () { + const writeNativeEvent = (event: ProviderEvent) => + Effect.gen(function* () { + if (!nativeEventLogger) { + return; + } + const orchestrationThreadId = yield* Effect.catch( + directory.getThreadId(event.sessionId).pipe( + Effect.map((threadIdOption) => + Option.isSome(threadIdOption) ? threadIdOption.value : null, + ), + ), + () => Effect.succeed(null), + ); + yield* nativeEventLogger.write(event, orchestrationThreadId); + }); - const listener = (event: ProviderEvent) => { - writeNativeEvent(event); - Queue.offerAllUnsafe(runtimeEventQueue, mapToRuntimeEvents(event)); - }; + const services = yield* Effect.services(); + const listener = (event: ProviderEvent) => + Effect.gen(function* () { + yield* writeNativeEvent(event); + yield* Queue.offerAll(runtimeEventQueue, mapToRuntimeEvents(event)); + }).pipe(Effect.runPromiseWith(services)); manager.on("event", listener); return listener; }), diff --git a/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts b/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts index f2ef4dcdc939c1db5e59a51d1b0b9775b7ecaffd..28894a5405202780812806e8f29ec8a404731b43 100644 --- a/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts +++ b/apps/server/src/provider/Layers/EventNdjsonLogger.test.ts @@ -2,116 +2,157 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { describe, expect, it } from "vitest"; +import { ThreadId } from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; +import { Effect } from "effect"; import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; +function parseLogLine(line: string) { + const match = /^\[([^\]]+)\] ([A-Z]+): (.+)$/.exec(line); + assert.notEqual(match, null); + if (!match) { + throw new Error(`invalid log line: ${line}`); + } + const observedAt = match[1]; + const stream = match[2]; + const payload = match[3]; + if (!observedAt || !stream || payload === undefined) { + throw new Error(`invalid log line: ${line}`); + } + return { + observedAt, + stream, + payload, + }; +} + describe("EventNdjsonLogger", () => { - it("writes effect-style lines to thread-scoped files", () => { - const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); - const basePath = path.join(tempDir, "provider-native.ndjson"); + it.effect("writes effect-style lines to thread-scoped files", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); + const basePath = path.join(tempDir, "provider-native.ndjson"); - try { - const logger = makeEventNdjsonLogger(basePath); - expect(logger).toBeDefined(); - if (!logger) { - return; - } + try { + const logger = yield* makeEventNdjsonLogger(basePath, { stream: "native" }); + assert.notEqual(logger, undefined); + if (!logger) { + return; + } - logger.write({ - observedAt: "2026-03-03T12:00:00.000Z", - stream: "native", - orchestrationThreadId: "thread-1", - event: { threadId: "provider-thread-1", id: "evt-1" }, - }); - logger.write({ - observedAt: "2026-03-03T12:00:01.000Z", - stream: "canonical", - orchestrationThreadId: "thread-2", - event: { type: "turn.completed", threadId: "provider-thread-2", id: "evt-2" }, - }); - logger.close(); + yield* logger.write( + { threadId: "provider-thread-1", id: "evt-1" }, + ThreadId.makeUnsafe("thread-1"), + ); + yield* logger.write( + { type: "turn.completed", threadId: "provider-thread-2", id: "evt-2" }, + ThreadId.makeUnsafe("thread-2"), + ); + yield* logger.close(); - const threadOnePath = path.join(tempDir, "thread-1.log"); - const threadTwoPath = path.join(tempDir, "thread-2.log"); - expect(fs.existsSync(threadOnePath)).toBe(true); - expect(fs.existsSync(threadTwoPath)).toBe(true); + const threadOnePath = path.join(tempDir, "thread-1.log"); + const threadTwoPath = path.join(tempDir, "thread-2.log"); + assert.equal(fs.existsSync(threadOnePath), true); + assert.equal(fs.existsSync(threadTwoPath), true); - const first = fs.readFileSync(threadOnePath, "utf8").trim(); - const second = fs.readFileSync(threadTwoPath, "utf8").trim(); - expect(first).toBe( - '[2026-03-03T12:00:00.000Z] NATIVE: {"threadId":"provider-thread-1","id":"evt-1"}', - ); - expect(second).toBe( - '[2026-03-03T12:00:01.000Z] ORCHESTRATION: {"type":"turn.completed","threadId":"provider-thread-2","id":"evt-2"}', - ); - } finally { - fs.rmSync(tempDir, { recursive: true, force: true }); - } - }); + const first = parseLogLine(fs.readFileSync(threadOnePath, "utf8").trim()); + const second = parseLogLine(fs.readFileSync(threadTwoPath, "utf8").trim()); - it("drops records without orchestration thread ids", () => { - const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); - const basePath = path.join(tempDir, "provider-canonical.ndjson"); + assert.equal(Number.isNaN(Date.parse(first.observedAt)), false); + assert.equal(first.stream, "NATIVE"); + assert.equal(first.payload, '{"threadId":"provider-thread-1","id":"evt-1"}'); - try { - const logger = makeEventNdjsonLogger(basePath); - expect(logger).toBeDefined(); - if (!logger) { - return; + assert.equal(Number.isNaN(Date.parse(second.observedAt)), false); + assert.equal(second.stream, "NATIVE"); + assert.equal( + second.payload, + '{"type":"turn.completed","threadId":"provider-thread-2","id":"evt-2"}', + ); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); } + }), + ); - logger.write({ stream: "orchestration", event: { id: "evt-no-thread" } }); - logger.close(); + it.effect("falls back to a global segment when orchestration thread id is missing or invalid", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); + const basePath = path.join(tempDir, "provider-canonical.ndjson"); - const logFiles = fs.readdirSync(tempDir).filter((entry) => entry.endsWith(".log")); - expect(logFiles).toEqual([]); - } finally { - fs.rmSync(tempDir, { recursive: true, force: true }); - } - }); + try { + const logger = yield* makeEventNdjsonLogger(basePath, { stream: "orchestration" }); + assert.notEqual(logger, undefined); + if (!logger) { + return; + } - it("rotates per-thread files when max size is exceeded", () => { - const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); - const basePath = path.join(tempDir, "provider-native.ndjson"); + yield* logger.write({ id: "evt-no-thread" }, null); + yield* logger.write({ id: "evt-invalid-thread" }, "!!!" as unknown as ThreadId); + yield* logger.close(); - try { - const logger = makeEventNdjsonLogger(basePath, { - maxBytes: 120, - maxFiles: 2, - }); - expect(logger).toBeDefined(); - if (!logger) { - return; + const globalPath = path.join(tempDir, "_global.log"); + assert.equal(fs.existsSync(globalPath), true); + const lines = fs + .readFileSync(globalPath, "utf8") + .trim() + .split("\n") + .map((line) => parseLogLine(line)); + assert.equal(lines.length, 2); + assert.equal(Number.isNaN(Date.parse(lines[0]?.observedAt ?? "")), false); + assert.equal(Number.isNaN(Date.parse(lines[1]?.observedAt ?? "")), false); + assert.equal(lines[0]?.stream, "ORCHESTRATION"); + assert.equal(lines[0]?.payload, '{"id":"evt-no-thread"}'); + assert.equal(lines[1]?.stream, "ORCHESTRATION"); + assert.equal(lines[1]?.payload, '{"id":"evt-invalid-thread"}'); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); } + }), + ); + + it.effect("rotates per-thread files when max size is exceeded", () => + Effect.gen(function* () { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-log-")); + const basePath = path.join(tempDir, "provider-native.ndjson"); - for (let index = 0; index < 10; index += 1) { - logger.write({ + try { + const logger = yield* makeEventNdjsonLogger(basePath, { stream: "native", - orchestrationThreadId: "thread-rotate", - event: { - threadId: "provider-thread-rotate", - id: `evt-${index}`, - payload: "x".repeat(40), - }, + maxBytes: 120, + maxFiles: 2, }); + assert.notEqual(logger, undefined); + if (!logger) { + return; + } + + for (let index = 0; index < 10; index += 1) { + yield* logger.write( + { + threadId: "provider-thread-rotate", + id: `evt-${index}`, + payload: "x".repeat(40), + }, + ThreadId.makeUnsafe("thread-rotate"), + ); + } + yield* logger.close(); + + const fileStem = "thread-rotate.log"; + const matchingFiles = fs + .readdirSync(tempDir) + .filter((entry) => entry === fileStem || entry.startsWith(`${fileStem}.`)) + .toSorted(); + + assert.equal(matchingFiles.some((entry) => entry === `${fileStem}.1`), true); + assert.equal( + matchingFiles.some((entry) => entry === fileStem || entry === `${fileStem}.2`), + true, + ); + assert.equal(matchingFiles.some((entry) => entry === `${fileStem}.3`), false); + } finally { + fs.rmSync(tempDir, { recursive: true, force: true }); } - logger.close(); - - const fileStem = "thread-rotate.log"; - const matchingFiles = fs - .readdirSync(tempDir) - .filter((entry) => entry === fileStem || entry.startsWith(`${fileStem}.`)) - .toSorted(); - - expect(matchingFiles.some((entry) => entry === `${fileStem}.1`)).toBe(true); - expect(matchingFiles.some((entry) => entry === fileStem || entry === `${fileStem}.2`)).toBe( - true, - ); - expect(matchingFiles.some((entry) => entry === `${fileStem}.3`)).toBe(false); - } finally { - fs.rmSync(tempDir, { recursive: true, force: true }); - } - }); - + }), + ); }); diff --git a/apps/server/src/provider/Layers/EventNdjsonLogger.ts b/apps/server/src/provider/Layers/EventNdjsonLogger.ts index 42dac0f9cbfd7ce2a69186f12b77fa68fe21231c..28b7ecf6117f2c948e5ab57def53616a9a4f6231 100644 --- a/apps/server/src/provider/Layers/EventNdjsonLogger.ts +++ b/apps/server/src/provider/Layers/EventNdjsonLogger.ts @@ -8,178 +8,250 @@ import fs from "node:fs"; import path from "node:path"; +import type { ThreadId } from "@t3tools/contracts"; import { RotatingFileSink } from "@t3tools/shared/logging"; +import { Effect, Exit, Logger, Scope } from "effect"; + import { toSafeThreadAttachmentSegment } from "../../attachmentStore.ts"; -import { createLogger } from "../../logger.ts"; -const logger = createLogger("provider-observability"); const DEFAULT_MAX_BYTES = 10 * 1024 * 1024; const DEFAULT_MAX_FILES = 10; +const DEFAULT_BATCH_WINDOW_MS = 200; +const GLOBAL_THREAD_SEGMENT = "_global"; +const LOG_SCOPE = "provider-observability"; + +export type EventNdjsonStream = "native" | "canonical" | "orchestration"; export interface EventNdjsonLogger { readonly filePath: string; - write: (record: unknown) => void; - close: () => void; + write: (event: unknown, threadId: ThreadId | null) => Effect.Effect; + close: () => Effect.Effect; } export interface EventNdjsonLoggerOptions { + readonly stream: EventNdjsonStream; readonly maxBytes?: number; readonly maxFiles?: number; + readonly batchWindowMs?: number; } -function stringifyRecord(record: unknown): string | undefined { - try { - return JSON.stringify(record); - } catch (error) { - logger.warn("failed to serialize provider event log record", { error }); - return undefined; - } +interface ThreadWriter { + writeMessage: (message: string) => Effect.Effect; + close: () => Effect.Effect; } -function normalizeThreadSegment(raw: unknown): string | null { - if (typeof raw !== "string") { - return null; - } - return toSafeThreadAttachmentSegment(raw); +function logWarning(message: string, context: Record): Effect.Effect { + return Effect.logWarning(message, context).pipe(Effect.annotateLogs({ scope: LOG_SCOPE })); } -function asObject(value: unknown): Record | undefined { - if (!value || typeof value !== "object" || Array.isArray(value)) { - return undefined; - } - return value as Record; +function resolveThreadSegment(raw: string | null | undefined): string { + const normalized = typeof raw === "string" ? toSafeThreadAttachmentSegment(raw) : null; + return normalized ?? GLOBAL_THREAD_SEGMENT; } -function resolveThreadSegment(record: unknown): string | null { - const root = asObject(record); - return normalizeThreadSegment(root?.orchestrationThreadId); +function formatLoggerMessage(message: unknown): string { + if (Array.isArray(message)) { + return message.map((part) => (typeof part === "string" ? part : String(part))).join(" "); + } + return typeof message === "string" ? message : String(message); } -function resolveThreadFilePath(baseFilePath: string, threadSegment: string): string { - const dir = path.dirname(baseFilePath); - return path.join(dir, `${threadSegment}.log`); +function makeLineLogger(streamLabel: string): Logger.Logger { + return Logger.make( + ({ date, message }) => + `[${date.toISOString()}] ${streamLabel}: ${formatLoggerMessage(message)}\n`, + ); } -function resolveObservedAt(raw: unknown): string { - if (typeof raw !== "string") { - return new Date().toISOString(); - } - const trimmed = raw.trim(); - if (trimmed.length === 0) { - return new Date().toISOString(); +function resolveStreamLabel(stream: EventNdjsonStream): string { + switch (stream) { + case "native": + return "NTIVE"; + case "canonical": + case "orchestration": + default: + return "CANON"; } - return trimmed; } -function resolveStreamLabel(raw: unknown): string { - if (typeof raw !== "string") { - return "ORCHESTRATION"; - } - const normalized = raw.trim().toLowerCase(); - if (normalized === "native") { - return "NATIVE"; - } - if (normalized === "canonical" || normalized === "orchestration") { - return "ORCHESTRATION"; - } - if (normalized.length === 0) { - return "ORCHESTRATION"; - } - return normalized.toUpperCase(); +function toLogMessage(event: unknown): Effect.Effect { + return Effect.gen(function* () { + const serialized = yield* Effect.sync(() => { + try { + return { ok: true as const, value: JSON.stringify(event) }; + } catch (error) { + return { ok: false as const, error }; + } + }); + + if (!serialized.ok) { + yield* logWarning("failed to serialize provider event log record", { + error: serialized.error, + }); + return undefined; + } + + if (typeof serialized.value !== "string") { + return undefined; + } + + return serialized.value; + }); } -function toLogLine(record: unknown): string | undefined { - const root = asObject(record); - const payload = root && "event" in root ? root.event : record; - const serializedPayload = stringifyRecord(payload); - if (!serializedPayload) { - return undefined; - } - const observedAt = resolveObservedAt(root?.observedAt); - const streamLabel = resolveStreamLabel(root?.stream); - return `[${observedAt}] ${streamLabel}: ${serializedPayload}\n`; +function makeThreadWriter(input: { + readonly filePath: string; + readonly maxBytes: number; + readonly maxFiles: number; + readonly batchWindowMs: number; + readonly streamLabel: string; +}): Effect.Effect { + return Effect.gen(function* () { + const sinkResult = yield* Effect.sync(() => { + try { + return { + ok: true as const, + sink: new RotatingFileSink({ + filePath: input.filePath, + maxBytes: input.maxBytes, + maxFiles: input.maxFiles, + throwOnError: true, + }), + }; + } catch (error) { + return { ok: false as const, error }; + } + }); + + if (!sinkResult.ok) { + yield* logWarning("failed to initialize provider thread log file", { + filePath: input.filePath, + error: sinkResult.error, + }); + return undefined; + } + + const sink = sinkResult.sink; + const scope = yield* Scope.make(); + const lineLogger = makeLineLogger(input.streamLabel); + const batchedLogger = yield* Logger.batched(lineLogger, { + window: input.batchWindowMs, + flush: (messages) => + Effect.gen(function* () { + const flushResult = yield* Effect.sync(() => { + try { + for (const message of messages) { + sink.write(message); + } + return { ok: true as const }; + } catch (error) { + return { ok: false as const, error }; + } + }); + + if (!flushResult.ok) { + yield* logWarning("provider event log batch flush failed", { + filePath: input.filePath, + error: flushResult.error, + }); + } + }), + }).pipe(Effect.provideService(Scope.Scope, scope)); + + const loggerLayer = Logger.layer([batchedLogger], { mergeWithExisting: false }); + + return { + writeMessage(message: string) { + return Effect.log(message).pipe(Effect.provide(loggerLayer)); + }, + close() { + return Scope.close(scope, Exit.void); + }, + } satisfies ThreadWriter; + }); } export function makeEventNdjsonLogger( filePath: string, - options?: EventNdjsonLoggerOptions, -): EventNdjsonLogger | undefined { - const maxBytes = options?.maxBytes ?? DEFAULT_MAX_BYTES; - const maxFiles = options?.maxFiles ?? DEFAULT_MAX_FILES; + options: EventNdjsonLoggerOptions, +): Effect.Effect { + return Effect.gen(function* () { + const maxBytes = options.maxBytes ?? DEFAULT_MAX_BYTES; + const maxFiles = options.maxFiles ?? DEFAULT_MAX_FILES; + const batchWindowMs = options.batchWindowMs ?? DEFAULT_BATCH_WINDOW_MS; + const streamLabel = resolveStreamLabel(options.stream); - try { - fs.mkdirSync(path.dirname(filePath), { recursive: true }); - } catch (error) { - logger.warn("failed to create provider event log directory", { filePath, error }); - return undefined; - } + const directoryReady = yield* Effect.sync(() => { + try { + fs.mkdirSync(path.dirname(filePath), { recursive: true }); + return true; + } catch (error) { + return { ok: false as const, error }; + } + }); + if (directoryReady !== true) { + yield* logWarning("failed to create provider event log directory", { + filePath, + error: directoryReady.error, + }); + return undefined; + } - try { - let broken = false; - const sinks = new Map(); + const threadWriters = new Map(); const failedSegments = new Set(); - const resolveSink = (threadSegment: string): RotatingFileSink | undefined => { - if (failedSegments.has(threadSegment)) { - return undefined; - } - const existingSink = sinks.get(threadSegment); - if (existingSink) { - return existingSink; - } + const resolveThreadWriter = (threadSegment: string): Effect.Effect => + Effect.gen(function* () { + if (failedSegments.has(threadSegment)) { + return undefined; + } + const existing = threadWriters.get(threadSegment); + if (existing) { + return existing; + } - const threadFilePath = resolveThreadFilePath(filePath, threadSegment); - try { - const sink = new RotatingFileSink({ - filePath: threadFilePath, + const writer = yield* makeThreadWriter({ + filePath: path.join(path.dirname(filePath), `${threadSegment}.log`), maxBytes, maxFiles, - throwOnError: true, + batchWindowMs, + streamLabel, }); - sinks.set(threadSegment, sink); - return sink; - } catch (error) { - failedSegments.add(threadSegment); - logger.warn("failed to initialize provider thread log file", { - filePath: threadFilePath, - threadSegment, - error, - }); - return undefined; - } - }; + if (!writer) { + failedSegments.add(threadSegment); + return undefined; + } + + threadWriters.set(threadSegment, writer); + return writer; + }); return { filePath, - write(record: unknown) { - if (broken) { - return; - } - const threadSegment = resolveThreadSegment(record); - if (!threadSegment) { - return; - } - const line = toLogLine(record); - if (!line) { - return; - } - const sink = resolveSink(threadSegment); - if (!sink) { - return; - } - try { - sink.write(line); - } catch (error) { - broken = true; - logger.warn("provider event log write failed", { filePath, threadSegment, error }); - } + write(event: unknown, threadId: ThreadId | null) { + return Effect.gen(function* () { + const threadSegment = resolveThreadSegment(threadId); + const message = yield* toLogMessage(event); + if (!message) { + return; + } + + const writer = yield* resolveThreadWriter(threadSegment); + if (!writer) { + return; + } + + yield* writer.writeMessage(message); + }); }, close() { - sinks.clear(); + return Effect.gen(function* () { + for (const writer of threadWriters.values()) { + yield* writer.close(); + } + threadWriters.clear(); + }); }, - }; - } catch (error) { - logger.warn("failed to create provider event log file", { filePath, error }); - return undefined; - } + } satisfies EventNdjsonLogger; + }); } diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 70972d0648137719ec85179b8bf9dab90fea6deb..bb58a7d93f1f14aee0b56189b219d133e610c124 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -111,7 +111,9 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => const canonicalEventLogger = options?.canonicalEventLogger ?? (options?.canonicalEventLogPath !== undefined - ? makeEventNdjsonLogger(options.canonicalEventLogPath) + ? yield* makeEventNdjsonLogger(options.canonicalEventLogPath, { + stream: "canonical", + }) : undefined); const registry = yield* ProviderAdapterRegistry; @@ -143,22 +145,17 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => canonicalizeRuntimeEventSession(event).pipe( Effect.tap((canonicalEvent) => canonicalEventLogger - ? Effect.tap( + ? Effect.flatMap( Effect.catch( directory.getThreadId(canonicalEvent.sessionId).pipe( - Effect.map(Option.getOrUndefined), + Effect.map((threadIdOption) => + Option.isSome(threadIdOption) ? threadIdOption.value : null, + ), ), - () => Effect.void, + () => Effect.succeed(null), ), (orchestrationThreadId) => - Effect.sync(() => { - canonicalEventLogger.write({ - observedAt: new Date().toISOString(), - stream: "canonical", - orchestrationThreadId: orchestrationThreadId ?? null, - event: canonicalEvent, - }); - }), + canonicalEventLogger.write(canonicalEvent, orchestrationThreadId), ) : Effect.void, ), diff --git a/apps/server/src/serverLayers.ts b/apps/server/src/serverLayers.ts index dd7d2593a296bf9eef5ef5f62aca79837900debc..f91710ea9719c2b7575d4bdafbe825950766d6ff 100644 --- a/apps/server/src/serverLayers.ts +++ b/apps/server/src/serverLayers.ts @@ -43,19 +43,25 @@ export function makeServerProviderLayer(): Layer.Layer< return Effect.gen(function* () { const { stateDir } = yield* ServerConfig; const providerLogsDir = path.join(stateDir, "logs", "provider"); - const providerEventLogger = makeEventNdjsonLogger(path.join(providerLogsDir, "events.log")); + const providerEventLogPath = path.join(providerLogsDir, "events.log"); + const nativeEventLogger = yield* makeEventNdjsonLogger(providerEventLogPath, { + stream: "native", + }); + const canonicalEventLogger = yield* makeEventNdjsonLogger(providerEventLogPath, { + stream: "canonical", + }); const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( Layer.provide(ProviderSessionRuntimeRepositoryLive), ); const codexAdapterLayer = makeCodexAdapterLive( - providerEventLogger ? { nativeEventLogger: providerEventLogger } : undefined, + nativeEventLogger ? { nativeEventLogger } : undefined, ); const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe( Layer.provide(codexAdapterLayer), Layer.provideMerge(providerSessionDirectoryLayer), ); return makeProviderServiceLive( - providerEventLogger ? { canonicalEventLogger: providerEventLogger } : undefined, + canonicalEventLogger ? { canonicalEventLogger } : undefined, ).pipe(Layer.provide(adapterRegistryLayer), Layer.provide(providerSessionDirectoryLayer)); }).pipe(Layer.unwrap); } -- 2.54.0