authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-19 13:15:19-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-19 13:15:19-07:00
logcdaba7fa8b9c94dba783b7161e265a6db03b3d93
treed2ccc4c5b00cf9a20ff6dc2068da76028a601484
parentfbf6263876fd9b6f8bcbbf6b9ea097ed8fc7cb60
signature Signed by PGP key B5690EEEBB952194

Share thread state idle TTL across client atoms (#3163)

4 files changed, 49 insertions(+), 14 deletions(-)

packages/client-runtime/src/state/threadDetail.ts+11-11
...@@ -15,12 +15,12 @@ import type { EnvironmentThread, EnvironmentThreadShell } from "./models.ts";...@@ -15,12 +15,12 @@ import type { EnvironmentThread, EnvironmentThreadShell } from "./models.ts";
15import { scopeThread } from "./models.ts";15import { scopeThread } from "./models.ts";
16import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threads.ts";16import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threads.ts";
17import { parseThreadKey, threadKey } from "./entities.ts";17import { parseThreadKey, threadKey } from "./entities.ts";
18import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
1819
19const EMPTY_MESSAGES: ReadonlyArray<OrchestrationMessage> = Object.freeze([]);20const EMPTY_MESSAGES: ReadonlyArray<OrchestrationMessage> = Object.freeze([]);
20const EMPTY_ACTIVITIES: ReadonlyArray<OrchestrationThreadActivity> = Object.freeze([]);21const EMPTY_ACTIVITIES: ReadonlyArray<OrchestrationThreadActivity> = Object.freeze([]);
21const EMPTY_PROPOSED_PLANS: ReadonlyArray<OrchestrationProposedPlan> = Object.freeze([]);22const EMPTY_PROPOSED_PLANS: ReadonlyArray<OrchestrationProposedPlan> = Object.freeze([]);
22const EMPTY_CHECKPOINTS: ReadonlyArray<OrchestrationCheckpointSummary> = Object.freeze([]);23const EMPTY_CHECKPOINTS: ReadonlyArray<OrchestrationCheckpointSummary> = Object.freeze([]);
23const THREAD_DETAIL_IDLE_TTL_MS = 5 * 60_000;
2424
25/**25/**
26 * Combine detail-only collections with the shell's authoritative thread metadata.26 * Combine detail-only collections with the shell's authoritative thread metadata.
...@@ -75,7 +75,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -75,7 +75,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
75 () => EMPTY_ENVIRONMENT_THREAD_STATE,75 () => EMPTY_ENVIRONMENT_THREAD_STATE,
76 ),76 ),
77 ).pipe(77 ).pipe(
78 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),78 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
79 Atom.withLabel(`environment-thread-state-value:${key}`),79 Atom.withLabel(`environment-thread-state-value:${key}`),
80 );80 );
81 });81 });
...@@ -93,21 +93,21 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -93,21 +93,21 @@ export function createEnvironmentThreadDetailAtoms<E>(
93 previousValue = source === null ? null : scopeThread(ref.environmentId, source);93 previousValue = source === null ? null : scopeThread(ref.environmentId, source);
94 return previousValue;94 return previousValue;
95 }).pipe(95 }).pipe(
96 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),96 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
97 Atom.withLabel(`environment-thread-detail:${key}`),97 Atom.withLabel(`environment-thread-detail:${key}`),
98 );98 );
99 });99 });
100100
101 const threadStatusAtomFamily = Atom.family((key: string) =>101 const threadStatusAtomFamily = Atom.family((key: string) =>
102 Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(102 Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(
103 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),103 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
104 Atom.withLabel(`environment-thread-status:${key}`),104 Atom.withLabel(`environment-thread-status:${key}`),
105 ),105 ),
106 );106 );
107107
108 const threadErrorAtomFamily = Atom.family((key: string) =>108 const threadErrorAtomFamily = Atom.family((key: string) =>
109 Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(109 Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(
110 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),110 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
111 Atom.withLabel(`environment-thread-error:${key}`),111 Atom.withLabel(`environment-thread-error:${key}`),
112 ),112 ),
113 );113 );
...@@ -117,7 +117,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -117,7 +117,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
117 (get): ReadonlyArray<OrchestrationMessage> =>117 (get): ReadonlyArray<OrchestrationMessage> =>
118 get(threadDetailAtomFamily(key))?.messages ?? EMPTY_MESSAGES,118 get(threadDetailAtomFamily(key))?.messages ?? EMPTY_MESSAGES,
119 ).pipe(119 ).pipe(
120 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),120 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
121 Atom.withLabel(`environment-thread-messages:${key}`),121 Atom.withLabel(`environment-thread-messages:${key}`),
122 ),122 ),
123 );123 );
...@@ -127,7 +127,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -127,7 +127,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
127 (get): ReadonlyArray<OrchestrationThreadActivity> =>127 (get): ReadonlyArray<OrchestrationThreadActivity> =>
128 get(threadDetailAtomFamily(key))?.activities ?? EMPTY_ACTIVITIES,128 get(threadDetailAtomFamily(key))?.activities ?? EMPTY_ACTIVITIES,
129 ).pipe(129 ).pipe(
130 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),130 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
131 Atom.withLabel(`environment-thread-activities:${key}`),131 Atom.withLabel(`environment-thread-activities:${key}`),
132 ),132 ),
133 );133 );
...@@ -137,7 +137,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -137,7 +137,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
137 (get): ReadonlyArray<OrchestrationProposedPlan> =>137 (get): ReadonlyArray<OrchestrationProposedPlan> =>
138 get(threadDetailAtomFamily(key))?.proposedPlans ?? EMPTY_PROPOSED_PLANS,138 get(threadDetailAtomFamily(key))?.proposedPlans ?? EMPTY_PROPOSED_PLANS,
139 ).pipe(139 ).pipe(
140 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),140 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
141 Atom.withLabel(`environment-thread-proposed-plans:${key}`),141 Atom.withLabel(`environment-thread-proposed-plans:${key}`),
142 ),142 ),
143 );143 );
...@@ -147,7 +147,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -147,7 +147,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
147 (get): ReadonlyArray<OrchestrationCheckpointSummary> =>147 (get): ReadonlyArray<OrchestrationCheckpointSummary> =>
148 get(threadDetailAtomFamily(key))?.checkpoints ?? EMPTY_CHECKPOINTS,148 get(threadDetailAtomFamily(key))?.checkpoints ?? EMPTY_CHECKPOINTS,
149 ).pipe(149 ).pipe(
150 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),150 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
151 Atom.withLabel(`environment-thread-checkpoints:${key}`),151 Atom.withLabel(`environment-thread-checkpoints:${key}`),
152 ),152 ),
153 );153 );
...@@ -156,7 +156,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -156,7 +156,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
156 Atom.make(156 Atom.make(
157 (get): OrchestrationSession | null => get(threadDetailAtomFamily(key))?.session ?? null,157 (get): OrchestrationSession | null => get(threadDetailAtomFamily(key))?.session ?? null,
158 ).pipe(158 ).pipe(
159 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),159 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
160 Atom.withLabel(`environment-thread-session:${key}`),160 Atom.withLabel(`environment-thread-session:${key}`),
161 ),161 ),
162 );162 );
...@@ -165,7 +165,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -165,7 +165,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
165 Atom.make(165 Atom.make(
166 (get): OrchestrationLatestTurn | null => get(threadDetailAtomFamily(key))?.latestTurn ?? null,166 (get): OrchestrationLatestTurn | null => get(threadDetailAtomFamily(key))?.latestTurn ?? null,
167 ).pipe(167 ).pipe(
168 Atom.setIdleTTL(THREAD_DETAIL_IDLE_TTL_MS),168 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
169 Atom.withLabel(`environment-thread-latest-turn:${key}`),169 Atom.withLabel(`environment-thread-latest-turn:${key}`),
170 ),170 ),
171 );171 );
packages/client-runtime/src/state/threadRetention.ts created+3
...@@ -0,0 +1,3 @@
1// Mobile thread routes unmount during back navigation. Retain the stream-backed
2// state across short subscriber gaps without keeping every opened thread alive.
3export const THREAD_STATE_IDLE_TTL_MS = 5 * 60_000;
packages/client-runtime/src/state/threads-atoms.test.ts created+26
...@@ -0,0 +1,26 @@
1import { EnvironmentId, ThreadId } from "@t3tools/contracts";
2import { describe, expect, it } from "@effect/vitest";
3import * as Layer from "effect/Layer";
4import { Atom } from "effect/unstable/reactivity";
5
6import type { EnvironmentRegistry } from "../connection/registry.ts";
7import type { EnvironmentCacheStore } from "../platform/persistence.ts";
8import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
9import { createEnvironmentThreadStateAtoms } from "./threads.ts";
10
11describe("createEnvironmentThreadStateAtoms", () => {
12 it("retains thread state across short subscriber gaps", () => {
13 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
14 EnvironmentRegistry | EnvironmentCacheStore,
15 never
16 >;
17 const threads = createEnvironmentThreadStateAtoms(runtime);
18 const environmentId = EnvironmentId.make("environment-1");
19 const threadId = ThreadId.make("thread-1");
20 const atom = threads.stateAtom(environmentId, threadId);
21
22 expect(atom.idleTTL).toBe(THREAD_STATE_IDLE_TTL_MS);
23 expect(threads.stateAtom(environmentId, threadId)).toBe(atom);
24 expect(threads.stateAtom(environmentId, ThreadId.make("thread-2"))).not.toBe(atom);
25 });
26});
packages/client-runtime/src/state/threads.ts+9-3
...@@ -21,6 +21,7 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts";...@@ -21,6 +21,7 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts";
21import { EnvironmentCacheStore } from "../platform/persistence.ts";21import { EnvironmentCacheStore } from "../platform/persistence.ts";
22import { subscribe } from "../rpc/client.ts";22import { subscribe } from "../rpc/client.ts";
23import { applyThreadDetailEvent } from "./threadReducer.ts";23import { applyThreadDetailEvent } from "./threadReducer.ts";
24import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
24import { followStreamInEnvironment } from "./runtime.ts";25import { followStreamInEnvironment } from "./runtime.ts";
2526
26export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted";27export type EnvironmentThreadStatus = "empty" | "cached" | "synchronizing" | "live" | "deleted";
...@@ -249,9 +250,14 @@ export function createEnvironmentThreadStateAtoms<R, E>(...@@ -249,9 +250,14 @@ export function createEnvironmentThreadStateAtoms<R, E>(
249) {250) {
250 const family = Atom.family((key: string) => {251 const family = Atom.family((key: string) => {
251 const { environmentId, threadId } = parseThreadAtomKey(key);252 const { environmentId, threadId } = parseThreadAtomKey(key);
252 return runtime.atom(threadStateChanges(environmentId, threadId), {253 return runtime
253 initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,254 .atom(threadStateChanges(environmentId, threadId), {
254 });255 initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
256 })
257 .pipe(
258 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
259 Atom.withLabel(`environment-thread-state:${key}`),
260 );
255 });261 });
256262
257 return {263 return {