From 41e2a8876d32869af905056eb45e4f6829172bb0 Mon Sep 17 00:00:00 2001 From: tae_98 Date: Sun, 25 Oct 2026 18:08:11 +0900 Subject: [PATCH 1/2] fix(tui): drop redelivered stream events so streamed text renders once The runtime forwards every session event to each live subscription (the active prompt, steers, and subscriptions held open while background-task notification turns run), so one stream event can reach the TUI several times. AssistantStream.append is not idempotent, so each redelivery rendered again: after a background task completed, notification-turn responses displayed every token once per delivery (e.g. three deliveries rendered each syllable three times) even though the persisted transcript stayed clean. Drop exact redeliveries at the onEvent boundary: an event is skipped when its eventId was already seen, or when the delivered object itself was (subscriptions forward the same object reference). Distinct events never share a key, so legitimately repeated delta content still renders. The session-subscription channel is left untouched; it does not append streamed content. --- packages/zcode-tui/src/events.ts | 35 +++++++++++++++++++++ packages/zcode-tui/src/index.ts | 5 +++ test/events.test.ts | 54 ++++++++++++++++++++++++++++++++ 3 files changed, 94 insertions(+) diff --git a/packages/zcode-tui/src/events.ts b/packages/zcode-tui/src/events.ts index 2e80d3d7..058891a3 100644 --- a/packages/zcode-tui/src/events.ts +++ b/packages/zcode-tui/src/events.ts @@ -228,6 +228,41 @@ export function isModelCancellationEvent(event: StreamEvent): boolean { ); } +export interface StreamEventRedeliveryFilter { + duplicate(value: object, event: StreamEvent): boolean; +} + +/** + * The runtime forwards every session event to each live subscription (the + * active prompt, steers, watchers kept open around background-task + * notification turns), so one event can reach the TUI several times. + * Streamed deltas are appended rather than replaced, so each redelivery + * would render again — exact redeliveries of an already-seen event + * (same eventId, or the same delivered object when no id is attached) + * must be dropped. Distinct events never share a key: separate deliveries + * are separate objects, and legitimately repeated delta content produces + * new event objects each time. + */ +export function createStreamEventRedeliveryFilter(capacity = 256): StreamEventRedeliveryFilter { + const recent = new Map(); + return { + duplicate(value: object, event: StreamEvent): boolean { + const key: string | object = event.eventId ?? value; + if (recent.has(key)) { + recent.delete(key); + recent.set(key, true); + return true; + } + recent.set(key, true); + if (recent.size > capacity) { + const oldest = recent.keys().next().value; + if (oldest !== undefined) recent.delete(oldest); + } + return false; + } + }; +} + const toolCancellationValues = new Set([ "aborterror", "cancelled", diff --git a/packages/zcode-tui/src/index.ts b/packages/zcode-tui/src/index.ts index 7cb036ff..26ad46fe 100644 --- a/packages/zcode-tui/src/index.ts +++ b/packages/zcode-tui/src/index.ts @@ -63,6 +63,7 @@ import { type ZCodeThemePreference } from "./color-scheme.ts"; import { + createStreamEventRedeliveryFilter, historyText, isModelCancellationEvent, isToolCancellation, @@ -648,6 +649,9 @@ class ZCodeTui { private readonly attachmentBar: AttachmentBar; private editor: PlanEditor; private readonly assistantStream: AssistantStream; + // Subscription callbacks each forward every session event; only the + // onEvent channel appends streamed content, so only it needs the guard. + private readonly deliveredStreamEvents = createStreamEventRedeliveryFilter(); private readonly notifications: TurnNotifier; private readonly skillCatalog: SkillCatalog; private readonly done: Promise; @@ -2319,6 +2323,7 @@ class ZCodeTui { if (turnEpoch !== undefined && turnEpoch !== this.activeTurnEpoch) return; const event = normalizeEvent(value); if (!event || this.isForeignSessionEvent(event)) return; + if (this.deliveredStreamEvents.duplicate(event.raw, event)) return; if (runtimeContextRefreshNeeded(event)) this.runtimeContextCache.invalidate(); if (this.handleDynamicWorkflowEvent(value, event.type)) return; const taskScoped = this.backgroundTaskEvents.isTaskScoped(event); diff --git a/test/events.test.ts b/test/events.test.ts index c8dc9716..05bb099b 100644 --- a/test/events.test.ts +++ b/test/events.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test"; import { + createStreamEventRedeliveryFilter, isToolCancellation, historyText, isModelCancellationEvent, @@ -523,3 +524,56 @@ describe("ZCode event adapter", () => { }); }); }); + +describe("stream event redelivery filter", () => { + test("flags an exact redelivery of the same delivered object", () => { + const filter = createStreamEventRedeliveryFilter(); + const delivered = { kind: "text_delta", delta: "방", messageId: "message_1" }; + expect(filter.duplicate(delivered, normalizeEvent(delivered)!)).toBe(false); + expect(filter.duplicate(delivered, normalizeEvent(delivered)!)).toBe(true); + }); + + test("keeps legitimately repeated deltas with identical content", () => { + const filter = createStreamEventRedeliveryFilter(); + const first = normalizeEvent({ kind: "text_delta", delta: "-", messageId: "message_1" })!; + const second = normalizeEvent({ kind: "text_delta", delta: "-", messageId: "message_1" })!; + expect(filter.duplicate({}, first)).toBe(false); + expect(filter.duplicate({}, second)).toBe(false); + }); + + test("flags redeliveries that carry the same eventId in fresh objects", () => { + const filter = createStreamEventRedeliveryFilter(); + expect(filter.duplicate( + {}, + normalizeEvent({ eventId: "event_1", kind: "part.delta", partId: "part_1", field: "text", delta: "!" })! + )).toBe(false); + expect(filter.duplicate( + {}, + normalizeEvent({ eventId: "event_1", kind: "part.delta", partId: "part_1", field: "text", delta: "!" })! + )).toBe(true); + }); + + test("renders a triple-delivered notification-turn stream once per token", () => { + const filter = createStreamEventRedeliveryFilter(); + const tokens = ["방", "금", "완", "료"]; + let rendered = ""; + for (const token of tokens) { + // Each subscriber forwards the same event object, back to back. + const delivered = { kind: "text_delta", delta: token, messageId: "message_1" }; + for (let delivery = 0; delivery < 3; delivery += 1) { + const event = normalizeEvent(delivered)!; + if (!filter.duplicate(delivered, event)) rendered += event.delta; + } + } + expect(rendered).toBe("방금완료"); + }); + + test("forgets the oldest keys once capacity is exceeded", () => { + const filter = createStreamEventRedeliveryFilter(2); + const first = { kind: "text_delta", delta: "a" }; + expect(filter.duplicate(first, normalizeEvent(first)!)).toBe(false); + filter.duplicate({}, normalizeEvent({ kind: "text_delta", delta: "b" })!); + filter.duplicate({}, normalizeEvent({ kind: "text_delta", delta: "c" })!); + expect(filter.duplicate(first, normalizeEvent(first)!)).toBe(false); + }); +}); From 861be036f078d3298732fa3c034b90cf95446976 Mon Sep 17 00:00:00 2001 From: Kingsword Date: Mon, 26 Oct 2026 12:36:44 +0800 Subject: [PATCH 2/2] fix(tui): avoid deduping by payload message id --- packages/zcode-tui/src/events.ts | 3 +-- test/events.test.ts | 18 ++++++++++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/packages/zcode-tui/src/events.ts b/packages/zcode-tui/src/events.ts index 058891a3..448c4f74 100644 --- a/packages/zcode-tui/src/events.ts +++ b/packages/zcode-tui/src/events.ts @@ -134,8 +134,7 @@ export function normalizeEvent(value: unknown): StreamEvent | null { ?? (params && asString(params.eventId)) ?? asString(value.id) ?? (params && asString(params.id)) - ?? asString(body.eventId) - ?? asString(body.id), + ?? asString(body.eventId), inputSource: envelopeString("inputSource"), taskId: envelopeString("taskId") ?? envelopeString("taskID") diff --git a/test/events.test.ts b/test/events.test.ts index 05bb099b..64db68c6 100644 --- a/test/events.test.ts +++ b/test/events.test.ts @@ -77,6 +77,24 @@ describe("ZCode event adapter", () => { }); }); + test("does not treat a payload message id as a stream event id", () => { + const first = normalizeEvent({ + type: "model.streaming", + payload: { id: "message_1", kind: "text_delta", delta: "hello" } + }); + const second = normalizeEvent({ + type: "model.streaming", + payload: { id: "message_1", kind: "text_delta", delta: " world" } + }); + + expect(first?.eventId).toBeUndefined(); + expect(second?.eventId).toBeUndefined(); + + const filter = createStreamEventRedeliveryFilter(); + expect(filter.duplicate(first!.raw, first!)).toBe(false); + expect(filter.duplicate(second!.raw, second!)).toBe(false); + }); + test("uses the official assistant message identity and result duration", () => { expect(normalizeEvent({ type: "model.streaming",