perf(server): thread opens and live events send each message body once

apps/server/src/orchestration-v2/ThreadStream.ts

@@ -4,6 +4,7 @@ import type {
 } from "@t3tools/contracts";
 
 import { omitLocalVisibleTurnItems } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
+import { dedupeSnapshotBodies } from "@t3tools/shared/orchestrationV2DedupedBodies";
 
 import {
   buildBoundedThreadProjection,
@@ -42,13 +43,16 @@ type ThreadSnapshotStreamItem = Extract<
 export function boundedSnapshotResponseFields(input: {
   readonly bounded: BoundedProjectionResult;
   readonly compactTurnItems: boolean;
+  readonly dedupeBodies: boolean;
 }) {
   const { bounded } = input;
-  const compactProjection = input.compactTurnItems
-    ? omitLocalVisibleTurnItems(bounded.projection)
-    : null;
+  // Dedupe keeps shared items one object, which the compaction's identity check needs.
+  const projection = input.dedupeBodies
+    ? dedupeSnapshotBodies(bounded.projection)
+    : bounded.projection;
+  const compactProjection = input.compactTurnItems ? omitLocalVisibleTurnItems(projection) : null;
   return {
-    projection: compactProjection ?? bounded.projection,
+    projection: compactProjection ?? projection,
     historyCursor: bounded.historyCursor,
     hasMoreHistory: bounded.hasMoreHistory,
     latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
@@ -60,6 +64,7 @@ export function buildBoundedThreadStreamSnapshot(input: {
   readonly snapshotSequence: number;
   readonly projection: OrchestrationV2ThreadProjection;
   readonly compactTurnItems?: boolean;
+  readonly dedupeBodies?: boolean;
 }): ThreadSnapshotStreamItem {
   const bounded = buildBoundedThreadProjection({
     snapshotSequence: input.snapshotSequence,
@@ -71,6 +76,7 @@ export function buildBoundedThreadStreamSnapshot(input: {
     ...boundedSnapshotResponseFields({
       bounded,
       compactTurnItems: input.compactTurnItems === true,
+      dedupeBodies: input.dedupeBodies === true,
     }),
   };
 }

apps/server/src/orchestration-v2/WireProjection.test.ts

@@ -9,15 +9,19 @@ import {
   ProviderInstanceId,
   ThreadId,
   TurnItemId,
+  type OrchestrationV2DomainEvent,
   type OrchestrationV2ThreadProjection,
+  type OrchestrationV2ThreadStreamItem,
   OrchestrationV2TurnItem,
   OrchestrationV2TurnItemJson,
 } from "@t3tools/contracts";
 import { describe, expect, it } from "@effect/vitest";
 import * as DateTime from "effect/DateTime";
 import * as Schema from "effect/Schema";
 
+import { coalesceLiveUpdatedEvents } from "./ThreadLiveEventCoalescer.ts";
 import {
+  makeLiveBodyDeduper,
   projectTurnItemForWire,
   projectTurnItemForDetail,
   projectDomainEventForWire,
@@ -302,3 +306,74 @@ describe("orchestration V2 wire projection", () => {
     expect(projectDomainEventForWire(projected)).toEqual(projected);
   });
 });
+
+describe("live body deduper", () => {
+  const at = base.updatedAt;
+  const messageId = MessageId.make("message-live");
+  const event = (
+    sequence: number,
+    rest: Pick<OrchestrationV2DomainEvent, "type" | "payload">,
+  ): Extract<OrchestrationV2ThreadStreamItem, { kind: "event" }> => ({
+    kind: "event",
+    sequence,
+    event: {
+      id: EventId.make(`event-${sequence}`),
+      threadId: base.threadId,
+      occurredAt: at,
+      ...rest,
+    } as OrchestrationV2DomainEvent,
+  });
+  const message = (sequence: number, text: string) =>
+    event(sequence, {
+      type: "message.updated",
+      payload: {
+        id: messageId,
+        threadId: base.threadId,
+        runId: null,
+        nodeId: null,
+        role: "assistant",
+        text,
+        attachments: [],
+        streaming: true,
+        createdBy: "agent",
+        creationSource: "provider",
+        createdAt: at,
+        updatedAt: at,
+      },
+    });
+  const reply = (sequence: number, text: string) =>
+    event(sequence, {
+      type: "turn-item.updated",
+      payload: {
+        ...base,
+        status: "running",
+        type: "assistant_message",
+        messageId,
+        text,
+        streaming: true,
+      },
+    });
+  const payloadOf = (item: OrchestrationV2ThreadStreamItem) =>
+    item.kind === "event" ? item.event.payload : undefined;
+
+  it("blanks a reply that repeats the message delivered before it, after coalescing", () => {
+    const delivered = coalesceLiveUpdatedEvents([
+      message(1, "a"),
+      reply(2, "a"),
+      message(3, "ab"),
+      reply(4, "ab"),
+    ]).map(makeLiveBodyDeduper());
+
+    expect(
+      delivered.map((item) => [item.kind === "event" && item.sequence, payloadOf(item)]),
+    ).toEqual([
+      [3, message(3, "ab").event.payload],
+      [4, { ...reply(4, "ab").event.payload, text: "", textOmitted: true }],
+    ]);
+  });
+
+  it("ships a reply whole when its message arrives after it", () => {
+    const first = reply(1, "a");
+    expect([first, message(2, "a")].map(makeLiveBodyDeduper())[0]).toBe(first);
+  });
+});

apps/server/src/orchestration-v2/WireProjection.ts

@@ -1,9 +1,19 @@
 import type {
+  CheckpointId,
+  MessageId,
+  NodeId,
+  OrchestrationV2Checkpoint,
   OrchestrationV2DomainEvent,
   OrchestrationV2ContextHandoff,
+  OrchestrationV2Subagent,
   OrchestrationV2ThreadProjection,
+  OrchestrationV2ThreadStreamItem,
   OrchestrationV2TurnItem,
 } from "@t3tools/contracts";
+import {
+  dedupeSubagentBodies,
+  dedupeTurnItemBodies,
+} from "@t3tools/shared/orchestrationV2DedupedBodies";
 import {
   compactDynamicToolOutput,
   omitToolOutputImageData,
@@ -245,3 +255,65 @@ export function projectDomainEventForWire(
       ? { ...event, payload: projectContextHandoffForWire(event.payload) }
       : event;
 }
+
+const LIVE_DEDUPE_MEMORY = 32;
+
+function remember<K, V>(memory: Map<K, V>, key: K, value: V): void {
+  memory.delete(key);
+  memory.set(key, value);
+  if (memory.size > LIVE_DEDUPE_MEMORY) memory.delete(memory.keys().next().value as K);
+}
+
+/**
+ * One per subscription, applied after coalescing so it only remembers records the client
+ * received, in the snapshot or in events. A record event comes before the turn item that repeats
+ * its body (adapters and the orchestrator emit them in that order), so the item can ship blank
+ * and marked; so can a repeated subagent record. Emitting in the other order, or forgetting a
+ * record, only costs bytes: the repeat ships whole.
+ */
+export function makeLiveBodyDeduper(): (
+  item: OrchestrationV2ThreadStreamItem,
+) => OrchestrationV2ThreadStreamItem {
+  const messageText = new Map<MessageId, string>();
+  const checkpointFiles = new Map<CheckpointId, OrchestrationV2Checkpoint["files"]>();
+  const subagents = new Map<NodeId, Pick<OrchestrationV2Subagent, "prompt" | "result">>();
+  const source = {
+    messageText: (id: MessageId) => messageText.get(id),
+    checkpointFiles: (id: CheckpointId) => checkpointFiles.get(id),
+    subagent: (id: NodeId) => subagents.get(id),
+  };
+  const rememberSubagent = ({ id, prompt, result }: OrchestrationV2Subagent) =>
+    remember(subagents, id, { prompt, result });
+  return (item) => {
+    if (item.kind === "snapshot") {
+      // The client now holds exactly these records; their bodies are never blanked.
+      const { messages, checkpoints, subagents: records } = item.projection;
+      for (const message of messages) remember(messageText, message.id, message.text);
+      for (const checkpoint of checkpoints)
+        remember(checkpointFiles, checkpoint.id, checkpoint.files);
+      for (const subagent of records) rememberSubagent(subagent);
+      return item;
+    }
+    if (item.kind !== "event") return item;
+    const { event } = item;
+    switch (event.type) {
+      case "message.updated":
+        remember(messageText, event.payload.id, event.payload.text);
+        return item;
+      case "checkpoint.captured":
+        remember(checkpointFiles, event.payload.id, event.payload.files);
+        return item;
+      case "subagent.updated": {
+        const payload = dedupeSubagentBodies(event.payload, subagents.get(event.payload.id));
+        rememberSubagent(event.payload);
+        return payload === event.payload ? item : { ...item, event: { ...event, payload } };
+      }
+      case "turn-item.updated": {
+        const payload = dedupeTurnItemBodies(event.payload, source);
+        return payload === event.payload ? item : { ...item, event: { ...event, payload } };
+      }
+      default:
+        return item;
+    }
+  };
+}