fix(core): publish batched deltas before the next block starts (anomalyco#51105)

Author: jlongster

packages/core/src/session/runner/publish-llm-event.ts

@@ -193,7 +193,14 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
     const flush = Effect.fnUntraced(function* () {
       for (const id of Array.from(chunks.keys())) yield* end(id)
     })
-    return { start, append, end, flush, has: (id: string) => chunks.has(id) }
+    /** Publish batched deltas now, keeping every fragment open. */
+    const publishPending = Effect.fnUntraced(function* () {
+      for (const [id, current] of Array.from(chunks)) {
+        if (current.timer) yield* Fiber.interrupt(current.timer)
+        yield* publishDelta(id)
+      }
+    })
+    return { start, append, end, flush, publishPending, has: (id: string) => chunks.has(id) }
   }
 
   const text = fragments(
@@ -255,6 +262,13 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
     }),
   )
 
+  // Deltas are batched, but block starts are not. Publishing held deltas first keeps each
+  // block's content ahead of the next block, so the published order matches the model's.
+  const publishPendingDeltas = Effect.fnUntraced(function* () {
+    yield* text.publishPending()
+    yield* reasoning.publishPending()
+  })
+
   const flushFragments = Effect.fnUntraced(function* () {
     yield* text.flush()
     yield* reasoning.flush()
@@ -276,6 +290,7 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
     }
     tools.set(event.id, tool)
     yield* toolInput.start(event.id)
+    yield* publishPendingDeltas()
     yield* bus.publish(SessionEvent.Tool.Input.Started, {
       sessionID: input.sessionID,
       assistantMessageID,
@@ -390,6 +405,7 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
       case "text-start":
         outputStarted = true
         const startedTextOrdinal = yield* text.start(event.id, providerState(event.providerMetadata))
+        yield* publishPendingDeltas()
         yield* bus.publish(SessionEvent.Text.Started, {
           sessionID: input.sessionID,
           assistantMessageID: yield* startAssistant(),
@@ -405,6 +421,7 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
       case "reasoning-start":
         outputStarted = true
         const startedReasoningOrdinal = yield* reasoning.start(event.id, providerState(event.providerMetadata))
+        yield* publishPendingDeltas()
         yield* bus.publish(SessionEvent.Reasoning.Started, {
           sessionID: input.sessionID,
           assistantMessageID: yield* startAssistant(),

packages/core/test/session-runner-tool-events.test.ts

@@ -387,6 +387,46 @@ it.effect("batches text deltas and flushes pending text before the terminal even
   }),
 )
 
+it.effect("publishes batched deltas before the next block starts", () =>
+  Effect.gen(function* () {
+    const { published, publisher } = capture()
+    const types = () =>
+      published
+        .map((event) => event.type)
+        .filter((type) => type !== "session.step.started.1" && type !== "session.step.streamed")
+    yield* Effect.forEach(
+      [
+        LLMEvent.reasoningStart({ id: "reasoning" }),
+        LLMEvent.reasoningDelta({ id: "reasoning", text: "Plan the edits." }),
+        LLMEvent.textStart({ id: "text" }),
+        LLMEvent.textDelta({ id: "text", text: "Now the edits:" }),
+        LLMEvent.toolInputStart({ id: "call", name: "edit" }),
+      ],
+      publisher.publish,
+      { discard: true },
+    )
+    expect(types()).toEqual([
+      "session.reasoning.started.1",
+      "session.reasoning.delta",
+      "session.text.started.1",
+      "session.text.delta",
+      "session.tool.input.started.1",
+    ])
+
+    // Blocks stay open: later chunks still batch, and nothing is published twice.
+    yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " more" }))
+    yield* TestClock.adjust("1 second")
+    yield* publisher.publish(LLMEvent.textEnd({ id: "text" }))
+    expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
+      { delta: "Now the edits:" },
+      { delta: " more" },
+    ])
+    expect(published.find((event) => event.type === "session.text.ended.1")?.data).toMatchObject({
+      text: "Now the edits: more",
+    })
+  }),
+)
+
 it.effect("retains new chunks and orders text-end behind an in-flight timer publication", () =>
   Effect.gen(function* () {
     const entered = yield* Deferred.make<void>()