authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-04-13 13:55:20-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-04-13 13:55:20-07:00
log569fea87a90f1d9a5d4d9b6c7c1c59801324a561
tree8cc52ba16fc900048f6568aac3345ab76956a276
parent33dadb5aeca1ac5201682daed6e1412d57e14287
signature Signed by PGP key B5690EEEBB952194

Warm sidebar thread detail subscriptions (#2001)

Co-authored-by: Cursor Agent <cursoragent@cursor.com>

5 files changed, 315 insertions(+), 9 deletions(-)

apps/web/src/components/Sidebar.logic.test.ts+15
......@@ -2,6 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
22
33import {
44 createThreadJumpHintVisibilityController,
5 getSidebarThreadIdsToPrewarm,
56 getVisibleSidebarThreadIds,
67 resolveAdjacentThreadId,
78 getFallbackThreadIdAfterDelete,
......@@ -121,6 +122,20 @@ describe("createThreadJumpHintVisibilityController", () => {
121122 });
122123});
123124
125describe("getSidebarThreadIdsToPrewarm", () => {
126 it("returns only the first visible thread ids up to the prewarm limit", () => {
127 expect(getSidebarThreadIdsToPrewarm(["t1", "t2", "t3"], 2)).toEqual(["t1", "t2"]);
128 });
129
130 it("returns all visible thread ids when they fit within the limit", () => {
131 expect(getSidebarThreadIdsToPrewarm(["t1", "t2"], 10)).toEqual(["t1", "t2"]);
132 });
133
134 it("returns no thread ids when the limit is zero", () => {
135 expect(getSidebarThreadIdsToPrewarm(["t1", "t2"], 0)).toEqual([]);
136 });
137});
138
124139describe("shouldClearThreadSelectionOnMouseDown", () => {
125140 it("preserves selection for thread items", () => {
126141 const child = {
apps/web/src/components/Sidebar.logic.ts+10
......@@ -12,6 +12,9 @@ import { isLatestTurnSettled } from "../session-logic";
1212
1313export const THREAD_SELECTION_SAFE_SELECTOR = "[data-thread-item], [data-thread-selection-safe]";
1414export const THREAD_JUMP_HINT_SHOW_DELAY_MS = 100;
15// Visible sidebar rows are prewarmed into the thread-detail cache so opening a
16// nearby thread usually reuses an already-hot subscription.
17export const SIDEBAR_THREAD_PREWARM_LIMIT = 10;
1518export type SidebarNewThreadEnvMode = "local" | "worktree";
1619type SidebarProject = {
1720 id: string;
......@@ -243,6 +246,13 @@ export function getVisibleSidebarThreadIds<TThreadId>(
243246 );
244247}
245248
249export function getSidebarThreadIdsToPrewarm<TThreadId>(
250 visibleThreadIds: readonly TThreadId[],
251 limit = SIDEBAR_THREAD_PREWARM_LIMIT,
252): TThreadId[] {
253 return visibleThreadIds.slice(0, Math.max(0, limit));
254}
255
246256export function resolveAdjacentThreadId<T>(input: {
247257 threadIds: readonly T[];
248258 currentThreadId: T | null;
apps/web/src/components/Sidebar.tsx+27
......@@ -43,6 +43,7 @@ import {
4343 type GitStatusResult,
4444} from "@t3tools/contracts";
4545import {
46 parseScopedThreadKey,
4647 scopedProjectKey,
4748 scopedThreadKey,
4849 scopeProjectRef,
......@@ -81,6 +82,7 @@ import { useGitStatus } from "../lib/gitStatusState";
8182import { readLocalApi } from "../localApi";
8283import { useComposerDraftStore } from "../composerDraftStore";
8384import { useNewThreadHandler } from "../hooks/useHandleNewThread";
85import { retainThreadDetailSubscription } from "../environments/runtime/service";
8486
8587import { useThreadActions } from "../hooks/useThreadActions";
8688import {
......@@ -122,6 +124,7 @@ import {
122124import { useThreadSelectionStore } from "../threadSelectionStore";
123125import { isNonEmpty as isNonEmptyString } from "effect/String";
124126import {
127 getSidebarThreadIdsToPrewarm,
125128 resolveAdjacentThreadId,
126129 isContextMenuPointerDown,
127130 resolveProjectStatusIndicator,
......@@ -2878,6 +2881,30 @@ export default function Sidebar() {
28782881 ? threadJumpLabelByKey
28792882 : EMPTY_THREAD_JUMP_LABELS;
28802883 const orderedSidebarThreadKeys = visibleSidebarThreadKeys;
2884 const prewarmedSidebarThreadKeys = useMemo(
2885 () => getSidebarThreadIdsToPrewarm(visibleSidebarThreadKeys),
2886 [visibleSidebarThreadKeys],
2887 );
2888 const prewarmedSidebarThreadRefs = useMemo(
2889 () =>
2890 prewarmedSidebarThreadKeys.flatMap((threadKey) => {
2891 const ref = parseScopedThreadKey(threadKey);
2892 return ref ? [ref] : [];
2893 }),
2894 [prewarmedSidebarThreadKeys],
2895 );
2896
2897 useEffect(() => {
2898 const releases = prewarmedSidebarThreadRefs.map((ref) =>
2899 retainThreadDetailSubscription(ref.environmentId, ref.threadId),
2900 );
2901
2902 return () => {
2903 for (const release of releases) {
2904 release();
2905 }
2906 };
2907 }, [prewarmedSidebarThreadRefs]);
28812908
28822909 useEffect(() => {
28832910 const clearThreadJumpHints = () => {
apps/web/src/environments/runtime/service.threadSubscriptions.test.ts+150-3
......@@ -1,5 +1,11 @@
11import { QueryClient } from "@tanstack/react-query";
2import { EnvironmentId, ThreadId } from "@t3tools/contracts";
2import {
3 EnvironmentId,
4 ProjectId,
5 ThreadId,
6 TurnId,
7 type OrchestrationShellSnapshot,
8} from "@t3tools/contracts";
39import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
410
511const mockSubscribeThread = vi.fn();
......@@ -65,6 +71,74 @@ vi.mock("../../rpc/wsTransport", () => ({
6571 WsTransport: MockWsTransport,
6672}));
6773
74function makeThreadShellSnapshot(params: {
75 readonly threadId: ThreadId;
76 readonly sessionStatus?:
77 | "idle"
78 | "starting"
79 | "running"
80 | "ready"
81 | "interrupted"
82 | "stopped"
83 | "error";
84 readonly hasPendingApprovals?: boolean;
85 readonly hasPendingUserInput?: boolean;
86 readonly hasActionableProposedPlan?: boolean;
87}): OrchestrationShellSnapshot {
88 const projectId = ProjectId.make("project-1");
89 const turnId = TurnId.make("turn-1");
90
91 return {
92 snapshotSequence: 1,
93 projects: [],
94 updatedAt: "2026-04-13T00:00:00.000Z",
95 threads: [
96 {
97 id: params.threadId,
98 projectId,
99 title: "Thread",
100 modelSelection: {
101 provider: "codex",
102 model: "gpt-5-codex",
103 },
104 runtimeMode: "full-access",
105 interactionMode: "default",
106 branch: null,
107 worktreePath: null,
108 latestTurn:
109 params.sessionStatus === "running"
110 ? {
111 turnId,
112 state: "running",
113 requestedAt: "2026-04-13T00:00:00.000Z",
114 startedAt: "2026-04-13T00:00:01.000Z",
115 completedAt: null,
116 assistantMessageId: null,
117 }
118 : null,
119 createdAt: "2026-04-13T00:00:00.000Z",
120 updatedAt: "2026-04-13T00:00:00.000Z",
121 archivedAt: null,
122 session: params.sessionStatus
123 ? {
124 threadId: params.threadId,
125 status: params.sessionStatus,
126 providerName: "codex",
127 runtimeMode: "full-access",
128 activeTurnId: params.sessionStatus === "running" ? turnId : null,
129 lastError: null,
130 updatedAt: "2026-04-13T00:00:00.000Z",
131 }
132 : null,
133 latestUserMessageAt: null,
134 hasPendingApprovals: params.hasPendingApprovals ?? false,
135 hasPendingUserInput: params.hasPendingUserInput ?? false,
136 hasActionableProposedPlan: params.hasActionableProposedPlan ?? false,
137 },
138 ],
139 };
140}
141
68142describe("retainThreadDetailSubscription", () => {
69143 beforeEach(() => {
70144 vi.useFakeTimers();
......@@ -119,16 +193,89 @@ describe("retainThreadDetailSubscription", () => {
119193 expect(mockSubscribeThread).toHaveBeenCalledTimes(1);
120194
121195 releaseSecond();
122 await vi.advanceTimersByTimeAsync(2 * 60 * 1000 - 1);
196 await vi.advanceTimersByTimeAsync(2 * 60 * 1000);
123197 expect(mockThreadUnsubscribe).not.toHaveBeenCalled();
124198
125 await vi.advanceTimersByTimeAsync(1);
199 await vi.advanceTimersByTimeAsync(28 * 60 * 1000);
126200 expect(mockThreadUnsubscribe).toHaveBeenCalledTimes(1);
127201
128202 stop();
129203 await resetEnvironmentServiceForTests();
130204 });
131205
206 it("keeps non-idle thread detail subscriptions attached until the thread becomes idle", async () => {
207 const {
208 retainThreadDetailSubscription,
209 startEnvironmentConnectionService,
210 resetEnvironmentServiceForTests,
211 } = await import("./service");
212
213 const stop = startEnvironmentConnectionService(new QueryClient());
214 const environmentId = EnvironmentId.make("env-1");
215 const threadId = ThreadId.make("thread-active");
216
217 const connectionInput = mockCreateEnvironmentConnection.mock.calls[0]?.[0];
218 expect(connectionInput).toBeDefined();
219
220 connectionInput.syncShellSnapshot(
221 makeThreadShellSnapshot({
222 threadId,
223 sessionStatus: "ready",
224 hasPendingApprovals: true,
225 }),
226 environmentId,
227 );
228
229 const release = retainThreadDetailSubscription(environmentId, threadId);
230 expect(mockSubscribeThread).toHaveBeenCalledTimes(1);
231
232 release();
233 await vi.advanceTimersByTimeAsync(30 * 60 * 1000);
234 expect(mockThreadUnsubscribe).not.toHaveBeenCalled();
235
236 connectionInput.applyShellEvent(
237 {
238 kind: "thread-upserted",
239 sequence: 2,
240 thread: makeThreadShellSnapshot({
241 threadId,
242 sessionStatus: "idle",
243 }).threads[0]!,
244 },
245 environmentId,
246 );
247
248 await vi.advanceTimersByTimeAsync(30 * 60 * 1000);
249 expect(mockThreadUnsubscribe).toHaveBeenCalledTimes(1);
250
251 stop();
252 await resetEnvironmentServiceForTests();
253 });
254
255 it("allows a larger idle cache before capacity eviction starts", async () => {
256 const {
257 retainThreadDetailSubscription,
258 startEnvironmentConnectionService,
259 resetEnvironmentServiceForTests,
260 } = await import("./service");
261
262 const stop = startEnvironmentConnectionService(new QueryClient());
263 const environmentId = EnvironmentId.make("env-1");
264
265 for (let index = 0; index < 12; index += 1) {
266 const release = retainThreadDetailSubscription(
267 environmentId,
268 ThreadId.make(`thread-${index + 1}`),
269 );
270 release();
271 }
272
273 expect(mockThreadUnsubscribe).not.toHaveBeenCalled();
274
275 stop();
276 await resetEnvironmentServiceForTests();
277 });
278
132279 it("disposes cached thread detail subscriptions when the environment service resets", async () => {
133280 const {
134281 retainThreadDetailSubscription,
apps/web/src/environments/runtime/service.ts+113-6
......@@ -54,6 +54,7 @@ import { createEnvironmentConnection, type EnvironmentConnection } from "./conne
5454import {
5555 useStore,
5656 selectProjectsAcrossEnvironments,
57 selectSidebarThreadSummaryByRef,
5758 selectThreadByRef,
5859 selectThreadsAcrossEnvironments,
5960} from "~/store";
......@@ -86,8 +87,15 @@ const threadDetailSubscriptions = new Map<string, ThreadDetailSubscriptionEntry>
8687let activeService: EnvironmentServiceState | null = null;
8788let needsProviderInvalidation = false;
8889
89const THREAD_DETAIL_SUBSCRIPTION_IDLE_EVICTION_MS = 2 * 60 * 1000;
90const MAX_CACHED_THREAD_DETAIL_SUBSCRIPTIONS = 8;
90// Thread detail subscription cache policy:
91// - Active consumers keep a subscription retained via refCount.
92// - Released subscriptions stay warm for a longer idle TTL to avoid churn
93// while moving around the UI.
94// - Threads with active work or pending user action are sticky and are never
95// evicted while they remain non-idle.
96// - Capacity eviction only targets idle cached subscriptions.
97const THREAD_DETAIL_SUBSCRIPTION_IDLE_EVICTION_MS = 15 * 60 * 1000;
98const MAX_CACHED_THREAD_DETAIL_SUBSCRIPTIONS = 32;
9199const NOOP = () => undefined;
92100
93101function getThreadDetailSubscriptionKey(environmentId: EnvironmentId, threadId: ThreadId): string {
......@@ -104,6 +112,55 @@ function clearThreadDetailSubscriptionEviction(
104112 return entry;
105113}
106114
115function isNonIdleThreadDetailSubscription(entry: ThreadDetailSubscriptionEntry): boolean {
116 const threadRef = scopeThreadRef(entry.environmentId, entry.threadId);
117 const state = useStore.getState();
118 const sidebarThread = selectSidebarThreadSummaryByRef(state, threadRef);
119
120 // Prefer shell/sidebar state first because it carries the coarse thread
121 // readiness flags used throughout the UI (pending approvals/input/plan).
122 if (sidebarThread) {
123 if (
124 sidebarThread.hasPendingApprovals ||
125 sidebarThread.hasPendingUserInput ||
126 sidebarThread.hasActionableProposedPlan
127 ) {
128 return true;
129 }
130
131 const orchestrationStatus = sidebarThread.session?.orchestrationStatus;
132 if (
133 orchestrationStatus &&
134 orchestrationStatus !== "idle" &&
135 orchestrationStatus !== "stopped"
136 ) {
137 return true;
138 }
139
140 if (sidebarThread.latestTurn?.state === "running") {
141 return true;
142 }
143 }
144
145 const thread = selectThreadByRef(state, threadRef);
146 if (!thread) {
147 return false;
148 }
149
150 const orchestrationStatus = thread.session?.orchestrationStatus;
151 return (
152 Boolean(
153 orchestrationStatus && orchestrationStatus !== "idle" && orchestrationStatus !== "stopped",
154 ) ||
155 thread.latestTurn?.state === "running" ||
156 thread.pendingSourceProposedPlan !== undefined
157 );
158}
159
160function shouldEvictThreadDetailSubscription(entry: ThreadDetailSubscriptionEntry): boolean {
161 return entry.refCount === 0 && !isNonIdleThreadDetailSubscription(entry);
162}
163
107164function attachThreadDetailSubscription(entry: ThreadDetailSubscriptionEntry): boolean {
108165 if (entry.unsubscribeConnectionListener !== null) {
109166 entry.unsubscribeConnectionListener();
......@@ -181,11 +238,20 @@ function reconcileThreadDetailSubscriptionsForEnvironment(
181238
182239function scheduleThreadDetailSubscriptionEviction(entry: ThreadDetailSubscriptionEntry): void {
183240 clearThreadDetailSubscriptionEviction(entry);
241 if (!shouldEvictThreadDetailSubscription(entry)) {
242 return;
243 }
244
184245 entry.evictionTimeoutId = setTimeout(() => {
185246 const currentEntry = threadDetailSubscriptions.get(
186247 getThreadDetailSubscriptionKey(entry.environmentId, entry.threadId),
187248 );
188 if (!currentEntry || currentEntry.refCount > 0) {
249 if (!currentEntry) {
250 return;
251 }
252
253 currentEntry.evictionTimeoutId = null;
254 if (!shouldEvictThreadDetailSubscription(currentEntry)) {
189255 return;
190256 }
191257 disposeThreadDetailSubscriptionByKey(
......@@ -200,7 +266,7 @@ function evictIdleThreadDetailSubscriptionsToCapacity(): void {
200266 }
201267
202268 const idleEntries = [...threadDetailSubscriptions.entries()]
203 .filter(([, entry]) => entry.refCount === 0)
269 .filter(([, entry]) => shouldEvictThreadDetailSubscription(entry))
204270 .toSorted(([, left], [, right]) => left.lastAccessedAt - right.lastAccessedAt);
205271
206272 for (const [key] of idleEntries) {
......@@ -211,6 +277,42 @@ function evictIdleThreadDetailSubscriptionsToCapacity(): void {
211277 }
212278}
213279
280function reconcileThreadDetailSubscriptionEvictionState(
281 entry: ThreadDetailSubscriptionEntry,
282): void {
283 clearThreadDetailSubscriptionEviction(entry);
284 if (!shouldEvictThreadDetailSubscription(entry)) {
285 return;
286 }
287
288 scheduleThreadDetailSubscriptionEviction(entry);
289}
290
291function reconcileThreadDetailSubscriptionEvictionForThread(
292 environmentId: EnvironmentId,
293 threadId: ThreadId,
294): void {
295 const entry = threadDetailSubscriptions.get(
296 getThreadDetailSubscriptionKey(environmentId, threadId),
297 );
298 if (!entry) {
299 return;
300 }
301
302 reconcileThreadDetailSubscriptionEvictionState(entry);
303}
304
305function reconcileThreadDetailSubscriptionEvictionForEnvironment(
306 environmentId: EnvironmentId,
307): void {
308 for (const entry of threadDetailSubscriptions.values()) {
309 if (entry.environmentId === environmentId) {
310 reconcileThreadDetailSubscriptionEvictionState(entry);
311 }
312 }
313 evictIdleThreadDetailSubscriptionsToCapacity();
314}
315
214316export function retainThreadDetailSubscription(
215317 environmentId: EnvironmentId,
216318 threadId: ThreadId,
......@@ -233,7 +335,7 @@ export function retainThreadDetailSubscription(
233335 existing.refCount = Math.max(0, existing.refCount - 1);
234336 existing.lastAccessedAt = Date.now();
235337 if (existing.refCount === 0) {
236 scheduleThreadDetailSubscriptionEviction(existing);
338 reconcileThreadDetailSubscriptionEvictionState(existing);
237339 evictIdleThreadDetailSubscriptionsToCapacity();
238340 }
239341 };
......@@ -263,7 +365,7 @@ export function retainThreadDetailSubscription(
263365 entry.refCount = Math.max(0, entry.refCount - 1);
264366 entry.lastAccessedAt = Date.now();
265367 if (entry.refCount === 0) {
266 scheduleThreadDetailSubscriptionEviction(entry);
368 reconcileThreadDetailSubscriptionEvictionState(entry);
267369 evictIdleThreadDetailSubscriptionsToCapacity();
268370 }
269371 };
......@@ -473,6 +575,8 @@ function applyRecoveredEventBatch(
473575 for (const threadId of batchEffects.removeTerminalStateThreadIds) {
474576 useTerminalStateStore.getState().removeTerminalState(scopeThreadRef(environmentId, threadId));
475577 }
578
579 reconcileThreadDetailSubscriptionEvictionForEnvironment(environmentId);
476580}
477581
478582export function applyEnvironmentThreadDetailEvent(
......@@ -507,6 +611,8 @@ function applyShellEvent(event: OrchestrationShellStreamEvent, environmentId: En
507611 if (previousThread?.archivedAt === null && event.thread.archivedAt !== null && threadRef) {
508612 useTerminalStateStore.getState().removeTerminalState(threadRef);
509613 }
614 reconcileThreadDetailSubscriptionEvictionForThread(environmentId, event.thread.id);
615 evictIdleThreadDetailSubscriptionsToCapacity();
510616 return;
511617 case "thread-removed":
512618 if (threadRef) {
......@@ -529,6 +635,7 @@ function createEnvironmentConnectionHandlers() {
529635 environmentId,
530636 snapshot.threads.map((thread) => thread.id),
531637 );
638 reconcileThreadDetailSubscriptionEvictionForEnvironment(environmentId);
532639 reconcileSnapshotDerivedState();
533640 },
534641 applyTerminalEvent: (event: TerminalEvent, environmentId: EnvironmentId) => {