Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 36 additions & 2 deletions packages/zcode-tui/src/events.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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<string | object, true>();
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",
Expand Down
5 changes: 5 additions & 0 deletions packages/zcode-tui/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ import {
type ZCodeThemePreference
} from "./color-scheme.ts";
import {
createStreamEventRedeliveryFilter,
historyText,
isModelCancellationEvent,
isToolCancellation,
Expand Down Expand Up @@ -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<void>;
Expand Down Expand Up @@ -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);
Expand Down
72 changes: 72 additions & 0 deletions test/events.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { describe, expect, test } from "bun:test";

import {
createStreamEventRedeliveryFilter,
isToolCancellation,
historyText,
isModelCancellationEvent,
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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);
});
});
Loading