perf(client-runtime): reuse projection entity lookups

packages/client-runtime/src/remotePerformance.bench.ts

@@ -2,7 +2,11 @@ import {
   EnvironmentId,
   EventId,
   MessageId,
+  NodeId,
+  ProviderThreadId,
+  ProviderTurnId,
   type OrchestrationV2DomainEvent,
+  type OrchestrationV2ProviderTurn,
   type OrchestrationV2ThreadProjection,
 } from "@t3tools/contracts";
 import * as Effect from "effect/Effect";
@@ -133,3 +137,50 @@ describe("remote message replay", () => {
     });
   }
 });
+
+const providerTurn: OrchestrationV2ProviderTurn = {
+  id: ProviderTurnId.make("provider-turn-0"),
+  providerThreadId: ProviderThreadId.make("provider-thread"),
+  nodeId: NodeId.make("provider-node"),
+  runAttemptId: null,
+  nativeTurnRef: null,
+  ordinal: 1,
+  status: "running",
+  startedAt: v2Now,
+  completedAt: null,
+};
+// The bench counts module export getter reads as iteration overhead, so read it once.
+const applyProviderTurnProjectionEvent = applyOrchestrationV2ProjectionEvent;
+
+describe("remote provider turn replay", () => {
+  for (const count of [100, 1_000]) {
+    const loaded = {
+      ...thread,
+      providerTurns: Array.from({ length: count }, (_, index) => ({
+        ...providerTurn,
+        id: ProviderTurnId.make(`provider-turn-${index}`),
+        ordinal: index + 1,
+      })),
+    };
+    const event: Extract<OrchestrationV2DomainEvent, { type: "provider-turn.updated" }> = {
+      id: EventId.make("provider-turn"),
+      type: "provider-turn.updated",
+      threadId: thread.thread.id,
+      occurredAt: v2Now,
+      payload: loaded.providerTurns.at(-1)!,
+    };
+    const name = `apply 200 provider turn updates to ${count} loaded turns`;
+    test(name, async ({ bench }) => {
+      await bench(name, () => {
+        let current: OrchestrationV2ThreadProjection = loaded;
+        for (let index = 0; index < 200; index += 1) {
+          current =
+            applyProviderTurnProjectionEvent(current, {
+              ...event,
+              payload: { ...event.payload, status: index % 2 === 0 ? "running" : "completed" },
+            }) ?? current;
+        }
+      }).run(runOptions);
+    });
+  }
+});

packages/client-runtime/src/state/orchestrationV2Projection.test.ts

@@ -2,12 +2,15 @@ import { describe, expect, it } from "vite-plus/test";
 import {
   type ThreadPullRequestLink,
   type OrchestrationV2DomainEvent,
+  type OrchestrationV2ProviderTurn,
   type OrchestrationV2Run,
+  type OrchestrationV2Subagent,
   type OrchestrationV2ThreadProjection,
   type OrchestrationV2TurnItem,
   MessageId,
   ProjectId,
   ProviderInstanceId,
+  ProviderDriverKind,
   NodeId,
   ProviderThreadId,
   ProviderTurnId,
@@ -62,6 +65,48 @@ function commandItem(id: string, output = "done", ordinal = 1): OrchestrationV2T
     exitCode: 0,
   };
 }
+
+function subagentRecord(id: string): OrchestrationV2Subagent {
+  return {
+    id: NodeId.make(id),
+    threadId,
+    runId,
+    parentNodeId: NodeId.make("node-root"),
+    origin: "app_owned",
+    createdBy: "agent",
+    driver: ProviderDriverKind.make("codex"),
+    providerInstanceId: ProviderInstanceId.make("codex"),
+    providerThreadId: null,
+    childThreadId: null,
+    nativeTaskRef: null,
+    prompt: `brief for ${id}`,
+    title: null,
+    model: null,
+    status: "running",
+    result: `result for ${id}`,
+    startedAt: now,
+    completedAt: null,
+    updatedAt: now,
+  };
+}
+
+function providerTurnRecord(
+  id: string,
+  status: OrchestrationV2ProviderTurn["status"] = "running",
+): OrchestrationV2ProviderTurn {
+  return {
+    id: ProviderTurnId.make(id),
+    providerThreadId: ProviderThreadId.make("provider-thread-reducer"),
+    nodeId: NodeId.make("provider-node-reducer"),
+    runAttemptId: null,
+    nativeTurnRef: null,
+    ordinal: 1,
+    status,
+    startedAt: now,
+    completedAt: null,
+  };
+}
+
 const emptyProjection = {
   thread: {
     id: threadId,
@@ -173,6 +218,105 @@ describe("applyOrchestrationV2ProjectionEvent", () => {
     expect(next?.providerTurns[0]?.tokenUsage).toEqual(running.tokenUsage);
   });
 
+  it("updates the first provider turn with a repeated id from the one index it locates", () => {
+    const usage = {
+      usedTokens: 50_000,
+      maxTokens: 200_000,
+      updatedAt: "2026-08-29T00:00:00.000Z",
+    };
+    const first = { ...providerTurnRecord("turn-repeated"), tokenUsage: usage };
+    const duplicate = providerTurnRecord("turn-repeated", "failed");
+    const other = providerTurnRecord("turn-other");
+    let turnIdReads = 0;
+    const stored = [first, duplicate, other].map((turn) => ({
+      ...turn,
+      get id() {
+        turnIdReads += 1;
+        return turn.id;
+      },
+    }));
+
+    const next = applyOrchestrationV2ProjectionEvent(
+      { ...emptyProjection, providerTurns: stored },
+      {
+        id: "event-provider-turn-repeated",
+        type: "provider-turn.updated",
+        threadId,
+        occurredAt: now,
+        payload: { ...first, status: "completed", completedAt: now, tokenUsage: undefined },
+      } as OrchestrationV2DomainEvent,
+    );
+
+    expect(turnIdReads).toBe(1);
+    expect(next?.providerTurns.map((turn) => turn.tokenUsage)).toEqual([
+      usage,
+      undefined,
+      undefined,
+    ]);
+    expect(next?.providerTurns[0]?.status).toBe("completed");
+    expect(next?.providerTurns[1]).toBe(stored[1]);
+    expect(next?.providerTurns[2]).toBe(stored[2]);
+  });
+
+  it("appends a new provider turn with the token usage it carries", () => {
+    const usage = {
+      usedTokens: 1_200,
+      maxTokens: 200_000,
+      updatedAt: "2026-08-29T00:00:00.000Z",
+    };
+    const stored = [providerTurnRecord("turn-existing", "completed")];
+    const fresh = { ...providerTurnRecord("turn-new"), tokenUsage: usage };
+
+    const next = applyOrchestrationV2ProjectionEvent(
+      { ...emptyProjection, providerTurns: stored },
+      {
+        id: "event-provider-turn-new",
+        type: "provider-turn.updated",
+        threadId,
+        occurredAt: now,
+        payload: fresh,
+      } as OrchestrationV2DomainEvent,
+    );
+
+    expect(next?.providerTurns.map((turn) => turn.id)).toEqual([stored[0]!.id, fresh.id]);
+    expect(next?.providerTurns[0]).toBe(stored[0]);
+    expect(next?.providerTurns[1]?.tokenUsage).toEqual(usage);
+  });
+
+  it("restores a repeated subagent body from the one record it locates", () => {
+    const records = ["node-a", "node-b", "node-c"].map((id) => subagentRecord(id));
+    const target = records[2]!;
+    let subagentIdReads = 0;
+    const stored = records.map((record) => ({
+      ...record,
+      get id() {
+        subagentIdReads += 1;
+        return record.id;
+      },
+    }));
+
+    const next = applyOrchestrationV2ProjectionEvent({ ...emptyProjection, subagents: stored }, {
+      id: "event-subagent-repeated",
+      type: "subagent.updated",
+      threadId,
+      runId,
+      occurredAt: now,
+      payload: {
+        ...target,
+        prompt: "",
+        promptOmitted: true,
+        result: null,
+        resultOmitted: true,
+        progress: "step 2",
+      },
+    } as OrchestrationV2DomainEvent);
+
+    expect(subagentIdReads).toBe(records.length);
+    expect(next?.subagents).toEqual([records[0], records[1], { ...target, progress: "step 2" }]);
+    expect(next?.subagents[0]).toBe(stored[0]);
+    expect(next?.subagents[1]).toBe(stored[1]);
+  });
+
   it("applies thread lifecycle payloads instead of leaving stale metadata", () => {
     const archivedAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");
     const event = {

packages/client-runtime/src/state/orchestrationV2Projection.ts

@@ -50,17 +50,28 @@ function replaceProjectionArrayItem<T>(
   return next;
 }
 
-function upsertEntity<T extends { readonly id: unknown }>(
+function upsertEntityAt<T extends { readonly id: unknown }>(
   items: ReadonlyArray<T>,
+  index: number,
   item: T,
 ): ReadonlyArray<T> {
-  const index = items.findIndex((candidate) => candidate.id === item.id);
   if (index === -1) return [...items, item];
   const next = [...items];
   next[index] = item;
   return next;
 }
 
+function upsertEntity<T extends { readonly id: unknown }>(
+  items: ReadonlyArray<T>,
+  item: T,
+): ReadonlyArray<T> {
+  return upsertEntityAt(
+    items,
+    items.findIndex((candidate) => candidate.id === item.id),
+    item,
+  );
+}
+
 // A deduped item repeats a record the stream just sent, so search newest first.
 function latestBodies(projection: OrchestrationV2ThreadProjection): DedupedBodySource {
   return {
@@ -321,17 +332,18 @@ export function applyOrchestrationV2ProjectionEvent(
     }
     case "node.updated":
       return { ...base, nodes: upsertEntity(base.nodes, event.payload) };
-    case "subagent.updated":
+    case "subagent.updated": {
+      const index = base.subagents.findIndex((subagent) => subagent.id === event.payload.id);
+      const previous = index === -1 ? undefined : base.subagents[index];
       return {
         ...base,
-        subagents: upsertEntity(
+        subagents: upsertEntityAt(
           base.subagents,
-          restoreSubagentBodies(
-            event.payload,
-            base.subagents.find((subagent) => subagent.id === event.payload.id),
-          ),
+          index,
+          restoreSubagentBodies(event.payload, previous),
         ),
       };
+    }
     case "provider-session.attached":
     case "provider-session.updated":
       return {
@@ -347,22 +359,18 @@ export function applyOrchestrationV2ProjectionEvent(
       };
     case "provider-thread.updated":
       return { ...base, providerThreads: upsertEntity(base.providerThreads, event.payload) };
-    case "provider-turn.updated":
+    case "provider-turn.updated": {
+      const index = base.providerTurns.findIndex((turn) => turn.id === event.payload.id);
+      const previous = index === -1 ? undefined : base.providerTurns[index];
+      const tokenUsage = event.payload.tokenUsage ?? previous?.tokenUsage;
       return {
         ...base,
-        providerTurns: upsertEntity(base.providerTurns, {
+        providerTurns: upsertEntityAt(base.providerTurns, index, {
           ...event.payload,
-          ...((event.payload.tokenUsage ??
-            base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage) ===
-          undefined
-            ? {}
-            : {
-                tokenUsage:
-                  event.payload.tokenUsage ??
-                  base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage,
-              }),
+          ...(tokenUsage === undefined ? {} : { tokenUsage }),
         }),
       };
+    }
     case "runtime-request.updated":
       return { ...base, runtimeRequests: upsertEntity(base.runtimeRequests, event.payload) };
     case "message.updated":