authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-27 23:53:47-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-03-05 23:00:41-08:00
log1542b00db21f7586354adcc2ffd4f1e2a20bd02b
treef703fe82747ac87c57f27eb8a93b12cee64ff625
parentf79f8bac8072f4c634144cca2b7380a3020f7fec

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

5 files changed, 489 insertions(+), 28 deletions(-)

apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+61
...@@ -225,6 +225,67 @@ describe("ProviderRuntimeIngestion", () => {...@@ -225,6 +225,67 @@ describe("ProviderRuntimeIngestion", () => {
225 expect(thread.session?.lastError).toBe("turn failed");225 expect(thread.session?.lastError).toBe("turn failed");
226 });226 });
227227
228 it("does not clear active turn when session/thread started arrives mid-turn", async () => {
229 const harness = await createHarness();
230 const now = new Date().toISOString();
231
232 harness.emit({
233 type: "turn.started",
234 eventId: asEventId("evt-turn-started-midturn-lifecycle"),
235 provider: "codex",
236 sessionId: asSessionId("sess-1"),
237 createdAt: now,
238 threadId: asProviderThreadId("provider-thread-1"),
239 turnId: asProviderTurnId("turn-midturn-lifecycle"),
240 });
241
242 await waitForThread(
243 harness.engine,
244 (thread) =>
245 thread.session?.status === "running" &&
246 thread.session?.activeTurnId === "turn-midturn-lifecycle",
247 );
248
249 harness.emit({
250 type: "thread.started",
251 eventId: asEventId("evt-thread-started-midturn-lifecycle"),
252 provider: "codex",
253 sessionId: asSessionId("sess-1"),
254 createdAt: new Date().toISOString(),
255 threadId: asProviderThreadId("provider-thread-1"),
256 });
257 harness.emit({
258 type: "session.started",
259 eventId: asEventId("evt-session-started-midturn-lifecycle"),
260 provider: "codex",
261 sessionId: asSessionId("sess-1"),
262 createdAt: new Date().toISOString(),
263 threadId: asProviderThreadId("provider-thread-1"),
264 });
265
266 await Effect.runPromise(Effect.sleep("40 millis"));
267 const midReadModel = await Effect.runPromise(harness.engine.getReadModel());
268 const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1"));
269 expect(midThread?.session?.status).toBe("running");
270 expect(midThread?.session?.activeTurnId).toBe("turn-midturn-lifecycle");
271
272 harness.emit({
273 type: "turn.completed",
274 eventId: asEventId("evt-turn-completed-midturn-lifecycle"),
275 provider: "codex",
276 sessionId: asSessionId("sess-1"),
277 createdAt: new Date().toISOString(),
278 threadId: asProviderThreadId("provider-thread-1"),
279 turnId: asProviderTurnId("turn-midturn-lifecycle"),
280 status: "completed",
281 });
282
283 await waitForThread(
284 harness.engine,
285 (thread) => thread.session?.status === "ready" && thread.session?.activeTurnId === null,
286 );
287 });
288
228 it("ignores auxiliary turn completions from a different provider thread", async () => {289 it("ignores auxiliary turn completions from a different provider thread", async () => {
229 const harness = await createHarness();290 const harness = await createHarness();
230 const now = new Date().toISOString();291 const now = new Date().toISOString();
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+23-11
...@@ -407,8 +407,8 @@ const make = Effect.gen(function* () {...@@ -407,8 +407,8 @@ const make = Effect.gen(function* () {
407 if (activeTurnId !== null && eventTurnId !== undefined) {407 if (activeTurnId !== null && eventTurnId !== undefined) {
408 return sameId(activeTurnId, eventTurnId);408 return sameId(activeTurnId, eventTurnId);
409 }409 }
410 // Without an active turn, only accept completion when no thread mismatch signal exists.410 // If no active turn is tracked, accept completion scoped to this thread.
411 return eventProviderThreadId === null || scopedSessionProviderThreadId === null;411 return true;
412 default:412 default:
413 return true;413 return true;
414 }414 }
...@@ -421,7 +421,12 @@ const make = Effect.gen(function* () {...@@ -421,7 +421,12 @@ const make = Effect.gen(function* () {
421 event.type === "turn.started" ||421 event.type === "turn.started" ||
422 event.type === "turn.completed"422 event.type === "turn.completed"
423 ) {423 ) {
424 const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) : null;424 const nextActiveTurnId =
425 event.type === "turn.started"
426 ? (eventTurnId ?? null)
427 : event.type === "turn.completed" || event.type === "session.exited"
428 ? null
429 : activeTurnId;
425 const providerThreadIdFromEvent =430 const providerThreadIdFromEvent =
426 event.type === "thread.started"431 event.type === "thread.started"
427 ? ProviderThreadId.makeUnsafe(event.threadId)432 ? ProviderThreadId.makeUnsafe(event.threadId)
...@@ -430,14 +435,21 @@ const make = Effect.gen(function* () {...@@ -430,14 +435,21 @@ const make = Effect.gen(function* () {
430 : null;435 : null;
431 const providerThreadId =436 const providerThreadId =
432 providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null;437 providerThreadIdFromEvent ?? scopedSessionProviderThreadId ?? null;
433 const status =438 const status = (() => {
434 event.type === "turn.started"439 switch (event.type) {
435 ? "running"440 case "turn.started":
436 : event.type === "session.exited"441 return "running";
437 ? "stopped"442 case "session.exited":
438 : event.type === "turn.completed" && event.status === "failed"443 return "stopped";
439 ? "error"444 case "turn.completed":
440 : "ready";445 return event.status === "failed" ? "error" : "ready";
446 case "session.started":
447 case "thread.started":
448 // Provider thread/session start notifications can arrive during an
449 // active turn; preserve turn-running state in that case.
450 return activeTurnId !== null ? "running" : "ready";
451 }
452 })();
441 const lastError =453 const lastError =
442 event.type === "turn.completed" && event.status === "failed"454 event.type === "turn.completed" && event.status === "failed"
443 ? (event.errorMessage ?? thread.session?.lastError ?? "Turn failed")455 ? (event.errorMessage ?? thread.session?.lastError ?? "Turn failed")
apps/server/src/provider/Layers/ClaudeCodeAdapter.test.ts+248-3
...@@ -1,3 +1,7 @@...@@ -1,3 +1,7 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
1import type {5import type {
2 Options as ClaudeQueryOptions,6 Options as ClaudeQueryOptions,
3 PermissionMode,7 PermissionMode,
...@@ -109,7 +113,7 @@ interface Harness {...@@ -109,7 +113,7 @@ interface Harness {
109 | undefined;113 | undefined;
110}114}
111115
112function makeHarness(): Harness {116function makeHarness(config?: { readonly nativeEventLogPath?: string }): Harness {
113 const query = new FakeClaudeQuery();117 const query = new FakeClaudeQuery();
114 let createInput:118 let createInput:
115 | {119 | {
...@@ -118,15 +122,20 @@ function makeHarness(): Harness {...@@ -118,15 +122,20 @@ function makeHarness(): Harness {
118 }122 }
119 | undefined;123 | undefined;
120124
121 const options: ClaudeCodeAdapterLiveOptions = {125 const adapterOptions: ClaudeCodeAdapterLiveOptions = {
122 createQuery: (input) => {126 createQuery: (input) => {
123 createInput = input;127 createInput = input;
124 return query;128 return query;
125 },129 },
130 ...(config?.nativeEventLogPath
131 ? {
132 nativeEventLogPath: config.nativeEventLogPath,
133 }
134 : {}),
126 };135 };
127136
128 return {137 return {
129 layer: makeClaudeCodeAdapterLive(options),138 layer: makeClaudeCodeAdapterLive(adapterOptions),
130 query,139 query,
131 getLastCreateQueryInput: () => createInput,140 getLastCreateQueryInput: () => createInput,
132 };141 };
...@@ -305,6 +314,155 @@ describe("ClaudeCodeAdapterLive", () => {...@@ -305,6 +314,155 @@ describe("ClaudeCodeAdapterLive", () => {
305 );314 );
306 });315 });
307316
317 it.effect("emits completion only after turn result when assistant frames arrive before deltas", () => {
318 const harness = makeHarness();
319 return Effect.gen(function* () {
320 const adapter = yield* ClaudeCodeAdapter;
321
322 const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 6).pipe(
323 Stream.runCollect,
324 Effect.forkChild,
325 );
326
327 const session = yield* adapter.startSession({
328 provider: "claudeCode",
329 });
330
331 const turn = yield* adapter.sendTurn({
332 sessionId: session.sessionId,
333 input: "hello",
334 attachments: [],
335 });
336
337 harness.query.emit({
338 type: "assistant",
339 session_id: "sdk-session-early-assistant",
340 uuid: "assistant-early",
341 parent_tool_use_id: null,
342 message: {
343 id: "assistant-message-early",
344 content: [{ type: "tool_use", id: "tool-early", name: "Read", input: { path: "a.ts" } }],
345 },
346 } as unknown as SDKMessage);
347
348 harness.query.emit({
349 type: "stream_event",
350 session_id: "sdk-session-early-assistant",
351 uuid: "stream-early",
352 parent_tool_use_id: null,
353 event: {
354 type: "content_block_delta",
355 index: 0,
356 delta: {
357 type: "text_delta",
358 text: "Late text",
359 },
360 },
361 } as unknown as SDKMessage);
362
363 harness.query.emit({
364 type: "result",
365 subtype: "success",
366 is_error: false,
367 errors: [],
368 session_id: "sdk-session-early-assistant",
369 uuid: "result-early",
370 } as unknown as SDKMessage);
371
372 const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber));
373 assert.deepEqual(
374 runtimeEvents.map((event) => event.type),
375 [
376 "session.started",
377 "turn.started",
378 "thread.started",
379 "message.delta",
380 "message.completed",
381 "turn.completed",
382 ],
383 );
384
385 const deltaIndex = runtimeEvents.findIndex((event) => event.type === "message.delta");
386 const completedIndex = runtimeEvents.findIndex((event) => event.type === "message.completed");
387 assert.equal(deltaIndex >= 0 && completedIndex >= 0 && deltaIndex < completedIndex, true);
388
389 const deltaEvent = runtimeEvents[deltaIndex];
390 assert.equal(deltaEvent?.type, "message.delta");
391 if (deltaEvent?.type === "message.delta") {
392 assert.equal(deltaEvent.delta, "Late text");
393 assert.equal(deltaEvent.turnId, turn.turnId);
394 }
395 }).pipe(
396 Effect.provideService(Random.Random, makeDeterministicRandomService()),
397 Effect.provide(harness.layer),
398 );
399 });
400
401 it.effect("falls back to assistant payload text when stream deltas are absent", () => {
402 const harness = makeHarness();
403 return Effect.gen(function* () {
404 const adapter = yield* ClaudeCodeAdapter;
405
406 const runtimeEventsFiber = yield* Stream.take(adapter.streamEvents, 6).pipe(
407 Stream.runCollect,
408 Effect.forkChild,
409 );
410
411 const session = yield* adapter.startSession({
412 provider: "claudeCode",
413 });
414
415 const turn = yield* adapter.sendTurn({
416 sessionId: session.sessionId,
417 input: "hello",
418 attachments: [],
419 });
420
421 harness.query.emit({
422 type: "assistant",
423 session_id: "sdk-session-fallback-text",
424 uuid: "assistant-fallback",
425 parent_tool_use_id: null,
426 message: {
427 id: "assistant-message-fallback",
428 content: [{ type: "text", text: "Fallback hello" }],
429 },
430 } as unknown as SDKMessage);
431
432 harness.query.emit({
433 type: "result",
434 subtype: "success",
435 is_error: false,
436 errors: [],
437 session_id: "sdk-session-fallback-text",
438 uuid: "result-fallback",
439 } as unknown as SDKMessage);
440
441 const runtimeEvents = Array.from(yield* Fiber.join(runtimeEventsFiber));
442 assert.deepEqual(
443 runtimeEvents.map((event) => event.type),
444 [
445 "session.started",
446 "turn.started",
447 "thread.started",
448 "message.delta",
449 "message.completed",
450 "turn.completed",
451 ],
452 );
453
454 const deltaEvent = runtimeEvents.find((event) => event.type === "message.delta");
455 assert.equal(deltaEvent?.type, "message.delta");
456 if (deltaEvent?.type === "message.delta") {
457 assert.equal(deltaEvent.delta, "Fallback hello");
458 assert.equal(deltaEvent.turnId, turn.turnId);
459 }
460 }).pipe(
461 Effect.provideService(Random.Random, makeDeterministicRandomService()),
462 Effect.provide(harness.layer),
463 );
464 });
465
308 it.effect("bridges approval request/response lifecycle through canUseTool", () => {466 it.effect("bridges approval request/response lifecycle through canUseTool", () => {
309 const harness = makeHarness();467 const harness = makeHarness();
310 return Effect.gen(function* () {468 return Effect.gen(function* () {
...@@ -527,4 +685,91 @@ describe("ClaudeCodeAdapterLive", () => {...@@ -527,4 +685,91 @@ describe("ClaudeCodeAdapterLive", () => {
527 Effect.provide(harness.layer),685 Effect.provide(harness.layer),
528 );686 );
529 });687 });
688
689 it.effect("writes provider-native observability records when enabled", () => {
690 const nativeLogDir = fs.mkdtempSync(path.join(os.tmpdir(), "claude-native-events-"));
691 const nativeEventLogPath = path.join(nativeLogDir, "provider-native.ndjson");
692 const harness = makeHarness({ nativeEventLogPath });
693 return Effect.gen(function* () {
694 const adapter = yield* ClaudeCodeAdapter;
695
696 const session = yield* adapter.startSession({
697 provider: "claudeCode",
698 });
699 const turn = yield* adapter.sendTurn({
700 sessionId: session.sessionId,
701 input: "hello",
702 attachments: [],
703 });
704
705 const turnCompletedFiber = yield* Stream.filter(
706 adapter.streamEvents,
707 (event) => event.type === "turn.completed",
708 ).pipe(Stream.runHead, Effect.forkChild);
709
710 harness.query.emit({
711 type: "stream_event",
712 session_id: "sdk-session-native-log",
713 uuid: "stream-native-log",
714 parent_tool_use_id: null,
715 event: {
716 type: "content_block_delta",
717 index: 0,
718 delta: {
719 type: "text_delta",
720 text: "hi",
721 },
722 },
723 } as unknown as SDKMessage);
724
725 harness.query.emit({
726 type: "result",
727 subtype: "success",
728 is_error: false,
729 errors: [],
730 session_id: "sdk-session-native-log",
731 uuid: "result-native-log",
732 } as unknown as SDKMessage);
733
734 const turnCompleted = yield* Fiber.join(turnCompletedFiber);
735 assert.equal(turnCompleted._tag, "Some");
736
737 const lines = fs
738 .readFileSync(nativeEventLogPath, "utf8")
739 .split("\n")
740 .map((line) => line.trim())
741 .filter((line) => line.length > 0);
742 assert.equal(lines.length > 0, true);
743
744 const records = lines.map((line) => JSON.parse(line) as {
745 event?: {
746 provider?: string;
747 method?: string;
748 sessionId?: string;
749 turnId?: string;
750 };
751 });
752
753 assert.equal(records.some((record) => record.event?.provider === "claudeCode"), true);
754 assert.equal(records.some((record) => record.event?.sessionId === session.sessionId), true);
755 assert.equal(records.some((record) => record.event?.turnId === turn.turnId), true);
756 assert.equal(
757 records.some(
758 (record) => record.event?.method === "claude/stream_event/content_block_delta/text_delta",
759 ),
760 true,
761 );
762 }).pipe(
763 Effect.ensuring(
764 Effect.sync(() => {
765 fs.rmSync(nativeLogDir, {
766 recursive: true,
767 force: true,
768 });
769 }),
770 ),
771 Effect.provideService(Random.Random, makeDeterministicRandomService()),
772 Effect.provide(harness.layer),
773 );
774 });
530});775});
apps/server/src/provider/Layers/ClaudeCodeAdapter.ts+154-13
...@@ -42,6 +42,7 @@ import {...@@ -42,6 +42,7 @@ import {
42 type ProviderAdapterError,42 type ProviderAdapterError,
43} from "../Errors.ts";43} from "../Errors.ts";
44import { ClaudeCodeAdapter, type ClaudeCodeAdapterShape } from "../Services/ClaudeCodeAdapter.ts";44import { ClaudeCodeAdapter, type ClaudeCodeAdapterShape } from "../Services/ClaudeCodeAdapter.ts";
45import { makeEventNdjsonLogger } from "./EventNdjsonLogger.ts";
4546
46const PROVIDER = "claudeCode" as const;47const PROVIDER = "claudeCode" as const;
4748
...@@ -67,6 +68,8 @@ interface ClaudeTurnState {...@@ -67,6 +68,8 @@ interface ClaudeTurnState {
67 readonly startedAt: string;68 readonly startedAt: string;
68 readonly items: Array<unknown>;69 readonly items: Array<unknown>;
69 readonly messageCompleted: boolean;70 readonly messageCompleted: boolean;
71 readonly emittedTextDelta: boolean;
72 readonly fallbackAssistantText: string;
70}73}
7174
72interface PendingApproval {75interface PendingApproval {
...@@ -114,6 +117,7 @@ export interface ClaudeCodeAdapterLiveOptions {...@@ -114,6 +117,7 @@ export interface ClaudeCodeAdapterLiveOptions {
114 readonly prompt: AsyncIterable<SDKUserMessage>;117 readonly prompt: AsyncIterable<SDKUserMessage>;
115 readonly options: ClaudeQueryOptions;118 readonly options: ClaudeQueryOptions;
116 }) => ClaudeQueryRuntime;119 }) => ClaudeQueryRuntime;
120 readonly nativeEventLogPath?: string;
117}121}
118122
119function isUuid(value: string): boolean {123function isUuid(value: string): boolean {
...@@ -272,6 +276,30 @@ function turnStatusFromResult(result: SDKResultMessage): ProviderRuntimeTurnStat...@@ -272,6 +276,30 @@ function turnStatusFromResult(result: SDKResultMessage): ProviderRuntimeTurnStat
272 return "failed";276 return "failed";
273}277}
274278
279function extractAssistantText(message: SDKMessage): string {
280 if (message.type !== "assistant") {
281 return "";
282 }
283
284 const content = (message.message as { content?: unknown } | undefined)?.content;
285 if (!Array.isArray(content)) {
286 return "";
287 }
288
289 const fragments: string[] = [];
290 for (const block of content) {
291 if (!block || typeof block !== "object") {
292 continue;
293 }
294 const candidate = block as { type?: unknown; text?: unknown };
295 if (candidate.type === "text" && typeof candidate.text === "string" && candidate.text.length > 0) {
296 fragments.push(candidate.text);
297 }
298 }
299
300 return fragments.join("");
301}
302
275function toSessionError(303function toSessionError(
276 sessionId: ReturnType<typeof ProviderSessionId.makeUnsafe>,304 sessionId: ReturnType<typeof ProviderSessionId.makeUnsafe>,
277 cause: unknown,305 cause: unknown,
...@@ -311,8 +339,72 @@ function toRequestError(...@@ -311,8 +339,72 @@ function toRequestError(
311 });339 });
312}340}
313341
342function sdkMessageType(value: unknown): string | undefined {
343 if (!value || typeof value !== "object") {
344 return undefined;
345 }
346 const record = value as { type?: unknown };
347 return typeof record.type === "string" ? record.type : undefined;
348}
349
350function sdkMessageSubtype(value: unknown): string | undefined {
351 if (!value || typeof value !== "object") {
352 return undefined;
353 }
354 const record = value as { subtype?: unknown };
355 return typeof record.subtype === "string" ? record.subtype : undefined;
356}
357
358function sdkNativeMethod(message: SDKMessage): string {
359 const subtype = sdkMessageSubtype(message);
360 if (subtype) {
361 return `claude/${message.type}/${subtype}`;
362 }
363
364 if (message.type === "stream_event") {
365 const streamType = sdkMessageType(message.event);
366 if (streamType) {
367 const deltaType =
368 streamType === "content_block_delta" ? sdkMessageType((message.event as { delta?: unknown }).delta) : undefined;
369 if (deltaType) {
370 return `claude/${message.type}/${streamType}/${deltaType}`;
371 }
372 return `claude/${message.type}/${streamType}`;
373 }
374 }
375
376 return `claude/${message.type}`;
377}
378
379function sdkNativeItemId(message: SDKMessage): string | undefined {
380 if (message.type === "assistant") {
381 const maybeId = (message.message as { id?: unknown }).id;
382 if (typeof maybeId === "string") {
383 return maybeId;
384 }
385 return undefined;
386 }
387
388 if (message.type === "stream_event") {
389 const event = message.event as {
390 type?: unknown;
391 content_block?: { id?: unknown };
392 };
393 if (event.type === "content_block_start" && typeof event.content_block?.id === "string") {
394 return event.content_block.id;
395 }
396 }
397
398 return undefined;
399}
400
314function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {401function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
315 return Effect.gen(function* () {402 return Effect.gen(function* () {
403 const nativeEventLogger =
404 options?.nativeEventLogPath !== undefined
405 ? makeEventNdjsonLogger(options.nativeEventLogPath)
406 : undefined;
407
316 const createQuery =408 const createQuery =
317 options?.createQuery ??409 options?.createQuery ??
318 ((input: {410 ((input: {
...@@ -330,6 +422,38 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {...@@ -330,6 +422,38 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
330 const offerRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>422 const offerRuntimeEvent = (event: ProviderRuntimeEvent): Effect.Effect<void> =>
331 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid);423 Queue.offer(runtimeEventQueue, event).pipe(Effect.asVoid);
332424
425 const logNativeSdkMessage = (
426 context: ClaudeSessionContext,
427 message: SDKMessage,
428 ): Effect.Effect<void> =>
429 Effect.sync(() => {
430 if (!nativeEventLogger) {
431 return;
432 }
433
434 const observedAt = new Date().toISOString();
435 const itemId = sdkNativeItemId(message);
436
437 nativeEventLogger.write({
438 observedAt,
439 event: {
440 id:
441 "uuid" in message && typeof message.uuid === "string"
442 ? message.uuid
443 : crypto.randomUUID(),
444 kind: "notification",
445 provider: PROVIDER,
446 sessionId: context.session.sessionId,
447 createdAt: observedAt,
448 method: sdkNativeMethod(message),
449 ...(typeof message.session_id === "string" ? { threadId: message.session_id } : {}),
450 ...(context.turnState ? { turnId: context.turnState.turnId } : {}),
451 ...(itemId ? { itemId: ProviderItemId.makeUnsafe(itemId) } : {}),
452 payload: message,
453 },
454 });
455 });
456
333 const snapshotThread = (457 const snapshotThread = (
334 context: ClaudeSessionContext,458 context: ClaudeSessionContext,
335 ): Effect.Effect<{459 ): Effect.Effect<{
...@@ -450,6 +574,21 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {...@@ -450,6 +574,21 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
450 }574 }
451575
452 if (!turnState.messageCompleted) {576 if (!turnState.messageCompleted) {
577 if (!turnState.emittedTextDelta && turnState.fallbackAssistantText.length > 0) {
578 const deltaStamp = yield* makeEventStamp();
579 yield* offerRuntimeEvent({
580 type: "message.delta",
581 eventId: deltaStamp.eventId,
582 provider: PROVIDER,
583 sessionId: context.session.sessionId,
584 createdAt: deltaStamp.createdAt,
585 ...(context.session.threadId ? { threadId: context.session.threadId } : {}),
586 turnId: turnState.turnId,
587 itemId: ProviderItemId.makeUnsafe(turnState.assistantItemId),
588 delta: turnState.fallbackAssistantText,
589 });
590 }
591
453 const stamp = yield* makeEventStamp();592 const stamp = yield* makeEventStamp();
454 yield* offerRuntimeEvent({593 yield* offerRuntimeEvent({
455 type: "message.completed",594 type: "message.completed",
...@@ -510,6 +649,12 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {...@@ -510,6 +649,12 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
510 event.delta.text.length > 0 &&649 event.delta.text.length > 0 &&
511 context.turnState650 context.turnState
512 ) {651 ) {
652 if (!context.turnState.emittedTextDelta) {
653 context.turnState = {
654 ...context.turnState,
655 emittedTextDelta: true,
656 };
657 }
513 const stamp = yield* makeEventStamp();658 const stamp = yield* makeEventStamp();
514 yield* offerRuntimeEvent({659 yield* offerRuntimeEvent({
515 type: "message.delta",660 type: "message.delta",
...@@ -606,22 +751,15 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {...@@ -606,22 +751,15 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
606751
607 if (context.turnState) {752 if (context.turnState) {
608 context.turnState.items.push(message.message);753 context.turnState.items.push(message.message);
609 if (!context.turnState.messageCompleted) {754 const fallbackAssistantText = extractAssistantText(message);
755 if (
756 fallbackAssistantText.length > 0 &&
757 fallbackAssistantText !== context.turnState.fallbackAssistantText
758 ) {
610 context.turnState = {759 context.turnState = {
611 ...context.turnState,760 ...context.turnState,
612 messageCompleted: true,761 fallbackAssistantText,
613 };762 };
614 const stamp = yield* makeEventStamp();
615 yield* offerRuntimeEvent({
616 type: "message.completed",
617 eventId: stamp.eventId,
618 provider: PROVIDER,
619 sessionId: context.session.sessionId,
620 createdAt: stamp.createdAt,
621 itemId: ProviderItemId.makeUnsafe(context.turnState.assistantItemId),
622 ...(context.session.threadId ? { threadId: context.session.threadId } : {}),
623 turnId: context.turnState.turnId,
624 });
625 }763 }
626 }764 }
627765
...@@ -653,6 +791,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {...@@ -653,6 +791,7 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
653 message: SDKMessage,791 message: SDKMessage,
654 ): Effect.Effect<void> =>792 ): Effect.Effect<void> =>
655 Effect.gen(function* () {793 Effect.gen(function* () {
794 yield* logNativeSdkMessage(context, message);
656 yield* ensureThreadId(context, message);795 yield* ensureThreadId(context, message);
657796
658 switch (message.type) {797 switch (message.type) {
...@@ -1014,6 +1153,8 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {...@@ -1014,6 +1153,8 @@ function makeClaudeCodeAdapter(options?: ClaudeCodeAdapterLiveOptions) {
1014 startedAt: yield* nowIso,1153 startedAt: yield* nowIso,
1015 items: [],1154 items: [],
1016 messageCompleted: false,1155 messageCompleted: false,
1156 emittedTextDelta: false,
1157 fallbackAssistantText: "",
1017 };1158 };
10181159
1019 const updatedAt = yield* nowIso;1160 const updatedAt = yield* nowIso;
apps/server/src/serverLayers.ts+3-1
...@@ -57,7 +57,9 @@ export function makeServerProviderLayer(): Layer.Layer<...@@ -57,7 +57,9 @@ export function makeServerProviderLayer(): Layer.Layer<
57 const codexAdapterLayer = makeCodexAdapterLive(57 const codexAdapterLayer = makeCodexAdapterLive(
58 nativeEventLogger ? { nativeEventLogger } : undefined,58 nativeEventLogger ? { nativeEventLogger } : undefined,
59 );59 );
60 const claudeAdapterLayer = makeClaudeCodeAdapterLive();60 const claudeAdapterLayer = makeClaudeCodeAdapterLive({
61 nativeEventLogPath: path.join(providerLogsDir, "provider-native.ndjson"),
62 });
61 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(63 const adapterRegistryLayer = ProviderAdapterRegistryLive.pipe(
62 Layer.provide(Layer.mergeAll(codexAdapterLayer, claudeAdapterLayer)),64 Layer.provide(Layer.mergeAll(codexAdapterLayer, claudeAdapterLayer)),
63 Layer.provideMerge(providerSessionDirectoryLayer),65 Layer.provideMerge(providerSessionDirectoryLayer),