History

Harden remote connect and subscription handling

- Scope cleartext mobile config to development - Fix stale loads, ordering, and terminal reuse - Clean up terminal metadata subscriptions on failure

Julius Marminge committed 6 months ago

Commit 7f03f4d1 parent 4ac261815 files · +276 −47Browse files

Unified · Split

15 files
  1. apps/mobile/app.config.ts +9 −4
  2. apps/mobile/modules/t3-review-diff/ios/T3ReviewDiffView.swift +11 −1
  3. apps/mobile/src/features/review/reviewWordDiffs.test.ts +31 −0
  4. apps/mobile/src/features/review/reviewWordDiffs.ts +1 −1
  5. apps/mobile/src/features/terminal/ThreadTerminalPanel.tsx +5 −5
  6. apps/mobile/src/features/threads/new-task-flow-provider.tsx +17 −3
  7. apps/mobile/src/lib/repositoryGroups.test.ts +68 −4
  8. apps/mobile/src/lib/repositoryGroups.ts +5 −3
  9. apps/server/src/terminal/Layers/Manager.test.ts +21 −0
  10. apps/server/src/terminal/Layers/Manager.ts +11 −1
  11. apps/web/src/components/ChatView.tsx +13 −17
  12. packages/client-runtime/src/vcsRefState.test.ts +25 −0
  13. packages/client-runtime/src/vcsRefState.ts +8 −0
  14. packages/client-runtime/src/wsRpcClient.test.ts +33 −0
  15. packages/client-runtime/src/wsRpcClient.ts +18 −8
apps/mobile/app.config.ts+9−4
@@ -49,6 +49,7 @@ function resolveAppVariant(value: string | undefined): AppVariant { · 48 unmodified lines
4949 }
5050
5151 const variant = VARIANT_CONFIG[APP_VARIANT];
52+const allowsCleartextTraffic = APP_VARIANT === "development";
5253
5354 const config: ExpoConfig = {
5455 name: variant.appName,
@@ -77,9 +78,13 @@ const config: ExpoConfig = { · 22 unmodified lines
7778 supportsTablet: true,
7879 bundleIdentifier: variant.iosBundleIdentifier,
7980 infoPlist: {
80− NSAppTransportSecurity: {
81− NSAllowsArbitraryLoads: true,
82− },
81+ ...(allowsCleartextTraffic
82+ ? {
83+ NSAppTransportSecurity: {
84+ NSAllowsArbitraryLoads: true,
85+ },
86+ }
87+ : {}),
8388 ITSAppUsesNonExemptEncryption: false,
8489 },
8590 },
@@ -128,7 +133,7 @@ const config: ExpoConfig = { · 42 unmodified lines
128133 ],
129134 "expo-secure-store",
130135 "expo-router",
131− "./plugins/withAndroidCleartextTraffic.cjs",
136+ ...(allowsCleartextTraffic ? ["./plugins/withAndroidCleartextTraffic.cjs"] : []),
132137 ],
133138 extra: {
134139 appVariant: APP_VARIANT,
apps/mobile/modules/t3-review-diff/ios/T3ReviewDiffView.swift+11−1
@@ -768,6 +768,7 @@ private final class ReviewDiffContentView: UIView, UIGestureRecognizerDelegate { · 767 unmodified lines
768768 didSet {
769769 tokenColorsByHex.removeAll()
770770 tokenAttributedStringsByRowId.removeAll()
771+ setNeedsDisplayForVisibleBounds()
771772 }
772773 }
773774 private(set) var contentHeight: CGFloat = 0
@@ -1007,6 +1008,10 @@ private final class ReviewDiffContentView: UIView, UIGestureRecognizerDelegate { · 233 unmodified lines
10071008 }
10081009
10091010 let point = gesture.location(in: self)
1011+ if let stickyHeader = stickyFileHeaderTarget(), stickyHeader.rect.contains(point) {
1012+ return
1013+ }
1014+
10101015 guard let rowIndex = rowIndex(at: verticalOffset + point.y),
10111016 rows.indices.contains(rowIndex) else {
10121017 return
@@ -1420,6 +1425,10 @@ private final class ReviewDiffContentView: UIView, UIGestureRecognizerDelegate { · 407 unmodified lines
14201425 lastDecelerationTimestamp = 0
14211426 }
14221427
1428+ deinit {
1429+ stopHorizontalDeceleration()
1430+ }
1431+
14231432 override func draw(_ rect: CGRect) {
14241433 let drawStartedAt = CACurrentMediaTime()
14251434 guard let context = UIGraphicsGetCurrentContext() else {
@@ -1914,7 +1923,6 @@ private final class ReviewDiffContentView: UIView, UIGestureRecognizerDelegate { · 488 unmodified lines
19141923 let change = row.change ?? "context"
19151924 rowBackground(for: change).setFill()
19161925 context.fill(rect)
1917− drawSelectionOverlay(row, rect: rect, context: context)
19181926
19191927 if change == "add" {
19201928 theme.addBar.setFill()
@@ -1926,6 +1934,8 @@ private final class ReviewDiffContentView: UIView, UIGestureRecognizerDelegate { · 5 unmodified lines
19261934 )
19271935 }
19281936
1937+ drawSelectionOverlay(row, rect: rect, context: context)
1938+
19291939 let lineNumber = row.newLineNumber ?? row.oldLineNumber
19301940 if let lineNumber {
19311941 drawRightAlignedText(
apps/mobile/src/features/review/reviewWordDiffs.test.ts+31−0
@@ -59,4 +59,35 @@ describe("applyDiffRangesToTokens", () => { · 58 unmodified lines
5959 },
6060 ]);
6161 });
62+
63+ it("skips exhausted ranges when later tokens start after the last range", () => {
64+ const tokens = [
65+ {
66+ content: "a",
67+ color: "#fff",
68+ fontStyle: null,
69+ },
70+ {
71+ content: "b",
72+ color: "#fff",
73+ fontStyle: null,
74+ },
75+ ];
76+
77+ const nextTokens = applyDiffRangesToTokens(tokens, [{ start: 0, end: 1 }]);
78+
79+ expect(nextTokens).toEqual([
80+ {
81+ content: "a",
82+ color: "#fff",
83+ fontStyle: null,
84+ diffHighlight: true,
85+ },
86+ {
87+ content: "b",
88+ color: "#fff",
89+ fontStyle: null,
90+ },
91+ ]);
92+ });
6293 });
apps/mobile/src/features/review/reviewWordDiffs.ts+1−1
@@ -208,7 +208,7 @@ export function applyDiffRangesToTokens( · 207 unmodified lines
208208 return;
209209 }
210210
211− while ((ranges[rangeIndex]?.end ?? 0) <= tokenStart) {
211+ while (rangeIndex < ranges.length && ranges[rangeIndex]!.end <= tokenStart) {
212212 rangeIndex += 1;
213213 }
214214
apps/mobile/src/features/terminal/ThreadTerminalPanel.tsx+5−5
@@ -1,6 +1,6 @@
11 import { DEFAULT_TERMINAL_ID, type EnvironmentId, type ThreadId } from "@t3tools/contracts";
22 import { SymbolView } from "expo-symbols";
3−import { memo, useCallback, useEffect, useState } from "react";
3+import { memo, useCallback, useEffect, useRef, useState } from "react";
44 import { Pressable, View } from "react-native";
55
66 import { AppText as Text } from "../../components/AppText";
@@ -41,6 +41,8 @@ export const ThreadTerminalPanel = memo(function ThreadTerminalPanel( · 34 unmodified lines
4141 cols: DEFAULT_TERMINAL_COLS,
4242 rows: DEFAULT_TERMINAL_ROWS,
4343 });
44+ const lastGridSizeRef = useRef(lastGridSize);
45+ lastGridSizeRef.current = lastGridSize;
4446
4547 const terminalKey = `${props.environmentId}:${props.threadId}:${terminalId}`;
4648 const isRunning = terminal.status === "running" || terminal.status === "starting";
@@ -73,13 +75,11 @@ export const ThreadTerminalPanel = memo(function ThreadTerminalPanel( · 26 unmodified lines
7375 terminalId,
7476 cwd: props.cwd,
7577 worktreePath: props.worktreePath,
76− cols: lastGridSize.cols,
77− rows: lastGridSize.rows,
78+ cols: lastGridSizeRef.current.cols,
79+ rows: lastGridSizeRef.current.rows,
7880 },
7981 });
8082 }, [
81− lastGridSize.cols,
82− lastGridSize.rows,
8383 props.cwd,
8484 props.environmentId,
8585 props.threadId,
apps/mobile/src/features/threads/new-task-flow-provider.tsx+17−3
@@ -1,4 +1,4 @@
1−import React, { useCallback, useEffect, useMemo, useState } from "react";
1+import React, { useCallback, useEffect, useMemo, useRef, useState } from "react";
22
33 import type {
44 EnvironmentId,
@@ -152,6 +152,7 @@ export function NewTaskFlowProvider(props: React.PropsWithChildren) { · 147 unmodified lines
152152 const [workspaceMode, setWorkspaceMode] = useState<WorkspaceMode>("local");
153153 const [selectedBranchName, setSelectedBranchName] = useState<string | null>(null);
154154 const [selectedWorktreePath, setSelectedWorktreePath] = useState<string | null>(null);
155+ const branchLoadVersionRef = useRef(0);
155156 const [prompt, setPrompt] = useState("");
156157 const [attachments, setAttachments] = useState<ReadonlyArray<DraftComposerImageAttachment>>([]);
157158 const [submitting, setSubmitting] = useState(false);
@@ -329,11 +330,16 @@ export function NewTaskFlowProvider(props: React.PropsWithChildren) { · 171 unmodified lines
329330 }, [availableBranches, branchQuery]);
330331
331332 const setProject = useCallback((project: EnvironmentScopedProjectShell) => {
333+ const nextProjectKey = scopedProjectKey(project.environmentId, project.id);
334+ branchLoadVersionRef.current += 1;
332335 setSelectedEnvironmentId(project.environmentId);
333− setSelectedProjectKey(scopedProjectKey(project.environmentId, project.id));
336+ setSelectedProjectKey(nextProjectKey);
337+ setSelectedBranchName(null);
338+ setSelectedWorktreePath(null);
334339 }, []);
335340
336341 const selectEnvironment = useCallback((environmentId: EnvironmentId) => {
342+ branchLoadVersionRef.current += 1;
337343 setSelectedEnvironmentId(environmentId);
338344 setSelectedProjectKey(null);
339345 setSelectedBranchName(null);
@@ -355,12 +361,17 @@ export function NewTaskFlowProvider(props: React.PropsWithChildren) { · 15 unmodified lines
355361 return;
356362 }
357363
364+ const loadVersion = ++branchLoadVersionRef.current;
365+ const projectKey = scopedProjectKey(selectedProject.environmentId, selectedProject.id);
358366 try {
359367 const result = await vcsRefManager.load({
360368 environmentId: selectedProject.environmentId,
361369 cwd: selectedProject.workspaceRoot,
362370 query: null,
363371 });
372+ if (loadVersion !== branchLoadVersionRef.current || selectedProjectKey !== projectKey) {
373+ return;
374+ }
364375 setPendingConnectionError(null);
365376 const branches = pipe(
366377 result?.refs ?? [],
@@ -377,9 +388,12 @@ export function NewTaskFlowProvider(props: React.PropsWithChildren) { · 10 unmodified lines
377388 }
378389 }
379390 } catch {
391+ if (loadVersion !== branchLoadVersionRef.current) {
392+ return;
393+ }
380394 setPendingConnectionError("Failed to load branches.");
381395 }
382− }, [selectedBranchName, selectedProject, workspaceMode]);
396+ }, [selectedBranchName, selectedProject, selectedProjectKey, workspaceMode]);
383397
384398 const value = useMemo<NewTaskFlowContextValue>(
385399 () => ({
apps/mobile/src/lib/repositoryGroups.test.ts+68−4
@@ -108,10 +108,74 @@ describe("groupProjectsByRepository", () => { · 107 unmodified lines
108108 threadCount: 2,
109109 });
110110 expect(
111− groups[0]?.projects
112− .map((entry) => entry.project.environmentId)
113− .toSorted((left, right) => left.localeCompare(right)),
114− ).toEqual(["env-local", "env-staging"]);
111+ groups[0]?.projects.map((entry) => ({
112+ environmentId: entry.project.environmentId,
113+ latestActivityAt: entry.latestActivityAt,
114+ threads: entry.threads.map((thread) => thread.id),
115+ })),
116+ ).toEqual([
117+ {
118+ environmentId: "env-local",
119+ latestActivityAt: "2026-04-03T12:00:00.000Z",
120+ threads: ["thread-1"],
121+ },
122+ {
123+ environmentId: "env-staging",
124+ latestActivityAt: "2026-04-02T12:00:00.000Z",
125+ threads: ["thread-2"],
126+ },
127+ ]);
128+ expect(groups[0]?.latestActivityAt).toBe("2026-04-03T12:00:00.000Z");
129+ });
130+
131+ it("orders threads, projects, and repository groups by latest activity", () => {
132+ const projects = [
133+ makeProject({
134+ environmentId: EnvironmentId.make("env-local"),
135+ id: ProjectId.make("older-project"),
136+ title: "Older",
137+ }),
138+ makeProject({
139+ environmentId: EnvironmentId.make("env-local"),
140+ id: ProjectId.make("newer-project"),
141+ title: "Newer",
142+ }),
143+ ];
144+
145+ const threads = [
146+ makeThread({
147+ environmentId: EnvironmentId.make("env-local"),
148+ id: ThreadId.make("older-thread"),
149+ projectId: ProjectId.make("older-project"),
150+ title: "Older thread",
151+ modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
152+ updatedAt: "2026-04-02T12:00:00.000Z",
153+ }),
154+ makeThread({
155+ environmentId: EnvironmentId.make("env-local"),
156+ id: ThreadId.make("newer-thread"),
157+ projectId: ProjectId.make("older-project"),
158+ title: "Newer thread",
159+ modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
160+ updatedAt: "2026-04-04T12:00:00.000Z",
161+ }),
162+ makeThread({
163+ environmentId: EnvironmentId.make("env-local"),
164+ id: ThreadId.make("newest-thread"),
165+ projectId: ProjectId.make("newer-project"),
166+ title: "Newest thread",
167+ modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
168+ updatedAt: "2026-04-05T12:00:00.000Z",
169+ }),
170+ ];
171+
172+ const groups = groupProjectsByRepository({ projects, threads });
173+
174+ expect(groups.map((group) => group.title)).toEqual(["Newer", "Older"]);
175+ expect(groups[1]?.projects[0]?.threads.map((thread) => thread.id)).toEqual([
176+ "newer-thread",
177+ "older-thread",
178+ ]);
115179 });
116180
117181 it("falls back to a scoped project key when repository identity is unavailable", () => {
apps/mobile/src/lib/repositoryGroups.ts+5−3
@@ -8,6 +8,8 @@ import { · 7 unmodified lines
88 EnvironmentScopedThreadShell,
99 } from "@t3tools/client-runtime";
1010
11+const DateDescending = Order.flip(Order.Date);
12+
1113 export interface MobileRepositoryProjectGroup {
1214 readonly key: string;
1315 readonly project: EnvironmentScopedProjectShell;
@@ -83,7 +85,7 @@ export function groupProjectsByRepository(input: { · 69 unmodified lines
8385 const threads = Arr.sortWith(
8486 threadsByProjectKey.get(projectKey) ?? [],
8587 (s) => new Date(s.updatedAt ?? s.createdAt),
86− Order.Date,
88+ DateDescending,
8789 );
8890
8991 const latestActivityAt = deriveProjectLatestActivity(project, threads);
@@ -123,10 +125,10 @@ export function groupProjectsByRepository(input: { · 33 unmodified lines
123125 projects: Arr.sortWith(
124126 [...existing.projects, projectGroup],
125127 (s) => new Date(s.latestActivityAt),
126− Order.Date,
128+ DateDescending,
127129 ),
128130 });
129131 }
130132
131− return Arr.sortWith([...grouped.values()], (s) => new Date(s.latestActivityAt), Order.Date);
133+ return Arr.sortWith(grouped.values(), (s) => new Date(s.latestActivityAt), DateDescending);
132134 }
apps/server/src/terminal/Layers/Manager.test.ts+21−0
@@ -1351,6 +1351,27 @@ it.layer( · 1350 unmodified lines
13511351 }),
13521352 );
13531353
1354+ it.effect("removes terminal metadata subscriptions when initial delivery fails", () =>
1355+ Effect.gen(function* () {
1356+ const { manager } = yield* createManager();
1357+ yield* manager.open(openInput({ threadId: "existing-thread" }));
1358+
1359+ const leakedLiveEvents = yield* Ref.make(0);
1360+ const exit = yield* Effect.exit(
1361+ manager.subscribeMetadata((event) =>
1362+ event.type === "snapshot"
1363+ ? Effect.die("snapshot listener failed")
1364+ : Ref.update(leakedLiveEvents, (count) => count + 1),
1365+ ),
1366+ );
1367+
1368+ expect(Exit.isFailure(exit)).toBe(true);
1369+
1370+ yield* manager.open(openInput({ threadId: "new-thread" }));
1371+ expect(yield* Ref.get(leakedLiveEvents)).toBe(0);
1372+ }),
1373+ );
1374+
13541375 it.effect(
13551376 "streams attach snapshots followed by live events without duplicate start snapshots",
13561377 () =>
apps/server/src/terminal/Layers/Manager.ts+11−1
@@ -2223,7 +2223,17 @@ export const makeTerminalManagerWithOptions = Effect.fn("makeTerminalManagerWith · 2222 unmodified lines
22232223 unsubscribe?.();
22242224 unsubscribe = null;
22252225 };
2226− });
2226+ }).pipe(
2227+ Effect.catchCause((cause) =>
2228+ Effect.flatMap(
2229+ Effect.sync(() => {
2230+ unsubscribe?.();
2231+ unsubscribe = null;
2232+ }),
2233+ () => Effect.failCause(cause),
2234+ ),
2235+ ),
2236+ );
22272237 };
22282238
22292239 const write: TerminalManagerShape["write"] = Effect.fn("terminal.write")(function* (input) {
apps/web/src/components/ChatView.tsx+13−17
@@ -989,6 +989,10 @@ export default function ChatView(props: ChatViewProps) { · 988 unmodified lines
989989 () => activeThreadKnownSessions.map((session) => session.target.terminalId),
990990 [activeThreadKnownSessions],
991991 );
992+ const activeKnownTerminalIds = useMemo(
993+ () => [...new Set([...activeServerOrderedTerminalIds, ...terminalUiState.terminalIds])],
994+ [activeServerOrderedTerminalIds, terminalUiState.terminalIds],
995+ );
992996 const reconcileTerminalIds = useTerminalUiStateStore((state) => state.reconcileTerminalIds);
993997 const activeThreadRef = useMemo(
994998 () => (activeThread ? scopeThreadRef(activeThread.environmentId, activeThread.id) : null),
@@ -2011,7 +2015,7 @@ export default function ChatView(props: ChatViewProps) { · 1016 unmodified lines
20112015 if (!api) {
20122016 return;
20132017 }
2014− const terminalId = nextTerminalId(activeServerOrderedTerminalIds);
2018+ const terminalId = nextTerminalId(activeKnownTerminalIds);
20152019 storeSplitTerminal(activeThreadRef, terminalId);
20162020 setTerminalFocusRequestId((value) => value + 1);
20172021 void (async () => {
@@ -2032,7 +2036,7 @@ export default function ChatView(props: ChatViewProps) { · 14 unmodified lines
20322036 })();
20332037 }, [
20342038 activeProject,
2035− activeServerOrderedTerminalIds,
2039+ activeKnownTerminalIds,
20362040 activeThreadId,
20372041 activeThreadRef,
20382042 activeThreadWorktreePath,
@@ -2053,7 +2057,7 @@ export default function ChatView(props: ChatViewProps) { · 14 unmodified lines
20532057 if (!api) {
20542058 return;
20552059 }
2056− const terminalId = nextTerminalId(activeServerOrderedTerminalIds);
2060+ const terminalId = nextTerminalId(activeKnownTerminalIds);
20572061 storeNewTerminal(activeThreadRef, terminalId);
20582062 setTerminalFocusRequestId((value) => value + 1);
20592063 void (async () => {
@@ -2074,7 +2078,7 @@ export default function ChatView(props: ChatViewProps) { · 14 unmodified lines
20742078 })();
20752079 }, [
20762080 activeProject,
2077− activeServerOrderedTerminalIds,
2081+ activeKnownTerminalIds,
20782082 activeThreadId,
20792083 activeThreadRef,
20802084 activeThreadWorktreePath,
@@ -2086,7 +2090,7 @@ export default function ChatView(props: ChatViewProps) { · 5 unmodified lines
20862090 (terminalId: string) => {
20872091 const api = readEnvironmentApi(environmentId);
20882092 if (!activeThreadId || !api || !activeThreadRef) return;
2089− const isFinalTerminal = activeServerOrderedTerminalIds.length <= 1;
2093+ const isFinalTerminal = activeKnownTerminalIds.length <= 1;
20902094 const fallbackExitWrite = () =>
20912095 api.terminal
20922096 .write({ threadId: activeThreadId, terminalId, data: "exit\n" })
@@ -2110,13 +2114,7 @@ export default function ChatView(props: ChatViewProps) { · 17 unmodified lines
21102114 storeCloseTerminal(activeThreadRef, terminalId);
21112115 setTerminalFocusRequestId((value) => value + 1);
21122116 },
2113− [
2114− activeThreadId,
2115− activeThreadRef,
2116− activeServerOrderedTerminalIds,
2117− environmentId,
2118− storeCloseTerminal,
2119− ],
2117+ [activeThreadId, activeThreadRef, activeKnownTerminalIds, environmentId, storeCloseTerminal],
21202118 );
21212119 const runProjectScript = useCallback(
21222120 async (
@@ -2139,9 +2137,7 @@ export default function ChatView(props: ChatViewProps) { · 16 unmodified lines
21392137 }
21402138 const targetCwd = options?.cwd ?? gitCwd ?? activeProject.cwd;
21412139 const baseTerminalId =
2142− terminalUiState.activeTerminalId ||
2143− activeServerOrderedTerminalIds[0] ||
2144− DEFAULT_THREAD_TERMINAL_ID;
2140+ terminalUiState.activeTerminalId || activeKnownTerminalIds[0] || DEFAULT_THREAD_TERMINAL_ID;
21452141 const isBaseTerminalBusy = runningTerminalIds.includes(baseTerminalId);
21462142 const wantsNewTerminal = Boolean(options?.preferNewTerminal) || isBaseTerminalBusy;
21472143 const shouldCreateNewTerminal = wantsNewTerminal;
@@ -2166,7 +2162,7 @@ export default function ChatView(props: ChatViewProps) { · 18 unmodified lines
21662162 ...(options?.env ? { extraEnv: options.env } : {}),
21672163 });
21682164 const targetTerminalId = shouldCreateNewTerminal
2169− ? nextTerminalId(activeServerOrderedTerminalIds)
2165+ ? nextTerminalId(activeKnownTerminalIds)
21702166 : baseTerminalId;
21712167 const openTerminalInput: TerminalOpenInput = shouldCreateNewTerminal
21722168 ? {
@@ -2218,7 +2214,7 @@ export default function ChatView(props: ChatViewProps) { · 45 unmodified lines
22182214 storeSetActiveTerminal,
22192215 setLastInvokedScriptByProjectId,
22202216 environmentId,
2221− activeServerOrderedTerminalIds,
2217+ activeKnownTerminalIds,
22222218 runningTerminalIds,
22232219 terminalUiState.activeTerminalId,
22242220 ],
packages/client-runtime/src/vcsRefState.test.ts+25−0
@@ -252,6 +252,31 @@ describe("createVcsRefManager", () => { · 251 unmodified lines
252252 expect(manager.getSnapshot(queriedTarget)).toEqual(EMPTY_VCS_REF_STATE);
253253 });
254254
255+ it("invalidates target in-flight loads before they can write stale data", async () => {
256+ const firstLoad = deferred<VcsListRefsResult>();
257+ let callCount = 0;
258+ const listRefs = vi.fn((async () => {
259+ callCount += 1;
260+ return callCount === 1 ? firstLoad.promise : SECOND_PAGE;
261+ }) satisfies VcsRefClient["listRefs"]);
262+ const client = { listRefs } satisfies VcsRefClient;
263+ const manager = createVcsRefManager({
264+ getRegistry: () => atomRegistry,
265+ getClient: () => client,
266+ });
267+
268+ const staleLoad = manager.load(TARGET, client);
269+ manager.invalidate(TARGET);
270+ const freshLoad = manager.load(TARGET, client);
271+
272+ expect(listRefs).toHaveBeenCalledTimes(2);
273+
274+ firstLoad.resolve(FIRST_PAGE);
275+ await expect(staleLoad).resolves.toEqual(FIRST_PAGE);
276+ await expect(freshLoad).resolves.toEqual(SECOND_PAGE);
277+ expect(manager.getSnapshot(TARGET).data).toEqual(SECOND_PAGE);
278+ });
279+
255280 it("watches refs with a ref-counted client-change subscription", async () => {
256281 const mock = createMockClient();
257282 let listener: () => void = noop;
packages/client-runtime/src/vcsRefState.ts+8−0
@@ -389,14 +389,22 @@ export function createVcsRefManager(config: VcsRefManagerConfig) { · 388 unmodified lines
389389 if (target) {
390390 const targetKey = getVcsRefTargetKey(target);
391391 if (targetKey !== null) {
392+ bumpLoadVersion(targetKey);
392393 setState(targetKey, EMPTY_VCS_REF_STATE);
394+ for (const key of inFlight.keys()) {
395+ if (key.startsWith(`${targetKey}:`)) {
396+ inFlight.delete(key);
397+ }
398+ }
393399 }
394400 return;
395401 }
396402
397403 for (const key of knownVcsRefKeys) {
404+ bumpLoadVersion(key);
398405 setState(key, EMPTY_VCS_REF_STATE);
399406 }
407+ inFlight.clear();
400408 }
401409
402410 function invalidateScope(scope: VcsRefScope): void {
packages/client-runtime/src/wsRpcClient.test.ts+33−0
@@ -3,6 +3,7 @@ import type { · 2 unmodified lines
33 VcsStatusRemoteResult,
44 VcsStatusStreamEvent,
55 } from "@t3tools/contracts";
6+import { ORCHESTRATION_WS_METHODS, ThreadId, WS_METHODS } from "@t3tools/contracts";
67 import { describe, expect, it, vi } from "vitest";
78
89 vi.mock("./wsTransport.ts", () => ({
@@ -127,4 +128,36 @@ describe("createWsRpcClient", () => { · 118 unmodified lines
127128 ],
128129 ]);
129130 });
131+
132+ it("tags stream subscriptions for targeted resubscribe handling", () => {
133+ const subscribe = vi.fn(() => () => undefined);
134+ const transport = {
135+ dispose: vi.fn(async () => undefined),
136+ reconnect: vi.fn(async () => undefined),
137+ request: vi.fn(),
138+ requestStream: vi.fn(),
139+ subscribe,
140+ } satisfies Pick<
141+ WsTransport,
142+ "dispose" | "reconnect" | "request" | "requestStream" | "subscribe"
143+ >;
144+
145+ const client = createWsRpcClient(transport as unknown as WsTransport);
146+ const listener = vi.fn();
147+
148+ client.terminal.onMetadata(listener);
149+ client.vcs.onStatus({ cwd: "/repo" }, listener);
150+ client.server.subscribeConfig(listener);
151+ client.orchestration.subscribeThread({ threadId: ThreadId.make("thread-1") }, listener);
152+
153+ const subscribeCalls = subscribe.mock.calls as unknown as Array<
154+ readonly [unknown, unknown, { readonly tag?: string }?]
155+ >;
156+ expect(subscribeCalls.map((call) => call[2]?.tag)).toEqual([
157+ WS_METHODS.subscribeTerminalMetadata,
158+ WS_METHODS.subscribeVcsStatus,
159+ WS_METHODS.subscribeServerConfig,
160+ ORCHESTRATION_WS_METHODS.subscribeThread,
161+ ]);
162+ });
130163 });
packages/client-runtime/src/wsRpcClient.ts+18−8
@@ -23,6 +23,16 @@ interface StreamSubscriptionOptions { · 22 unmodified lines
2323 readonly onResubscribe?: () => void;
2424 }
2525
26+function subscriptionOptions(
27+ options: StreamSubscriptionOptions | undefined,
28+ tag: string,
29+): StreamSubscriptionOptions & { readonly tag: string } {
30+ return {
31+ ...options,
32+ tag,
33+ };
34+}
35+
2636 type RpcUnaryMethod<TTag extends RpcTag> =
2737 RpcMethod<TTag> extends (input: any, options?: any) => Effect.Effect<infer TSuccess, any, any>
2838 ? (input: RpcInput<TTag>) => Promise<TSuccess>
@@ -145,7 +155,7 @@ export function createWsRpcClient( · 116 unmodified lines
145155 transport.subscribe(
146156 (client) => client[WS_METHODS.terminalAttach](input),
147157 listener,
148− options,
158+ subscriptionOptions(options, WS_METHODS.terminalAttach),
149159 ),
150160 write: (input) => transport.request((client) => client[WS_METHODS.terminalWrite](input)),
151161 resize: (input) => transport.request((client) => client[WS_METHODS.terminalResize](input)),
@@ -156,7 +166,7 @@ export function createWsRpcClient( · 4 unmodified lines
156166 transport.subscribe(
157167 (client) => client[WS_METHODS.subscribeTerminalMetadata]({}),
158168 listener,
159− options,
169+ subscriptionOptions(options, WS_METHODS.subscribeTerminalMetadata),
160170 ),
161171 },
162172 projects: {
@@ -184,7 +194,7 @@ export function createWsRpcClient( · 21 unmodified lines
184194 current = applyGitStatusStreamEvent(current, event);
185195 listener(current);
186196 },
187− options,
197+ subscriptionOptions(options, WS_METHODS.subscribeVcsStatus),
188198 );
189199 },
190200 runStackedAction: async (input, options) => {
@@ -238,19 +248,19 @@ export function createWsRpcClient( · 47 unmodified lines
238248 transport.subscribe(
239249 (client) => client[WS_METHODS.subscribeServerConfig]({}),
240250 listener,
241− options,
251+ subscriptionOptions(options, WS_METHODS.subscribeServerConfig),
242252 ),
243253 subscribeLifecycle: (listener, options) =>
244254 transport.subscribe(
245255 (client) => client[WS_METHODS.subscribeServerLifecycle]({}),
246256 listener,
247− options,
257+ subscriptionOptions(options, WS_METHODS.subscribeServerLifecycle),
248258 ),
249259 subscribeAuthAccess: (listener, options) =>
250260 transport.subscribe(
251261 (client) => client[WS_METHODS.subscribeAuthAccess]({}),
252262 listener,
253− options,
263+ subscriptionOptions(options, WS_METHODS.subscribeAuthAccess),
254264 ),
255265 },
256266 orchestration: {
@@ -264,13 +274,13 @@ export function createWsRpcClient( · 7 unmodified lines
264274 transport.subscribe(
265275 (client) => client[ORCHESTRATION_WS_METHODS.subscribeShell]({}),
266276 listener,
267− options,
277+ subscriptionOptions(options, ORCHESTRATION_WS_METHODS.subscribeShell),
268278 ),
269279 subscribeThread: (input, listener, options) =>
270280 transport.subscribe(
271281 (client) => client[ORCHESTRATION_WS_METHODS.subscribeThread](input),
272282 listener,
273− options,
283+ subscriptionOptions(options, ORCHESTRATION_WS_METHODS.subscribeThread),
274284 ),
275285 },
276286 };