diff --git a/packages/zcode-tui/src/events.ts b/packages/zcode-tui/src/events.ts index 2e80d3d..448c4f7 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") @@ -228,6 +227,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 7cb036f..26ad46f 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 c8dc971..64db68c 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, @@ -76,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", @@ -523,3 +542,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); + }); +});