authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-04 11:56:05-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-04 11:56:05-07:00
logd7cf8aaa8d4fbcbdd523b4f4bc86fda5c47b4a70
tree808a3ffb20e7291d4dad60746d2c976b828acadf
parent0de956ed2f92b8a6ead56e1566801db232d365dc
signature Signed by PGP key B5690EEEBB952194

perf(client): stop thread streams when unused (#9740)

Close a thread's live detail stream when its last consumer leaves. Retain a registry-local completed state and replay cursor for five idle minutes. Keep warm lookup independent of collectible raw atom definitions. Preserve paging and deletion state, reject stale owners, and cache only completed data/cursor updates. Skip repeated saves of unchanged data while retaining dirty-data flushes and failed-write retries. Verify real GC/remount behavior and focused lifecycle cases. Created with GPT-6 Astra (preview) in Codex.

6 files changed, 911 insertions(+), 129 deletions(-)

docs/internals/connection-runtime.md+15
...@@ -170,6 +170,20 @@ Application code must not construct RPC clients, retry loops, or raw...@@ -170,6 +170,20 @@ Application code must not construct RPC clients, retry loops, or raw
170orchestration commands. Persistence paths belong to the platform registration170orchestration commands. Persistence paths belong to the platform registration
171and cache stores, with explicit migration or invalidation policy.171and cache stores, with explicit migration or invalidation policy.
172172
173### Thread detail ownership
174
175Mounted detail consumers share one live thread subscription. The subscription
176ends when the last consumer leaves. Hidden routes that remain mounted still
177count as consumers.
178
179A separate registry-local atom retains the last thread state and replay cursor
180for five idle minutes. A warm mount renders that state and resumes the stream
181without another snapshot download. The cache holds completed state and cursor
182updates together. It retains pagination data but clears canceled loading state.
183An ownership token prevents an old scope from replacing its successor's cache.
184Closing a thread skips snapshots that were already saved. Changed snapshots and
185failed background writes still get a final save attempt.
186
173## Verification187## Verification
174188
175Core state-machine tests use `@effect/vitest` and deterministic service layers.189Core state-machine tests use `@effect/vitest` and deterministic service layers.
...@@ -184,6 +198,7 @@ Required coverage includes:...@@ -184,6 +198,7 @@ Required coverage includes:
184- relay token reuse and refresh;198- relay token reuse and refresh;
185- progressive relay discovery;199- progressive relay discovery;
186- shell and thread cache hydration;200- shell and thread cache hydration;
201- thread stream shutdown, warm resume, and snapshot expiry;
187- durable subscriptions switching sessions;202- durable subscriptions switching sessions;
188- command metadata and idempotent queued-command metadata.203- command metadata and idempotent queued-command metadata.
189204
packages/client-runtime/src/state/threadDetail.ts+10-35
...@@ -15,7 +15,6 @@ import type { EnvironmentThread, EnvironmentThreadShell } from "./models.ts";...@@ -15,7 +15,6 @@ 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 "./threadState.ts";16import { EMPTY_ENVIRONMENT_THREAD_STATE, type EnvironmentThreadState } from "./threadState.ts";
17import { parseThreadKey, threadKey } from "./entities.ts";17import { parseThreadKey, threadKey } from "./entities.ts";
18import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
1918
20const EMPTY_MESSAGES: ReadonlyArray<OrchestrationMessage> = Object.freeze([]);19const EMPTY_MESSAGES: ReadonlyArray<OrchestrationMessage> = Object.freeze([]);
21const EMPTY_ACTIVITIES: ReadonlyArray<OrchestrationThreadActivity> = Object.freeze([]);20const EMPTY_ACTIVITIES: ReadonlyArray<OrchestrationThreadActivity> = Object.freeze([]);
...@@ -80,10 +79,7 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -80,10 +79,7 @@ export function createEnvironmentThreadDetailAtoms<E>(
80 AsyncResult.value(get(threadStateAtom(ref.environmentId, ref.threadId))),79 AsyncResult.value(get(threadStateAtom(ref.environmentId, ref.threadId))),
81 () => EMPTY_ENVIRONMENT_THREAD_STATE,80 () => EMPTY_ENVIRONMENT_THREAD_STATE,
82 ),81 ),
83 ).pipe(82 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state-value:${key}`));
84 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
85 Atom.withLabel(`environment-thread-state-value:${key}`),
86 );
87 });83 });
8884
89 const threadDetailAtomFamily = Atom.family((key: string) => {85 const threadDetailAtomFamily = Atom.family((key: string) => {
...@@ -98,22 +94,19 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -98,22 +94,19 @@ export function createEnvironmentThreadDetailAtoms<E>(
98 previousSource = source;94 previousSource = source;
99 previousValue = source === null ? null : scopeThread(ref.environmentId, source);95 previousValue = source === null ? null : scopeThread(ref.environmentId, source);
100 return previousValue;96 return previousValue;
101 }).pipe(97 }).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-detail:${key}`));
102 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
103 Atom.withLabel(`environment-thread-detail:${key}`),
104 );
105 });98 });
10699
107 const threadStatusAtomFamily = Atom.family((key: string) =>100 const threadStatusAtomFamily = Atom.family((key: string) =>
108 Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(101 Atom.make((get) => get(threadStateValueAtomFamily(key)).status).pipe(
109 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),102 Atom.setIdleTTL(0),
110 Atom.withLabel(`environment-thread-status:${key}`),103 Atom.withLabel(`environment-thread-status:${key}`),
111 ),104 ),
112 );105 );
113106
114 const threadErrorAtomFamily = Atom.family((key: string) =>107 const threadErrorAtomFamily = Atom.family((key: string) =>
115 Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(108 Atom.make((get) => Option.getOrNull(get(threadStateValueAtomFamily(key)).error)).pipe(
116 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),109 Atom.setIdleTTL(0),
117 Atom.withLabel(`environment-thread-error:${key}`),110 Atom.withLabel(`environment-thread-error:${key}`),
118 ),111 ),
119 );112 );
...@@ -122,58 +115,40 @@ export function createEnvironmentThreadDetailAtoms<E>(...@@ -122,58 +115,40 @@ export function createEnvironmentThreadDetailAtoms<E>(
122 Atom.make(115 Atom.make(
123 (get): ReadonlyArray<OrchestrationMessage> =>116 (get): ReadonlyArray<OrchestrationMessage> =>
124 get(threadDetailAtomFamily(key))?.messages ?? EMPTY_MESSAGES,117 get(threadDetailAtomFamily(key))?.messages ?? EMPTY_MESSAGES,
125 ).pipe(118 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-messages:${key}`)),
126 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
127 Atom.withLabel(`environment-thread-messages:${key}`),
128 ),
129 );119 );
130120
131 const threadActivitiesAtomFamily = Atom.family((key: string) =>121 const threadActivitiesAtomFamily = Atom.family((key: string) =>
132 Atom.make(122 Atom.make(
133 (get): ReadonlyArray<OrchestrationThreadActivity> =>123 (get): ReadonlyArray<OrchestrationThreadActivity> =>
134 get(threadDetailAtomFamily(key))?.activities ?? EMPTY_ACTIVITIES,124 get(threadDetailAtomFamily(key))?.activities ?? EMPTY_ACTIVITIES,
135 ).pipe(125 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-activities:${key}`)),
136 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
137 Atom.withLabel(`environment-thread-activities:${key}`),
138 ),
139 );126 );
140127
141 const threadProposedPlansAtomFamily = Atom.family((key: string) =>128 const threadProposedPlansAtomFamily = Atom.family((key: string) =>
142 Atom.make(129 Atom.make(
143 (get): ReadonlyArray<OrchestrationProposedPlan> =>130 (get): ReadonlyArray<OrchestrationProposedPlan> =>
144 get(threadDetailAtomFamily(key))?.proposedPlans ?? EMPTY_PROPOSED_PLANS,131 get(threadDetailAtomFamily(key))?.proposedPlans ?? EMPTY_PROPOSED_PLANS,
145 ).pipe(132 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-proposed-plans:${key}`)),
146 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
147 Atom.withLabel(`environment-thread-proposed-plans:${key}`),
148 ),
149 );133 );
150134
151 const threadCheckpointsAtomFamily = Atom.family((key: string) =>135 const threadCheckpointsAtomFamily = Atom.family((key: string) =>
152 Atom.make(136 Atom.make(
153 (get): ReadonlyArray<OrchestrationCheckpointSummary> =>137 (get): ReadonlyArray<OrchestrationCheckpointSummary> =>
154 get(threadDetailAtomFamily(key))?.checkpoints ?? EMPTY_CHECKPOINTS,138 get(threadDetailAtomFamily(key))?.checkpoints ?? EMPTY_CHECKPOINTS,
155 ).pipe(139 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-checkpoints:${key}`)),
156 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
157 Atom.withLabel(`environment-thread-checkpoints:${key}`),
158 ),
159 );140 );
160141
161 const threadSessionAtomFamily = Atom.family((key: string) =>142 const threadSessionAtomFamily = Atom.family((key: string) =>
162 Atom.make(143 Atom.make(
163 (get): OrchestrationSession | null => get(threadDetailAtomFamily(key))?.session ?? null,144 (get): OrchestrationSession | null => get(threadDetailAtomFamily(key))?.session ?? null,
164 ).pipe(145 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-session:${key}`)),
165 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
166 Atom.withLabel(`environment-thread-session:${key}`),
167 ),
168 );146 );
169147
170 const threadLatestTurnAtomFamily = Atom.family((key: string) =>148 const threadLatestTurnAtomFamily = Atom.family((key: string) =>
171 Atom.make(149 Atom.make(
172 (get): OrchestrationLatestTurn | null => get(threadDetailAtomFamily(key))?.latestTurn ?? null,150 (get): OrchestrationLatestTurn | null => get(threadDetailAtomFamily(key))?.latestTurn ?? null,
173 ).pipe(151 ).pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-latest-turn:${key}`)),
174 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),
175 Atom.withLabel(`environment-thread-latest-turn:${key}`),
176 ),
177 );152 );
178153
179 return {154 return {
packages/client-runtime/src/state/threadRetention.ts+3-3
...@@ -1,3 +1,3 @@...@@ -1,3 +1,3 @@
1// Mobile thread routes unmount during back navigation. Retain the stream-backed1// Keep recent thread snapshots for back navigation. Live subscriptions end
2// state across short subscriber gaps without keeping every opened thread alive.2// when the last detail consumer leaves.
3export const THREAD_STATE_IDLE_TTL_MS = 5 * 60_000;3export const THREAD_SNAPSHOT_IDLE_TTL_MS = 5 * 60_000;
packages/client-runtime/src/state/threads-atoms.test.ts+410-20
...@@ -1,26 +1,416 @@...@@ -1,26 +1,416 @@
1import { EnvironmentId, ThreadId } from "@t3tools/contracts";1import {
2import { describe, expect, it } from "@effect/vitest";2 EnvironmentId,
3 EventId,
4 MessageId,
5 ORCHESTRATION_WS_METHODS,
6 ProjectId,
7 ProviderInstanceId,
8 ThreadId,
9 type OrchestrationThread,
10 type OrchestrationThreadDetailSnapshot,
11 type OrchestrationThreadStreamItem,
12} from "@t3tools/contracts";
13import { afterEach, describe, expect, it, vi } from "@effect/vitest";
14import * as Deferred from "effect/Deferred";
15import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";16import * as Layer from "effect/Layer";
4import { Atom } from "effect/unstable/reactivity";17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Stream from "effect/Stream";
20import * as SubscriptionRef from "effect/SubscriptionRef";
21import { Atom, AtomRegistry } from "effect/unstable/reactivity";
522
6import type { EnvironmentRegistry } from "../connection/registry.ts";23import type { ConnectionCatalogEntry } from "../connection/catalog.ts";
7import type { EnvironmentCacheStore } from "../platform/persistence.ts";24import { EnvironmentRegistry } from "../connection/registry.ts";
8import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";25import {
9import { createEnvironmentThreadStateAtoms, type ThreadSnapshotLoader } from "./threads.ts";26 AVAILABLE_CONNECTION_STATE,
27 PrimaryConnectionTarget,
28 type NetworkStatus,
29 type PreparedConnection,
30} from "../connection/model.ts";
31import { EnvironmentSupervisor } from "../connection/supervisor.ts";
32import { EnvironmentCacheStore } from "../platform/persistence.ts";
33import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
34import type { RpcSession } from "../rpc/session.ts";
35import { createEnvironmentThreadDetailAtoms } from "./threadDetail.ts";
36import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";
37import type { ThreadSnapshotWindow } from "./threadSnapshotHttp.ts";
38import {
39 createEnvironmentThreadStateAtoms,
40 requestOlderThreadTurns,
41 ThreadSnapshotLoader,
42 type EnvironmentThreadState,
43} from "./threads.ts";
44
45const TARGET = new PrimaryConnectionTarget({
46 environmentId: EnvironmentId.make("environment-1"),
47 label: "Test environment",
48 httpBaseUrl: "https://environment.example.test",
49 wsBaseUrl: "wss://environment.example.test",
50});
51const THREAD_ID = ThreadId.make("thread-1");
52const THREAD: OrchestrationThread = {
53 id: THREAD_ID,
54 projectId: ProjectId.make("project-1"),
55 title: "Cached thread",
56 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
57 runtimeMode: "full-access",
58 interactionMode: "default",
59 branch: "main",
60 worktreePath: null,
61 latestTurn: null,
62 createdAt: "2026-04-01T00:00:00.000Z",
63 updatedAt: "2026-04-01T00:00:00.000Z",
64 archivedAt: null,
65 settledOverride: null,
66 settledAt: null,
67 deletedAt: null,
68 messages: [],
69 proposedPlans: [],
70 activities: [],
71 checkpoints: [],
72 session: null,
73};
74const SNAPSHOT: OrchestrationThreadDetailSnapshot = { snapshotSequence: 7, thread: THREAD };
75
76const makeHarness = Effect.fn("TestThreadAtoms.makeHarness")(function* (options?: {
77 readonly snapshot?: OrchestrationThreadDetailSnapshot;
78}) {
79 const subscriptions = yield* Queue.unbounded<{
80 readonly afterSequence: number | undefined;
81 readonly events: Queue.Queue<OrchestrationThreadStreamItem>;
82 readonly closed: Deferred.Deferred<void>;
83 }>();
84 const olderLoads = yield* Queue.unbounded<{
85 readonly window: ThreadSnapshotWindow;
86 readonly response: Deferred.Deferred<Option.Option<OrchestrationThreadDetailSnapshot>>;
87 readonly closed: Deferred.Deferred<void>;
88 }>();
89 const snapshot = options?.snapshot ?? SNAPSHOT;
90 let httpLoads = 0;
91 let diskLoads = 0;
92 let opened = 0;
93 let active = 0;
94 const client = {
95 [ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) =>
96 Stream.unwrap(
97 Effect.gen(function* () {
98 const events = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
99 const closed = yield* Deferred.make<void>();
100 yield* Effect.acquireRelease(
101 Effect.sync(() => {
102 opened += 1;
103 active += 1;
104 }),
105 () =>
106 Effect.sync(() => {
107 active -= 1;
108 }).pipe(Effect.andThen(Deferred.succeed(closed, undefined))),
109 );
110 yield* Queue.offer(subscriptions, { afterSequence: input.afterSequence, events, closed });
111 return Stream.fromQueue(events);
112 }),
113 ),
114 } as unknown as WsRpcProtocolClient;
115 const session: RpcSession = {
116 client,
117 initialConfig: Effect.succeed({
118 threadResumeCompletionMarker: true,
119 threadSnapshotPagination: true,
120 } as never),
121 subscribeServerConfig: (input) => client.subscribeServerConfig(input),
122 ready: Effect.void,
123 probe: Effect.void,
124 closed: Effect.never,
125 };
126 const supervisor = EnvironmentSupervisor.of({
127 target: TARGET,
128 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
129 session: yield* SubscriptionRef.make(Option.some(session)),
130 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
131 Option.some({
132 environmentId: TARGET.environmentId,
133 label: TARGET.label,
134 httpBaseUrl: TARGET.httpBaseUrl,
135 socketUrl: TARGET.wsBaseUrl,
136 httpAuthorization: null,
137 target: TARGET,
138 }),
139 ),
140 connect: Effect.void,
141 disconnect: Effect.void,
142 retryNow: Effect.void,
143 });
144 const environmentRegistry = EnvironmentRegistry.of({
145 entries: yield* SubscriptionRef.make<ReadonlyMap<EnvironmentId, ConnectionCatalogEntry>>(
146 new Map(),
147 ),
148 networkStatus: yield* SubscriptionRef.make<NetworkStatus>("online"),
149 start: Effect.void,
150 register: () => Effect.die("Unexpected environment registration"),
151 registerPlatform: () => Effect.die("Unexpected environment registration"),
152 reconcilePlatform: () => Effect.die("Unexpected environment reconciliation"),
153 remove: () => Effect.die("Unexpected environment removal"),
154 removeRelayEnvironments: () => Effect.die("Unexpected environment removal"),
155 retryNow: () => Effect.void,
156 state: () => SubscriptionRef.get(supervisor.state),
157 stateChanges: () => SubscriptionRef.changes(supervisor.state),
158 run: (_environmentId, effect) =>
159 Effect.provideService(effect, EnvironmentSupervisor, supervisor),
160 runStream: (_environmentId, stream) =>
161 Stream.provideService(stream, EnvironmentSupervisor, supervisor),
162 followStream: (_environmentId, stream) =>
163 Stream.provideService(stream, EnvironmentSupervisor, supervisor),
164 });
165 const runtime = Atom.runtime(
166 Layer.mergeAll(
167 Layer.succeed(EnvironmentRegistry, environmentRegistry),
168 Layer.succeed(
169 EnvironmentCacheStore,
170 EnvironmentCacheStore.of({
171 loadShell: () => Effect.succeed(Option.none()),
172 saveShell: () => Effect.void,
173 loadThread: () =>
174 Effect.sync(() => {
175 diskLoads += 1;
176 return Option.none();
177 }),
178 saveThread: () => Effect.void,
179 removeThread: () => Effect.void,
180 loadServerConfig: () => Effect.succeed(Option.none()),
181 saveServerConfig: () => Effect.void,
182 loadVcsRefs: () => Effect.succeed(Option.none()),
183 saveVcsRefs: () => Effect.void,
184 removeVcsRefs: () => Effect.void,
185 clearVcsRefs: () => Effect.void,
186 clear: () => Effect.void,
187 }),
188 ),
189 Layer.succeed(
190 ThreadSnapshotLoader,
191 ThreadSnapshotLoader.of({
192 load: (_prepared, _threadId, window) => {
193 if (window?.beforeCursor === undefined) {
194 return Effect.sync(() => {
195 httpLoads += 1;
196 return Option.some(snapshot);
197 });
198 }
199 return Effect.gen(function* () {
200 const response =
201 yield* Deferred.make<Option.Option<OrchestrationThreadDetailSnapshot>>();
202 const closed = yield* Deferred.make<void>();
203 yield* Effect.addFinalizer(() => Deferred.succeed(closed, undefined));
204 yield* Queue.offer(olderLoads, { window, response, closed });
205 return yield* Deferred.await(response);
206 }).pipe(Effect.scoped);
207 },
208 }),
209 ),
210 ),
211 );
212 const raw = createEnvironmentThreadStateAtoms(runtime);
213 const details = createEnvironmentThreadDetailAtoms(raw.stateAtom);
214 const ref = { environmentId: TARGET.environmentId, threadId: THREAD_ID };
215 const stateAtom = details.stateAtom(ref);
216 const makeRegistry = Effect.acquireRelease(
217 Effect.sync(() => AtomRegistry.make({ defaultIdleTTL: 60_000, timeoutResolution: 1 })),
218 (registry) => Effect.sync(() => registry.dispose()),
219 );
220 const registry = yield* makeRegistry;
221
222 return {
223 registry,
224 makeRegistry,
225 rawAtoms: raw,
226 stateAtom,
227 details,
228 ref,
229 subscriptions,
230 olderLoads,
231 counts: () => ({ httpLoads, diskLoads, opened, active }),
232 };
233});
234
235function observeState(
236 registry: AtomRegistry.AtomRegistry,
237 atom: Atom.Atom<EnvironmentThreadState>,
238 predicate: (state: EnvironmentThreadState) => boolean,
239) {
240 return AtomRegistry.toStream(registry, atom).pipe(
241 Stream.filter(predicate),
242 Stream.runHead,
243 Effect.map(Option.getOrThrow),
244 );
245}
246
247function currentThread(
248 registry: AtomRegistry.AtomRegistry,
249 atom: Atom.Atom<EnvironmentThreadState>,
250) {
251 return Option.getOrThrow(registry.get(atom).data);
252}
10253
11describe("createEnvironmentThreadStateAtoms", () => {254describe("createEnvironmentThreadStateAtoms", () => {
12 it("retains thread state across short subscriber gaps", () => {255 afterEach(() => {
13 const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<256 vi.useRealTimers();
14 EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader,257 vi.restoreAllMocks();
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 });258 });
259
260 it.effect("shares one live stream and closes it after the last detail consumer leaves", () =>
261 Effect.gen(function* () {
262 const h = yield* makeHarness();
263 const unmountMessages = h.registry.mount(h.details.messagesAtom(h.ref));
264 const first = yield* Queue.take(h.subscriptions);
265 const unmountStatus = h.registry.mount(h.details.statusAtom(h.ref));
266 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 1, active: 1 });
267 unmountMessages();
268 yield* Queue.offer(first.events, { kind: "synchronized" });
269 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
270 expect(h.counts().active).toBe(1);
271 unmountStatus();
272 yield* Deferred.await(first.closed);
273 expect(h.counts().active).toBe(0);
274 }),
275 );
276
277 it.effect("keeps warm data and resumes a completed cursor without loading another snapshot", () =>
278 Effect.gen(function* () {
279 const h = yield* makeHarness();
280 const unmount = h.registry.mount(h.stateAtom);
281 const first = yield* Queue.take(h.subscriptions);
282 yield* Queue.offer(first.events, {
283 kind: "event",
284 event: {
285 type: "thread.message-sent",
286 sequence: 8,
287 eventId: EventId.make("message-1"),
288 aggregateKind: "thread",
289 aggregateId: THREAD_ID,
290 occurredAt: THREAD.createdAt,
291 commandId: null,
292 causationEventId: null,
293 correlationId: null,
294 metadata: {},
295 payload: {
296 threadId: THREAD_ID,
297 messageId: MessageId.make("message-1"),
298 role: "assistant",
299 text: "Retained text",
300 turnId: null,
301 streaming: true,
302 createdAt: THREAD.createdAt,
303 updatedAt: THREAD.createdAt,
304 },
305 },
306 });
307 yield* Queue.offer(first.events, { kind: "synchronized" });
308 yield* observeState(h.registry, h.stateAtom, (state) => state.status === "live");
309 const before = currentThread(h.registry, h.stateAtom);
310 unmount();
311 yield* Deferred.await(first.closed);
312 const remount = h.registry.mount(h.stateAtom);
313 expect(currentThread(h.registry, h.stateAtom)).toBe(before);
314 const next = yield* Queue.take(h.subscriptions);
315 expect(next.afterSequence).toBe(8);
316 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 });
317 remount();
318 yield* Deferred.await(next.closed);
319 }),
320 );
321
322 it.effect("keeps warm data when the raw atom family's weak entry is collected", () =>
323 Effect.gen(function* () {
324 const h = yield* makeHarness();
325 const oldRaw = h.rawAtoms.stateAtom(TARGET.environmentId, THREAD_ID);
326 const unmount = h.registry.mount(h.stateAtom);
327 const first = yield* Queue.take(h.subscriptions);
328 const latest = { ...THREAD, title: "Newer cached thread" };
329 yield* Queue.offer(first.events, {
330 kind: "snapshot",
331 snapshot: { snapshotSequence: 8, thread: latest },
332 });
333 yield* Queue.offer(first.events, { kind: "synchronized" });
334 yield* observeState(h.registry, h.stateAtom, (value) => value.status === "live");
335 unmount();
336 yield* Deferred.await(first.closed);
337
338 // Force the weak-family miss without depending on host GC timing.
339 const deref = WeakRef.prototype.deref;
340 vi.spyOn(WeakRef.prototype, "deref").mockImplementation(function (this: WeakRef<object>) {
341 const value = deref.call(this);
342 return value === oldRaw ? undefined : value;
343 });
344 const remount = h.registry.mount(h.stateAtom);
345 expect(currentThread(h.registry, h.stateAtom)).toBe(latest);
346 const next = yield* Queue.take(h.subscriptions);
347 expect(next.afterSequence).toBe(8);
348 expect(h.counts()).toEqual({ httpLoads: 1, diskLoads: 1, opened: 2, active: 1 });
349 remount();
350 yield* Deferred.await(next.closed);
351 }),
352 );
353
354 it.effect("keeps cached snapshots local to each registry", () =>
355 Effect.gen(function* () {
356 const h = yield* makeHarness();
357 const unmount = h.registry.mount(h.stateAtom);
358 const first = yield* Queue.take(h.subscriptions);
359 unmount();
360 yield* Deferred.await(first.closed);
361 const otherRegistry = yield* h.makeRegistry;
362 const unmountOther = otherRegistry.mount(h.stateAtom);
363 const other = yield* Queue.take(h.subscriptions);
364 expect(h.counts()).toEqual({ httpLoads: 2, diskLoads: 2, opened: 2, active: 1 });
365 unmountOther();
366 yield* Deferred.await(other.closed);
367 }),
368 );
369
370 it.effect("expires the plain snapshot after five idle minutes", () =>
371 Effect.gen(function* () {
372 vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] });
373 const h = yield* makeHarness();
374 const unmount = h.registry.mount(h.stateAtom);
375 const first = yield* Queue.take(h.subscriptions);
376 unmount();
377 yield* Deferred.await(first.closed);
378 yield* Effect.yieldNow;
379 yield* Effect.promise(() => vi.advanceTimersByTimeAsync(THREAD_SNAPSHOT_IDLE_TTL_MS + 1));
380 const remount = h.registry.mount(h.stateAtom);
381 const next = yield* Queue.take(h.subscriptions);
382 expect(h.counts()).toEqual({ httpLoads: 2, diskLoads: 2, opened: 2, active: 1 });
383 remount();
384 yield* Deferred.await(next.closed);
385 }),
386 );
387
388 it.effect("cancels older-page work on unmount and permits it again on a warm return", () =>
389 Effect.gen(function* () {
390 const h = yield* makeHarness({
391 snapshot: {
392 ...SNAPSHOT,
393 page: { beforeCursor: "older-1", hasMore: true, snapshotSequence: 7 },
394 },
395 });
396 const unmount = h.registry.mount(h.stateAtom);
397 const first = yield* Queue.take(h.subscriptions);
398 expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true);
399 const older = yield* Queue.take(h.olderLoads);
400 expect(Option.getOrThrow(h.registry.get(h.stateAtom).page).loadingOlder).toBe(true);
401 unmount();
402 yield* Deferred.await(first.closed);
403 yield* Deferred.await(older.closed);
404 const remount = h.registry.mount(h.stateAtom);
405 const next = yield* Queue.take(h.subscriptions);
406 expect(Option.getOrThrow(h.registry.get(h.stateAtom).page).loadingOlder).toBe(false);
407 expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true);
408 const retried = yield* Queue.take(h.olderLoads);
409 expect(retried.window.beforeCursor).toBe("older-1");
410 expect(h.counts().httpLoads).toBe(1);
411 remount();
412 yield* Deferred.await(next.closed);
413 yield* Deferred.await(retried.closed);
414 }),
415 );
26});416});
packages/client-runtime/src/state/threads-sync.test.ts+279-3
...@@ -11,10 +11,14 @@ import {...@@ -11,10 +11,14 @@ import {
11 type OrchestrationThreadStreamItem,11 type OrchestrationThreadStreamItem,
12} from "@t3tools/contracts";12} from "@t3tools/contracts";
13import { describe, expect, it } from "@effect/vitest";13import { describe, expect, it } from "@effect/vitest";
14import * as Deferred from "effect/Deferred";
14import * as Effect from "effect/Effect";15import * as Effect from "effect/Effect";
16import * as Exit from "effect/Exit";
17import * as Fiber from "effect/Fiber";
15import * as Option from "effect/Option";18import * as Option from "effect/Option";
16import * as Queue from "effect/Queue";19import * as Queue from "effect/Queue";
17import * as Ref from "effect/Ref";20import * as Ref from "effect/Ref";
21import * as Scope from "effect/Scope";
18import * as Stream from "effect/Stream";22import * as Stream from "effect/Stream";
19import * as SubscriptionRef from "effect/SubscriptionRef";23import * as SubscriptionRef from "effect/SubscriptionRef";
20import * as TestClock from "effect/testing/TestClock";24import * as TestClock from "effect/testing/TestClock";
...@@ -134,6 +138,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -134,6 +138,9 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
134 readonly cached?: OrchestrationThread;138 readonly cached?: OrchestrationThread;
135 readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;139 readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;
136 readonly completionMarker?: boolean;140 readonly completionMarker?: boolean;
141 readonly resumeCache?: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]>;
142 readonly loadCached?: Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>>;
143 readonly saveThread?: Persistence.EnvironmentCacheStore["Service"]["saveThread"];
137}) {144}) {
138 const inputs = yield* Queue.unbounded<TestThreadInput>();145 const inputs = yield* Queue.unbounded<TestThreadInput>();
139 const observed = yield* Queue.unbounded<EnvironmentThreadState>();146 const observed = yield* Queue.unbounded<EnvironmentThreadState>();
...@@ -202,6 +209,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -202,6 +209,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
202 loadShell: () => Effect.succeed(Option.none()),209 loadShell: () => Effect.succeed(Option.none()),
203 saveShell: () => Effect.void,210 saveShell: () => Effect.void,
204 loadThread: (_environmentId, threadId) =>211 loadThread: (_environmentId, threadId) =>
212 options?.loadCached ??
205 Effect.succeed(213 Effect.succeed(
206 threadId === THREAD_ID && options?.cached !== undefined214 threadId === THREAD_ID && options?.cached !== undefined
207 ? Option.some({215 ? Option.some({
...@@ -210,8 +218,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -210,8 +218,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
210 })218 })
211 : Option.none(),219 : Option.none(),
212 ),220 ),
213 saveThread: (_environmentId, thread) =>221 saveThread: (environmentId, thread) =>
214 Ref.update(savedThreads, (current) => [...current, thread]),222 Ref.update(savedThreads, (current) => [...current, thread]).pipe(
223 Effect.andThen(options?.saveThread?.(environmentId, thread) ?? Effect.void),
224 ),
215 removeThread: (_environmentId, threadId) =>225 removeThread: (_environmentId, threadId) =>
216 Ref.update(removedThreads, (current) => [...current, threadId]),226 Ref.update(removedThreads, (current) => [...current, threadId]),
217 loadServerConfig: () => Effect.succeed(Option.none()),227 loadServerConfig: () => Effect.succeed(Option.none()),
...@@ -222,7 +232,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -222,7 +232,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
222 clearVcsRefs: () => Effect.void,232 clearVcsRefs: () => Effect.void,
223 clear: () => Effect.void,233 clear: () => Effect.void,
224 });234 });
225 const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe(235 const threadState = yield* makeEnvironmentThreadState(THREAD_ID, options?.resumeCache).pipe(
226 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),236 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
227 Effect.provideService(Persistence.EnvironmentCacheStore, cache),237 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
228 Effect.provideService(ThreadSnapshotLoader, snapshotLoader),238 Effect.provideService(ThreadSnapshotLoader, snapshotLoader),
...@@ -239,6 +249,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o...@@ -239,6 +249,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
239 );249 );
240250
241 return {251 return {
252 threadState,
242 inputs,253 inputs,
243 observed,254 observed,
244 latest,255 latest,
...@@ -316,6 +327,271 @@ const deleted = (): OrchestrationThreadStreamItem => ({...@@ -316,6 +327,271 @@ const deleted = (): OrchestrationThreadStreamItem => ({
316});327});
317328
318describe("EnvironmentThreads", () => {329describe("EnvironmentThreads", () => {
330 for (const source of ["disk", "HTTP"] as const) {
331 it.effect(`does not rewrite an unchanged ${source} snapshot on navigation or warm return`, () =>
332 Effect.gen(function* () {
333 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
334 snapshot: undefined,
335 owner: undefined,
336 };
337 const firstSaved = yield* Effect.scoped(
338 Effect.gen(function* () {
339 const h = yield* makeHarness({
340 resumeCache,
341 ...(source === "disk"
342 ? { cached: BASE_THREAD }
343 : { httpSnapshot: Option.some({ snapshotSequence: 7, thread: BASE_THREAD }) }),
344 });
345 yield* awaitThreadState(h.observed, (value) => value.status === "live");
346 if (source === "HTTP") yield* TestClock.adjust("500 millis");
347 return h.savedThreads;
348 }),
349 );
350 expect(yield* Ref.get(firstSaved)).toHaveLength(source === "disk" ? 0 : 1);
351 const nextSaved = yield* Effect.scoped(
352 Effect.gen(function* () {
353 const h = yield* makeHarness({ resumeCache });
354 yield* awaitThreadState(h.observed, (value) => value.status === "live");
355 return h.savedThreads;
356 }),
357 );
358 expect(yield* Ref.get(nextSaved)).toEqual([]);
359 }),
360 );
361 }
362
363 it.effect("retries a failed background cache write when the scope closes", () =>
364 Effect.gen(function* () {
365 let attempts = 0;
366 yield* Effect.scoped(
367 Effect.gen(function* () {
368 const h = yield* makeHarness({
369 httpSnapshot: Option.some({ snapshotSequence: 7, thread: BASE_THREAD }),
370 saveThread: () =>
371 Effect.suspend(() => {
372 attempts += 1;
373 return attempts === 1
374 ? Effect.fail(
375 new Persistence.ConnectionPersistenceError({
376 operation: "save-thread",
377 message: "Test storage failure",
378 }),
379 )
380 : Effect.void;
381 }),
382 });
383 yield* awaitThreadState(h.observed, (value) => value.status === "live");
384 yield* TestClock.adjust("500 millis");
385 expect(attempts).toBe(1);
386 }),
387 );
388 expect(attempts).toBe(2);
389 }),
390 );
391
392 it.effect("flushes newer data after an older background write completes", () =>
393 Effect.gen(function* () {
394 const writing = yield* Deferred.make<void>();
395 const written = yield* Deferred.make<void>();
396 const release = yield* Deferred.make<void>();
397 const saved = yield* Effect.scoped(
398 Effect.gen(function* () {
399 const h = yield* makeHarness({
400 cached: BASE_THREAD,
401 saveThread: (_environmentId, snapshot) =>
402 snapshot.snapshotSequence === 8
403 ? Deferred.succeed(writing, undefined).pipe(
404 Effect.andThen(Deferred.await(release)),
405 Effect.andThen(Deferred.succeed(written, undefined)),
406 )
407 : Effect.void,
408 });
409 yield* Queue.offer(h.inputs, titleUpdated("First update", 8));
410 yield* awaitThreadState(
411 h.observed,
412 (value) => Option.getOrNull(value.data)?.title === "First update",
413 );
414 yield* TestClock.adjust("500 millis");
415 yield* Deferred.await(writing);
416 yield* Queue.offer(h.inputs, titleUpdated("Newer update", 9));
417 yield* awaitThreadState(
418 h.observed,
419 (value) => Option.getOrNull(value.data)?.title === "Newer update",
420 );
421 yield* Deferred.succeed(release, undefined);
422 yield* Deferred.await(written);
423 return h.savedThreads;
424 }),
425 );
426 expect(
427 (yield* Ref.get(saved)).map((snapshot) => [
428 snapshot.snapshotSequence,
429 snapshot.thread.title,
430 ]),
431 ).toEqual([
432 [8, "First update"],
433 [9, "Newer update"],
434 ]);
435 }),
436 );
437
438 it.effect("does not resume past a canceled event whose data was not applied", () =>
439 Effect.gen(function* () {
440 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
441 snapshot: undefined,
442 owner: undefined,
443 };
444 const scope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
445 Scope.close(scope, Exit.void),
446 );
447 const first = yield* makeHarness({
448 httpSnapshot: Option.some({
449 thread: BASE_THREAD,
450 snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,
451 }),
452 resumeCache,
453 }).pipe(Effect.provideService(Scope.Scope, scope));
454 yield* awaitThreadState(first.observed, (value) => value.status === "live");
455 const applying = yield* Deferred.make<void>();
456 const update = titleUpdated("Not applied", 8);
457 if (update.kind !== "event") return yield* Effect.die("Expected an event");
458 Object.defineProperty(update.event.payload, "title", {
459 get: () => {
460 Deferred.doneUnsafe(applying, Exit.void);
461 return "Not applied";
462 },
463 });
464 // The reducer reads the title after it advances the cursor. Hold only
465 // the data write so closing the scope interrupts that exact interval.
466 yield* first.threadState.semaphore.take(1);
467 yield* Queue.offer(first.inputs, update);
468 yield* Deferred.await(applying);
469 yield* Scope.close(scope, Exit.void);
470 yield* first.threadState.semaphore.release(1);
471
472 const resumed = yield* makeHarness({ resumeCache });
473 const state = yield* awaitThreadState(resumed.observed, (value) => value.status === "live");
474 expect(Option.getOrThrow(state.data).title).toBe(BASE_THREAD.title);
475 expect(yield* Ref.get(resumed.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
476 expect(yield* Ref.get(first.savedThreads)).toEqual([
477 {
478 thread: BASE_THREAD,
479 snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,
480 },
481 ]);
482 }),
483 );
484
485 it.effect("prevents an old scope from replacing its successor's cached data", () =>
486 Effect.gen(function* () {
487 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
488 snapshot: undefined,
489 owner: undefined,
490 };
491 const oldScope = yield* Effect.acquireRelease(Scope.make(), (scope) =>
492 Scope.close(scope, Exit.void),
493 );
494 const old = yield* makeHarness({ cached: BASE_THREAD, resumeCache }).pipe(
495 Effect.provideService(Scope.Scope, oldScope),
496 );
497 yield* awaitThreadState(old.observed, (value) => value.status === "live");
498 const successor = yield* makeHarness({ resumeCache });
499 yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9));
500 yield* awaitThreadState(
501 successor.observed,
502 (value) => Option.getOrNull(value.data)?.title === "Successor title",
503 );
504 yield* Queue.offer(old.inputs, titleUpdated("Old scope title", 8));
505 yield* awaitThreadState(
506 old.observed,
507 (value) => Option.getOrNull(value.data)?.title === "Old scope title",
508 );
509 yield* Scope.close(oldScope, Exit.void);
510
511 expect(resumeCache.snapshot?.sequence).toBe(9);
512 expect(Option.getOrThrow(resumeCache.snapshot!.state.data).title).toBe("Successor title");
513 expect(yield* Ref.get(old.savedThreads)).toEqual([]);
514 }),
515 );
516
517 it.effect("does not let a delayed cache read replace an initialized successor", () =>
518 Effect.gen(function* () {
519 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
520 snapshot: undefined,
521 owner: undefined,
522 };
523 const loading = yield* Deferred.make<void>();
524 const response = yield* Deferred.make<Option.Option<OrchestrationThreadDetailSnapshot>>();
525 const oldFiber = yield* makeHarness({
526 resumeCache,
527 loadCached: Deferred.succeed(loading, undefined).pipe(
528 Effect.andThen(Deferred.await(response)),
529 ),
530 }).pipe(Effect.forkScoped);
531 yield* Deferred.await(loading);
532 const successor = yield* makeHarness({ cached: BASE_THREAD, resumeCache });
533 yield* awaitThreadState(successor.observed, (value) => value.status === "live");
534 yield* Deferred.succeed(
535 response,
536 Option.some({ snapshotSequence: 3, thread: { ...BASE_THREAD, title: "Old disk data" } }),
537 );
538 const old = yield* Fiber.join(oldFiber);
539 yield* awaitThreadState(old.observed, (value) => value.status === "live");
540
541 expect(resumeCache.snapshot?.sequence).toBe(CACHED_SNAPSHOT_SEQUENCE);
542 expect(Option.getOrThrow(resumeCache.snapshot!.state.data).title).toBe(BASE_THREAD.title);
543 }),
544 );
545
546 it.effect("does not let an old deletion remove its successor's persisted cache", () =>
547 Effect.gen(function* () {
548 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
549 snapshot: undefined,
550 owner: undefined,
551 };
552 const old = yield* makeHarness({ cached: BASE_THREAD, resumeCache });
553 yield* awaitThreadState(old.observed, (value) => value.status === "live");
554 const successor = yield* makeHarness({ resumeCache });
555 yield* Queue.offer(successor.inputs, titleUpdated("Successor title", 9));
556 yield* awaitThreadState(
557 successor.observed,
558 (value) => Option.getOrNull(value.data)?.title === "Successor title",
559 );
560 const update = deleted();
561 if (update.kind !== "event") return yield* Effect.die("Expected an event");
562 yield* Queue.offer(old.inputs, { ...update, event: { ...update.event, sequence: 8 } });
563 yield* awaitThreadState(old.observed, (value) => value.status === "deleted");
564
565 expect(resumeCache.snapshot?.sequence).toBe(9);
566 expect(yield* Ref.get(old.removedThreads)).toEqual([]);
567 }),
568 );
569
570 it.effect("retains a deletion instead of restoring the old disk snapshot", () =>
571 Effect.gen(function* () {
572 const resumeCache: NonNullable<Parameters<typeof makeEnvironmentThreadState>[1]> = {
573 snapshot: undefined,
574 owner: undefined,
575 };
576 yield* Effect.scoped(
577 Effect.gen(function* () {
578 const first = yield* makeHarness({ cached: BASE_THREAD, resumeCache });
579 const update = deleted();
580 if (update.kind !== "event") return yield* Effect.die("Expected an event");
581 yield* Queue.offer(first.inputs, { ...update, event: { ...update.event, sequence: 8 } });
582 yield* awaitThreadState(first.observed, (value) => value.status === "deleted");
583 }),
584 );
585 const resumed = yield* makeHarness({ cached: BASE_THREAD, resumeCache });
586 const state = yield* awaitThreadState(
587 resumed.observed,
588 (value) => value.status === "deleted",
589 );
590 expect(Option.isNone(state.data)).toBe(true);
591 expect(yield* Ref.get(resumed.loaderCalls)).toBe(0);
592 }),
593 );
594
319 it.effect("publishes cached data immediately from a warm cache", () =>595 it.effect("publishes cached data immediately from a warm cache", () =>
320 Effect.gen(function* () {596 Effect.gen(function* () {
321 const harness = yield* makeHarness({ cached: BASE_THREAD });597 const harness = yield* makeHarness({ cached: BASE_THREAD });
packages/client-runtime/src/state/threads.ts+194-68
...@@ -27,7 +27,7 @@ import { subscribeDynamic } from "../rpc/client.ts";...@@ -27,7 +27,7 @@ import { subscribeDynamic } from "../rpc/client.ts";
27import { ThreadSnapshotLoader, type ThreadSnapshotWindow } from "./threadSnapshotHttp.ts";27import { ThreadSnapshotLoader, type ThreadSnapshotWindow } from "./threadSnapshotHttp.ts";
28import { parseThreadKey, threadKey } from "./entities.ts";28import { parseThreadKey, threadKey } from "./entities.ts";
29import { applyThreadDetailEvent } from "./threadReducer.ts";29import { applyThreadDetailEvent } from "./threadReducer.ts";
30import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";30import { THREAD_SNAPSHOT_IDLE_TTL_MS } from "./threadRetention.ts";
31import { followStreamInEnvironment } from "./runtime.ts";31import { followStreamInEnvironment } from "./runtime.ts";
32import {32import {
33 EMPTY_ENVIRONMENT_THREAD_STATE,33 EMPTY_ENVIRONMENT_THREAD_STATE,
...@@ -131,40 +131,93 @@ function shouldPersistThread(thread: OrchestrationThread): boolean {...@@ -131,40 +131,93 @@ function shouldPersistThread(thread: OrchestrationThread): boolean {
131 return status !== "starting" && status !== "running";131 return status !== "starting" && status !== "running";
132}132}
133133
134interface ThreadResumeSnapshot {
135 readonly state: EnvironmentThreadState;
136 readonly sequence: number;
137 readonly persisted: boolean;
138}
139
140interface ThreadResumeCache {
141 snapshot: ThreadResumeSnapshot | undefined;
142 owner: object | undefined;
143}
144
145function matchesThreadSnapshot(
146 current: ThreadResumeSnapshot,
147 thread: OrchestrationThread | null,
148 sequence: number,
149 page: Pick<EnvironmentThreadPageState, "beforeCursor" | "hasMore"> | undefined,
150): boolean {
151 if (current.sequence !== sequence || Option.getOrNull(current.state.data) !== thread)
152 return false;
153 const currentPage = Option.getOrUndefined(current.state.page);
154 return currentPage === undefined
155 ? page === undefined
156 : page !== undefined &&
157 currentPage.beforeCursor === page.beforeCursor &&
158 currentPage.hasMore === page.hasMore;
159}
160
161function cachedThreadState(value: EnvironmentThreadState): EnvironmentThreadState {
162 return {
163 ...value,
164 status: value.status === "deleted" ? "deleted" : statusWithoutLiveData(value.data),
165 error: Option.none(),
166 page: Option.map(value.page, (page) => ({ ...page, loadingOlder: false })),
167 };
168}
169
134export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (170export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make")(function* (
135 threadId: ThreadIdType,171 threadId: ThreadIdType,
172 resumeCache?: ThreadResumeCache,
136) {173) {
137 const supervisor = yield* EnvironmentSupervisor;174 const supervisor = yield* EnvironmentSupervisor;
138 const cache = yield* EnvironmentCacheStore;175 const cache = yield* EnvironmentCacheStore;
139 const snapshotLoader = yield* ThreadSnapshotLoader;176 const snapshotLoader = yield* ThreadSnapshotLoader;
140 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);177 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
141 const environmentId = supervisor.target.environmentId;178 const environmentId = supervisor.target.environmentId;
142 const cached = yield* cache.loadThread(environmentId, threadId).pipe(179 const retained = resumeCache?.snapshot;
143 Effect.catch((error) =>180 const owner = {};
144 Effect.logWarning("Could not load cached thread.").pipe(181 if (resumeCache) resumeCache.owner = owner;
145 Effect.annotateLogs({182 const cached =
146 environmentId,183 retained === undefined
147 threadId,184 ? yield* cache.loadThread(environmentId, threadId).pipe(
148 error: error.message,185 Effect.catch((error) =>
149 }),186 Effect.logWarning("Could not load cached thread.").pipe(
150 Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),187 Effect.annotateLogs({
151 ),188 environmentId,
152 ),189 threadId,
153 );190 error: error.message,
191 }),
192 Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
193 ),
194 ),
195 )
196 : Option.none<OrchestrationThreadDetailSnapshot>();
154 const cachedThread = Option.map(cached, (snapshot) => snapshot.thread);197 const cachedThread = Option.map(cached, (snapshot) => snapshot.thread);
155 const state = yield* SubscriptionRef.make<EnvironmentThreadState>({198 const initialState: EnvironmentThreadState = retained
156 data: cachedThread,199 ? cachedThreadState(retained.state)
157 status: statusWithoutLiveData(cachedThread),200 : {
158 error: Option.none(),201 data: cachedThread,
159 // A cached windowed snapshot restores its page cursor so "load earlier"202 status: statusWithoutLiveData(cachedThread),
160 // works while rendering from cache; a cached full snapshot has no page.203 error: Option.none(),
161 page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)),204 // A cached windowed snapshot restores its page cursor so "load earlier"
162 });205 // works while rendering from cache; a cached full snapshot has no page.
206 page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)),
207 };
208 const state = yield* SubscriptionRef.make(initialState);
163 // Seed the resume cursor from the cached snapshot so a warm cache can catch up209 // Seed the resume cursor from the cached snapshot so a warm cache can catch up
164 // via `afterSequence` instead of re-downloading the full thread body.210 // via `afterSequence` instead of re-downloading the full thread body.
165 const lastSequence = yield* SubscriptionRef.make(211 const initialSequence =
166 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }),212 retained?.sequence ??
167 );213 Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence });
214 const lastSequence = yield* SubscriptionRef.make(initialSequence);
215 let committed: ThreadResumeSnapshot = {
216 state: initialState,
217 sequence: initialSequence,
218 persisted: retained?.persisted ?? Option.isSome(cached),
219 };
220 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
168 const awaitingCompletion = yield* Ref.make(false);221 const awaitingCompletion = yield* Ref.make(false);
169 // Bumped whenever loaded history may have been rewritten out from under an222 // Bumped whenever loaded history may have been rewritten out from under an
170 // in-flight older-page fetch (snapshot replacement, revert, deletion). A223 // in-flight older-page fetch (snapshot replacement, revert, deletion). A
...@@ -174,6 +227,25 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -174,6 +227,25 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
174 // merges. Without it, a revert or snapshot processed between loadOlderTurns'227 // merges. Without it, a revert or snapshot processed between loadOlderTurns'
175 // epoch check and its merge could still slip resurrected history in.228 // epoch check and its merge could still slip resurrected history in.
176 const applyLock = yield* Semaphore.make(1);229 const applyLock = yield* Semaphore.make(1);
230 // Save only completed data/cursor updates. A canceled scope must not cache
231 // a cursor whose event has not reached the data yet.
232 const remember = Effect.gen(function* () {
233 const current = yield* SubscriptionRef.get(state);
234 const sequence = yield* SubscriptionRef.get(lastSequence);
235 committed = {
236 state: current,
237 sequence,
238 persisted:
239 committed.persisted &&
240 matchesThreadSnapshot(
241 committed,
242 Option.getOrNull(current.data),
243 sequence,
244 Option.getOrUndefined(current.page),
245 ),
246 };
247 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
248 });
177 // Whether the connected server accepts windowed reads; set per subscription249 // Whether the connected server accepts windowed reads; set per subscription
178 // from the session config. Gates loadOlderTurns so a reconnect to a250 // from the session config. Gates loadOlderTurns so a reconnect to a
179 // pre-pagination server never sends unsupported window parameters.251 // pre-pagination server never sends unsupported window parameters.
...@@ -190,7 +262,28 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -190,7 +262,28 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
190 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (262 const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
191 snapshot: OrchestrationThreadDetailSnapshot,263 snapshot: OrchestrationThreadDetailSnapshot,
192 ) {264 ) {
265 if (resumeCache !== undefined && resumeCache.owner !== owner) return;
266 if (
267 committed.persisted &&
268 matchesThreadSnapshot(committed, snapshot.thread, snapshot.snapshotSequence, snapshot.page)
269 )
270 return;
193 yield* cache.saveThread(environmentId, snapshot).pipe(271 yield* cache.saveThread(environmentId, snapshot).pipe(
272 Effect.tap(() =>
273 Effect.sync(() => {
274 if (
275 !matchesThreadSnapshot(
276 committed,
277 snapshot.thread,
278 snapshot.snapshotSequence,
279 snapshot.page,
280 )
281 )
282 return;
283 committed = { ...committed, persisted: true };
284 if (resumeCache?.owner === owner) resumeCache.snapshot = committed;
285 }),
286 ),
194 Effect.catch((error) =>287 Effect.catch((error) =>
195 Effect.logWarning("Could not persist the thread cache.").pipe(288 Effect.logWarning("Could not persist the thread cache.").pipe(
196 Effect.annotateLogs({289 Effect.annotateLogs({
...@@ -300,6 +393,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -300,6 +393,8 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
300 error: Option.none(),393 error: Option.none(),
301 page: Option.none(),394 page: Option.none(),
302 });395 });
396 yield* remember;
397 if (resumeCache !== undefined && resumeCache.owner !== owner) return;
303 yield* cache.removeThread(environmentId, threadId).pipe(398 yield* cache.removeThread(environmentId, threadId).pipe(
304 Effect.catch((error) =>399 Effect.catch((error) =>
305 Effect.logWarning("Could not remove the cached thread.").pipe(400 Effect.logWarning("Could not remove the cached thread.").pipe(
...@@ -402,7 +497,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -402,7 +497,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
402 const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (497 const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (
403 item: OrchestrationThreadStreamItem,498 item: OrchestrationThreadStreamItem,
404 ) {499 ) {
405 yield* applyLock.withPermits(1)(applyItemLocked(item));500 yield* applyLock.withPermits(1)(applyItemLocked(item).pipe(Effect.andThen(remember)));
406 });501 });
407502
408 // Merges an older disjoint page below the currently loaded window. All four503 // Merges an older disjoint page below the currently loaded window. All four
...@@ -459,6 +554,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -459,6 +554,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
459 ...(snapshot.page === undefined ? {} : { page: { ...snapshot.page, snapshotSequence } }),554 ...(snapshot.page === undefined ? {} : { page: { ...snapshot.page, snapshotSequence } }),
460 });555 });
461 }556 }
557 yield* remember;
462 });558 });
463559
464 const loadOlderTurns = Effect.fn("EnvironmentThreadState.loadOlderTurns")(function* () {560 const loadOlderTurns = Effect.fn("EnvironmentThreadState.loadOlderTurns")(function* () {
...@@ -579,15 +675,21 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -579,15 +675,21 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
579 // trap: afterSequence resume keeps only the window, and the missing675 // trap: afterSequence resume keeps only the window, and the missing
580 // older turns can never be loaded (the server has no cursor reads).676 // older turns can never be loaded (the server has no cursor reads).
581 // Drop the window marker and treat the data as needing a full reload.677 // Drop the window marker and treat the data as needing a full reload.
582 if (!supportsPagination && Option.isSome(current.page)) {678 if (!supportsPagination) {
583 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);679 yield* applyLock.withPermits(1)(
584 yield* SubscriptionRef.update(state, (value) => ({680 Effect.gen(function* () {
585 ...value,681 if (Option.isNone((yield* SubscriptionRef.get(state)).page)) return;
586 data: Option.none(),682 yield* Ref.update(historyEpoch, (epoch) => epoch + 1);
587 status: value.status === "deleted" ? value.status : ("empty" as const),683 yield* SubscriptionRef.update(state, (value) => ({
588 page: Option.none(),684 ...value,
589 }));685 data: Option.none(),
590 yield* SubscriptionRef.set(lastSequence, 0);686 status: value.status === "deleted" ? value.status : ("empty" as const),
687 page: Option.none(),
688 }));
689 yield* SubscriptionRef.set(lastSequence, 0);
690 yield* remember;
691 }),
692 );
591 current = yield* SubscriptionRef.get(state);693 current = yield* SubscriptionRef.get(state);
592 }694 }
593 if (Option.isNone(current.data) && current.status !== "deleted") {695 if (Option.isNone(current.data) && current.status !== "deleted") {
...@@ -663,40 +765,45 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make...@@ -663,40 +765,45 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
663 yield* Effect.addFinalizer(() => Effect.sync(deregister));765 yield* Effect.addFinalizer(() => Effect.sync(deregister));
664766
665 yield* Effect.addFinalizer(() =>767 yield* Effect.addFinalizer(() =>
666 Effect.all([SubscriptionRef.get(state), SubscriptionRef.get(lastSequence)]).pipe(768 Effect.suspend(() => {
667 Effect.flatMap(([current, snapshotSequence]) =>769 const { state: current, sequence: snapshotSequence } = committed;
668 Option.match(current.data, {770 return Option.match(current.data, {
669 onNone: () => Effect.void,771 onNone: () => Effect.void,
670 onSome: (thread) =>772 onSome: (thread) =>
671 shouldPersistThread(thread)773 shouldPersistThread(thread)
672 ? persist({774 ? persist({
673 snapshotSequence,775 snapshotSequence,
674 thread,776 thread,
675 ...Option.match(current.page, {777 ...Option.match(current.page, {
676 onNone: () => ({}),778 onNone: () => ({}),
677 onSome: (page) =>779 onSome: (page) =>
678 ({780 ({
679 page: {781 page: {
680 beforeCursor: page.beforeCursor,782 beforeCursor: page.beforeCursor,
681 hasMore: page.hasMore,783 hasMore: page.hasMore,
682 snapshotSequence,784 snapshotSequence,
683 },785 },
684 }) as const,786 }) as const,
685 }),787 }),
686 })788 })
687 : Effect.void,789 : Effect.void,
688 }),790 });
689 ),791 }),
690 ),
691 );792 );
692793
693 return state;794 return state;
694});795});
695796
696export function threadStateChanges(environmentId: EnvironmentIdType, threadId: ThreadIdType) {797export function threadStateChanges(
798 environmentId: EnvironmentIdType,
799 threadId: ThreadIdType,
800 resumeCache?: ThreadResumeCache,
801) {
697 return followStreamInEnvironment(802 return followStreamInEnvironment(
698 environmentId,803 environmentId,
699 Stream.unwrap(makeEnvironmentThreadState(threadId).pipe(Effect.map(SubscriptionRef.changes))),804 Stream.unwrap(
805 makeEnvironmentThreadState(threadId, resumeCache).pipe(Effect.map(SubscriptionRef.changes)),
806 ),
700 );807 );
701}808}
702809
...@@ -706,16 +813,35 @@ export function createEnvironmentThreadStateAtoms<R, E>(...@@ -706,16 +813,35 @@ export function createEnvironmentThreadStateAtoms<R, E>(
706 E813 E
707 >,814 >,
708) {815) {
816 // Cache definitions must outlive collectible live-atom definitions. The
817 // registry retains these nodes without retaining environment or RPC scopes.
818 const resumeFamily = Atom.family((key: string) =>
819 Atom.make((): ThreadResumeCache => ({
820 snapshot: undefined,
821 owner: undefined,
822 })).pipe(
823 Atom.setIdleTTL(THREAD_SNAPSHOT_IDLE_TTL_MS),
824 Atom.withLabel(`environment-thread-resume:${key}`),
825 ),
826 );
709 const family = Atom.family((key: string) => {827 const family = Atom.family((key: string) => {
710 const { environmentId, threadId } = parseThreadKey(key);828 const { environmentId, threadId } = parseThreadKey(key);
829 const resumeAtom = resumeFamily(key);
711 return runtime830 return runtime
712 .atom(threadStateChanges(environmentId, threadId), {831 .atom(
713 initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,832 (get) => {
714 })833 get.mount(resumeAtom);
715 .pipe(834 const resume = get.once(resumeAtom);
716 Atom.setIdleTTL(THREAD_STATE_IDLE_TTL_MS),835 const live = threadStateChanges(environmentId, threadId, resume);
717 Atom.withLabel(`environment-thread-state:${key}`),836 return resume.snapshot === undefined
718 );837 ? live
838 : Stream.concat(Stream.succeed(cachedThreadState(resume.snapshot.state)), live);
839 },
840 {
841 initialValue: EMPTY_ENVIRONMENT_THREAD_STATE,
842 },
843 )
844 .pipe(Atom.setIdleTTL(0), Atom.withLabel(`environment-thread-state:${key}`));
719 });845 });
720846
721 return {847 return {