From 1542b00db21f7586354adcc2ffd4f1e2a20bd02b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Fri, 27 Feb 2026 23:53:47 -0800 Subject: [PATCH] Stabilize Claude turn event ordering and native event logging - Preserve active turn state when `session.started`/`thread.started` arrive mid-turn - Emit `message.delta` from assistant text when stream deltas are missing, then complete the message - Add Claude native SDK NDJSON observability logging and wire its log path in server layers - Expand ingestion/adapter tests to cover mid-turn lifecycle and delta fallback behavior --- .../Layers/ProviderRuntimeIngestion.test.ts | 61 +++++ .../Layers/ProviderRuntimeIngestion.ts | 34 ++- .../provider/Layers/ClaudeCodeAdapter.test.ts | 251 +++++++++++++++++- .../src/provider/Layers/ClaudeCodeAdapter.ts | 167 +++++++++++- apps/server/src/serverLayers.ts | 4 +- 5 files changed, 489 insertions(+), 28 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index caeb1d95182cac967b5a32311de004ff66274b8e..228bc1d1db7c0b9dbce7d1416c0c22f6c12dc7cb 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -225,6 +225,67 @@ describe("ProviderRuntimeIngestion", () => { expect(thread.session?.lastError).toBe("turn failed"); }); + it("does not clear active turn when session/thread started arrives mid-turn", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-turn-started-midturn-lifecycle"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: now, + threadId: asProviderThreadId("provider-thread-1"), + turnId: asProviderTurnId("turn-midturn-lifecycle"), + }); + + await waitForThread( + harness.engine, + (thread) => + thread.session?.status === "running" && + thread.session?.activeTurnId === "turn-midturn-lifecycle", + ); + + harness.emit({ + type: "thread.started", + eventId: asEventId("evt-thread-started-midturn-lifecycle"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + }); + harness.emit({ + type: "session.started", + eventId: asEventId("evt-session-started-midturn-lifecycle"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + }); + + await Effect.runPromise(Effect.sleep("40 millis")); + const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); + const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(midThread?.session?.status).toBe("running"); + expect(midThread?.session?.activeTurnId).toBe("turn-midturn-lifecycle"); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-turn-completed-midturn-lifecycle"), + provider: "codex", + sessionId: asSessionId("sess-1"), + createdAt: new Date().toISOString(), + threadId: asProviderThreadId("provider-thread-1"), + turnId: asProviderTurnId("turn-midturn-lifecycle"), + status: "completed", + }); + + await waitForThread( + harness.engine, + (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null, + ); + }); + it("ignores auxiliary turn completions from a different provider thread", async () => { const harness = await createHarness(); const now = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index b0611e38bdea17baa5db730fa336b50edf89d0c6..4451a6b83b526a4ac29eab7cd8bcc26779d03885 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -407,8 +407,8 @@ const make = Effect.gen(function* () { if (activeTurnId !== null && eventTurnId !== undefined) { return sameId(activeTurnId, eventTurnId); } - // Without an active turn, only accept completion when no thread mismatch signal exists. - return eventProviderThreadId === null || scopedSessionProviderThreadId === null; + // If no active turn is tracked, accept completion scoped to this thread. + return true; default: return true; } @@ -421,7 +421,12 @@ const make = Effect.gen(function* () { event.type === "turn.started" || event.type === "turn.completed" ) { - const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) : null; + const nextActiveTurnId = + event.type === "turn.started" + ? (eventTurnId ?? null) + : event.type === "turn.completed" || event.type === "session.exited" + ? null + : activeTurnId; const providerThreadIdFromEvent = event.type === "thread.started" ? ProviderThreadId.makeUnsafe(event.threadId) @@ -430,14 +435,21 @@ const make = Effect.gen(function* () { : null; const providerThreadId = providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null; - const status = - event.type === "turn.started" - ? "running" - : event.type === "session.exited" - ? "stopped" - : event.type === "turn.completed" && event.status === "failed" - ? "error" - : "ready"; + const status = (() => { + switch (event.type) { + case "turn.started": + return "running"; + case "session.exited": + return "stopped"; + case "turn.completed": + return event.status === "failed" ? "error" : "ready"; + case "session.started": + case "thread.started": + // Provider thread/session start notifications can arrive during an + // active turn; preserve turn-running state in that case. + return activeTurnId !== null ? "running" : "ready"; + } + })(); const lastError = event.type === "turn.completed" && event.status === "failed" ? (event.errorMessage ?? thread.session?.lastError ?? "Turn failed") diff --git a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts index 21b3254b6d63546c63f9350aa54b67e11d7496f6..21bf2376e07121eb9c4a6b6a0ad6f2db518b7f22 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts @@ -1,3 +1,7 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + import type { Options as ClaudeQueryOptions, PermissionMode, @@ -109,7 +113,7 @@ interface Harness { | undefined; } -function makeHarness(): Harness { +function makeHarness(config?: { readonly nativeEventLogPath?: string }): Harness { const query = new FakeClaudeQuery(); let createInput: | { @@ -118,15 +122,20 @@ function makeHarness(): Harness { } | undefined; - const options: ClaudeCodeAdapterLiveOptions = { + const adapterOptions: ClaudeCodeAdapterLiveOptions = { createQuery: (input) => { createInput = input; return query; }, + ...(config?.nativeEventLogPath + ? { + nativeEventLogPath: config.nativeEventLogPath, + } + : {}), }; return { - layer: makeClaudeCodeAdapterLive(options), + layer: makeClaudeCodeAdapterLive(adapterOptions), query, getLastCreateQueryInput: () => createInput, }; @@ -305,6 +314,155 @@ describe("ClaudeCodeAdapterLive", () => { ); }); + it.effect("emits completion only after turn result when assistant frames arrive before deltas", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeCodeAdapter; + + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 6).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + provider: "claudeCode", + }); + + const turn = yield* adapter.sendTurn({ + sessionId: session.sessionId, + input: "hello", + attachments: [], + }); + + harness.query.emit({ + type: "assistant", + session_id: "sdk-session-early-assistant", + uuid: "assistant-early", + parent_tool_use_id: null, + message: { + id: "assistant-message-early", + content: [{ type: "tool_use", id: "tool-early", name: "Read", input: { path: "a.ts" } }], + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "stream_event", + session_id: "sdk-session-early-assistant", + uuid: "stream-early", + parent_tool_use_id: null, + event: { + type: "content_block_delta", + index: 0, + delta: { + type: "text_delta", + text: "Late text", + }, + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-early-assistant", + uuid: "result-early", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + assert.deepEqual( + runtimeEvents.map((event) => event.type), + [ + "session.started", + "turn.started", + "thread.started", + "message.delta", + "message.completed", + "turn.completed", + ], + ); + + const deltaIndex = runtimeEvents.findIndex((event) => event.type === "message.delta"); + const completedIndex = runtimeEvents.findIndex((event) => event.type === "message.completed"); + assert.equal(deltaIndex >= 0 && completedIndex >= 0 && deltaIndex < completedIndex, true); + + const deltaEvent = runtimeEvents[deltaIndex]; + assert.equal(deltaEvent?.type, "message.delta"); + if (deltaEvent?.type === "message.delta") { + assert.equal(deltaEvent.delta, "Late text"); + assert.equal(deltaEvent.turnId, turn.turnId); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("falls back to assistant payload text when stream deltas are absent", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeCodeAdapter; + + const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 6).pipe( + Stream.runCollect, + Effect.forkChild, + ); + + const session = yield* adapter.startSession({ + provider: "claudeCode", + }); + + const turn = yield* adapter.sendTurn({ + sessionId: session.sessionId, + input: "hello", + attachments: [], + }); + + harness.query.emit({ + type: "assistant", + session_id: "sdk-session-fallback-text", + uuid: "assistant-fallback", + parent_tool_use_id: null, + message: { + id: "assistant-message-fallback", + content: [{ type: "text", text: "Fallback hello" }], + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-fallback-text", + uuid: "result-fallback", + } as unknown as SDKMessage); + + const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber)); + assert.deepEqual( + runtimeEvents.map((event) => event.type), + [ + "session.started", + "turn.started", + "thread.started", + "message.delta", + "message.completed", + "turn.completed", + ], + ); + + const deltaEvent = runtimeEvents.find((event) => event.type === "message.delta"); + assert.equal(deltaEvent?.type, "message.delta"); + if (deltaEvent?.type === "message.delta") { + assert.equal(deltaEvent.delta, "Fallback hello"); + assert.equal(deltaEvent.turnId, turn.turnId); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("bridges approval request/response lifecycle through canUseTool", () => { const harness = makeHarness(); return Effect.gen(function* () { @@ -527,4 +685,91 @@ describe("ClaudeCodeAdapterLive", () => { Effect.provide(harness.layer), ); }); + + 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 }); + return Effect.gen(function* () { + const adapter = yield* ClaudeCodeAdapter; + + const session = yield* adapter.startSession({ + provider: "claudeCode", + }); + const turn = yield* adapter.sendTurn({ + sessionId: session.sessionId, + input: "hello", + attachments: [], + }); + + const turnCompletedFiber = yield* Stream.filter( + adapter.streamEvents, + (event) => event.type === "turn.completed", + ).pipe(Stream.runHead, Effect.forkChild); + + harness.query.emit({ + type: "stream_event", + session_id: "sdk-session-native-log", + uuid: "stream-native-log", + parent_tool_use_id: null, + event: { + type: "content_block_delta", + index: 0, + delta: { + type: "text_delta", + text: "hi", + }, + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + errors: [], + session_id: "sdk-session-native-log", + uuid: "result-native-log", + } as unknown as SDKMessage); + + 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) => record.event?.turnId === turn.turnId), true); + assert.equal( + records.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 6975b5ee7a67d4fcc1f64aed2740018ea374b8f8..be42594ed0464e1866c4bbe6947d92d7410539c1 100644 --- a/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeCodeAdapter.ts @@ -42,6 +42,7 @@ import { type ProviderAdapterError, } from "../Errors.ts"; import { ClaudeCodeAdapter, type ClaudeCodeAdapterShape } from "../Services/ClaudeCodeAdapter.ts"; +import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = "claudeCode" as const; @@ -67,6 +68,8 @@ interface ClaudeTurnState { readonly startedAt: string; readonly items: Array; readonly messageCompleted: boolean; + readonly emittedTextDelta: boolean; + readonly fallbackAssistantText: string; } interface PendingApproval { @@ -114,6 +117,7 @@ export interface ClaudeCodeAdapterLiveOptions { readonly prompt: AsyncIterable; readonly options: ClaudeQueryOptions; }) => ClaudeQueryRuntime; + readonly nativeEventLogPath?: string; } function isUuid(value: string): boolean { @@ -272,6 +276,30 @@ function turnStatusFromResult(result: SDKResultMessage): ProviderRuntimeTurnStat return "failed"; } +function extractAssistantText(message: SDKMessage): string { + if (message.type !== "assistant") { + return ""; + } + + const content = (message.message as { content?: unknown } | undefined)?.content; + if (!Array.isArray(content)) { + return ""; + } + + const fragments: string[] = []; + for (const block of content) { + if (!block || typeof block !== "object") { + continue; + } + const candidate = block as { type?: unknown; text?: unknown }; + if (candidate.type === "text" && typeof candidate.text === "string" && candidate.text.length > 0) { + fragments.push(candidate.text); + } + } + + return fragments.join(""); +} + function toSessionError( sessionId: ReturnType, cause: unknown, @@ -311,8 +339,72 @@ function toRequestError( }); } +function sdkMessageType(value: unknown): string | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + const record = value as { type?: unknown }; + return typeof record.type === "string" ? record.type : undefined; +} + +function sdkMessageSubtype(value: unknown): string | undefined { + if (!value || typeof value !== "object") { + return undefined; + } + const record = value as { subtype?: unknown }; + return typeof record.subtype === "string" ? record.subtype : undefined; +} + +function sdkNativeMethod(message: SDKMessage): string { + const subtype = sdkMessageSubtype(message); + if (subtype) { + return `claude/${message.type}/${subtype}`; + } + + if (message.type === "stream_event") { + const streamType = sdkMessageType(message.event); + if (streamType) { + const deltaType = + streamType === "content_block_delta" ? sdkMessageType((message.event as { delta?: unknown }).delta) : undefined; + if (deltaType) { + return `claude/${message.type}/${streamType}/${deltaType}`; + } + return `claude/${message.type}/${streamType}`; + } + } + + return `claude/${message.type}`; +} + +function sdkNativeItemId(message: SDKMessage): string | undefined { + if (message.type === "assistant") { + const maybeId = (message.message as { id?: unknown }).id; + if (typeof maybeId === "string") { + return maybeId; + } + return undefined; + } + + if (message.type === "stream_event") { + const event = message.event as { + type?: unknown; + content_block?: { id?: unknown }; + }; + if (event.type === "content_block_start" && typeof event.content_block?.id === "string") { + return event.content_block.id; + } + } + + return undefined; +} + function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { return Effect.gen(function* () { + const nativeEventLogger = + options?.nativeEventLogPath !== undefined + ? makeEventNdjsonLogger(options.nativeEventLogPath) + : undefined; + const createQuery = options?.createQuery ?? ((input: { @@ -330,6 +422,38 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { const offerRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect => Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid); + const logNativeSdkMessage = ( + context: ClaudeSessionContext, + message: SDKMessage, + ): Effect.Effect => + Effect.sync(() => { + if (!nativeEventLogger) { + return; + } + + 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: context.session.sessionId, + createdAt: observedAt, + method: sdkNativeMethod(message), + ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}), + ...(context.turnState ? { turnId: context.turnState.turnId } : {}), + ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}), + payload: message, + }, + }); + }); + const snapshotThread = ( context: ClaudeSessionContext, ): Effect.Effect<{ @@ -450,6 +574,21 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { } if (!turnState.messageCompleted) { + if (!turnState.emittedTextDelta && turnState.fallbackAssistantText.length > 0) { + const deltaStamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "message.delta", + eventId: deltaStamp.eventId, + provider: PROVIDER, + sessionId: context.session.sessionId, + createdAt: deltaStamp.createdAt, + ...(context.session.threadId ? { threadId: context.session.threadId } : {}), + turnId: turnState.turnId, + itemId: ProviderItemId.makeUnsafe(turnState.assistantItemId), + delta: turnState.fallbackAssistantText, + }); + } + const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "message.completed", @@ -510,6 +649,12 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { event.delta.text.length > 0 && context.turnState ) { + if (!context.turnState.emittedTextDelta) { + context.turnState = { + ...context.turnState, + emittedTextDelta: true, + }; + } const stamp = yield* makeEventStamp(); yield* offerRuntimeEvent({ type: "message.delta", @@ -606,22 +751,15 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { if (context.turnState) { context.turnState.items.push(message.message); - if (!context.turnState.messageCompleted) { + const fallbackAssistantText = extractAssistantText(message); + if ( + fallbackAssistantText.length > 0 && + fallbackAssistantText !== context.turnState.fallbackAssistantText + ) { context.turnState = { ...context.turnState, - messageCompleted: true, + fallbackAssistantText, }; - const stamp = yield* makeEventStamp(); - yield* offerRuntimeEvent({ - type: "message.completed", - eventId: stamp.eventId, - provider: PROVIDER, - sessionId: context.session.sessionId, - createdAt: stamp.createdAt, - itemId: ProviderItemId.makeUnsafe(context.turnState.assistantItemId), - ...(context.session.threadId ? { threadId: context.session.threadId } : {}), - turnId: context.turnState.turnId, - }); } } @@ -653,6 +791,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { message: SDKMessage, ): Effect.Effect => Effect.gen(function* () { + yield* logNativeSdkMessage(context, message); yield* ensureThreadId(context, message); switch (message.type) { @@ -1014,6 +1153,8 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) { startedAt: yield* nowIso, items: [], messageCompleted: false, + emittedTextDelta: false, + fallbackAssistantText: "", }; const updatedAt = yield* nowIso; diff --git a/apps/server/src/serverLayers.ts b/apps/server/src/serverLayers.ts index 2fff0dee330f1b4a4246a1dc40978a374ea631f0..58b7354765d1f72e7b78287cf2181e0f38ec7f6a 100644 --- a/apps/server/src/serverLayers.ts +++ b/apps/server/src/serverLayers.ts @@ -57,7 +57,9 @@ export function makeServerProviderLayer(): Layer.Layer< const codexAdapterLayer = makeCodexAdapterLive( nativeEventLogger ? { nativeEventLogger } : undefined, ); - const claudeAdapterLayer = makeClaudeCodeAdapterLive(); + const claudeAdapterLayer = makeClaudeCodeAdapterLive({ + nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"), + }); const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe( Layer.provide(Layer.mergeAll(codexAdapterLayer, claudeAdapterLayer)), Layer.provideMerge(providerSessionDirectoryLayer), -- 2.54.0