From 16229b8b0ab8cf5bac1ce542a1986ab670eb1b8f Mon Sep 17 00:00:00 2001 From: David Kaya Date: Sat, 21 Mar 2026 21:39:46 +0100 Subject: [PATCH] fix: stabilize streamed agent text Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Services/CopilotWorkflowRunner.cs | 8 +- .../Services/StreamingTextMerger.cs | 87 +++++++++++++++++++ .../StreamingTextMergerTests.cs | 39 +++++++++ src/main/KopayaAppService.ts | 3 +- src/renderer/lib/sessionWorkspace.ts | 3 +- src/shared/utils/streamingText.ts | 67 ++++++++++++++ tests/shared/streamingText.test.ts | 28 ++++++ 7 files changed, 232 insertions(+), 3 deletions(-) create mode 100644 sidecar/src/Kopaya.AgentHost/Services/StreamingTextMerger.cs create mode 100644 sidecar/tests/Kopaya.AgentHost.Tests/StreamingTextMergerTests.cs create mode 100644 src/shared/utils/streamingText.ts create mode 100644 tests/shared/streamingText.test.ts diff --git a/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs b/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs index eacee7d..c4353af 100644 --- a/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs +++ b/sidecar/src/Kopaya.AgentHost/Services/CopilotWorkflowRunner.cs @@ -118,7 +118,7 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner string messageId = update.Update.MessageId ?? $"{command.RequestId}-delta-{fallbackMessageIndex++}"; StreamingSegment segment = GetOrCreateSegment(segments, messageId, authorName); - segment.Content.Append(update.Update.Text); + segment.SetContent(StreamingTextMerger.Merge(segment.Content.ToString(), update.Update.Text)); await onDelta(new TurnDeltaEventDto { @@ -416,6 +416,12 @@ public sealed class CopilotWorkflowRunner : ITurnWorkflowRunner public string AuthorName { get; } public StringBuilder Content { get; } = new(); + + public void SetContent(string value) + { + Content.Clear(); + Content.Append(value); + } } private sealed class AgentBundle : IAsyncDisposable diff --git a/sidecar/src/Kopaya.AgentHost/Services/StreamingTextMerger.cs b/sidecar/src/Kopaya.AgentHost/Services/StreamingTextMerger.cs new file mode 100644 index 0000000..7910d4d --- /dev/null +++ b/sidecar/src/Kopaya.AgentHost/Services/StreamingTextMerger.cs @@ -0,0 +1,87 @@ +using System.Text.RegularExpressions; + +namespace Kopaya.AgentHost.Services; + +internal static partial class StreamingTextMerger +{ + public static string Merge(string current, string incoming) + { + if (string.IsNullOrEmpty(current)) + { + return incoming; + } + + if (string.IsNullOrEmpty(incoming)) + { + return current; + } + + if (incoming.StartsWith(current, StringComparison.Ordinal) + || incoming.Contains(current, StringComparison.Ordinal)) + { + return incoming; + } + + if (current.Contains(incoming, StringComparison.Ordinal)) + { + return current; + } + + int overlap = ComputeSuffixPrefixOverlap(current, incoming); + if (overlap > 0) + { + return current + incoming[overlap..]; + } + + if (ShouldReplaceWithSnapshot(current, incoming)) + { + return incoming; + } + + return current + incoming; + } + + private static int ComputeSuffixPrefixOverlap(string current, string incoming) + { + int maxOverlap = Math.Min(current.Length, incoming.Length); + for (int length = maxOverlap; length > 0; length--) + { + if (string.CompareOrdinal(current, current.Length - length, incoming, 0, length) == 0) + { + return length; + } + } + + return 0; + } + + private static bool ShouldReplaceWithSnapshot(string current, string incoming) + { + if (incoming.Length < Math.Floor(current.Length * 0.6)) + { + return false; + } + + HashSet currentTokens = Tokenize(current); + HashSet incomingTokens = Tokenize(incoming); + if (currentTokens.Count < 3 || incomingTokens.Count < 3) + { + return false; + } + + int shared = incomingTokens.Count(token => currentTokens.Contains(token)); + return shared / (double)Math.Min(currentTokens.Count, incomingTokens.Count) >= 0.5; + } + + private static HashSet Tokenize(string value) + { + return TokenRegex() + .Matches(value.ToLowerInvariant()) + .Select(match => match.Value) + .Where(token => token.Length > 0) + .ToHashSet(StringComparer.Ordinal); + } + + [GeneratedRegex("[a-z0-9]+", RegexOptions.IgnoreCase)] + private static partial Regex TokenRegex(); +} diff --git a/sidecar/tests/Kopaya.AgentHost.Tests/StreamingTextMergerTests.cs b/sidecar/tests/Kopaya.AgentHost.Tests/StreamingTextMergerTests.cs new file mode 100644 index 0000000..fa391a4 --- /dev/null +++ b/sidecar/tests/Kopaya.AgentHost.Tests/StreamingTextMergerTests.cs @@ -0,0 +1,39 @@ +using Kopaya.AgentHost.Services; + +namespace Kopaya.AgentHost.Tests; + +public sealed class StreamingTextMergerTests +{ + [Fact] + public void Merge_AppendsPlainDeltas() + { + Assert.Equal("I am going", StreamingTextMerger.Merge("I am", " going")); + } + + [Fact] + public void Merge_ReplacesWithGrowingSnapshot() + { + Assert.Equal("I am going", StreamingTextMerger.Merge("I am", "I am going")); + } + + [Fact] + public void Merge_PreservesCurrentTextForDuplicateSubset() + { + Assert.Equal("I am going", StreamingTextMerger.Merge("I am going", "going")); + } + + [Fact] + public void Merge_UsesOverlapToAvoidDuplicateJoins() + { + Assert.Equal("Hello world", StreamingTextMerger.Merge("Hello wor", "world")); + } + + [Fact] + public void Merge_ReplacesWithRevisedSnapshotWhenMostTokensOverlap() + { + const string current = "I mirror the existing button pattern and add brief toggle docs."; + const string incoming = "I found the standalone component pattern and I am updating toggle docs next."; + + Assert.Equal(incoming, StreamingTextMerger.Merge(current, incoming)); + } +} diff --git a/src/main/KopayaAppService.ts b/src/main/KopayaAppService.ts index 6a00cde..5e1295e 100644 --- a/src/main/KopayaAppService.ts +++ b/src/main/KopayaAppService.ts @@ -10,6 +10,7 @@ import type { SessionEventRecord } from '@shared/domain/event'; import type { ChatMessageRecord, SessionRecord } from '@shared/domain/session'; import type { WorkspaceState } from '@shared/domain/workspace'; import { createId, nowIso } from '@shared/utils/ids'; +import { mergeStreamingText } from '@shared/utils/streamingText'; import { WorkspaceRepository } from '@main/persistence/workspaceRepository'; import { SecretStore } from '@main/secrets/secretStore'; @@ -278,7 +279,7 @@ export class KopayaAppService extends EventEmitter { const existing = session.messages.find((message) => message.id === event.messageId); if (existing) { - existing.content += event.contentDelta; + existing.content = mergeStreamingText(existing.content, event.contentDelta); existing.pending = true; } else { session.messages.push({ diff --git a/src/renderer/lib/sessionWorkspace.ts b/src/renderer/lib/sessionWorkspace.ts index 584a513..421bef5 100644 --- a/src/renderer/lib/sessionWorkspace.ts +++ b/src/renderer/lib/sessionWorkspace.ts @@ -1,6 +1,7 @@ import type { SessionEventRecord } from '@shared/domain/event'; import type { ChatMessageRecord, SessionRecord } from '@shared/domain/session'; import type { WorkspaceState } from '@shared/domain/workspace'; +import { mergeStreamingText } from '@shared/utils/streamingText'; export function applySessionEventWorkspace( current: WorkspaceState | undefined, @@ -86,7 +87,7 @@ function applyMessageDeltaEvent(session: SessionRecord, event: SessionEventRecor const nextMessage: ChatMessageRecord = { ...existing, authorName: event.authorName ?? existing.authorName, - content: `${existing.content}${event.contentDelta}`, + content: mergeStreamingText(existing.content, event.contentDelta), pending: true, }; diff --git a/src/shared/utils/streamingText.ts b/src/shared/utils/streamingText.ts new file mode 100644 index 0000000..9e34acd --- /dev/null +++ b/src/shared/utils/streamingText.ts @@ -0,0 +1,67 @@ +function tokenize(value: string): string[] { + return value + .toLowerCase() + .split(/[^a-z0-9]+/i) + .filter((token) => token.length > 0); +} + +function computeSuffixPrefixOverlap(current: string, incoming: string): number { + const maxOverlap = Math.min(current.length, incoming.length); + for (let length = maxOverlap; length > 0; length -= 1) { + if (current.slice(-length) === incoming.slice(0, length)) { + return length; + } + } + + return 0; +} + +function shouldReplaceWithSnapshot(current: string, incoming: string): boolean { + if (incoming.length < Math.floor(current.length * 0.6)) { + return false; + } + + const currentTokens = new Set(tokenize(current)); + const incomingTokens = new Set(tokenize(incoming)); + if (currentTokens.size < 3 || incomingTokens.size < 3) { + return false; + } + + let shared = 0; + for (const token of incomingTokens) { + if (currentTokens.has(token)) { + shared += 1; + } + } + + return shared / Math.min(currentTokens.size, incomingTokens.size) >= 0.5; +} + +export function mergeStreamingText(current: string, incoming: string): string { + if (!current) { + return incoming; + } + + if (!incoming) { + return current; + } + + if (incoming.startsWith(current) || incoming.includes(current)) { + return incoming; + } + + if (current.includes(incoming)) { + return current; + } + + const overlap = computeSuffixPrefixOverlap(current, incoming); + if (overlap > 0) { + return current + incoming.slice(overlap); + } + + if (shouldReplaceWithSnapshot(current, incoming)) { + return incoming; + } + + return current + incoming; +} diff --git a/tests/shared/streamingText.test.ts b/tests/shared/streamingText.test.ts new file mode 100644 index 0000000..b655d83 --- /dev/null +++ b/tests/shared/streamingText.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, test } from 'bun:test'; + +import { mergeStreamingText } from '@shared/utils/streamingText'; + +describe('streaming text merge', () => { + test('appends plain deltas', () => { + expect(mergeStreamingText('I am', ' going')).toBe('I am going'); + }); + + test('replaces with a growing snapshot when the incoming text already contains the current text', () => { + expect(mergeStreamingText('I am', 'I am going')).toBe('I am going'); + }); + + test('preserves the current text when the incoming update is a duplicate subset', () => { + expect(mergeStreamingText('I am going', 'going')).toBe('I am going'); + }); + + test('uses overlap matching to avoid duplicated joins', () => { + expect(mergeStreamingText('Hello wor', 'world')).toBe('Hello world'); + }); + + test('replaces with a revised snapshot when the updates share most of the same tokens', () => { + const current = 'I mirror the existing button pattern and add brief toggle docs.'; + const incoming = 'I found the standalone component pattern and I am updating toggle docs next.'; + + expect(mergeStreamingText(current, incoming)).toBe(incoming); + }); +});