From fba2ca100cf3683f14385e175157dbb601b6cb76 Mon Sep 17 00:00:00 2001 From: Narayan201120 Date: Mon, 26 Oct 2026 12:15:53 +0530 Subject: [PATCH] perf(client-runtime): repeated subagent and turn updates locate their entry once A repeated `subagent.updated` or `provider-turn.updated` found the stored record to restore a deduped body or token usage, then searched the same list again to replace it, and once more for token usage. Both branches now locate the index and previous record once and replace at that index, keeping first-match and append behavior. OpenCode with deepseek-v4.1-flash-free in T3 Code. --- .../src/remotePerformance.bench.ts | 51 +++++++ .../state/orchestrationV2Projection.test.ts | 144 ++++++++++++++++++ .../src/state/orchestrationV2Projection.ts | 46 +++--- 3 files changed, 222 insertions(+), 19 deletions(-) diff --git a/packages/client-runtime/src/remotePerformance.bench.ts b/packages/client-runtime/src/remotePerformance.bench.ts index e4452ffa04d14..f79d85036ae5c 100644 --- a/packages/client-runtime/src/remotePerformance.bench.ts +++ b/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 = { + 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); + }); + } +}); diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts b/packages/client-runtime/src/state/orchestrationV2Projection.test.ts index e0680c17c46f8..daa5c4b2b334d 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.test.ts +++ b/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 = { diff --git a/packages/client-runtime/src/state/orchestrationV2Projection.ts b/packages/client-runtime/src/state/orchestrationV2Projection.ts index ffe9eb14dfa2d..9ac7ca681757c 100644 --- a/packages/client-runtime/src/state/orchestrationV2Projection.ts +++ b/packages/client-runtime/src/state/orchestrationV2Projection.ts @@ -50,17 +50,28 @@ function replaceProjectionArrayItem( return next; } -function upsertEntity( +function upsertEntityAt( items: ReadonlyArray, + index: number, item: T, ): ReadonlyArray { - 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( + items: ReadonlyArray, + item: T, +): ReadonlyArray { + 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":