Skip to content

Commit fba2ca1

Browse files
Narayan201120Narayan201120
committed
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.
1 parent 10629af commit fba2ca1

3 files changed

Lines changed: 222 additions & 19 deletions

File tree

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

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,11 @@ import {
22
EnvironmentId,
33
EventId,
44
MessageId,
5+
NodeId,
6+
ProviderThreadId,
7+
ProviderTurnId,
58
type OrchestrationV2DomainEvent,
9+
type OrchestrationV2ProviderTurn,
610
type OrchestrationV2ThreadProjection,
711
} from "@t3tools/contracts";
812
import * as Effect from "effect/Effect";
@@ -133,3 +137,50 @@ describe("remote message replay", () => {
133137
});
134138
}
135139
});
140+
141+
const providerTurn: OrchestrationV2ProviderTurn = {
142+
id: ProviderTurnId.make("provider-turn-0"),
143+
providerThreadId: ProviderThreadId.make("provider-thread"),
144+
nodeId: NodeId.make("provider-node"),
145+
runAttemptId: null,
146+
nativeTurnRef: null,
147+
ordinal: 1,
148+
status: "running",
149+
startedAt: v2Now,
150+
completedAt: null,
151+
};
152+
// The bench counts module export getter reads as iteration overhead, so read it once.
153+
const applyProviderTurnProjectionEvent = applyOrchestrationV2ProjectionEvent;
154+
155+
describe("remote provider turn replay", () => {
156+
for (const count of [100, 1_000]) {
157+
const loaded = {
158+
...thread,
159+
providerTurns: Array.from({ length: count }, (_, index) => ({
160+
...providerTurn,
161+
id: ProviderTurnId.make(`provider-turn-${index}`),
162+
ordinal: index + 1,
163+
})),
164+
};
165+
const event: Extract<OrchestrationV2DomainEvent, { type: "provider-turn.updated" }> = {
166+
id: EventId.make("provider-turn"),
167+
type: "provider-turn.updated",
168+
threadId: thread.thread.id,
169+
occurredAt: v2Now,
170+
payload: loaded.providerTurns.at(-1)!,
171+
};
172+
const name = `apply 200 provider turn updates to ${count} loaded turns`;
173+
test(name, async ({ bench }) => {
174+
await bench(name, () => {
175+
let current: OrchestrationV2ThreadProjection = loaded;
176+
for (let index = 0; index < 200; index += 1) {
177+
current =
178+
applyProviderTurnProjectionEvent(current, {
179+
...event,
180+
payload: { ...event.payload, status: index % 2 === 0 ? "running" : "completed" },
181+
}) ?? current;
182+
}
183+
}).run(runOptions);
184+
});
185+
}
186+
});

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

Lines changed: 144 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,12 +2,15 @@ import { describe, expect, it } from "vite-plus/test";
22
import {
33
type ThreadPullRequestLink,
44
type OrchestrationV2DomainEvent,
5+
type OrchestrationV2ProviderTurn,
56
type OrchestrationV2Run,
7+
type OrchestrationV2Subagent,
68
type OrchestrationV2ThreadProjection,
79
type OrchestrationV2TurnItem,
810
MessageId,
911
ProjectId,
1012
ProviderInstanceId,
13+
ProviderDriverKind,
1114
NodeId,
1215
ProviderThreadId,
1316
ProviderTurnId,
@@ -62,6 +65,48 @@ function commandItem(id: string, output = "done", ordinal = 1): OrchestrationV2T
6265
exitCode: 0,
6366
};
6467
}
68+
69+
function subagentRecord(id: string): OrchestrationV2Subagent {
70+
return {
71+
id: NodeId.make(id),
72+
threadId,
73+
runId,
74+
parentNodeId: NodeId.make("node-root"),
75+
origin: "app_owned",
76+
createdBy: "agent",
77+
driver: ProviderDriverKind.make("codex"),
78+
providerInstanceId: ProviderInstanceId.make("codex"),
79+
providerThreadId: null,
80+
childThreadId: null,
81+
nativeTaskRef: null,
82+
prompt: `brief for ${id}`,
83+
title: null,
84+
model: null,
85+
status: "running",
86+
result: `result for ${id}`,
87+
startedAt: now,
88+
completedAt: null,
89+
updatedAt: now,
90+
};
91+
}
92+
93+
function providerTurnRecord(
94+
id: string,
95+
status: OrchestrationV2ProviderTurn["status"] = "running",
96+
): OrchestrationV2ProviderTurn {
97+
return {
98+
id: ProviderTurnId.make(id),
99+
providerThreadId: ProviderThreadId.make("provider-thread-reducer"),
100+
nodeId: NodeId.make("provider-node-reducer"),
101+
runAttemptId: null,
102+
nativeTurnRef: null,
103+
ordinal: 1,
104+
status,
105+
startedAt: now,
106+
completedAt: null,
107+
};
108+
}
109+
65110
const emptyProjection = {
66111
thread: {
67112
id: threadId,
@@ -173,6 +218,105 @@ describe("applyOrchestrationV2ProjectionEvent", () => {
173218
expect(next?.providerTurns[0]?.tokenUsage).toEqual(running.tokenUsage);
174219
});
175220

221+
it("updates the first provider turn with a repeated id from the one index it locates", () => {
222+
const usage = {
223+
usedTokens: 50_000,
224+
maxTokens: 200_000,
225+
updatedAt: "2026-08-29T00:00:00.000Z",
226+
};
227+
const first = { ...providerTurnRecord("turn-repeated"), tokenUsage: usage };
228+
const duplicate = providerTurnRecord("turn-repeated", "failed");
229+
const other = providerTurnRecord("turn-other");
230+
let turnIdReads = 0;
231+
const stored = [first, duplicate, other].map((turn) => ({
232+
...turn,
233+
get id() {
234+
turnIdReads += 1;
235+
return turn.id;
236+
},
237+
}));
238+
239+
const next = applyOrchestrationV2ProjectionEvent(
240+
{ ...emptyProjection, providerTurns: stored },
241+
{
242+
id: "event-provider-turn-repeated",
243+
type: "provider-turn.updated",
244+
threadId,
245+
occurredAt: now,
246+
payload: { ...first, status: "completed", completedAt: now, tokenUsage: undefined },
247+
} as OrchestrationV2DomainEvent,
248+
);
249+
250+
expect(turnIdReads).toBe(1);
251+
expect(next?.providerTurns.map((turn) => turn.tokenUsage)).toEqual([
252+
usage,
253+
undefined,
254+
undefined,
255+
]);
256+
expect(next?.providerTurns[0]?.status).toBe("completed");
257+
expect(next?.providerTurns[1]).toBe(stored[1]);
258+
expect(next?.providerTurns[2]).toBe(stored[2]);
259+
});
260+
261+
it("appends a new provider turn with the token usage it carries", () => {
262+
const usage = {
263+
usedTokens: 1_200,
264+
maxTokens: 200_000,
265+
updatedAt: "2026-08-29T00:00:00.000Z",
266+
};
267+
const stored = [providerTurnRecord("turn-existing", "completed")];
268+
const fresh = { ...providerTurnRecord("turn-new"), tokenUsage: usage };
269+
270+
const next = applyOrchestrationV2ProjectionEvent(
271+
{ ...emptyProjection, providerTurns: stored },
272+
{
273+
id: "event-provider-turn-new",
274+
type: "provider-turn.updated",
275+
threadId,
276+
occurredAt: now,
277+
payload: fresh,
278+
} as OrchestrationV2DomainEvent,
279+
);
280+
281+
expect(next?.providerTurns.map((turn) => turn.id)).toEqual([stored[0]!.id, fresh.id]);
282+
expect(next?.providerTurns[0]).toBe(stored[0]);
283+
expect(next?.providerTurns[1]?.tokenUsage).toEqual(usage);
284+
});
285+
286+
it("restores a repeated subagent body from the one record it locates", () => {
287+
const records = ["node-a", "node-b", "node-c"].map((id) => subagentRecord(id));
288+
const target = records[2]!;
289+
let subagentIdReads = 0;
290+
const stored = records.map((record) => ({
291+
...record,
292+
get id() {
293+
subagentIdReads += 1;
294+
return record.id;
295+
},
296+
}));
297+
298+
const next = applyOrchestrationV2ProjectionEvent({ ...emptyProjection, subagents: stored }, {
299+
id: "event-subagent-repeated",
300+
type: "subagent.updated",
301+
threadId,
302+
runId,
303+
occurredAt: now,
304+
payload: {
305+
...target,
306+
prompt: "",
307+
promptOmitted: true,
308+
result: null,
309+
resultOmitted: true,
310+
progress: "step 2",
311+
},
312+
} as OrchestrationV2DomainEvent);
313+
314+
expect(subagentIdReads).toBe(records.length);
315+
expect(next?.subagents).toEqual([records[0], records[1], { ...target, progress: "step 2" }]);
316+
expect(next?.subagents[0]).toBe(stored[0]);
317+
expect(next?.subagents[1]).toBe(stored[1]);
318+
});
319+
176320
it("applies thread lifecycle payloads instead of leaving stale metadata", () => {
177321
const archivedAt = DateTime.makeUnsafe("2026-06-20T01:00:00.000Z");
178322
const event = {

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

Lines changed: 27 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -50,17 +50,28 @@ function replaceProjectionArrayItem<T>(
5050
return next;
5151
}
5252

53-
function upsertEntity<T extends { readonly id: unknown }>(
53+
function upsertEntityAt<T extends { readonly id: unknown }>(
5454
items: ReadonlyArray<T>,
55+
index: number,
5556
item: T,
5657
): ReadonlyArray<T> {
57-
const index = items.findIndex((candidate) => candidate.id === item.id);
5858
if (index === -1) return [...items, item];
5959
const next = [...items];
6060
next[index] = item;
6161
return next;
6262
}
6363

64+
function upsertEntity<T extends { readonly id: unknown }>(
65+
items: ReadonlyArray<T>,
66+
item: T,
67+
): ReadonlyArray<T> {
68+
return upsertEntityAt(
69+
items,
70+
items.findIndex((candidate) => candidate.id === item.id),
71+
item,
72+
);
73+
}
74+
6475
// A deduped item repeats a record the stream just sent, so search newest first.
6576
function latestBodies(projection: OrchestrationV2ThreadProjection): DedupedBodySource {
6677
return {
@@ -321,17 +332,18 @@ export function applyOrchestrationV2ProjectionEvent(
321332
}
322333
case "node.updated":
323334
return { ...base, nodes: upsertEntity(base.nodes, event.payload) };
324-
case "subagent.updated":
335+
case "subagent.updated": {
336+
const index = base.subagents.findIndex((subagent) => subagent.id === event.payload.id);
337+
const previous = index === -1 ? undefined : base.subagents[index];
325338
return {
326339
...base,
327-
subagents: upsertEntity(
340+
subagents: upsertEntityAt(
328341
base.subagents,
329-
restoreSubagentBodies(
330-
event.payload,
331-
base.subagents.find((subagent) => subagent.id === event.payload.id),
332-
),
342+
index,
343+
restoreSubagentBodies(event.payload, previous),
333344
),
334345
};
346+
}
335347
case "provider-session.attached":
336348
case "provider-session.updated":
337349
return {
@@ -347,22 +359,18 @@ export function applyOrchestrationV2ProjectionEvent(
347359
};
348360
case "provider-thread.updated":
349361
return { ...base, providerThreads: upsertEntity(base.providerThreads, event.payload) };
350-
case "provider-turn.updated":
362+
case "provider-turn.updated": {
363+
const index = base.providerTurns.findIndex((turn) => turn.id === event.payload.id);
364+
const previous = index === -1 ? undefined : base.providerTurns[index];
365+
const tokenUsage = event.payload.tokenUsage ?? previous?.tokenUsage;
351366
return {
352367
...base,
353-
providerTurns: upsertEntity(base.providerTurns, {
368+
providerTurns: upsertEntityAt(base.providerTurns, index, {
354369
...event.payload,
355-
...((event.payload.tokenUsage ??
356-
base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage) ===
357-
undefined
358-
? {}
359-
: {
360-
tokenUsage:
361-
event.payload.tokenUsage ??
362-
base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage,
363-
}),
370+
...(tokenUsage === undefined ? {} : { tokenUsage }),
364371
}),
365372
};
373+
}
366374
case "runtime-request.updated":
367375
return { ...base, runtimeRequests: upsertEntity(base.runtimeRequests, event.payload) };
368376
case "message.updated":

0 commit comments

Comments
 (0)