diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts index d576c61f7a33d2fb3166fb92f57cdb6e6ff61059..de80b05313491bd1f8de5404cf66fa2608c4376d 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts @@ -1,7 +1,3 @@ -import fs from "node:fs"; -import os from "node:os"; -import path from "node:path"; - import type { Options as ClaudeQueryOptions, PermissionMode, @@ -113,7 +109,10 @@ interface Harness { | undefined; } -function makeHarness(config?: { readonly nativeEventLogPath?: string }): Harness { +function makeHarness(config?: { + readonly nativeEventLogPath?: string; + readonly nativeEventLogger?: ClaudeCodeAdapterLiveOptions["nativeEventLogger"]; +}): Harness { const query = new FakeClaudeQuery(); let createInput: | { @@ -127,6 +126,11 @@ function makeHarness(config?: { readonly nativeEventLogPath?: string }): Harness createInput = input; return query; }, + ...(config?.nativeEventLogger + ? { + nativeEventLogger: config.nativeEventLogger, + } + : {}), ...(config?.nativeEventLogPath ? { nativeEventLogPath: config.nativeEventLogPath, @@ -815,9 +819,24 @@ describe("ClaudeCodeAdapterLive", () => { }); it.effect("writes provider-native observability records when enabled", () => { - const nativeLogDir = fs.mkdtempSync(path.join(os.tmpdir(), "claude-native-events-")); - const nativeEventLogPath = path.join(nativeLogDir, "provider-native.ndjson"); - const harness = makeHarness({ nativeEventLogPath }); + const nativeEvents: Array<{ + event?: { + provider?: string; + method?: string; + sessionId?: string; + turnId?: string; + }; + }> = []; + const harness = makeHarness({ + nativeEventLogger: { + filePath: "memory://claude-native-events", + write: (event) => { + nativeEvents.push(event as (typeof nativeEvents)[number]); + return Effect.void; + }, + close: () => Effect.void, + }, + }); return Effect.gen(function* () { const adapter = yield* ClaudeCodeAdapter; @@ -862,40 +881,20 @@ describe("ClaudeCodeAdapterLive", () => { const turnCompleted = yield* Fiber.join(turnCompletedFiber); assert.equal(turnCompleted._tag, "Some"); - const lines = fs - .readFileSync(nativeEventLogPath, "utf8") - .split("\n") - .map((line) => line.trim()) - .filter((line) => line.length > 0); - assert.equal(lines.length > 0, true); - - const records = lines.map((line) => JSON.parse(line) as { - event?: { - provider?: string; - method?: string; - sessionId?: string; - turnId?: string; - }; - }); - - assert.equal(records.some((record) => record.event?.provider === "claudeCode"), true); - assert.equal(records.some((record) => record.event?.sessionId === session.sessionId), true); - assert.equal(records.some((record) => String(record.event?.turnId) === String(turn.turnId)), true); + assert.equal(nativeEvents.length > 0, true); + assert.equal(nativeEvents.some((record) => record.event?.provider === "claudeCode"), true); + assert.equal(nativeEvents.some((record) => record.event?.sessionId === session.sessionId), true); assert.equal( - records.some( + nativeEvents.some((record) => String(record.event?.turnId) === String(turn.turnId)), + true, + ); + assert.equal( + nativeEvents.some( (record) => record.event?.method === "claude/stream_event/content_block_delta/text_delta", ), true, ); }).pipe( - Effect.ensuring( - Effect.sync(() => { - fs.rmSync(nativeLogDir, { - recursive: true, - force: true, - }); - }), - ), Effect.provideService(Random.Random, makeDeterministicRandomService()), Effect.provide(harness.layer), ); diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts index 4e777c8524bd53f6d8d62009c264c08c1d1a4b5e..c32ba62e54362e458114cb5f82b2757503aa5269 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts @@ -48,7 +48,7 @@ import { type ProviderAdapterError, } from "../Errors.ts"; import { ClaudeCodeAdapter, type ClaudeCodeAdapterShape } from "../Services/ClaudeCodeAdapter.ts"; -import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; +import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = "claudeCode" as const; @@ -125,6 +125,7 @@ export interface ClaudeCodeAdapterLiveOptions { readonly options: ClaudeQueryOptions; }) => ClaudeQueryRuntime; readonly nativeEventLogPath?: string; + readonly nativeEventLogger?: EventNdjsonLogger; } function isUuid(value: string): boolean { @@ -453,9 +454,12 @@ function sdkNativeItemId(message: SDKMessage): string | undefined { function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { return Effect.gen(function* () { const nativeEventLogger = - options?.nativeEventLogPath !== undefined - ? makeEventNdjsonLogger(options.nativeEventLogPath) - : undefined; + options?.nativeEventLogger ?? + (options?.nativeEventLogPath !== undefined + ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { + stream: "native", + }) + : undefined); const createQuery = options?.createQuery ?? @@ -478,7 +482,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { context: ClaudeSessionContext, message: SDKMessage, ): Effect.Effect => - Effect.sync(() => { + Effect.gen(function* () { if (!nativeEventLogger) { return; } @@ -486,24 +490,28 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { const observedAt = new Date().toISOString(); const itemId = sdkNativeItemId(message); - nativeEventLogger.write({ - observedAt, - event: { - id: - "uuid" in message && typeof message.uuid === "string" - ? message.uuid - : crypto.randomUUID(), - kind: "notification", - provider: PROVIDER, - sessionId: asRuntimeSessionId(context.session.sessionId), - createdAt: observedAt, - method: sdkNativeMethod(message), - ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}), - ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), - ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}), - payload: message, - }, - }); + yield* nativeEventLogger + .write( + { + observedAt, + event: { + id: + "uuid" in message && typeof message.uuid === "string" + ? message.uuid + : crypto.randomUUID(), + kind: "notification", + provider: PROVIDER, + sessionId: asRuntimeSessionId(context.session.sessionId), + createdAt: observedAt, + method: sdkNativeMethod(message), + ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}), + ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), + ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}), + payload: message, + }, + }, + null, + ); }); const snapshotThread = ( diff --git a/apps/server/src/provider/Layers/CursorAdapter.test.ts b/apps/server/src/provider/Layers/CursorAdapter.test.ts index 36921b172be19fdd53164dab68bc48d4e117f3e6..fc09751dd158ac703ccf501ee1caf41eb088e2a8 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.test.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.test.ts @@ -359,6 +359,41 @@ describe("CursorAdapterLive", () => { }).pipe(Effect.provide(layer)); }); + it.effect("writes provider-native observability records when enabled", () => { + const nativeEvents: Array<{ + event?: { + provider?: string; + method?: string; + sessionId?: string; + }; + }> = []; + const fake = new FakeCursorAcpProcess(); + const layer = makeCursorAdapterLive({ + createProcess: () => fake as never, + nativeEventLogger: { + filePath: "memory://cursor-native-events", + write: (event) => { + nativeEvents.push(event as (typeof nativeEvents)[number]); + return Effect.void; + }, + close: () => Effect.void, + }, + }); + + return Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const session = yield* adapter.startSession({ + provider: "cursor", + cwd: "/tmp/project", + }); + + assert.equal(nativeEvents.length > 0, true); + assert.equal(nativeEvents.some((record) => record.event?.provider === "cursor"), true); + assert.equal(nativeEvents.some((record) => record.event?.sessionId === session.sessionId), true); + assert.equal(nativeEvents.some((record) => record.event?.method === "cursor/acp/response"), true); + }).pipe(Effect.provide(layer)); + }); + it.effect("resumes ACP session using resumeCursor.acpSessionId", () => { const fake = new FakeCursorAcpProcess(); const layer = makeCursorAdapterLive({ diff --git a/apps/server/src/provider/Layers/CursorAdapter.ts b/apps/server/src/provider/Layers/CursorAdapter.ts index fc2d1c3da3fa6a3ff32830179bc01a4ff0b61b86..47561e737a867b41cd2eae092fad0f0acbaee73b 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.ts @@ -48,7 +48,7 @@ import { CursorAcpSessionPromptResult, CursorAcpSessionUpdateNotification, } from "../Services/CursorAdapter.ts"; -import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; +import { type EventNdjsonLogger, makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = "cursor" as const; const DEFAULT_REQUEST_TIMEOUT_MS = 120_000; @@ -103,6 +103,7 @@ export interface CursorAdapterLiveOptions { readonly model?: string; }) => ChildProcessWithoutNullStreams; readonly nativeEventLogPath?: string; + readonly nativeEventLogger?: EventNdjsonLogger; } function toMessage(cause: unknown, fallback: string): string { @@ -323,9 +324,12 @@ function writeCursorMessage(context: CursorSessionContext, message: unknown): vo function makeCursorAdapter(options?: CursorAdapterLiveOptions) { return Effect.gen(function* () { const nativeEventLogger = - options?.nativeEventLogPath !== undefined - ? makeEventNdjsonLogger(options.nativeEventLogPath) - : undefined; + options?.nativeEventLogger ?? + (options?.nativeEventLogPath !== undefined + ? yield* makeEventNdjsonLogger(options.nativeEventLogPath, { + stream: "native", + }) + : undefined); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const nextEventId = Effect.map(Random.nextUUIDv4, (id) => EventId.makeUnsafe(id)); @@ -822,33 +826,43 @@ function makeCursorAdapter(options?: CursorAdapterLiveOptions) { } if (nativeEventLogger) { - const nativeMethod = - typeof message.method === "string" - ? message.method - : typeof message.id === "string" || typeof message.id === "number" - ? "cursor/acp/response" - : "cursor/acp/message"; - const nativeKind = - typeof message.method === "string" && - (typeof message.id === "string" || typeof message.id === "number") - ? "request" - : typeof message.method === "string" - ? "notification" - : "session"; - nativeEventLogger.write({ - observedAt: new Date().toISOString(), - event: { - id: EventId.makeUnsafe(randomUUID()), - kind: nativeKind, - provider: PROVIDER, - sessionId: asRuntimeSessionId(context.session.sessionId), - createdAt: new Date().toISOString(), - method: nativeMethod, - ...((context.session.threadId ? { threadId: asRuntimeThreadId(context.session.threadId) } : {})), - ...((context.turnState ? { turnId: asRuntimeTurnId(context.turnState.turnId) } : {})), - payload: message, - }, - }); + try { + const nativeMethod = + typeof message.method === "string" + ? message.method + : typeof message.id === "string" || typeof message.id === "number" + ? "cursor/acp/response" + : "cursor/acp/message"; + const nativeKind = + typeof message.method === "string" && + (typeof message.id === "string" || typeof message.id === "number") + ? "request" + : typeof message.method === "string" + ? "notification" + : "session"; + Effect.runFork( + nativeEventLogger + .write( + { + observedAt: new Date().toISOString(), + event: { + id: EventId.makeUnsafe(randomUUID()), + kind: nativeKind, + provider: PROVIDER, + sessionId: asRuntimeSessionId(context.session.sessionId), + createdAt: new Date().toISOString(), + method: nativeMethod, + ...(context.session.threadId ? { threadId: context.session.threadId } : {}), + ...(context.turnState ? { turnId: String(context.turnState.turnId) } : {}), + payload: message, + }, + }, + null, + ), + ); + } catch { + // Native logging must never block or break protocol handling. + } } if ( diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index c0e5bf921049054bc49ac3f4b00341c2cce8fe68..1422667733ed926ab1368900792b1213ca1a5380 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -126,6 +126,8 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => ); const asRuntimeSessionId = (sessionId: ProviderSessionId): RuntimeSessionId => RuntimeSessionId.makeUnsafe(sessionId); + const asProviderSessionId = (sessionId: RuntimeSessionId): ProviderSessionId => + ProviderSessionId.makeUnsafe(sessionId); const canonicalizeRuntimeEventSession = ( event: ProviderRuntimeEvent, @@ -150,7 +152,7 @@ const makeProviderService = (options?: ProviderServiceLiveOptions) => canonicalEventLogger ? Effect.flatMap( Effect.catch( - directory.getThreadId(canonicalEvent.sessionId).pipe( + directory.getThreadId(asProviderSessionId(canonicalEvent.sessionId)).pipe( Effect.map((threadIdOption) => Option.isSome(threadIdOption) ? threadIdOption.value : null, ),