# Files changed · PR \#24410 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/24410/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(server\): Claude answers stream as they are written](https://git.cafe/pingdotgg/t3code/pull/24410.md)

Version: 2

Head commit: a459583c71f7edee50f5ec0724db712f771398ed

Base commit: 6e792117436662208bf2b6eb71ee8d0d88817e43

Comparison base: 6e792117436662208bf2b6eb71ee8d0d88817e43

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts

Status: modified; +306; −0

```diff
@@ -3446,6 +3446,312 @@
       ),
   );
 
+  // regression: #17270 assistant text streams before the completed snapshot
+  it.effect.each([
+    { snapshot: "matching", snapshotText: "Movable type changed Europe." },
+    { snapshot: "corrected", snapshotText: "Movable type reshaped Europe." },
+    // An empty authoritative block keeps the streamed text rather than blanking it.
+    { snapshot: "empty", snapshotText: "" },
+  ])("streams Claude assistant text and settles it with a $snapshot snapshot", ({ snapshotText }) =>
+    Effect.gen(function* () {
+      const harness = yield* makeWakeHarness;
+      yield* harness.runtime.startTurn(
+        makeClaudeTestTurnInput({
+          threadId: harness.threadId,
+          providerThread: harness.providerThread,
+          now: yield* DateTime.now,
+          attemptId: RunAttemptId.make("text-stream-attempt"),
+          text: "Explain the printing press.",
+          attachments: [],
+        }),
+      );
+      const stream = (event: unknown) =>
+        claudeSdkFrame({
+          type: "stream_event",
+          event,
+          parent_tool_use_id: null,
+          session_id: WAKE_NATIVE_SESSION,
+          uuid: "text-stream-frame",
+        });
+      const assistantMessages = () =>
+        harness.events.flatMap((event) =>
+          event.type === "message.updated" && event.message.role === "assistant"
+            ? [event.message]
+            : [],
+        );
+      for (const frame of [
+        stream({ type: "message_start", message: { id: "answer-message" } }),
+        stream({
+          type: "content_block_start",
+          index: 0,
+          content_block: { type: "text", text: "" },
+        }),
+        stream({
+          type: "content_block_delta",
+          index: 0,
+          delta: { type: "text_delta", text: "Movable type " },
+        }),
+      ]) {
+        yield* harness.offerAndWait(frame);
+      }
+      yield* TestClock.adjust("1 second");
+      yield* harness.offerAndWait(
+        stream({
+          type: "content_block_delta",
+          index: 0,
+          delta: { type: "text_delta", text: "changed Europe." },
+        }),
+      );
+      yield* TestClock.adjust("1 second");
+      assert.deepEqual(
+        assistantMessages().map((message) => [message.text, message.streaming]),
+        [
+          ["Movable type ", true],
+          ["Movable type changed Europe.", true],
+        ],
+      );
+
+      yield* harness.offerAndWait(stream({ type: "content_block_stop", index: 0 }));
+      const snapshot = makeAssistantTextFrame({ uuid: "answer-snapshot", text: snapshotText });
+      assert(snapshot.type === "assistant");
+      const answerSnapshot = {
+        ...snapshot,
+        message: { ...snapshot.message, id: "answer-message" },
+      };
+      yield* harness.offerAndWait(answerSnapshot);
+      yield* harness.offerAndWait(answerSnapshot);
+      yield* Queue.offer(
+        harness.sdkMessages,
+        makeResultFrame({ uuid: "text-stream-result", result: "Movable type changed Europe." }),
+      );
+      yield* Queue.take(harness.terminalReceipts);
+      const messages = assistantMessages();
+      assert.lengthOf(messages, 3);
+      assert.equal(new Set(messages.map((message) => message.id)).size, 1);
+      assert.equal(
+        new Set(messages.map((message) => DateTime.toEpochMillis(message.createdAt))).size,
+        1,
+      );
+      assert.equal(messages.at(-1)?.text, snapshotText || "Movable type changed Europe.");
+      assert.isFalse(messages.at(-1)?.streaming);
+      const items = harness.events.flatMap((event) =>
+        event.type === "turn_item.updated" && event.turnItem.type === "assistant_message"
+          ? [event.turnItem]
+          : [],
+      );
+      assert.deepEqual(
+        items.map((item) => item.status),
+        ["running", "running", "completed"],
+      );
+      assert.equal(new Set(items.map((item) => `${item.id}:${item.ordinal}`)).size, 1);
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(
+        Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+      ),
+    ),
+  );
+
+  // regression: #17270 streamed text without a snapshot settles with the turn
+  it.effect.each([
+    { outcome: "result", flushed: true, turnStatus: "completed" },
+    { outcome: "error_result", flushed: true, turnStatus: "failed" },
+    { outcome: "query_exit", flushed: true, turnStatus: "failed" },
+    // Stop inside the 50ms coalescer window, before any delta was emitted.
+    { outcome: "interrupt", flushed: false, turnStatus: "interrupted" },
+  ] as const)(
+    "settles streamed Claude text without a snapshot on $outcome",
+    ({ outcome, flushed, turnStatus }) =>
+      Effect.gen(function* () {
+        const harness = yield* makeWakeHarnessWithOptions({
+          close: (sdkMessages) => Queue.shutdown(sdkMessages),
+        });
+        const attemptId = RunAttemptId.make(`text-stream-${outcome}`);
+        yield* harness.runtime.startTurn(
+          makeClaudeTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now: yield* DateTime.now,
+            attemptId,
+            text: "Explain the printing press.",
+            attachments: [],
+          }),
+        );
+        const stream = (event: unknown) =>
+          claudeSdkFrame({
+            type: "stream_event",
+            event,
+            parent_tool_use_id: null,
+            session_id: WAKE_NATIVE_SESSION,
+            uuid: "text-stream-frame",
+          });
+        for (const frame of [
+          stream({ type: "message_start", message: { id: "partial-answer" } }),
+          stream({
+            type: "content_block_start",
+            index: 0,
+            content_block: { type: "text", text: "Movable " },
+          }),
+          stream({
+            type: "content_block_delta",
+            index: 0,
+            delta: { type: "text_delta", text: "type" },
+          }),
+        ]) {
+          yield* harness.offerAndWait(frame);
+        }
+        if (flushed) yield* TestClock.adjust("1 second");
+        const assistantMessages = () =>
+          harness.events.flatMap((event) =>
+            event.type === "message.updated" && event.message.role === "assistant"
+              ? [event.message]
+              : [],
+          );
+        assert.lengthOf(assistantMessages(), flushed ? 1 : 0);
+
+        if (outcome === "query_exit") {
+          yield* Queue.shutdown(harness.sdkMessages);
+        } else if (outcome === "interrupt") {
+          const idAllocator = yield* IdAllocator.IdAllocatorV2;
+          yield* harness.runtime.interruptTurn({
+            providerThread: harness.providerThread,
+            providerTurnId: idAllocator.derive.providerTurn({
+              driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
+              nativeTurnId: `turn:${attemptId}`,
+            }),
+          });
+        } else {
+          yield* Queue.offer(
+            harness.sdkMessages,
+            makeResultFrame({
+              uuid: `text-stream-${outcome}-result`,
+              result: "Movable type",
+              ...(outcome === "error_result"
+                ? { subtype: "error_during_execution", isError: true }
+                : {}),
+            }),
+          );
+        }
+        const terminal = yield* Queue.take(harness.terminalReceipts);
+        assert.equal(terminal.status, turnStatus);
+        const messages = assistantMessages();
+        assert.equal(new Set(messages.map((message) => message.id)).size, 1);
+        assert.equal(messages.at(-1)?.text, "Movable type");
+        assert.isFalse(messages.at(-1)?.streaming);
+        const items = harness.events.flatMap((event) =>
+          event.type === "turn_item.updated" && event.turnItem.type === "assistant_message"
+            ? [event.turnItem]
+            : [],
+        );
+        assert.equal(items.at(-1)?.status, "completed");
+        assert.isNotNull(items.at(-1)?.completedAt);
+      }).pipe(
+        Effect.scoped,
+        Effect.provide(
+          Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+        ),
+      ),
+  );
+
+  // regression: #17270 stream frames that carry no main-thread text emit nothing
+  it.effect.each([
+    {
+      response: "result-only",
+      result: "Only the result.",
+      emptySnapshot: false,
+      expected: ["Only the result."],
+    },
+    {
+      response: "empty-snapshot result",
+      result: "Only the result.",
+      emptySnapshot: true,
+      expected: ["Only the result."],
+    },
+    { response: "empty", result: "", emptySnapshot: true, expected: [] },
+  ])(
+    "projects a $response Claude response without phantom text",
+    ({ response, result, emptySnapshot, expected }) =>
+      Effect.gen(function* () {
+        const harness = yield* makeWakeHarness;
+        yield* harness.runtime.startTurn(
+          makeClaudeTestTurnInput({
+            threadId: harness.threadId,
+            providerThread: harness.providerThread,
+            now: yield* DateTime.now,
+            attemptId: RunAttemptId.make(`text-stream-${response}`),
+            text: "Explain the printing press.",
+            attachments: [],
+          }),
+        );
+        const stream = (event: unknown, parent: string | null = null) =>
+          claudeSdkFrame({
+            type: "stream_event",
+            event,
+            parent_tool_use_id: parent,
+            session_id: WAKE_NATIVE_SESSION,
+            uuid: "text-stream-frame",
+          });
+        for (const frame of [
+          stream({ type: "message_start", message: { id: "quiet-message" } }),
+          stream({
+            type: "content_block_start",
+            index: 0,
+            content_block: { type: "text", text: "" },
+          }),
+          stream({ type: "content_block_stop", index: 0 }),
+          stream({
+            type: "content_block_delta",
+            index: 5,
+            delta: { type: "text_delta", text: "Stray delta" },
+          }),
+          stream({ type: "message_start", message: { id: "child-message" } }, "child-tool"),
+          stream(
+            {
+              type: "content_block_start",
+              index: 0,
+              content_block: { type: "text", text: "Child " },
+            },
+            "child-tool",
+          ),
+          stream(
+            { type: "content_block_delta", index: 0, delta: { type: "text_delta", text: "text" } },
+            "child-tool",
+          ),
+        ]) {
+          yield* harness.offerAndWait(frame);
+        }
+        if (emptySnapshot) {
+          const snapshot = makeAssistantTextFrame({ uuid: "quiet-snapshot", text: "" });
+          assert(snapshot.type === "assistant");
+          yield* harness.offerAndWait({
+            ...snapshot,
+            message: { ...snapshot.message, id: "quiet-message" },
+          });
+        }
+        yield* TestClock.adjust("1 second");
+        yield* Queue.offer(
+          harness.sdkMessages,
+          makeResultFrame({ uuid: `text-stream-${response}-result`, result }),
+        );
+        yield* Queue.take(harness.terminalReceipts);
+        const messages = harness.events.flatMap((event) =>
+          event.type === "message.updated" && event.message.role === "assistant"
+            ? [event.message]
+            : [],
+        );
+        assert.deepEqual(
+          messages.map((message) => message.text),
+          expected,
+        );
+        for (const message of messages) assert.isFalse(message.streaming);
+      }).pipe(
+        Effect.scoped,
+        Effect.provide(
+          Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer),
+        ),
+      ),
+  );
+
   // regression: #15892 non-execution metadata is carried through, not guessed from result text
   it.effect.each(["cancelled", "denied", "permission_denied", undefined])(
     "preserves native tool non-execution metadata %s without inferring a denial from text",
```

## apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts

Status: modified; +128; −12

```diff
@@ -3279,6 +3279,8 @@
   readonly ordinal: number;
   readonly startedAt: DateTime.Utc;
   readonly completedAt: DateTime.Utc;
+  // A streamed block keeps its start as createdAt across updates.
+  readonly streaming?: boolean;
 }): {
   readonly node: OrchestrationV2ExecutionNode;
   readonly message: OrchestrationV2ConversationMessage;
@@ -3310,7 +3312,7 @@
       parentNodeId: input.turnInput.rootNodeId,
       rootNodeId: input.turnInput.rootNodeId,
       kind: "assistant_message",
+      status: input.streaming === true ? "running" : "completed",
-      status: "completed",
       countsForRun: false,
       providerThreadId: input.turnInput.providerThread.id,
       providerTurnId: input.providerTurnId,
@@ -3318,7 +3320,7 @@
       runtimeRequestId: null,
       checkpointScopeId: null,
       startedAt: input.startedAt,
+      completedAt: input.streaming === true ? null : input.completedAt,
-      completedAt: input.completedAt,
     },
     message: {
       createdBy: "agent",
@@ -3330,8 +3332,8 @@
       role: "assistant",
       text: input.text,
       attachments: [],
-      streaming: false,
-      createdAt: input.completedAt,
+      streaming: input.streaming === true,
+      createdAt: input.streaming === undefined ? input.completedAt : input.startedAt,
       updatedAt: input.completedAt,
     },
     turnItem: {
@@ -3344,15 +3346,15 @@
       nativeItemRef,
       parentItemId: null,
       ordinal: input.ordinal,
+      status: input.streaming === true ? "running" : "completed",
-      status: "completed",
       title: null,
       startedAt: input.startedAt,
-      completedAt: input.completedAt,
+      completedAt: input.streaming === true ? null : input.completedAt,
       updatedAt: input.completedAt,
       type: "assistant_message",
       messageId,
       text: input.text,
-      streaming: false,
+      streaming: input.streaming === true,
     },
   };
 }
@@ -3413,6 +3415,15 @@
     fallbackText: string;
     fallbackNativeItemId: string;
     emittedNativeItemIds: Set<string>;
+    // Text blocks streamed under `${message.id}:text:${n}`; the completed
+    // snapshot has its own uuid, so it finishes the streamed item instead.
+    readonly streamBlocks: Map<number, string>;
+    readonly nextBlockIndex: Map<string, number>;
+    readonly snapshotBlockIndex: Map<string, number>;
+    readonly streamItems: Map<
+      string,
+      { readonly startedAt: DateTime.Utc; readonly ordinal: number }
+    >;
     // The CLI repeats a notice that ends a prompt as the turn's result text.
     noticeTexts: Set<string>;
   };
@@ -6579,6 +6590,46 @@
             }),
         });
 
+        const assistantDeltas = yield* makeProviderTextDeltaCoalescer({
+          flushIntervalMs: 50,
+          emit:
+            /* Provider events include streaming deltas; avoid a span for each update. */
+            Effect.fnUntraced(function* (update) {
+              const context = yield* Ref.get(activeTurn);
+              if (context === null || context.nativeTurnId !== update.turnId) return;
+              const item = context.assistant.streamItems.get(update.itemId);
+              if (item === undefined || update.text.length === 0) return;
+              if (update.completed) context.assistant.streamItems.delete(update.itemId);
+              context.assistant.emittedNativeItemIds.add(update.itemId);
+              const artifacts = buildAssistantArtifacts({
+                idAllocator,
+                turnInput: context.input,
+                providerTurnId: context.providerTurnId,
+                nativeItemId: update.itemId,
+                text: update.text,
+                ordinal: item.ordinal,
+                startedAt: item.startedAt,
+                completedAt: yield* DateTime.now,
+                streaming: !update.completed,
+              });
+              yield* emitProviderEvent({
+                type: "node.updated",
+                driver: CLAUDE_PROVIDER,
+                node: artifacts.node,
+              });
+              yield* emitProviderEvent({
+                type: "message.updated",
+                driver: CLAUDE_PROVIDER,
+                message: artifacts.message,
+              });
+              yield* emitProviderEvent({
+                type: "turn_item.updated",
+                driver: CLAUDE_PROVIDER,
+                turnItem: artifacts.turnItem,
+              });
+            }),
+        });
+
         const endToolCalls = Effect.fnUntraced(function* (input: {
           readonly context: ActiveClaudeTurnContext | null;
           readonly toolCalls: ReadonlyArray<ActiveClaudeToolCall>;
@@ -6640,6 +6691,7 @@
           readonly result?: SDKResultMessage;
         }) {
           yield* reasoningDeltas.flushTurn(input.context.nativeTurnId);
+          yield* assistantDeltas.flushTurn(input.context.nativeTurnId);
           if (input.context.compaction?.status === "running") {
             yield* emitProviderEvent({
               type: "turn_item.updated",
@@ -7611,6 +7663,7 @@
             if (event.type === "message_start") {
               reasoning.messageId = event.message.id;
               reasoning.streamBlocks.clear();
+              context.assistant.streamBlocks.clear();
               const usage: ClaudeMessageUsage | undefined = event.message.usage;
               context.streamUsage =
                 usage === undefined ? null : { messageId: event.message.id, usage };
@@ -7649,6 +7702,34 @@
                   delta: event.delta.thinking,
                 });
               }
+            } else if (
+              event.type === "content_block_start" &&
+              event.content_block.type === "text" &&
+              reasoning.messageId !== null
+            ) {
+              const assistant = context.assistant;
+              const index = assistant.nextBlockIndex.get(reasoning.messageId) ?? 0;
+              assistant.nextBlockIndex.set(reasoning.messageId, index + 1);
+              const itemId = `${reasoning.messageId}:text:${index}`;
+              assistant.streamBlocks.set(event.index, itemId);
+              assistant.streamItems.set(itemId, {
+                startedAt: yield* DateTime.now,
+                ordinal: yield* resolveItemOrdinal(context, itemId),
+              });
+              yield* assistantDeltas.append({
+                turnId: context.nativeTurnId,
+                itemId,
+                delta: event.content_block.text,
+              });
+            } else if (event.type === "content_block_delta" && event.delta.type === "text_delta") {
+              const itemId = context.assistant.streamBlocks.get(event.index);
+              if (itemId !== undefined) {
+                yield* assistantDeltas.append({
+                  turnId: context.nativeTurnId,
+                  itemId,
+                  delta: event.delta.text,
+                });
+              }
             } else if (event.type === "content_block_stop") {
               const itemId = reasoning.streamBlocks.get(event.index);
               if (itemId !== undefined) {
@@ -7662,6 +7743,8 @@
             }
             return;
           }
+          // Snapshot text the stream did not already deliver, when it delivered some.
+          let unstreamedSnapshotText: string | null = null;
           if (
             message.type === "assistant" &&
             !message.parent_tool_use_id &&
@@ -7670,7 +7753,26 @@
             context.reasoning.snapshots.add(message.uuid);
             // The SDK emits one assistant snapshot per completed content block.
             // Count thinking blocks separately so snapshots share the streamed ID.
+            const unstreamedText: Array<string> = [];
+            let streamedText = false;
             for (const block of message.message.content) {
+              if (block.type === "text") {
+                const index = context.assistant.snapshotBlockIndex.get(message.message.id) ?? 0;
+                context.assistant.snapshotBlockIndex.set(message.message.id, index + 1);
+                const itemId = `${message.message.id}:text:${index}`;
+                if (context.assistant.streamItems.has(itemId)) {
+                  const completedText = yield* assistantDeltas.complete({
+                    turnId: context.nativeTurnId,
+                    itemId,
+                    ...(block.text ? { finalText: block.text } : {}),
+                    emitEmpty: false,
+                  });
+                  // An empty streamed block delivered nothing, so the result fallback still applies.
+                  if (completedText.length > 0) streamedText = true;
+                } else {
+                  unstreamedText.push(block.text);
+                }
+              }
               if (block.type !== "thinking") {
                 const update = context.reasoning.lastThinking;
                 context.reasoning.lastThinking = null;
@@ -7710,6 +7812,13 @@
                   }
                 : null;
             }
+            if (streamedText) {
+              unstreamedSnapshotText = unstreamedText.join("");
+              // A repeated snapshot frame must not re-emit the streamed text under its uuid.
+              if (unstreamedSnapshotText.length === 0) {
+                context.assistant.emittedNativeItemIds.add(message.uuid);
+              }
+            }
           }
 
           if (message.type === "assistant" && input.replayed !== true) {
@@ -8485,11 +8594,14 @@
             return;
           }
           if (assistantText !== null && assistantText.text.length > 0) {
+            const text = unstreamedSnapshotText ?? assistantText.text;
+            if (text.length > 0) {
+              yield* emitAssistantTextArtifacts({
+                context,
+                nativeItemId: assistantText.nativeItemId,
-            yield* emitAssistantTextArtifacts({
+                text,
+              });
+            }
-              context,
-              nativeItemId: assistantText.nativeItemId,
-              text: assistantText.text,
-            });
             return;
           }
 
@@ -9515,6 +9627,10 @@
                 fallbackText: "",
                 fallbackNativeItemId: `assistant:${turnInput.runId}`,
                 emittedNativeItemIds: new Set(),
+                streamBlocks: new Map(),
+                nextBlockIndex: new Map(),
+                snapshotBlockIndex: new Map(),
+                streamItems: new Map(),
                 noticeTexts: new Set(),
               },
               reasoning: {
```