root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

FN-5803: normalize streamed sentence spacing across message boundaries

Browse Source 
Ensure streamed agent text preserves sentence-boundary spacing even when providers split output across message boundaries.

- add stateful streaming delta normalizer that tracks prior text/thinking tail when partial blocks reset
- update executor and PI session subscriptions to use per-session normalizer instances for text_delta and thinking_delta events
- expand streaming-delta tests to cover cross-message/tool-call boundary spacing regressions
- add task notes documenting cross-message spacing refinement

Files changed:
 .../fn-5803-streaming-cross-message-space.md       |  5 ++
 .../engine/src/__tests__/streaming-delta.test.ts   | 86 +++++++++++++++++++++-
 packages/engine/src/executor.ts                    | 13 ++--
 packages/engine/src/pi.ts                          | 24 +++---
 packages/engine/src/streaming-delta.ts             | 44 ++++++++++-
 5 files changed, 155 insertions(+), 17 deletions(-)

Fusion-Task-Id: FN-5803

Fusion-Task-Lineage: 8f88d54c-bbc0-44dd-99fc-c69738ce0ca8
This commit is contained in:
gsxdsm
2026-05-31 23:24:53 -07:00
parent a44b6102e3
commit feceedbf7e
5 changed files with 155 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
.changeset/fn-5803-streaming-cross-message-space.md Normal file
View File

@@ -0,0 +1,5 @@
---
"@runfusion/fusion": patch
---
Repair dropped spaces after sentence-ending punctuation when streamed agent text is split across separate assistant messages by tool-call round-trips (chat and agent logs), by tracking a per-session running tail at the shared engine streaming-delta chokepoints. Completes FN-5789, which only covered within-message boundaries.

86
packages/engine/src/__tests__/streaming-delta.test.ts
View File

@@ -1,5 +1,9 @@
import { describe, expect, it } from "vitest";
import { normalizeStreamingDelta, normalizeStreamingDeltaFromEvent } from "../streaming-delta.js";
import {
createStreamingDeltaNormalizer,
normalizeStreamingDelta,
normalizeStreamingDeltaFromEvent,
} from "../streaming-delta.js";
describe("normalizeStreamingDelta", () => {
it("repairs period + uppercase sentence boundaries across deltas", () => {
@@ -82,3 +86,83 @@ describe("normalizeStreamingDeltaFromEvent", () => {
).toBe(" Foundation");
});
});
describe("createStreamingDeltaNormalizer", () => {
it("repairs punctuation boundaries across separate assistant messages", () => {
const normalizer = createStreamingDeltaNormalizer();
normalizer.normalize(
{ content: [{ type: "text", text: "create the foundation task." }] },
0,
"create the foundation task.",
"text",
);
expect(
normalizer.normalize({ content: [{ type: "text", text: "Foundation" }] }, 0, "Foundation", "text"),
).toBe(" Foundation");
normalizer.normalize({ content: [{ type: "text", text: "dependent tasks." }] }, 0, "dependent tasks.", "text");
expect(normalizer.normalize({ content: [{ type: "text", text: "Let me add" }] }, 0, "Let me add", "text"))
.toBe(" Let me add");
normalizer.normalize({ content: [{ type: "text", text: "render." }] }, 0, "render.", "text");
expect(normalizer.normalize({ content: [{ type: "text", text: "Done. Filed 5" }] }, 0, "Done. Filed 5", "text"))
.toBe(" Done. Filed 5");
});
it("preserves same-message behavior and lower-case/property continuations", () => {
const normalizer = createStreamingDeltaNormalizer();
expect(
normalizer.normalize(
{
content: [
{ type: "text", text: "task." },
{ type: "text", text: "" },
],
},
1,
"Let us continue.",
"text",
),
).toBe(" Let us continue.");
expect(normalizer.normalize({ content: [{ type: "text", text: "obj.prop" }] }, 0, ".prop", "text")).toBe(".prop");
expect(normalizer.normalize({ content: [{ type: "text", text: "foo.bar" }] }, 0, "bar", "text")).toBe("bar");
});
it("is idempotent when incoming deltas already start with whitespace", () => {
const normalizer = createStreamingDeltaNormalizer();
normalizer.normalize({ content: [{ type: "text", text: "...task." }] }, 0, "...task.", "text");
expect(normalizer.normalize({ content: [{ type: "text", text: " Foundation" }] }, 0, " Foundation", "text"))
.toBe(" Foundation");
});
it("does not leak tails across text/thinking kinds", () => {
const thinkingFirst = createStreamingDeltaNormalizer();
thinkingFirst.normalize({ content: [{ type: "thinking", thinking: "reason." }] }, 0, "reason.", "thinking");
expect(thinkingFirst.normalize({ content: [{ type: "text", text: "Foundation" }] }, 0, "Foundation", "text"))
.toBe("Foundation");
const textFirst = createStreamingDeltaNormalizer();
textFirst.normalize({ content: [{ type: "text", text: "task." }] }, 0, "task.", "text");
expect(textFirst.normalize({ content: [{ type: "thinking", thinking: "Done" }] }, 0, "Done", "thinking"))
.toBe("Done");
});
it("starts fresh per instance", () => {
const normalizer = createStreamingDeltaNormalizer();
expect(normalizer.normalize(undefined, 0, "Foundation", "text")).toBe("Foundation");
});
it("is defensive for invalid partial/content index and wrong block type", () => {
const normalizer = createStreamingDeltaNormalizer();
expect(normalizer.normalize(undefined, 0, "Foundation", "text")).toBe("Foundation");
expect(normalizer.normalize({ content: [{ type: "text", text: "execution" }] }, 8, "Foundation", "text"))
.toBe("Foundation");
expect(normalizer.normalize({ content: [{ type: "thinking", thinking: "execution." }] }, 0, "Foundation", "text"))
.toBe("Foundation");
normalizer.normalize({ content: [{ type: "text", text: "task." }] }, 0, "task.", "text");
expect(normalizer.normalize(undefined, Number.NaN, "Foundation", "text")).toBe(" Foundation");
});
});

13
packages/engine/src/executor.ts
View File

@@ -134,7 +134,7 @@ import {
createTaskLogTool as sharedCreateTaskLogTool,
} from "./agent-tools.js";
import { getTaskCompletionBlockerForStore } from "./task-completion.js";
import { normalizeStreamingDeltaFromEvent } from "./streaming-delta.js";
import { createStreamingDeltaNormalizer } from "./streaming-delta.js";
import {
getEnabledPluginTools,
getResearchGuidanceForSurface,
@@ -8286,17 +8286,20 @@ Backward compat fallback: if JSON is unavailable, you may still begin output wit
this.setActiveWorkflowStepSession(task.id, session, worktreePath);
let output = "";
const deltaNormalizer = createStreamingDeltaNormalizer();
session.subscribe((event) => {
if (event.type === "message_update") {
const msgEvent = event.assistantMessageEvent;
if (msgEvent.type === "text_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");
// Repair dropped sentence-boundary spaces at the shared engine delta chokepoint,
// including tool-call cross-message boundaries (see streaming-delta.ts).
const delta = deltaNormalizer.normalize(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text");
output += delta;
agentLogger.onText(delta);
} else if (msgEvent.type === "thinking_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");
// Repair dropped sentence-boundary spaces at the shared engine delta chokepoint,
// including tool-call cross-message boundaries (see streaming-delta.ts).
const delta = deltaNormalizer.normalize(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "thinking");
agentLogger.onThinking(delta);
}
}

24
packages/engine/src/pi.ts
View File

@@ -58,7 +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";
import { createStreamingDeltaNormalizer } from "./streaming-delta.js";
const RTK_ACCEPTED_REWRITE_EXIT_CODES = new Set([0, 3]);
const RTK_EXPECTED_PASSTHROUGH_EXIT_CODES = new Set([1, 2]);
@@ -2118,15 +2118,18 @@ export async function createFnAgent(options: AgentOptions): Promise<AgentResult>
sessionManager as unknown as SessionManagerLike,
);
(targetSession as any).__fusionMemoryAppendAvailable = options.customTools?.some((tool) => tool.name === FN_MEMORY_APPEND_TOOL_NAME) === true;
const deltaNormalizer = createStreamingDeltaNormalizer();
targetSession.subscribe((event) => {
if (event.type === "message_update") {
const msgEvent = event.assistantMessageEvent;
if (msgEvent.type === "text_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"));
// Repair dropped sentence-boundary spaces at the shared engine delta chokepoint,
// including tool-call cross-message boundaries (see streaming-delta.ts).
options.onText?.(deltaNormalizer.normalize(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text"));
} else if (msgEvent.type === "thinking_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"));
// Repair dropped sentence-boundary spaces at the shared engine delta chokepoint,
// including tool-call cross-message boundaries (see streaming-delta.ts).
options.onThinking?.(deltaNormalizer.normalize(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "thinking"));
}
}
if (event.type === "tool_execution_start") {
@@ -2285,15 +2288,18 @@ export async function createFnAgent(options: AgentOptions): Promise<AgentResult>
);
// Wire up event listeners
const deltaNormalizer = createStreamingDeltaNormalizer();
promptableSession.subscribe((event) => {
if (event.type === "message_update") {
const msgEvent = event.assistantMessageEvent;
if (msgEvent.type === "text_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"));
// Repair dropped sentence-boundary spaces at the shared engine delta chokepoint,
// including tool-call cross-message boundaries (see streaming-delta.ts).
options.onText?.(deltaNormalizer.normalize(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "text"));
} else if (msgEvent.type === "thinking_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"));
// Repair dropped sentence-boundary spaces at the shared engine delta chokepoint,
// including tool-call cross-message boundaries (see streaming-delta.ts).
options.onThinking?.(deltaNormalizer.normalize(msgEvent.partial, msgEvent.contentIndex, msgEvent.delta, "thinking"));
}
}
if (event.type === "tool_execution_start") {

44
packages/engine/src/streaming-delta.ts
View File

@@ -69,7 +69,7 @@ function findPreviousBlockText(
return "";
}
export function normalizeStreamingDeltaFromEvent(
function derivePreviousTextFromEvent(
partial: StreamingPartialMessage | undefined,
contentIndex: number,
delta: string,
@@ -87,5 +87,45 @@ export function normalizeStreamingDeltaFromEvent(
previousText = findPreviousBlockText(partial, contentIndex, kind);
}
return normalizeStreamingDelta(previousText, delta);
return previousText;
}
export function createStreamingDeltaNormalizer(): {
normalize: (
partial: StreamingPartialMessage | undefined,
contentIndex: number,
delta: string,
kind: "text" | "thinking",
) => string;
} {
let lastTextTail = "";
let lastThinkingTail = "";
return {
normalize(partial, contentIndex, delta, kind) {
const derivedPreviousText = derivePreviousTextFromEvent(partial, contentIndex, delta, kind);
const previousText = derivedPreviousText || (kind === "text" ? lastTextTail : lastThinkingTail);
const result = normalizeStreamingDelta(previousText, delta);
if (result) {
const tail = result.slice(-1);
if (kind === "text") {
lastTextTail = tail;
} else {
lastThinkingTail = tail;
}
}
return result;
},
};
}
export function normalizeStreamingDeltaFromEvent(
partial: StreamingPartialMessage | undefined,
contentIndex: number,
delta: string,
kind: "text" | "thinking",
): string {
return normalizeStreamingDelta(derivePreviousTextFromEvent(partial, contentIndex, delta, kind), delta);
}