FN-5789: normalize missing sentence spaces in streamed deltas

Repair provider-agnostic streaming deltas so sentence boundaries keep expected spacing in chat and agent logs.

- add shared streaming-delta normalization helpers for text/thinking events
- apply normalization at executor and PI message_update chokepoints before emitting deltas
- add focused engine tests covering punctuation-boundary repairs and no-op cases
- add a patch changeset for @runfusion/fusion describing the fix

Files changed:
 .changeset/fn-5789-streaming-space-repair.md       |  5 ++
 .../engine/src/__tests__/streaming-delta.test.ts   | 84 ++++++++++++++++++++
 packages/engine/src/executor.ts                    | 11 ++-
 packages/engine/src/pi.ts                          | 13 +++-
 packages/engine/src/streaming-delta.ts             | 91 ++++++++++++++++++++++
 5 files changed, 197 insertions(+), 7 deletions(-)

Fusion-Task-Id: FN-5789

Fusion-Task-Lineage: e60ed2ae-c006-42f9-9ab0-f4e79560fb66

packages/engine/src/__tests__/streaming-delta.test.ts

@@ -0,0 +1,84 @@
+import { describe, expect, it } from "vitest";
+import { normalizeStreamingDelta, normalizeStreamingDeltaFromEvent } from "../streaming-delta.js";
+
+describe("normalizeStreamingDelta", () => {
+  it("repairs period + uppercase sentence boundaries across deltas", () => {
+    expect(normalizeStreamingDelta("Let's compare them.", "Good overview.")).toBe(" Good overview.");
+  });
+
+  it("repairs punctuation boundaries for quoted, bracketed, and numeric starts", () => {
+    expect(normalizeStreamingDelta("Done.", "\"Quoted\"")).toBe(" \"Quoted\"");
+    expect(normalizeStreamingDelta("Great!", "(Next)")).toBe(" (Next)");
+    expect(normalizeStreamingDelta("Ready?", "[Checklist]")).toBe(" [Checklist]");
+    expect(normalizeStreamingDelta("Phase complete.", "2 more items")).toBe(" 2 more items");
+    expect(normalizeStreamingDelta("Ready.", "'Single quote start'"))
+      .toBe(" 'Single quote start'");
+  });
+
+  it("does not alter lowercase continuations or property access", () => {
+    expect(normalizeStreamingDelta("foo.", "bar")).toBe("bar");
+    expect(normalizeStreamingDelta("obj", ".prop")).toBe(".prop");
+  });
+
+  it("is idempotent when whitespace already exists", () => {
+    expect(normalizeStreamingDelta("...task.", " Foundation")).toBe(" Foundation");
+  });
+});
+
+describe("normalizeStreamingDeltaFromEvent", () => {
+  it("derives previous text from same text block across deltas", () => {
+    const partial = {
+      content: [
+        { type: "text", text: "execution.Foundation" },
+      ],
+    };
+
+    expect(normalizeStreamingDeltaFromEvent(partial, 0, "Foundation", "text")).toBe(" Foundation");
+  });
+
+  it("repairs cross-block text boundaries when current block is empty", () => {
+    const partial = {
+      content: [
+        { type: "text", text: "task." },
+        { type: "text", text: "" },
+      ],
+    };
+
+    expect(normalizeStreamingDeltaFromEvent(partial, 1, "Let us continue.", "text")).toBe(" Let us continue.");
+  });
+
+  it("repairs thinking deltas across thinking blocks", () => {
+    const partial = {
+      content: [
+        { type: "thinking", thinking: "render." },
+        { type: "thinking", thinking: "" },
+      ],
+    };
+
+    expect(normalizeStreamingDeltaFromEvent(partial, 1, "Done", "thinking")).toBe(" Done");
+  });
+
+  it("returns delta unchanged for defensive edge cases", () => {
+    expect(normalizeStreamingDeltaFromEvent(undefined, 0, "Foundation", "text")).toBe("Foundation");
+
+    const outOfRange = { content: [{ type: "text", text: "execution" }] };
+    expect(normalizeStreamingDeltaFromEvent(outOfRange, 3, "Foundation", "text")).toBe("Foundation");
+
+    const wrongType = { content: [{ type: "thinking", thinking: "execution." }] };
+    expect(normalizeStreamingDeltaFromEvent(wrongType, 0, "Foundation", "text")).toBe("Foundation");
+  });
+
+  it("matches wiring payload shape for execution.Foundation event forwarding", () => {
+    const msgEvent = {
+      contentIndex: 0,
+      delta: "Foundation",
+      partial: {
+        content: [{ type: "text", text: "execution.Foundation" }],
+      },
+    };
+
+    expect(
+      normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text"),
+    ).toBe(" Foundation");
+  });
+});

packages/engine/src/executor.ts

@@ -134,6 +134,7 @@ import {
   createTaskLogTool as sharedCreateTaskLogTool,
 } from "./agent-tools.js";
 import { getTaskCompletionBlockerForStore } from "./task-completion.js";
+import { normalizeStreamingDeltaFromEvent } from "./streaming-delta.js";
 import {
   getEnabledPluginTools,
   getResearchGuidanceForSurface,
@@ -8289,10 +8290,14 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit
         if (event.type === "message_update") {
           const msgEvent = event.assistantMessageEvent;
           if (msgEvent.type === "text_delta") {
-            output += msgEvent.delta;
-            agentLogger.onText(msgEvent.delta);
+            // Repair dropped sentence-boundary spaces for all providers at the shared engine delta chokepoint.
+            const delta = normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text");
+            output += delta;
+            agentLogger.onText(delta);
           } else if (msgEvent.type === "thinking_delta") {
-            agentLogger.onThinking(msgEvent.delta);
+            // Repair dropped sentence-boundary spaces for all providers at the shared engine delta chokepoint.
+            const delta = normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "thinking");
+            agentLogger.onThinking(delta);
           }
         }
         if (event.type === "tool_execution_start") {

packages/engine/src/pi.ts

@@ -58,6 +58,7 @@ import {
 import { resolvePermanentAgentToolDecision } from "./permanent-agent-gating.js";
 import type { SystemPromptLayers } from "./prompt-layers.js";
 import { READONLY_ALLOWLIST, filterCustomToolsForReadonly, isReadonlyAllowed } from "./workflow-step-tool-policy.js";
+import { normalizeStreamingDeltaFromEvent } from "./streaming-delta.js";
 
 const RTK_ACCEPTED_REWRITE_EXIT_CODES = new Set([0, 3]);
 const RTK_EXPECTED_PASSTHROUGH_EXIT_CODES = new Set([1, 2]);
@@ -2121,9 +2122,11 @@ export async function createFnAgent(options: AgentOptions): Promise<AgentResult>
       if (event.type === "message_update") {
         const msgEvent = event.assistantMessageEvent;
         if (msgEvent.type === "text_delta") {
-          options.onText?.(msgEvent.delta);
+          // Repair dropped sentence-boundary spaces for all providers at the shared engine delta chokepoint.
+          options.onText?.(normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text"));
         } else if (msgEvent.type === "thinking_delta") {
-          options.onThinking?.(msgEvent.delta);
+          // Repair dropped sentence-boundary spaces for all providers at the shared engine delta chokepoint.
+          options.onThinking?.(normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "thinking"));
         }
       }
       if (event.type === "tool_execution_start") {
@@ -2286,9 +2289,11 @@ export async function createFnAgent(options: AgentOptions): Promise<AgentResult>
     if (event.type === "message_update") {
       const msgEvent = event.assistantMessageEvent;
       if (msgEvent.type === "text_delta") {
-        options.onText?.(msgEvent.delta);
+        // Repair dropped sentence-boundary spaces for all providers at the shared engine delta chokepoint.
+        options.onText?.(normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text"));
       } else if (msgEvent.type === "thinking_delta") {
-        options.onThinking?.(msgEvent.delta);
+        // Repair dropped sentence-boundary spaces for all providers at the shared engine delta chokepoint.
+        options.onThinking?.(normalizeStreamingDeltaFromEvent(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "thinking"));
       }
     }
     if (event.type === "tool_execution_start") {

packages/engine/src/streaming-delta.ts

@@ -0,0 +1,91 @@
+type StreamingContentBlock = {
+  type?: string;
+  text?: string;
+  thinking?: string;
+};
+
+type StreamingPartialMessage = {
+  content?: StreamingContentBlock[];
+};
+
+export function normalizeStreamingDelta(previousText: string, nextDelta: string): string {
+  if (!previousText || !nextDelta) {
+    return nextDelta;
+  }
+
+  const previousChar = previousText.slice(-1);
+  const nextChar = nextDelta[0] ?? "";
+
+  if (/\s/.test(previousChar) || /\s/.test(nextChar)) {
+    return nextDelta;
+  }
+
+  // Claude sometimes splits adjacent sentences across separate deltas or text
+  // blocks without preserving the separating space. Only repair the specific
+  // "sentence punctuation + uppercase/quoted sentence start" case so code,
+  // domains, and lowercase continuations remain untouched.
+  if (/[.!?]/.test(previousChar) && /[A-Z0-9"'([]/.test(nextChar)) {
+    return ` ${nextDelta}`;
+  }
+
+  return nextDelta;
+}
+
+function getContentText(block: StreamingContentBlock | undefined, kind: "text" | "thinking"): string {
+  if (!block || block.type !== kind) {
+    return "";
+  }
+  if (kind === "text") {
+    return typeof block.text === "string" ? block.text : "";
+  }
+  return typeof block.thinking === "string" ? block.thinking : "";
+}
+
+function derivePreviousText(accumulatedText: string, delta: string): string {
+  if (!accumulatedText || !delta) {
+    return accumulatedText;
+  }
+  return accumulatedText.endsWith(delta)
+    ? accumulatedText.slice(0, Math.max(0, accumulatedText.length - delta.length))
+    : accumulatedText;
+}
+
+function findPreviousBlockText(
+  partial: StreamingPartialMessage,
+  contentIndex: number,
+  kind: "text" | "thinking",
+): string {
+  const content = partial.content;
+  if (!Array.isArray(content)) {
+    return "";
+  }
+
+  for (let i = contentIndex - 1; i >= 0; i--) {
+    const text = getContentText(content[i], kind);
+    if (text) {
+      return text;
+    }
+  }
+  return "";
+}
+
+export function normalizeStreamingDeltaFromEvent(
+  partial: StreamingPartialMessage | undefined,
+  contentIndex: number,
+  delta: string,
+  kind: "text" | "thinking",
+): string {
+  const content = partial?.content;
+  const block = Array.isArray(content) && Number.isInteger(contentIndex) && contentIndex >= 0
+    ? content[contentIndex]
+    : undefined;
+
+  const accumulatedText = getContentText(block, kind);
+  let previousText = derivePreviousText(accumulatedText, delta);
+
+  if (!previousText && partial && Number.isInteger(contentIndex) && contentIndex > 0) {
+    previousText = findPreviousBlockText(partial, contentIndex, kind);
+  }
+
+  return normalizeStreamingDelta(previousText, delta);
+}