authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-03-03 19:03:27-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-05 23:00:41-08:00
log344e04b62351d005fbf973c1f6384fc162878db1
treed0076e52357b1f5f0701078c7ec64a4651bdd736
parent7b71e234e0c7d9e937cad940a6a2474d060ca3c6

Inject native event loggers into provider adapters

- allow Claude/Cursor adapters to accept an injected native event logger - make native event writes effectful/non-blocking so logging cannot break message handling - update adapter tests to assert native observability events via in-memory loggers - fix runtime session ID conversion when resolving canonical event thread IDs

5 files changed, 149 insertions(+), 91 deletions(-)

apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts+35-36
......@@ -1,7 +1,3 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
51import type {
62 Options as ClaudeQueryOptions,
73 PermissionMode,
......@@ -113,7 +109,10 @@ interface Harness {
113109 | undefined;
114110}
115111
116function makeHarness(config?: { readonly nativeEventLogPath?: string }): Harness {
112function makeHarness(config?: {
113 readonly nativeEventLogPath?: string;
114 readonly nativeEventLogger?: ClaudeCodeAdapterLiveOptions["nativeEventLogger"];
115}): Harness {
117116 const query = new FakeClaudeQuery();
118117 let createInput:
119118 | {
......@@ -127,6 +126,11 @@ function makeHarness(config?: { readonly nativeEventLogPath?: string }): Harness
127126 createInput = input;
128127 return query;
129128 },
129 ...(config?.nativeEventLogger
130 ? {
131 nativeEventLogger: config.nativeEventLogger,
132 }
133 : {}),
130134 ...(config?.nativeEventLogPath
131135 ? {
132136 nativeEventLogPath: config.nativeEventLogPath,
......@@ -815,9 +819,24 @@ describe("ClaudeCodeAdapterLive", () => {
815819 });
816820
817821 it.effect("writes provider-native observability records when enabled", () => {
818 const nativeLogDir = fs.mkdtempSync(path.join(os.tmpdir(), "claude-native-events-"));
819 const nativeEventLogPath = path.join(nativeLogDir, "provider-native.ndjson");
820 const harness = makeHarness({ nativeEventLogPath });
822 const nativeEvents: Array<{
823 event?: {
824 provider?: string;
825 method?: string;
826 sessionId?: string;
827 turnId?: string;
828 };
829 }> = [];
830 const harness = makeHarness({
831 nativeEventLogger: {
832 filePath: "memory://claude-native-events",
833 write: (event) => {
834 nativeEvents.push(event as (typeof nativeEvents)[number]);
835 return Effect.void;
836 },
837 close: () => Effect.void,
838 },
839 });
821840 return Effect.gen(function* () {
822841 const adapter = yield* ClaudeCodeAdapter;
823842
......@@ -862,40 +881,20 @@ describe("ClaudeCodeAdapterLive", () => {
862881 const turnCompleted = yield* Fiber.join(turnCompletedFiber);
863882 assert.equal(turnCompleted._tag, "Some");
864883
865 const lines = fs
866 .readFileSync(nativeEventLogPath, "utf8")
867 .split("\n")
868 .map((line) => line.trim())
869 .filter((line) => line.length > 0);
870 assert.equal(lines.length > 0, true);
871
872 const records = lines.map((line) => JSON.parse(line) as {
873 event?: {
874 provider?: string;
875 method?: string;
876 sessionId?: string;
877 turnId?: string;
878 };
879 });
880
881 assert.equal(records.some((record) => record.event?.provider === "claudeCode"), true);
882 assert.equal(records.some((record) => record.event?.sessionId === session.sessionId), true);
883 assert.equal(records.some((record) => String(record.event?.turnId) === String(turn.turnId)), true);
884 assert.equal(nativeEvents.length > 0, true);
885 assert.equal(nativeEvents.some((record) => record.event?.provider === "claudeCode"), true);
886 assert.equal(nativeEvents.some((record) => record.event?.sessionId === session.sessionId), true);
887 assert.equal(
888 nativeEvents.some((record) => String(record.event?.turnId) === String(turn.turnId)),
889 true,
890 );
884891 assert.equal(
885 records.some(
892 nativeEvents.some(
886893 (record) => record.event?.method === "claude/stream_event/content_block_delta/text_delta",
887894 ),
888895 true,
889896 );
890897 }).pipe(
891 Effect.ensuring(
892 Effect.sync(() => {
893 fs.rmSync(nativeLogDir, {
894 recursive: true,
895 force: true,
896 });
897 }),
898 ),
899898 Effect.provideService(Random.Random, makeDeterministicRandomService()),
900899 Effect.provide(harness.layer),
901900 );
apps/server/src/provider/Layers/ClaudeCodeAdapter.ts+31-23
......@@ -48,7 +48,7 @@ import {
4848 type ProviderAdapterError,
4949} from "../Errors.ts";
5050import { ClaudeCodeAdapter, type ClaudeCodeAdapterShape } from "../Services/ClaudeCodeAdapter.ts";
51import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
51import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
5252
5353const PROVIDER = "claudeCode" as const;
5454
......@@ -125,6 +125,7 @@ export interface ClaudeCodeAdapterLiveOptions {
125125 readonly options: ClaudeQueryOptions;
126126 }) => ClaudeQueryRuntime;
127127 readonly nativeEventLogPath?: string;
128 readonly nativeEventLogger?: EventNdjsonLogger;
128129}
129130
130131function isUuid(value: string): boolean {
......@@ -453,9 +454,12 @@ function sdkNativeItemId(message: SDKMessage): string | undefined {
453454function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
454455 return Effect.gen(function* () {
455456 const nativeEventLogger =
456 options?.nativeEventLogPath !== undefined
457 ? makeEventNdjsonLogger(options.nativeEventLogPath)
458 : undefined;
457 options?.nativeEventLogger ??
458 (options?.nativeEventLogPath !== undefined
459 ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {
460 stream: "native",
461 })
462 : undefined);
459463
460464 const createQuery =
461465 options?.createQuery ??
......@@ -478,7 +482,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
478482 context: ClaudeSessionContext,
479483 message: SDKMessage,
480484 ): Effect.Effect<void> =>
481 Effect.sync(() => {
485 Effect.gen(function* () {
482486 if (!nativeEventLogger) {
483487 return;
484488 }
......@@ -486,24 +490,28 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
486490 const observedAt = new Date().toISOString();
487491 const itemId = sdkNativeItemId(message);
488492
489 nativeEventLogger.write({
490 observedAt,
491 event: {
492 id:
493 "uuid" in message && typeof message.uuid === "string"
494 ? message.uuid
495 : crypto.randomUUID(),
496 kind: "notification",
497 provider: PROVIDER,
498 sessionId: asRuntimeSessionId(context.session.sessionId),
499 createdAt: observedAt,
500 method: sdkNativeMethod(message),
501 ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}),
502 ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})),
503 ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}),
504 payload: message,
505 },
506 });
493 yield* nativeEventLogger
494 .write(
495 {
496 observedAt,
497 event: {
498 id:
499 "uuid" in message && typeof message.uuid === "string"
500 ? message.uuid
501 : crypto.randomUUID(),
502 kind: "notification",
503 provider: PROVIDER,
504 sessionId: asRuntimeSessionId(context.session.sessionId),
505 createdAt: observedAt,
506 method: sdkNativeMethod(message),
507 ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}),
508 ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})),
509 ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}),
510 payload: message,
511 },
512 },
513 null,
514 );
507515 });
508516
509517 const snapshotThread = (
apps/server/src/provider/Layers/CursorAdapter.test.ts+35
......@@ -359,6 +359,41 @@ describe("CursorAdapterLive", () => {
359359 }).pipe(Effect.provide(layer));
360360 });
361361
362 it.effect("writes provider-native observability records when enabled", () => {
363 const nativeEvents: Array<{
364 event?: {
365 provider?: string;
366 method?: string;
367 sessionId?: string;
368 };
369 }> = [];
370 const fake = new FakeCursorAcpProcess();
371 const layer = makeCursorAdapterLive({
372 createProcess: () => fake as never,
373 nativeEventLogger: {
374 filePath: "memory://cursor-native-events",
375 write: (event) => {
376 nativeEvents.push(event as (typeof nativeEvents)[number]);
377 return Effect.void;
378 },
379 close: () => Effect.void,
380 },
381 });
382
383 return Effect.gen(function* () {
384 const adapter = yield* CursorAdapter;
385 const session = yield* adapter.startSession({
386 provider: "cursor",
387 cwd: "/tmp/project",
388 });
389
390 assert.equal(nativeEvents.length > 0, true);
391 assert.equal(nativeEvents.some((record) => record.event?.provider === "cursor"), true);
392 assert.equal(nativeEvents.some((record) => record.event?.sessionId === session.sessionId), true);
393 assert.equal(nativeEvents.some((record) => record.event?.method === "cursor/acp/response"), true);
394 }).pipe(Effect.provide(layer));
395 });
396
362397 it.effect("resumes ACP session using resumeCursor.acpSessionId", () => {
363398 const fake = new FakeCursorAcpProcess();
364399 const layer = makeCursorAdapterLive({
apps/server/src/provider/Layers/CursorAdapter.ts+45-31
......@@ -48,7 +48,7 @@ import {
4848 CursorAcpSessionPromptResult,
4949 CursorAcpSessionUpdateNotification,
5050} from "../Services/CursorAdapter.ts";
51import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
51import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
5252
5353const PROVIDER = "cursor" as const;
5454const DEFAULT_REQUEST_TIMEOUT_MS = 120_000;
......@@ -103,6 +103,7 @@ export interface CursorAdapterLiveOptions {
103103 readonly model?: string;
104104 }) => ChildProcessWithoutNullStreams;
105105 readonly nativeEventLogPath?: string;
106 readonly nativeEventLogger?: EventNdjsonLogger;
106107}
107108
108109function toMessage(cause: unknown, fallback: string): string {
......@@ -323,9 +324,12 @@ function writeCursorMessage(context: CursorSessionContext, message: unknown): vo
323324function makeCursorAdapter(options?: CursorAdapterLiveOptions) {
324325 return Effect.gen(function* () {
325326 const nativeEventLogger =
326 options?.nativeEventLogPath !== undefined
327 ? makeEventNdjsonLogger(options.nativeEventLogPath)
328 : undefined;
327 options?.nativeEventLogger ??
328 (options?.nativeEventLogPath !== undefined
329 ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, {
330 stream: "native",
331 })
332 : undefined);
329333
330334 const nowIso = Effect.map(DateTime.now, DateTime.formatIso);
331335 const nextEventId = Effect.map(Random.nextUUIDv4, (id) => EventId.makeUnsafe(id));
......@@ -822,33 +826,43 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) {
822826 }
823827
824828 if (nativeEventLogger) {
825 const nativeMethod =
826 typeof message.method === "string"
827 ? message.method
828 : typeof message.id === "string" || typeof message.id === "number"
829 ? "cursor/acp/response"
830 : "cursor/acp/message";
831 const nativeKind =
832 typeof message.method === "string" &&
833 (typeof message.id === "string" || typeof message.id === "number")
834 ? "request"
835 : typeof message.method === "string"
836 ? "notification"
837 : "session";
838 nativeEventLogger.write({
839 observedAt: new Date().toISOString(),
840 event: {
841 id: EventId.makeUnsafe(randomUUID()),
842 kind: nativeKind,
843 provider: PROVIDER,
844 sessionId: asRuntimeSessionId(context.session.sessionId),
845 createdAt: new Date().toISOString(),
846 method: nativeMethod,
847 ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})),
848 ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})),
849 payload: message,
850 },
851 });
829 try {
830 const nativeMethod =
831 typeof message.method === "string"
832 ? message.method
833 : typeof message.id === "string" || typeof message.id === "number"
834 ? "cursor/acp/response"
835 : "cursor/acp/message";
836 const nativeKind =
837 typeof message.method === "string" &&
838 (typeof message.id === "string" || typeof message.id === "number")
839 ? "request"
840 : typeof message.method === "string"
841 ? "notification"
842 : "session";
843 Effect.runFork(
844 nativeEventLogger
845 .write(
846 {
847 observedAt: new Date().toISOString(),
848 event: {
849 id: EventId.makeUnsafe(randomUUID()),
850 kind: nativeKind,
851 provider: PROVIDER,
852 sessionId: asRuntimeSessionId(context.session.sessionId),
853 createdAt: new Date().toISOString(),
854 method: nativeMethod,
855 ...(context.session.threadId ? { threadId: context.session.threadId } : {}),
856 ...(context.turnState ? { turnId: String(context.turnState.turnId) } : {}),
857 payload: message,
858 },
859 },
860 null,
861 ),
862 );
863 } catch {
864 // Native logging must never block or break protocol handling.
865 }
852866 }
853867
854868 if (
apps/server/src/provider/Layers/ProviderService.ts+3-1
......@@ -126,6 +126,8 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
126126 );
127127 const asRuntimeSessionId = (sessionId: ProviderSessionId): RuntimeSessionId =>
128128 RuntimeSessionId.makeUnsafe(sessionId);
129 const asProviderSessionId = (sessionId: RuntimeSessionId): ProviderSessionId =>
130 ProviderSessionId.makeUnsafe(sessionId);
129131
130132 const canonicalizeRuntimeEventSession = (
131133 event: ProviderRuntimeEvent,
......@@ -150,7 +152,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) =>
150152 canonicalEventLogger
151153 ? Effect.flatMap(
152154 Effect.catch(
153 directory.getThreadId(canonicalEvent.sessionId).pipe(
155 directory.getThreadId(asProviderSessionId(canonicalEvent.sessionId)).pipe(
154156 Effect.map((threadIdOption) =>
155157 Option.isSome(threadIdOption) ? threadIdOption.value : null,
156158 ),