diff --git a/packages/client-runtime/src/remotePerformance.bench.ts b/packages/client-runtime/src/remotePerformance.bench.ts index e4452ffa04d1..f79d85036ae5 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 e0680c17c46f..daa5c4b2b334 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 ffe9eb14dfa2..9ac7ca681757 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":