authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-24 21:57:35-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-24 21:57:35-07:00
log99641fd09a509feb644b3c8ef718c12c60cf436c
tree1a878e04d0c688342fae9e34c6631932376f9ff7
parent0c84b4289cc9dee9af770a823f812ea1e8bea05d
signature Signed by PGP key B5690EEEBB952194

feat(desktop): keep running threads synced in the background (#13554)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

6 files changed, 320 insertions(+), 3 deletions(-)

apps/web/src/components/desktop/RunningThreadKeepAlive.tsx created+14
......@@ -0,0 +1,14 @@
1import { useAtomMount } from "@effect/atom-react";
2
3import { runningThreadKeepAliveAtom } from "../../state/threads";
4
5/**
6 * Desktop only. Keeps each running thread subscribed in every enabled
7 * environment, so opening one shows live state without a replay or a sync
8 * status flash. It mounts the atom without reading it, so thread updates
9 * never re-render.
10 */
11export function RunningThreadKeepAlive() {
12 useAtomMount(runningThreadKeepAliveAtom);
13 return null;
14}
apps/web/src/routes/__root.tsx+3
......@@ -26,6 +26,7 @@ import { RelayClientInstallDialog } from "../components/cloud/RelayClientInstall
2626import { SshPasswordPromptDialog } from "../components/desktop/SshPasswordPromptDialog";
2727import { SnapShotCoordinator } from "../components/desktop/SnapShotCoordinator";
2828import { DesktopAppActivationCoordinator } from "../components/desktop/DesktopAppActivationCoordinator";
29import { RunningThreadKeepAlive } from "../components/desktop/RunningThreadKeepAlive";
2930import { ProviderUpdateLaunchNotification } from "../components/ProviderUpdateLaunchNotification";
3031import { ThreadNotificationCoordinator } from "../components/ThreadNotificationCoordinator";
3132import { ProjectCloneToastCoordinator } from "../components/ProjectCloneToastCoordinator";
......@@ -43,6 +44,7 @@ import {
4344 toastManager,
4445} from "../components/ui/toast";
4546import { resolveAndPersistPreferredEditor } from "../editorPreferences";
47import { isElectron } from "../env";
4648import { applyAppearanceFontVariables } from "~/appearanceFonts";
4749import { applyAppearanceContrast } from "~/appearanceContrast";
4850import { useClientSettings } from "../hooks/useSettings";
......@@ -219,6 +221,7 @@ function RootRouteView() {
219221 >
220222 {primaryEnvironmentAuthenticated ? <AuthenticatedTracingBootstrap /> : null}
221223 {primaryEnvironmentAuthenticated ? <DesktopAppActivationCoordinator /> : null}
224 {isElectron ? <RunningThreadKeepAlive /> : null}
222225 <RelayClientInstallDialog />
223226 <ConnectOnboardingDialog />
224227 <SshPasswordPromptDialog />
apps/web/src/state/threads.test.ts created+200
......@@ -0,0 +1,200 @@
1import {
2 EMPTY_ENVIRONMENT_THREAD_STATE,
3 type EnvironmentThreadState,
4} from "@t3tools/client-runtime/state/threads";
5import {
6 EnvironmentId,
7 ProjectId,
8 ProviderInstanceId,
9 ThreadId,
10 type OrchestrationSessionStatus,
11 type OrchestrationThread,
12 type OrchestrationThreadShell,
13} from "@t3tools/contracts";
14import * as Option from "effect/Option";
15import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
16import { describe, expect, it } from "vite-plus/test";
17
18import { createRunningThreadKeepAliveAtom } from "./threads";
19
20const LOCAL = EnvironmentId.make("local");
21const REMOTE = EnvironmentId.make("remote");
22
23function session(threadId: ThreadId, status: OrchestrationSessionStatus) {
24 return {
25 threadId,
26 status,
27 providerName: "codex",
28 runtimeMode: "full-access",
29 activeTurnId: null,
30 lastError: null,
31 updatedAt: "2026-09-24T00:00:00.000Z",
32 } satisfies OrchestrationThread["session"];
33}
34
35function shell(id: string, status: OrchestrationSessionStatus | null) {
36 const threadId = ThreadId.make(id);
37 return {
38 id: threadId,
39 session: status === null ? null : session(threadId, status),
40 } satisfies Pick<OrchestrationThreadShell, "id" | "session">;
41}
42
43function detail(
44 id: string,
45 status: OrchestrationSessionStatus,
46 overrides: Partial<EnvironmentThreadState> = {},
47) {
48 const threadId = ThreadId.make(id);
49 const thread: OrchestrationThread = {
50 id: threadId,
51 projectId: ProjectId.make("project"),
52 title: id,
53 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
54 runtimeMode: "full-access",
55 interactionMode: "default",
56 branch: null,
57 worktreePath: null,
58 latestTurn: null,
59 createdAt: "2026-09-24T00:00:00.000Z",
60 updatedAt: "2026-09-24T00:00:00.000Z",
61 archivedAt: null,
62 settledOverride: null,
63 settledAt: null,
64 pullRequests: [],
65 deletedAt: null,
66 messages: [],
67 proposedPlans: [],
68 activities: [],
69 checkpoints: [],
70 session: session(threadId, status),
71 };
72 return AsyncResult.success<EnvironmentThreadState>({
73 ...EMPTY_ENVIRONMENT_THREAD_STATE,
74 status: "live",
75 data: Option.some(thread),
76 ...overrides,
77 });
78}
79
80function makeHarness() {
81 // Registry cleanup runs only on `flush`, like the real deferred task.
82 const tasks: Array<() => void> = [];
83 const registry = AtomRegistry.make({
84 scheduleTask: (task) => {
85 tasks.push(task);
86 return () => {};
87 },
88 });
89 const flush = () => {
90 for (let task = tasks.shift(); task !== undefined; task = tasks.shift()) task();
91 };
92 const environmentIds = Atom.make<ReadonlyArray<EnvironmentId>>([LOCAL, REMOTE]).pipe(
93 Atom.keepAlive,
94 );
95 const threads = Atom.family((_environmentId: EnvironmentId) =>
96 Atom.make<ReadonlyArray<ReturnType<typeof shell>>>([]).pipe(Atom.keepAlive),
97 );
98 // Stand-ins for the thread state atoms. Each one lives only while mounted,
99 // as the real stream does.
100 const keys = new Set<string>();
101 const states = Atom.family((_key: string) =>
102 Atom.make<AsyncResult.AsyncResult<EnvironmentThreadState>>(
103 AsyncResult.success(EMPTY_ENVIRONMENT_THREAD_STATE),
104 ),
105 );
106 const stateAtom = (environmentId: EnvironmentId, threadId: string) => {
107 const key = `${environmentId}:${threadId}`;
108 keys.add(key);
109 return states(key);
110 };
111 const keepAlive = createRunningThreadKeepAliveAtom({
112 environmentIdsAtom: environmentIds,
113 threadsAtom: threads,
114 stateAtom,
115 });
116 registry.mount(keepAlive);
117 return {
118 registry,
119 environmentIds,
120 threads,
121 stateAtom,
122 keepAlive,
123 openStreams: () => {
124 flush();
125 return [...keys].filter((key) => registry.getNodes().has(states(key))).toSorted();
126 },
127 };
128}
129
130describe("createRunningThreadKeepAliveAtom", () => {
131 it("keeps running threads open across shell updates and thread view visits", () => {
132 const h = makeHarness();
133 h.registry.set(h.threads(LOCAL), [
134 shell("a", "running"),
135 shell("b", "ready"),
136 shell("c", null),
137 ]);
138 h.registry.set(h.threads(REMOTE), [shell("d", "starting")]);
139 expect(h.openStreams()).toEqual(["local:a", "remote:d"]);
140
141 // A thread view that comes and goes shares the kept stream.
142 const live = detail("a", "running");
143 h.registry.set(h.stateAtom(LOCAL, "a"), live);
144 h.registry.mount(h.stateAtom(LOCAL, "a"))();
145
146 // A shell update that starts or stops nothing does not rebuild the set.
147 const kept = h.registry.get(h.keepAlive);
148 h.registry.set(h.threads(LOCAL), [shell("a", "running"), shell("b", "ready")]);
149 expect(h.registry.get(h.keepAlive)).toBe(kept);
150 expect(h.openStreams()).toEqual(["local:a", "remote:d"]);
151 expect(h.registry.get(h.stateAtom(LOCAL, "a"))).toBe(live);
152 });
153
154 it("holds a stopped thread until its own stream is live and shows the stop", () => {
155 const h = makeHarness();
156 h.registry.set(h.threads(LOCAL), [
157 shell("a", "running"),
158 shell("b", "running"),
159 shell("c", "running"),
160 ]);
161 // "b" has not loaded yet. "c" hit a stream error.
162 h.registry.set(h.stateAtom(LOCAL, "a"), detail("a", "running"));
163 h.registry.set(
164 h.stateAtom(LOCAL, "c"),
165 detail("c", "running", { status: "cached", error: Option.some("Could not sync.") }),
166 );
167
168 // The shell reports the stops first. A failed stream cannot deliver its
169 // stop, so only it is released now.
170 h.registry.set(h.threads(LOCAL), [
171 shell("a", "ready"),
172 shell("b", "ready"),
173 shell("c", "ready"),
174 ]);
175 expect(h.openStreams()).toEqual(["local:a", "local:b"]);
176
177 h.registry.set(h.stateAtom(LOCAL, "a"), detail("a", "ready"));
178 h.registry.set(h.stateAtom(LOCAL, "b"), detail("b", "ready", { status: "synchronizing" }));
179 expect(h.openStreams()).toEqual(["local:b"]);
180 h.registry.set(h.stateAtom(LOCAL, "b"), detail("b", "ready"));
181 expect(h.openStreams()).toEqual([]);
182 });
183
184 it("follows environments that connect and go away", () => {
185 const h = makeHarness();
186 h.registry.set(h.environmentIds, [LOCAL]);
187 h.registry.set(h.threads(REMOTE), [shell("d", "running")]);
188 expect(h.openStreams()).toEqual([]);
189
190 h.registry.set(h.environmentIds, [LOCAL, REMOTE]);
191 expect(h.openStreams()).toEqual(["remote:d"]);
192
193 // Removal drops every mount, including one still waiting for its stop.
194 h.registry.set(h.stateAtom(REMOTE, "d"), detail("d", "running"));
195 h.registry.set(h.threads(REMOTE), [shell("d", "ready")]);
196 expect(h.openStreams()).toEqual(["remote:d"]);
197 h.registry.set(h.environmentIds, [LOCAL]);
198 expect(h.openStreams()).toEqual([]);
199 });
200});
apps/web/src/state/threads.ts+85-1
......@@ -1,4 +1,6 @@
11import { useAtomValue } from "@effect/atom-react";
2import { enabledEnvironmentIds } from "@t3tools/client-runtime/state/connections";
3import { arrayElementsEqual } from "@t3tools/client-runtime/state/entities";
24import {
35 createEnvironmentThreadDetailAtoms,
46 createEnvironmentThreadShellAtoms,
......@@ -6,8 +8,9 @@ import {
68 EMPTY_ENVIRONMENT_THREAD_STATE,
79 type EnvironmentThreadState,
810 createThreadEnvironmentAtoms,
11 isThreadSessionRunning,
912} from "@t3tools/client-runtime/state/threads";
10import type { EnvironmentId, ThreadId } from "@t3tools/contracts";
13import type { EnvironmentId, OrchestrationThreadShell, ThreadId } from "@t3tools/contracts";
1114import * as Option from "effect/Option";
1215import { AsyncResult, Atom } from "effect/unstable/reactivity";
1316
......@@ -46,3 +49,84 @@ export function useEnvironmentThread(
4649 () => EMPTY_ENVIRONMENT_THREAD_STATE,
4750 ) as EnvironmentThreadState;
4851}
52
53type KeptThreads = ReadonlyMap<EnvironmentId, ReadonlySet<ThreadId>>;
54
55// True once a thread's own stream no longer needs to stay open: it is in sync
56// and shows a settled session, or it cannot progress (deleted or failed). A
57// stream that is still loading or reconnecting keeps waiting for the stop.
58function isDetailDone<E>(result: AsyncResult.AsyncResult<EnvironmentThreadState, E>): boolean {
59 if (!AsyncResult.isSuccess(result)) return true;
60 const { status, data, error } = result.value;
61 if (status === "deleted" || Option.isSome(error)) return true;
62 return (
63 status === "live" && !Option.exists(data, (thread) => isThreadSessionRunning(thread.session))
64 );
65}
66
67/**
68 * Keeps the thread state atom mounted for each running thread in the listed
69 * environments. Mount the result; its value is only bookkeeping.
70 *
71 * The shell and detail streams are independent, so the shell can report a
72 * stop before the detail loads or catches up. A stopped thread stays mounted
73 * until its own detail is live and shows the stop too. Then the stream closes
74 * and saves the settled state to disk.
75 */
76export function createRunningThreadKeepAliveAtom<E>(input: {
77 readonly environmentIdsAtom: Atom.Atom<ReadonlyArray<EnvironmentId>>;
78 readonly threadsAtom: (
79 environmentId: EnvironmentId,
80 ) => Atom.Atom<ReadonlyArray<Pick<OrchestrationThreadShell, "id" | "session">>>;
81 readonly stateAtom: (
82 environmentId: EnvironmentId,
83 threadId: ThreadId,
84 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentThreadState, E>>;
85}) {
86 // Keeps its identity until a thread starts or stops, so ordinary shell
87 // updates do not rebuild the keep-alive set.
88 const runningThreadIdsAtom = Atom.family((environmentId: EnvironmentId) => {
89 let previous: ReadonlyArray<ThreadId> = [];
90 return Atom.make((get) => {
91 const running = get(input.threadsAtom(environmentId)).flatMap((thread) =>
92 isThreadSessionRunning(thread.session) ? [thread.id] : [],
93 );
94 if (arrayElementsEqual(previous, running)) return previous;
95 previous = running;
96 return running;
97 }).pipe(Atom.withLabel(`web-running-thread-ids:${environmentId}`));
98 });
99
100 return Atom.make((get): KeptThreads => {
101 const previous = Option.getOrUndefined(get.self<KeptThreads>());
102 const kept = new Map<EnvironmentId, ReadonlySet<ThreadId>>();
103 // An environment that leaves the list is not visited, so its mounts drop.
104 for (const environmentId of get(input.environmentIdsAtom)) {
105 const threadIds = new Set(get(runningThreadIdsAtom(environmentId)));
106 for (const threadId of previous?.get(environmentId) ?? []) {
107 if (threadIds.has(threadId)) continue;
108 const stateAtom = input.stateAtom(environmentId, threadId);
109 // `once`, not `get`: a dependency on a stopped thread would hold its
110 // stream open until some other change rebuilds this atom.
111 if (isDetailDone(get.once(stateAtom))) continue;
112 threadIds.add(threadId);
113 // Rebuild when this detail is done, not on each update.
114 get.subscribe(stateAtom, (state) => {
115 if (isDetailDone(state)) get.refreshSelf();
116 });
117 }
118 for (const threadId of threadIds) get.mount(input.stateAtom(environmentId, threadId));
119 kept.set(environmentId, threadIds);
120 }
121 return kept;
122 }).pipe(Atom.withLabel("web-running-thread-keep-alive"));
123}
124
125/** Mounted by `RunningThreadKeepAlive` on desktop, for every enabled environment. */
126export const runningThreadKeepAliveAtom = createRunningThreadKeepAliveAtom({
127 environmentIdsAtom: Atom.map(environmentCatalog.catalogValueAtom, (catalog) => [
128 ...enabledEnvironmentIds(catalog),
129 ]),
130 threadsAtom: environmentThreadShells.environmentThreadsAtom,
131 stateAtom: environmentThreads.stateAtom,
132});
docs/internals/connection-runtime.md+9
......@@ -61,6 +61,15 @@ stream, which stops when the last consumer unmounts; hidden mounted routes still
6161count. A registry-local cache retains state and its replay cursor for five idle
6262minutes so back navigation can resume without another snapshot download.
6363
64The desktop app adds one consumer: a
65[keep-alive](../../apps/web/src/state/threads.ts) mounts every thread whose
66session is starting or running, in each enabled environment. Opening a running
67thread then needs no replay. The shell and detail streams are independent, so
68the shell can report a stop before the detail loads or catches up. A stopped
69thread stays mounted until its own stream is live and shows the stop, and the
70stream then closes and saves the settled state.
71Web and mobile do not keep threads alive.
72
6473Retain state and cursor together only after an update finishes. Cancellation must
6574not advance the cached cursor beyond the applied data, and an old scope must not
6675overwrite its successor's cache. Preserve pagination data on reuse, but clear
packages/client-runtime/src/state/threads.ts+9-2
......@@ -126,9 +126,16 @@ function formatThreadError(cause: Cause.Cause<unknown>): string {
126126 : "Could not synchronize the thread.";
127127}
128128
129/**
130 * A starting or running session is mid-turn. Its detail can change many times
131 * per second, so the disk cache waits for it to settle.
132 */
133export function isThreadSessionRunning(session: OrchestrationThread["session"]): boolean {
134 return session?.status === "starting" || session?.status === "running";
135}
136
129137function shouldPersistThread(thread: OrchestrationThread): boolean {
130 const status = thread.session?.status;
131 return status !== "starting" && status !== "running";
138 return !isThreadSessionRunning(thread.session);
132139}
133140
134141interface ThreadResumeSnapshot {