From b90e6add03e973c3c645ac1ce97a08e236ad0f7f Mon Sep 17 00:00:00 2001 From: Rohan Godha Date: Sat, 14 Nov 2026 22:44:36 -0500 Subject: [PATCH] perf(client): reuse unchanged pr links while streaming A streaming thread resends its whole shell on each tick, with every linked pull request decoded into new objects. A thread with hundreds of links then re-compared every link, re-resolved its current pull request, and rebuilt the saved summaries by re-parsing every link URL in the environment, on each tick from any thread. The shell reducer now keeps the prior links array when the links are unchanged, the current pull request is cached per links array, and saved summaries are cached per link. --- apps/web/src/state/pullRequests.ts | 19 +++++++++++-- .../src/state/shellReducer.test.ts | 28 +++++++++++++++++++ .../client-runtime/src/state/shellReducer.ts | 25 +++++++++++++---- .../src/shared/threadPullRequests.ts | 17 +++++++++++ 4 files changed, 82 insertions(+), 7 deletions(-) diff --git a/apps/web/src/state/pullRequests.ts b/apps/web/src/state/pullRequests.ts index 5e75d32cad75d..a19d51042fc10 100644 --- a/apps/web/src/state/pullRequests.ts +++ b/apps/web/src/state/pullRequests.ts @@ -502,13 +502,28 @@ function savedPullRequestKey( : `${environmentId}:${threadPullRequestKeyOf({ ...reference, host: reference.host })}`; } +// Any thread's streaming tick rebuilds the map below from every linked pull request. Links keep +// their identity until they change, so each link's URL is parsed once, not once per tick. +const savedLinkSummaries = new WeakMap< + ThreadPullRequestLink, + { readonly projectId: ProjectId; readonly summary: PullRequestSummary | null } +>(); + +function savedLinkSummary(link: ThreadPullRequestLink, projectId: ProjectId) { + const cached = savedLinkSummaries.get(link); + if (cached?.projectId === projectId) return cached.summary; + const summary = threadPullRequestLinkToSummary(link, projectId); + savedLinkSummaries.set(link, { projectId, summary }); + return summary; +} + const savedPullRequestSummariesAtom = Atom.make((get) => { const summaries = new Map(); for (const thread of get(environmentThreadShells.threadShellsAtom)) { for (const link of visibleThreadPullRequests(thread.pullRequests)) { - const summary = threadPullRequestLinkToSummary(link, thread.projectId); + const summary = savedLinkSummary(link, thread.projectId); if (summary === null) continue; - const key = savedPullRequestKey(thread.environmentId, link); + const key = `${thread.environmentId}:${threadPullRequestKeyOf(link)}`; summaries.set(key, newestPullRequestSummary(summaries.get(key) ?? null, summary) ?? summary); } } diff --git a/packages/client-runtime/src/state/shellReducer.test.ts b/packages/client-runtime/src/state/shellReducer.test.ts index f98a14c3ed32d..352a13f7f3770 100644 --- a/packages/client-runtime/src/state/shellReducer.test.ts +++ b/packages/client-runtime/src/state/shellReducer.test.ts @@ -420,6 +420,34 @@ describe("applyShellStreamEvents", () => { expect(applyShellStreamEvents(seed, [])).toBe(seed); }); + it("keeps an unchanged links array when a streaming row changes", () => { + const link: ThreadPullRequestLink = { + host: "github.com", + repository: "pingdotgg/t3code", + number: 3, + url: "https://github.com/pingdotgg/t3code/pull/3", + source: "agent", + linkedAt: "2026-06-20T00:00:00.000Z", + snapshot: null, + stack: null, + }; + const linked = { ...a, pullRequests: [link] }; + const seed = { ...v2ShellSnapshot, threads: [linked, b], snapshotSequence: 4 }; + const tick = { ...linked, itemCount: 9, pullRequests: [{ ...link }] }; + const relinked = { ...linked, pullRequests: [{ ...link, number: 4 }] }; + + expect(applyShellStreamEvent(seed, update(5, tick)).threads[0]?.pullRequests).toBe( + linked.pullRequests, + ); + const batched = applyShellStreamEvents(seed, [update(5, tick), update(6, { ...b })]); + expect(batched.threads.find((thread) => thread.id === a.id)?.pullRequests).toBe( + linked.pullRequests, + ); + expect(applyShellStreamEvent(seed, update(5, relinked)).threads[0]?.pullRequests).toBe( + relinked.pullRequests, + ); + }); + it("does not advance the cursor for an unknown future event", () => { const events = [ { kind: "future.event", sequence: 99 } as never, diff --git a/packages/client-runtime/src/state/shellReducer.ts b/packages/client-runtime/src/state/shellReducer.ts index 2ef7dea533f3c..5421217410a37 100644 --- a/packages/client-runtime/src/state/shellReducer.ts +++ b/packages/client-runtime/src/state/shellReducer.ts @@ -22,6 +22,19 @@ export function sameThreadPullRequests( return left === right || sameLinks(left, right); } +// A streaming thread resends its whole shell on each tick. Keeping the prior links array when it +// has not changed lets later comparisons stop at identity and keeps per-link caches warm. +function retainUnchangedLinks( + previous: OrchestrationV2ThreadShell | undefined, + next: OrchestrationV2ThreadShell, +): OrchestrationV2ThreadShell { + if (previous?.pullRequests === undefined || next.pullRequests === undefined) return next; + if (previous.pullRequests === next.pullRequests) return next; + return sameLinks(previous.pullRequests, next.pullRequests) + ? { ...next, pullRequests: previous.pullRequests } + : next; +} + /** * Links are compared apart from the rest of the row because a deferred snapshot arrives without * them; such a row keeps its links until they fill in. @@ -39,7 +52,7 @@ export function reuseUnchangedThreadShells( const candidate = thread.pullRequests === undefined && prior.pullRequests !== undefined ? { ...thread, pullRequests: prior.pullRequests } - : thread; + : retainUnchangedLinks(prior, thread); const retained = sameThreadShell(prior, candidate) ? prior : candidate; if (retained !== thread) changed = true; return retained; @@ -160,7 +173,8 @@ export function applyShellStreamEvent( event.location === "active" ? snapshot.threads.find((thread) => thread.id === event.thread.id) : undefined; - if (existing !== undefined && sameThreadShell(existing, event.thread)) { + const thread = retainUnchangedLinks(existing, event.thread); + if (existing !== undefined && sameThreadShell(existing, thread)) { return { ...snapshot, snapshotSequence: event.sequence }; } const withoutThread = (threads: OrchestrationV2ShellSnapshot["threads"]) => @@ -169,7 +183,7 @@ export function applyShellStreamEvent( ...snapshot, threads: event.location === "active" - ? upsertById(snapshot.threads, event.thread) + ? upsertById(snapshot.threads, thread) : withoutThread(snapshot.threads), // The archive has its own bounded query/subscription. Older servers may // still send archive-located deltas here; remove them from the normal @@ -243,8 +257,9 @@ export function applyShellStreamEvents( const id = event.thread.id; if (event.location === "active") { const previous = threads.get(id); - if (previous !== undefined && sameThreadShell(previous, event.thread)) break; - threads.set(event.thread); + const thread = retainUnchangedLinks(previous, event.thread); + if (previous !== undefined && sameThreadShell(previous, thread)) break; + threads.set(thread); } else { threads.remove(id); } diff --git a/packages/source-control-core/src/shared/threadPullRequests.ts b/packages/source-control-core/src/shared/threadPullRequests.ts index c485bf24d759d..a6982d2fefc67 100644 --- a/packages/source-control-core/src/shared/threadPullRequests.ts +++ b/packages/source-control-core/src/shared/threadPullRequests.ts @@ -142,8 +142,25 @@ export function makeThreadPullRequests( return `${normalized.host}/${normalized.repository}#${normalized.number}`; } + // Shells keep their links array while a run streams, and several components read the + // current pull request on each shell change, so resolve each array's chains once. + const currentByLinks = new WeakMap< + ReadonlyArray, + ThreadCurrentPullRequest | null + >(); + function resolveThreadCurrentPullRequest( links: ReadonlyArray, + ): ThreadCurrentPullRequest | null { + const cached = currentByLinks.get(links); + if (cached !== undefined) return cached; + const current = resolveUncachedThreadCurrentPullRequest(links); + currentByLinks.set(links, current); + return current; + } + + function resolveUncachedThreadCurrentPullRequest( + links: ReadonlyArray, ): ThreadCurrentPullRequest | null { const visible = visibleThreadPullRequests(links); if (visible.length === 0) return null;