authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-07 19:21:59-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 19:21:59-07:00
logf9e16b3f66476ad1e213152708c9aa8b035085a3
treecd0df33893155d00031e409e840e4572553e3fb3
parent83a82a46b0631234ac85748b912566d6f4b1dee3
signature Signed by PGP key B5690EEEBB952194

perf(mobile): show the cached thread list sooner and stop the freeze after it (#16713)

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

11 files changed, 816 insertions(+), 21 deletions(-)

apps/mobile/src/connection/environment-cache-store.test.ts+75
...@@ -10,12 +10,14 @@ import {...@@ -10,12 +10,14 @@ import {
10 ThreadId,10 ThreadId,
11 type OrchestrationV2ShellSnapshot,11 type OrchestrationV2ShellSnapshot,
12 type OrchestrationV2ThreadDetailSnapshot,12 type OrchestrationV2ThreadDetailSnapshot,
13 type ThreadPullRequestLink,
13 type VcsListRefsResult,14 type VcsListRefsResult,
14} from "@t3tools/contracts";15} from "@t3tools/contracts";
15import { describe, expect, it } from "@effect/vitest";16import { describe, expect, it } from "@effect/vitest";
16import * as DateTime from "effect/DateTime";17import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";18import * as Effect from "effect/Effect";
18import * as Fiber from "effect/Fiber";19import * as Fiber from "effect/Fiber";
20import * as Logger from "effect/Logger";
19import * as Schema from "effect/Schema";21import * as Schema from "effect/Schema";
20import { vi } from "vite-plus/test";22import { vi } from "vite-plus/test";
21import * as Deferred from "effect/Deferred";23import * as Deferred from "effect/Deferred";
...@@ -254,6 +256,79 @@ describe("mobile SQLite environment cache store", () => {...@@ -254,6 +256,79 @@ describe("mobile SQLite environment cache store", () => {
254 }),256 }),
255 );257 );
256258
259 it.live("returns cached threads before their pull request links, which load after", () =>
260 Effect.gen(function* () {
261 const memory = makeDatabase();
262 const store = yield* make().pipe(
263 Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
264 );
265 const link: ThreadPullRequestLink = {
266 host: "github.com",
267 repository: "pingdotgg/t3code",
268 number: 42,
269 url: "https://github.com/pingdotgg/t3code/pull/42",
270 source: "agent",
271 linkedAt: "2026-07-29T12:00:00.000Z",
272 snapshot: null,
273 stack: null,
274 };
275 const linked = { ...SHELL_SNAPSHOT.threads[0]!, pullRequests: [link] };
276 const unlinked = { ...SHELL_SNAPSHOT.threads[0]!, id: ThreadId.make("thread-2") };
277 yield* store.saveShell(ENVIRONMENT_ID, {
278 ...SHELL_SNAPSHOT,
279 threads: [linked, unlinked],
280 });
281
282 const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID));
283 expect(shell.threads.map((thread) => thread.pullRequests)).toEqual([undefined, undefined]);
284 const links = yield* shell.loadPullRequests!;
285 expect([...links]).toEqual([[THREAD_ID, [link]]]);
286 }),
287 );
288
289 it.live("keeps cached threads when their pull request links cannot be read", () => {
290 const messages: Array<unknown> = [];
291 const logger = Logger.make(({ message }) => {
292 messages.push(message);
293 });
294 return Effect.gen(function* () {
295 const memory = makeDatabase();
296 const store = yield* make().pipe(
297 Effect.provideService(MobileDatabase.MobileDatabase, memory.database),
298 );
299 const link: ThreadPullRequestLink = {
300 host: "github.com",
301 repository: "pingdotgg/t3code",
302 number: 5,
303 url: "https://github.com/pingdotgg/t3code/pull/5",
304 source: "agent",
305 linkedAt: "2026-07-29T12:00:00.000Z",
306 snapshot: null,
307 stack: null,
308 };
309 const readable = {
310 ...SHELL_SNAPSHOT.threads[0]!,
311 id: ThreadId.make("thread-2"),
312 pullRequests: [link],
313 };
314 yield* store.saveShell(ENVIRONMENT_ID, {
315 ...SHELL_SNAPSHOT,
316 threads: [SHELL_SNAPSHOT.threads[0]!, readable],
317 });
318 const id = cacheId(ENVIRONMENT_ID, "shell", "snapshot");
319 const payload = JSON.parse(memory.values.get(id)!);
320 payload.snapshot.threads[0].pullRequests = [{ number: -1 }];
321 memory.values.set(id, JSON.stringify(payload));
322
323 const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID));
324 expect(shell.threads.map((thread) => thread.id)).toEqual([THREAD_ID, readable.id]);
325 // Only the thread with the unreadable link loses its links.
326 expect([...(yield* shell.loadPullRequests!)]).toEqual([[readable.id, [link]]]);
327 expect(messages).toHaveLength(1);
328 expect(memory.removed).toEqual([]);
329 }).pipe(Effect.provide(Logger.layer([logger], { mergeWithExisting: false })));
330 });
331
257 it.effect("round-trips schema-validated VCS refs", () =>332 it.effect("round-trips schema-validated VCS refs", () =>
258 Effect.gen(function* () {333 Effect.gen(function* () {
259 const memory = makeDatabase();334 const memory = makeDatabase();
apps/mobile/src/connection/environment-cache-store.ts+10-5
...@@ -1,6 +1,5 @@...@@ -1,6 +1,5 @@
1import {1import {
2 ORCHESTRATION_CACHE_SCHEMA_VERSION,2 ORCHESTRATION_CACHE_SCHEMA_VERSION,
3 StoredOrchestrationShellSnapshot,
4 StoredOrchestrationThreadSnapshot,3 StoredOrchestrationThreadSnapshot,
5 Persistence,4 Persistence,
6} from "@t3tools/client-runtime/platform";5} from "@t3tools/client-runtime/platform";
...@@ -16,6 +15,7 @@ import {...@@ -16,6 +15,7 @@ import {
16 attachProjectFaviconDatabase,15 attachProjectFaviconDatabase,
17 projectFaviconDatabaseCache,16 projectFaviconDatabaseCache,
18} from "../lib/projectFaviconDatabaseCache";17} from "../lib/projectFaviconDatabaseCache";
18import { decodeStoredShellSnapshot } from "./shell-cache-decoding";
1919
20const SERVER_CONFIG_CACHE_SCHEMA_VERSION = 1;20const SERVER_CONFIG_CACHE_SCHEMA_VERSION = 1;
21const VCS_REFS_CACHE_SCHEMA_VERSION = 1;21const VCS_REFS_CACHE_SCHEMA_VERSION = 1;
...@@ -32,9 +32,6 @@ const StoredVcsRefs = Schema.Struct({...@@ -32,9 +32,6 @@ const StoredVcsRefs = Schema.Struct({
32 refs: VcsListRefsResult,32 refs: VcsListRefsResult,
33});33});
3434
35const decodeStoredShellSnapshot = Schema.decodeUnknownEffect(
36 Schema.fromJsonString(StoredOrchestrationShellSnapshot),
37);
38const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect(35const decodeStoredThreadSnapshot = Schema.decodeUnknownEffect(
39 Schema.fromJsonString(StoredOrchestrationThreadSnapshot),36 Schema.fromJsonString(StoredOrchestrationThreadSnapshot),
40);37);
...@@ -113,7 +110,15 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {...@@ -113,7 +110,15 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () {
113 decode: decodeStoredShellSnapshot,110 decode: decodeStoredShellSnapshot,
114 select: (stored) =>111 select: (stored) =>
115 stored.environmentId === environmentId ? Option.some(stored.snapshot) : Option.none(),112 stored.environmentId === environmentId ? Option.some(stored.snapshot) : Option.none(),
116 }).pipe(Effect.tap(() => Effect.promise(() => projectFaviconDatabaseCache.hydrate()))),113 }).pipe(
114 // Read cached project icons in parallel with the shell so the first rows can show
115 // them, instead of making the thread list wait for the icon read after the decode.
116 Effect.zipWith(
117 Effect.promise(() => projectFaviconDatabaseCache.hydrate()),
118 (snapshot) => snapshot,
119 { concurrent: true },
120 ),
121 ),
117 ),122 ),
118 saveShell: Effect.fn("MobileEnvironmentCache.saveShell")(function* (environmentId, snapshot) {123 saveShell: Effect.fn("MobileEnvironmentCache.saveShell")(function* (environmentId, snapshot) {
119 const payload = yield* encodeStoredShellSnapshot({124 const payload = yield* encodeStoredShellSnapshot({
apps/mobile/src/connection/shell-cache-decoding.ts created+19
...@@ -0,0 +1,19 @@
1import { StoredOrchestrationShellSnapshot } from "@t3tools/client-runtime/platform";
2import { deferPullRequests, detachPullRequests } from "@t3tools/client-runtime/state/shell";
3import * as Effect from "effect/Effect";
4import * as Predicate from "effect/Predicate";
5import * as Schema from "effect/Schema";
6
7const decodeJson = Schema.decodeUnknownEffect(Schema.fromJsonString(Schema.Unknown));
8const decodeStoredShell = Schema.decodeUnknownEffect(StoredOrchestrationShellSnapshot);
9
10/**
11 * Decode the shell cache payload with each thread's pull request links left out, so the
12 * thread list can paint first; `loadPullRequests` decodes the links afterwards.
13 */
14export const decodeStoredShellSnapshot = Effect.fnUntraced(function* (raw: string) {
15 const parsed = yield* decodeJson(raw);
16 const rawLinks = Predicate.isObject(parsed) ? detachPullRequests(parsed.snapshot) : [];
17 const stored = yield* decodeStoredShell(parsed);
18 return { ...stored, snapshot: deferPullRequests(stored.snapshot, rawLinks) };
19});
packages/client-runtime/src/platform/persistence.ts+2-1
...@@ -12,6 +12,7 @@ import * as Option from "effect/Option";...@@ -12,6 +12,7 @@ import * as Option from "effect/Option";
12import * as Schema from "effect/Schema";12import * as Schema from "effect/Schema";
1313
14import type { ConnectionRegistration } from "../connection/catalog.ts";14import type { ConnectionRegistration } from "../connection/catalog.ts";
15import type { DeferredShellSnapshot } from "../state/shellPullRequests.ts";
15import type { ConnectionTarget, PersistedConnectionTarget } from "../connection/model.ts";16import type { ConnectionTarget, PersistedConnectionTarget } from "../connection/model.ts";
1617
17export class ConnectionPersistenceError extends Schema.TaggedError<ConnectionPersistenceError>()(18export class ConnectionPersistenceError extends Schema.TaggedError<ConnectionPersistenceError>()(
...@@ -82,7 +83,7 @@ export class EnvironmentCacheStore extends Context.Service<...@@ -82,7 +83,7 @@ export class EnvironmentCacheStore extends Context.Service<
82 {83 {
83 readonly loadShell: (84 readonly loadShell: (
84 environmentId: EnvironmentId,85 environmentId: EnvironmentId,
85 ) => Effect.Effect<Option.Option<OrchestrationV2ShellSnapshot>, ConnectionPersistenceError>;86 ) => Effect.Effect<Option.Option<DeferredShellSnapshot>, ConnectionPersistenceError>;
86 readonly saveShell: (87 readonly saveShell: (
87 environmentId: EnvironmentId,88 environmentId: EnvironmentId,
88 snapshot: OrchestrationV2ShellSnapshot,89 snapshot: OrchestrationV2ShellSnapshot,
packages/client-runtime/src/state/environmentHttpAuth.test.ts+19-1
...@@ -204,7 +204,11 @@ const LOADERS: ReadonlyArray<{...@@ -204,7 +204,11 @@ const LOADERS: ReadonlyArray<{
204 path: "/api/orchestration/shell",204 path: "/api/orchestration/shell",
205 response: SHELL,205 response: SHELL,
206 expected: SHELL,206 expected: SHELL,
207 load: fetchEnvironmentShellSnapshot,207 // Pull request links decode separately; the rows match the response on their own.
208 load: (input: HttpInput) =>
209 fetchEnvironmentShellSnapshot(input).pipe(
210 Effect.map(({ loadPullRequests: _links, ...snapshot }) => snapshot),
211 ),
208 },212 },
209 {213 {
210 name: "thread snapshot",214 name: "thread snapshot",
...@@ -254,6 +258,20 @@ describe("authenticated environment HTTP requests", () => {...@@ -254,6 +258,20 @@ describe("authenticated environment HTTP requests", () => {
254 }),258 }),
255 );259 );
256260
261 it.effect("keeps the status of a shell snapshot error that is not a declared error", () =>
262 Effect.gen(function* () {
263 const harness = makeHarness(() => Response.json({ error: "bad_gateway" }, { status: 502 }));
264 const error = yield* fetchEnvironmentShellSnapshot(harness.input).pipe(
265 Effect.provide(harness.httpLayer),
266 Effect.flip,
267 );
268 expect(error).toMatchObject({
269 _tag: "RemoteEnvironmentAuthUndeclaredStatusError",
270 status: 502,
271 });
272 }),
273 );
274
257 it.effect.each(LOADERS)("uses current relay authorization and endpoint for $name", (loader) =>275 it.effect.each(LOADERS)("uses current relay authorization and endpoint for $name", (loader) =>
258 Effect.gen(function* () {276 Effect.gen(function* () {
259 const harness = makeHarness(() => Response.json(loader.response));277 const harness = makeHarness(() => Response.json(loader.response));
packages/client-runtime/src/state/shell-sync.test.ts+345-1
...@@ -3,6 +3,8 @@ import {...@@ -3,6 +3,8 @@ import {
3 ORCHESTRATION_V2_WS_METHODS,3 ORCHESTRATION_V2_WS_METHODS,
4 type OrchestrationV2ShellSnapshot,4 type OrchestrationV2ShellSnapshot,
5 type OrchestrationV2ShellStreamItem,5 type OrchestrationV2ShellStreamItem,
6 ThreadId,
7 type ThreadPullRequestLink,
6} from "@t3tools/contracts";8} from "@t3tools/contracts";
7import { describe, expect, it } from "@effect/vitest";9import { describe, expect, it } from "@effect/vitest";
8import * as Deferred from "effect/Deferred";10import * as Deferred from "effect/Deferred";
...@@ -29,7 +31,7 @@ import * as RpcSession from "../rpc/session.ts";...@@ -29,7 +31,7 @@ import * as RpcSession from "../rpc/session.ts";
29import type { WsRpcProtocolClient } from "../rpc/protocol.ts";31import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
30import { makeEnvironmentShellState } from "./shell.ts";32import { makeEnvironmentShellState } from "./shell.ts";
31import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";33import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
32import { v2Project, v2ShellSnapshot } from "./orchestrationV2TestFixtures.ts";34import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
3335
34const TARGET = new PrimaryConnectionTarget({36const TARGET = new PrimaryConnectionTarget({
35 environmentId: EnvironmentId.make("environment-1"),37 environmentId: EnvironmentId.make("environment-1"),
...@@ -953,4 +955,346 @@ describe("environment shell synchronization", () => {...@@ -953,4 +955,346 @@ describe("environment shell synchronization", () => {
953 expect(Option.getOrThrow(state.snapshot).projects[0]?.title).toBe("Server reset");955 expect(Option.getOrThrow(state.snapshot).projects[0]?.title).toBe("Server reset");
954 }),956 }),
955 );957 );
958
959 it.effect("fills deferred cached pull request links without overwriting newer rows", () =>
960 Effect.gen(function* () {
961 const link = (number: number): ThreadPullRequestLink => ({
962 host: "github.com",
963 repository: "pingdotgg/t3code",
964 number,
965 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
966 source: "agent",
967 linkedAt: "2026-06-20T00:00:00.000Z",
968 snapshot: null,
969 stack: null,
970 });
971 const otherThreadId = ThreadId.make("thread-other");
972 const cachedSnapshot: OrchestrationV2ShellSnapshot = {
973 ...v2ShellSnapshot,
974 snapshotSequence: 5,
975 threads: [v2ThreadShell, { ...v2ThreadShell, id: otherThreadId }],
976 };
977 const linksReady = yield* Deferred.make<void>();
978 const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();
979 const client = {
980 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),
981 } as unknown as WsRpcProtocolClient;
982 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
983 target: TARGET,
984 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
985 session: yield* SubscriptionRef.make(Option.some(session(client))),
986 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
987 connect: Effect.void,
988 disconnect: Effect.void,
989 retryNow: Effect.void,
990 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
991 const cache = Persistence.EnvironmentCacheStore.of({
992 loadShell: () =>
993 Effect.succeedSome({
994 ...cachedSnapshot,
995 loadPullRequests: Deferred.await(linksReady).pipe(
996 Effect.as(
997 new Map([
998 [v2ThreadShell.id, [link(1)]],
999 [otherThreadId, [link(2)]],
1000 ]),
1001 ),
1002 ),
1003 }),
1004 saveShell: () => Effect.void,
1005 loadThread: () => Effect.succeedNone,
1006 saveThread: () => Effect.void,
1007 removeThread: () => Effect.void,
1008 loadServerConfig: () => Effect.succeedNone,
1009 saveServerConfig: () => Effect.void,
1010 loadVcsRefs: () => Effect.succeedNone,
1011 saveVcsRefs: () => Effect.void,
1012 removeVcsRefs: () => Effect.void,
1013 clearVcsRefs: () => Effect.void,
1014 clear: () => Effect.void,
1015 });
1016 // No HTTP snapshot, so the cached rows stay until the socket sends deltas.
1017 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1018 load: () => Effect.succeedNone,
1019 });
1020 const shellState = yield* makeEnvironmentShellState().pipe(
1021 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1022 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1023 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1024 );
1025
1026 const threadsOf = (state: {
1027 readonly snapshot: Option.Option<OrchestrationV2ShellSnapshot>;
1028 }) => Option.getOrThrow(state.snapshot).threads;
1029 expect(threadsOf(yield* SubscriptionRef.get(shellState))).toEqual(cachedSnapshot.threads);
1030
1031 // The server updates one row before the cached links finish decoding.
1032 const serverRow = { ...v2ThreadShell, title: "From server", pullRequests: [link(9)] };
1033 yield* Queue.offer(events, {
1034 kind: "thread.updated",
1035 sequence: 6,
1036 location: "active",
1037 thread: serverRow,
1038 });
1039 yield* SubscriptionRef.changes(shellState).pipe(
1040 Stream.filter((state) => threadsOf(state)[0]?.title === "From server"),
1041 Stream.runHead,
1042 );
1043
1044 yield* Deferred.succeed(linksReady, undefined);
1045 const filled = yield* SubscriptionRef.changes(shellState).pipe(
1046 Stream.map(threadsOf),
1047 Stream.filter((threads) => threads[1]?.pullRequests !== undefined),
1048 Stream.runHead,
1049 Effect.map(Option.getOrThrow),
1050 );
1051 expect(filled[0]).toBe(serverRow);
1052 expect(filled[1]?.pullRequests).toEqual([link(2)]);
1053 }),
1054 );
1055
1056 it.effect("shows HTTP rows before their pull request links and saves them once filled", () =>
1057 Effect.gen(function* () {
1058 const link: ThreadPullRequestLink = {
1059 host: "github.com",
1060 repository: "pingdotgg/t3code",
1061 number: 7,
1062 url: "https://github.com/pingdotgg/t3code/pull/7",
1063 source: "agent",
1064 linkedAt: "2026-06-20T00:00:00.000Z",
1065 snapshot: null,
1066 stack: null,
1067 };
1068 const httpSnapshot: OrchestrationV2ShellSnapshot = {
1069 ...v2ShellSnapshot,
1070 snapshotSequence: 3,
1071 };
1072 const linksReady = yield* Deferred.make<void>();
1073 const saved = yield* Queue.unbounded<OrchestrationV2ShellSnapshot>();
1074 const client = {
1075 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.never,
1076 } as unknown as WsRpcProtocolClient;
1077 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1078 target: TARGET,
1079 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1080 session: yield* SubscriptionRef.make(Option.some(session(client))),
1081 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1082 connect: Effect.void,
1083 disconnect: Effect.void,
1084 retryNow: Effect.void,
1085 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1086 const cache = Persistence.EnvironmentCacheStore.of({
1087 loadShell: () => Effect.succeedNone,
1088 saveShell: (_environmentId, snapshot) => Queue.offer(saved, snapshot),
1089 loadThread: () => Effect.succeedNone,
1090 saveThread: () => Effect.void,
1091 removeThread: () => Effect.void,
1092 loadServerConfig: () => Effect.succeedNone,
1093 saveServerConfig: () => Effect.void,
1094 loadVcsRefs: () => Effect.succeedNone,
1095 saveVcsRefs: () => Effect.void,
1096 removeVcsRefs: () => Effect.void,
1097 clearVcsRefs: () => Effect.void,
1098 clear: () => Effect.void,
1099 });
1100 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1101 load: () =>
1102 Effect.succeedSome({
1103 ...httpSnapshot,
1104 loadPullRequests: Deferred.await(linksReady).pipe(
1105 Effect.as(new Map([[v2ThreadShell.id, [link]]])),
1106 ),
1107 }),
1108 });
1109 const shellState = yield* makeEnvironmentShellState().pipe(
1110 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1111 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1112 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1113 );
1114
1115 const rows = yield* SubscriptionRef.changes(shellState).pipe(
1116 Stream.filter((state) => Option.isSome(state.snapshot)),
1117 Stream.runHead,
1118 Effect.map((state) => Option.getOrThrow(Option.getOrThrow(state).snapshot)),
1119 );
1120 expect(rows.threads[0]?.pullRequests).toBeUndefined();
1121 expect(rows).not.toHaveProperty("loadPullRequests");
1122
1123 yield* Deferred.succeed(linksReady, undefined);
1124 const filled = yield* SubscriptionRef.changes(shellState).pipe(
1125 Stream.map((state) => Option.getOrThrow(state.snapshot).threads[0]),
1126 Stream.filter((thread) => thread?.pullRequests !== undefined),
1127 Stream.runHead,
1128 Effect.map(Option.getOrThrow),
1129 );
1130 expect(filled?.pullRequests).toEqual([link]);
1131
1132 // The cache must end up with the links: the last save carries them.
1133 yield* TestClock.adjust("30 seconds");
1134 const saves = yield* Queue.takeAll(saved);
1135 expect(saves.at(-1)?.threads[0]?.pullRequests).toEqual([link]);
1136 }),
1137 );
1138
1139 it.effect("does not save rows whose deferred links are still pending", () =>
1140 Effect.gen(function* () {
1141 const link: ThreadPullRequestLink = {
1142 host: "github.com",
1143 repository: "pingdotgg/t3code",
1144 number: 8,
1145 url: "https://github.com/pingdotgg/t3code/pull/8",
1146 source: "agent",
1147 linkedAt: "2026-06-20T00:00:00.000Z",
1148 snapshot: null,
1149 stack: null,
1150 };
1151 const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);
1152 const client = {
1153 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.never,
1154 } as unknown as WsRpcProtocolClient;
1155 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1156 target: TARGET,
1157 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1158 session: yield* SubscriptionRef.make(Option.some(session(client))),
1159 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1160 connect: Effect.void,
1161 disconnect: Effect.void,
1162 retryNow: Effect.void,
1163 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1164 const cache = Persistence.EnvironmentCacheStore.of({
1165 loadShell: () => Effect.succeedNone,
1166 saveShell: (_environmentId, snapshot) => Ref.update(saved, (all) => [...all, snapshot]),
1167 loadThread: () => Effect.succeedNone,
1168 saveThread: () => Effect.void,
1169 removeThread: () => Effect.void,
1170 loadServerConfig: () => Effect.succeedNone,
1171 saveServerConfig: () => Effect.void,
1172 loadVcsRefs: () => Effect.succeedNone,
1173 saveVcsRefs: () => Effect.void,
1174 removeVcsRefs: () => Effect.void,
1175 clearVcsRefs: () => Effect.void,
1176 clear: () => Effect.void,
1177 });
1178 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1179 load: () =>
1180 Effect.succeedSome({
1181 ...v2ShellSnapshot,
1182 snapshotSequence: 3,
1183 // Links never finish decoding before the shell closes.
1184 loadPullRequests: Effect.never.pipe(Effect.as(new Map([[v2ThreadShell.id, [link]]]))),
1185 }),
1186 });
1187 const scope = yield* Scope.make();
1188 const shellState = yield* makeEnvironmentShellState().pipe(
1189 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1190 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1191 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1192 Scope.provide(scope),
1193 );
1194 yield* SubscriptionRef.changes(shellState).pipe(
1195 Stream.filter((state) => Option.isSome(state.snapshot)),
1196 Stream.runHead,
1197 );
1198
1199 // Neither the throttled worker nor the closing flush may save the rows without links.
1200 yield* TestClock.adjust("30 seconds");
1201 yield* Scope.close(scope, Exit.void);
1202 expect(yield* Ref.get(saved)).toEqual([]);
1203 }),
1204 );
1205
1206 it.effect("drops an older snapshot's links once a newer snapshot reuses its rows", () =>
1207 Effect.gen(function* () {
1208 const link = (number: number): ThreadPullRequestLink => ({
1209 host: "github.com",
1210 repository: "pingdotgg/t3code",
1211 number,
1212 url: `https://github.com/pingdotgg/t3code/pull/${number}`,
1213 source: "agent",
1214 linkedAt: "2026-06-20T00:00:00.000Z",
1215 snapshot: null,
1216 stack: null,
1217 });
1218 const firstLinks = yield* Deferred.make<void>();
1219 const secondLinks = yield* Deferred.make<void>();
1220 // Set once each snapshot's links start decoding, i.e. once its rows are applied.
1221 const firstFillStarted = yield* Deferred.make<void>();
1222 const loads = yield* Ref.make(0);
1223 const subscriptions = yield* Ref.make(0);
1224 const client = {
1225 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () =>
1226 Stream.unwrap(Ref.update(subscriptions, (n) => n + 1).pipe(Effect.as(Stream.never))),
1227 } as unknown as WsRpcProtocolClient;
1228 const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
1229 Option.some(session(client)),
1230 );
1231 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
1232 target: TARGET,
1233 state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
1234 session: activeSession,
1235 prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
1236 connect: Effect.void,
1237 disconnect: Effect.void,
1238 retryNow: Effect.void,
1239 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
1240 const cache = Persistence.EnvironmentCacheStore.of({
1241 loadShell: () => Effect.succeedNone,
1242 saveShell: () => Effect.void,
1243 loadThread: () => Effect.succeedNone,
1244 saveThread: () => Effect.void,
1245 removeThread: () => Effect.void,
1246 loadServerConfig: () => Effect.succeedNone,
1247 saveServerConfig: () => Effect.void,
1248 loadVcsRefs: () => Effect.succeedNone,
1249 saveVcsRefs: () => Effect.void,
1250 removeVcsRefs: () => Effect.void,
1251 clearVcsRefs: () => Effect.void,
1252 clear: () => Effect.void,
1253 });
1254 // Both snapshots carry the same row; only its pull request links differ.
1255 const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({
1256 load: () =>
1257 Ref.updateAndGet(loads, (n) => n + 1).pipe(
1258 Effect.map((n) =>
1259 Option.some({
1260 ...v2ShellSnapshot,
1261 snapshotSequence: n,
1262 loadPullRequests: (n === 1
1263 ? Deferred.succeed(firstFillStarted, undefined).pipe(
1264 Effect.andThen(Deferred.await(firstLinks)),
1265 )
1266 : Deferred.await(secondLinks)
1267 ).pipe(Effect.as(new Map([[v2ThreadShell.id, [link(n)]]]))),
1268 }),
1269 ),
1270 ),
1271 });
1272 const shellState = yield* makeEnvironmentShellState().pipe(
1273 Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
1274 Effect.provideService(Persistence.EnvironmentCacheStore, cache),
1275 Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader),
1276 );
1277 yield* Deferred.await(firstFillStarted);
1278
1279 // A new session loads a second snapshot that reuses the bare row.
1280 yield* SubscriptionRef.set(activeSession, Option.some(session(client)));
1281 yield* SubscriptionRef.changes(shellState).pipe(
1282 Stream.filter(
1283 (state) => Option.isSome(state.snapshot) && state.snapshot.value.snapshotSequence === 2,
1284 ),
1285 Stream.runHead,
1286 );
1287
1288 // The superseded first fill finishes first, then the second.
1289 yield* Deferred.succeed(firstLinks, undefined);
1290 yield* Deferred.succeed(secondLinks, undefined);
1291 const filled = yield* SubscriptionRef.changes(shellState).pipe(
1292 Stream.map((state) => Option.getOrThrow(state.snapshot).threads[0]?.pullRequests),
1293 Stream.filter((links) => links !== undefined),
1294 Stream.runHead,
1295 Effect.map(Option.getOrThrow),
1296 );
1297 expect(filled).toEqual([link(2)]);
1298 }),
1299 );
956});1300});
packages/client-runtime/src/state/shell.ts+126-5
...@@ -3,10 +3,14 @@ import {...@@ -3,10 +3,14 @@ import {
3 type EnvironmentId,3 type EnvironmentId,
4 type OrchestrationV2ShellSnapshot,4 type OrchestrationV2ShellSnapshot,
5 type OrchestrationV2ShellStreamItem,5 type OrchestrationV2ShellStreamItem,
6 type OrchestrationV2ThreadShell,
6 type ServerConfig,7 type ServerConfig,
8 type ThreadId,
9 type ThreadPullRequestLink,
7} from "@t3tools/contracts";10} from "@t3tools/contracts";
8import * as Cause from "effect/Cause";11import * as Cause from "effect/Cause";
9import * as Effect from "effect/Effect";12import * as Effect from "effect/Effect";
13import * as Fiber from "effect/Fiber";
10import * as Option from "effect/Option";14import * as Option from "effect/Option";
11import * as Queue from "effect/Queue";15import * as Queue from "effect/Queue";
12import * as Ref from "effect/Ref";16import * as Ref from "effect/Ref";
...@@ -25,7 +29,13 @@ import { runCachePersistence } from "./cachePersistence.ts";...@@ -25,7 +29,13 @@ import { runCachePersistence } from "./cachePersistence.ts";
25import { subscribeDynamic } from "../rpc/client.ts";29import { subscribeDynamic } from "../rpc/client.ts";
26import type { RpcSession } from "../rpc/session.ts";30import type { RpcSession } from "../rpc/session.ts";
27import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";31import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
28import { applyShellStreamEvent, mergeShellSnapshotProjects } from "./shellReducer.ts";32import type { DeferredShellSnapshot } from "./shellPullRequests.ts";
33import {
34 applyShellStreamEvent,
35 mergeShellSnapshotProjects,
36 reuseUnchangedThreadShells,
37 sameThreadPullRequests,
38} from "./shellReducer.ts";
29import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";39import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";
30import { followStreamInEnvironment } from "./runtime.ts";40import { followStreamInEnvironment } from "./runtime.ts";
3141
...@@ -49,6 +59,12 @@ function shellStatusForSnapshot(...@@ -49,6 +59,12 @@ function shellStatusForSnapshot(
49 return Option.isSome(snapshot) ? "cached" : "empty";59 return Option.isSome(snapshot) ? "cached" : "empty";
50}60}
5161
62function withoutDeferredPullRequests(value: DeferredShellSnapshot): OrchestrationV2ShellSnapshot {
63 if (value.loadPullRequests === undefined) return value;
64 const { loadPullRequests: _deferred, ...snapshot } = value;
65 return snapshot;
66}
67
52const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data.";68const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data.";
5369
54export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {70export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {
...@@ -57,17 +73,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -57,17 +73,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
57 const snapshotLoader = yield* ShellSnapshotLoader.ShellSnapshotLoader;73 const snapshotLoader = yield* ShellSnapshotLoader.ShellSnapshotLoader;
58 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);74 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
59 const environmentId = supervisor.target.environmentId;75 const environmentId = supervisor.target.environmentId;
60 const cachedSnapshot = yield* cache.loadShell(environmentId).pipe(76 const cached = yield* cache.loadShell(environmentId).pipe(
61 Effect.catch((error) =>77 Effect.catch((error) =>
62 Effect.logWarning("Could not load cached environment shell.").pipe(78 Effect.logWarning("Could not load cached environment shell.").pipe(
63 Effect.annotateLogs({79 Effect.annotateLogs({
64 environmentId,80 environmentId,
65 ...safeErrorLogAttributes(error),81 ...safeErrorLogAttributes(error),
66 }),82 }),
67 Effect.as(Option.none<OrchestrationV2ShellSnapshot>()),83 Effect.as(Option.none<DeferredShellSnapshot>()),
68 ),84 ),
69 ),85 ),
70 );86 );
87 const cachedSnapshot = Option.map(cached, withoutDeferredPullRequests);
71 const state = yield* SubscriptionRef.make<EnvironmentShellState>({88 const state = yield* SubscriptionRef.make<EnvironmentShellState>({
72 snapshot: cachedSnapshot,89 snapshot: cachedSnapshot,
73 status: shellStatusForSnapshot(cachedSnapshot),90 status: shellStatusForSnapshot(cachedSnapshot),
...@@ -83,12 +100,16 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -83,12 +100,16 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
83100
84 const persistenceLock = yield* Semaphore.make(1);101 const persistenceLock = yield* Semaphore.make(1);
85 let lastPersisted: OrchestrationV2ShellSnapshot | undefined;102 let lastPersisted: OrchestrationV2ShellSnapshot | undefined;
103 // Rows still waiting for their deferred pull request links, which must not be saved:
104 // the cache would keep them without links until the next server refresh.
105 let pendingLinkRows: ReadonlySet<OrchestrationV2ThreadShell> = new Set();
86 const persistLatest = Effect.fn("EnvironmentShellState.persistLatest")(function* (106 const persistLatest = Effect.fn("EnvironmentShellState.persistLatest")(function* (
87 flush: boolean,107 flush: boolean,
88 ) {108 ) {
89 while (true) {109 while (true) {
90 const latest = yield* Ref.get(latestLiveSnapshot);110 const latest = yield* Ref.get(latestLiveSnapshot);
91 if (Option.isNone(latest) || latest.value === lastPersisted) return;111 if (Option.isNone(latest) || latest.value === lastPersisted) return;
112 if (latest.value.threads.some((thread) => pendingLinkRows.has(thread))) return;
92 const snapshot = latest.value;113 const snapshot = latest.value;
93 const saved = yield* cache.saveShell(environmentId, snapshot).pipe(114 const saved = yield* cache.saveShell(environmentId, snapshot).pipe(
94 Effect.as(true),115 Effect.as(true),
...@@ -155,11 +176,17 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -155,11 +176,17 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
155 ),176 ),
156 );177 );
157178
179 // Server items and deferred link fills both read the snapshot and write it back; one at a
180 // time, so neither can overwrite what the other wrote in between.
181 const snapshotWriteLock = yield* Semaphore.make(1);
182
158 // Apply each received batch with one state write. The RPC client's bounded183 // Apply each received batch with one state write. The RPC client's bounded
159 // buffer can split a server chunk, so a bulk action can still need several184 // buffer can split a server chunk, so a bulk action can still need several
160 // writes, but each write includes every event in that batch.185 // writes, but each write includes every event in that batch.
161 const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* (186 const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* (
162 items: ReadonlyArray<OrchestrationV2ShellStreamItem>,187 items: ReadonlyArray<OrchestrationV2ShellStreamItem>,
188 // Ids of rows whose links will fill later; they are held out of saves before one queues.
189 pendingLinkIds?: ReadonlySet<ThreadId>,
163 ) {190 ) {
164 const initial = yield* SubscriptionRef.get(state);191 const initial = yield* SubscriptionRef.get(state);
165 let waiting = yield* Ref.get(awaitingCompletion);192 let waiting = yield* Ref.get(awaitingCompletion);
...@@ -201,6 +228,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -201,6 +228,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
201 }228 }
202 yield* Ref.set(awaitingCompletion, waiting);229 yield* Ref.set(awaitingCompletion, waiting);
203 if (next === initial) return;230 if (next === initial) return;
231 if (pendingLinkIds !== undefined && Option.isSome(next.snapshot)) {
232 pendingLinkRows = new Set(
233 next.snapshot.value.threads.filter((thread) => pendingLinkIds.has(thread.id)),
234 );
235 }
204 if (Option.isSome(next.snapshot)) {236 if (Option.isSome(next.snapshot)) {
205 yield* Ref.set(latestLiveSnapshot, next.snapshot);237 yield* Ref.set(latestLiveSnapshot, next.snapshot);
206 }238 }
...@@ -214,7 +246,81 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -214,7 +246,81 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
214 if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) {246 if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) {
215 yield* Queue.offer(persistence, next.snapshot.value);247 yield* Queue.offer(persistence, next.snapshot.value);
216 }248 }
249 }, snapshotWriteLock.withPermit);
250
251 // Rows can arrive before their pull request links (see DeferredShellSnapshot), so the list
252 // paints first. The links then fill into the rows that still hold the object they arrived
253 // with; a row the server has replaced since carries its own links.
254 // Captured so fills started from the subscription callback still end with this state.
255 const stateScope = yield* Effect.scope;
256 // Each full snapshot supersedes the previous one's fill; links from an older snapshot must
257 // not land on rows a newer snapshot reused. A cancelled fill's rows stay pending until the
258 // next snapshot replaces the pending set.
259 let activeFill: Fiber.Fiber<void> | undefined;
260 const cancelActiveFill = Effect.suspend(() => {
261 const fill = activeFill;
262 activeFill = undefined;
263 return fill === undefined ? Effect.void : Fiber.interrupt(fill);
217 });264 });
265 const fillDeferredPullRequests = Effect.fnUntraced(function* (
266 source: DeferredShellSnapshot,
267 applied: OrchestrationV2ShellSnapshot,
268 ) {
269 yield* cancelActiveFill;
270 if (source.loadPullRequests === undefined) return;
271 const arrivedRows = new Set(applied.threads);
272 const arrivedIds = new Set(source.threads.map((thread) => thread.id));
273 pendingLinkRows = new Set(applied.threads.filter((thread) => arrivedIds.has(thread.id)));
274 activeFill = yield* source.loadPullRequests.pipe(
275 Effect.flatMap((linksByThreadId) =>
276 writeFilledLinks(linksByThreadId, arrivedRows, arrivedIds),
277 ),
278 Effect.forkIn(stateScope),
279 );
280 });
281 const writeFilledLinks = Effect.fnUntraced(
282 function* (
283 linksByThreadId: ReadonlyMap<ThreadId, ReadonlyArray<ThreadPullRequestLink>>,
284 arrivedRows: ReadonlySet<OrchestrationV2ThreadShell>,
285 arrivedIds: ReadonlySet<ThreadId>,
286 ) {
287 const filled = yield* SubscriptionRef.modify(
288 state,
289 (current): [Option.Option<OrchestrationV2ShellSnapshot>, EnvironmentShellState] => {
290 if (Option.isNone(current.snapshot)) return [Option.none(), current];
291 const snapshot = current.snapshot.value;
292 let changed = false;
293 const threads = snapshot.threads.map((thread) => {
294 if (!arrivedRows.has(thread) || !arrivedIds.has(thread.id)) return thread;
295 const links = linksByThreadId.get(thread.id);
296 // A reused row may already hold these links from an earlier fill.
297 if (sameThreadPullRequests(thread.pullRequests, links)) return thread;
298 changed = true;
299 if (links === undefined) {
300 const { pullRequests: _previous, ...row } = thread;
301 return row;
302 }
303 return { ...thread, pullRequests: links };
304 });
305 if (!changed) return [Option.none(), current];
306 const next = { ...snapshot, threads };
307 return [Option.some(next), { ...current, snapshot: Option.some(next) }];
308 },
309 );
310 // Saves skipped while links were pending are queued again. Cached rows are never live:
311 // until a server snapshot arrives there is nothing new to save.
312 pendingLinkRows = new Set();
313 const live = yield* Ref.get(latestLiveSnapshot);
314 if (Option.isNone(live)) return;
315 const latest = Option.isSome(filled) ? filled : live;
316 yield* Ref.set(latestLiveSnapshot, latest);
317 yield* Queue.offer(persistence, latest.value);
318 },
319 (effect) => snapshotWriteLock.withPermit(effect),
320 );
321 if (Option.isSome(cached) && Option.isSome(cachedSnapshot)) {
322 yield* fillDeferredPullRequests(cached.value, cachedSnapshot.value);
323 }
218324
219 const foregroundResubscriptions = Option.match(wakeups, {325 const foregroundResubscriptions = Option.match(wakeups, {
220 onNone: () => Stream.never,326 onNone: () => Stream.never,
...@@ -258,9 +364,23 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -258,9 +364,23 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
258 );364 );
259 const httpSnapshot = yield* snapshotLoader.load(prepared);365 const httpSnapshot = yield* snapshotLoader.load(prepared);
260 if (Option.isSome(httpSnapshot)) {366 if (Option.isSome(httpSnapshot)) {
261 yield* applyItems([{ kind: "snapshot", snapshot: httpSnapshot.value }]);367 yield* cancelActiveFill;
368 const previous = yield* SubscriptionRef.get(state);
369 const snapshot = reuseUnchangedThreadShells(
370 Option.getOrNull(previous.snapshot),
371 withoutDeferredPullRequests(httpSnapshot.value),
372 );
373 const pendingLinkIds = new Set(
374 httpSnapshot.value.loadPullRequests === undefined
375 ? []
376 : httpSnapshot.value.threads.map((thread) => thread.id),
377 );
378 yield* applyItems([{ kind: "snapshot", snapshot }], pendingLinkIds);
262 canResume = true;379 canResume = true;
263 current = yield* SubscriptionRef.get(state);380 current = yield* SubscriptionRef.get(state);
381 if (Option.isSome(current.snapshot)) {
382 yield* fillDeferredPullRequests(httpSnapshot.value, current.snapshot.value);
383 }
264 }384 }
265 }385 }
266386
...@@ -288,7 +408,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")...@@ -288,7 +408,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
288 retryExpectedFailureAfter: "250 millis",408 retryExpectedFailureAfter: "250 millis",
289 resubscribe: foregroundResubscriptions,409 resubscribe: foregroundResubscriptions,
290 },410 },
291 ).pipe(Stream.runForEachArray(applyItems)),411 ).pipe(Stream.runForEachArray((items) => applyItems(items))),
292 );412 );
293 yield* SubscriptionRef.changes(supervisor.state).pipe(413 yield* SubscriptionRef.changes(supervisor.state).pipe(
294 Stream.runForEach((connectionState) => {414 Stream.runForEach((connectionState) => {
...@@ -449,5 +569,6 @@ export function createEnvironmentShellAtoms<R, E>(...@@ -449,5 +569,6 @@ export function createEnvironmentShellAtoms<R, E>(
449export * from "./models.ts";569export * from "./models.ts";
450export * from "./shellCommands.ts";570export * from "./shellCommands.ts";
451export * from "./shellReducer.ts";571export * from "./shellReducer.ts";
572export * from "./shellPullRequests.ts";
452export * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";573export * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
453export * from "./snapshots.ts";574export * from "./snapshots.ts";
packages/client-runtime/src/state/shellPullRequests.ts created+97
...@@ -0,0 +1,97 @@
1import {
2 OrchestrationV2ShellSnapshot,
3 type ThreadId,
4 ThreadPullRequestLink,
5} from "@t3tools/contracts";
6import * as Effect from "effect/Effect";
7import * as Predicate from "effect/Predicate";
8import * as Schema from "effect/Schema";
9
10/**
11 * A shell snapshot whose threads may arrive before their pull request links are decoded. Those
12 * links are most of a large shell's decode cost and only feed PR badges, so the thread list
13 * can paint first. `loadPullRequests` then decodes the links by thread id; until it runs,
14 * those threads carry no `pullRequests`.
15 */
16export interface DeferredShellSnapshot extends OrchestrationV2ShellSnapshot {
17 readonly loadPullRequests?: Effect.Effect<
18 ReadonlyMap<ThreadId, ReadonlyArray<ThreadPullRequestLink>>
19 >;
20}
21
22// A zero sleep is a microtask yield, which on React Native runs before the next frame.
23// A 1ms sleep takes a host timer, so the rows can render before the deferred decode.
24const yieldToHost = Effect.sleep("1 millis");
25
26const decodeThreadLinks = Schema.decodeUnknownEffect(
27 Schema.UndefinedOr(Schema.toCodecJson(Schema.Array(ThreadPullRequestLink))),
28);
29// Bounds each uninterrupted stretch of decoding, so taps and frames run between batches.
30const LINKS_PER_BATCH = 400;
31
32/**
33 * Removes each parsed thread row's pull request links from `snapshot` in place and returns
34 * them in row order, for `deferPullRequests` once the rows are decoded.
35 */
36export function detachPullRequests(snapshot: unknown): ReadonlyArray<unknown> {
37 if (!Predicate.isObject(snapshot)) return [];
38 const rows: unknown = snapshot.threads;
39 if (!Array.isArray(rows)) return [];
40 return rows.map((row: unknown) => {
41 if (!Predicate.isObject(row)) return undefined;
42 const links = row.pullRequests;
43 delete row.pullRequests;
44 return links;
45 });
46}
47
48/**
49 * Attaches a `loadPullRequests` that decodes the detached links after yielding to the host.
50 * A thread whose links cannot be read gets none; its row stays usable.
51 */
52export function deferPullRequests<S extends OrchestrationV2ShellSnapshot>(
53 snapshot: S,
54 rawLinks: ReadonlyArray<unknown>,
55): S & Required<Pick<DeferredShellSnapshot, "loadPullRequests">> {
56 const threadIds = snapshot.threads.map((thread) => thread.id);
57 // Schema decoding runs when called, so each thread's decode is built after a yield.
58 const loadPullRequests = Effect.gen(function* () {
59 const linksByThreadId = new Map<ThreadId, ReadonlyArray<ThreadPullRequestLink>>();
60 let batchSize = LINKS_PER_BATCH;
61 for (const [index, threadId] of threadIds.entries()) {
62 const raw = rawLinks[index];
63 if (raw === undefined) continue;
64 const size = Array.isArray(raw) ? raw.length : 1;
65 if (batchSize + size > LINKS_PER_BATCH) {
66 yield* yieldToHost;
67 batchSize = 0;
68 }
69 batchSize += size;
70 // An unreadable thread keeps no links; the others still fill in.
71 const links = yield* decodeThreadLinks(raw).pipe(
72 Effect.catch((cause) =>
73 Effect.logWarning("Discarding unreadable shell pull request links.", {
74 threadId,
75 cause: String(cause),
76 }).pipe(Effect.as(undefined)),
77 ),
78 );
79 if (links !== undefined) linksByThreadId.set(threadId, links);
80 }
81 return linksByThreadId as ReadonlyMap<ThreadId, ReadonlyArray<ThreadPullRequestLink>>;
82 });
83 return { ...snapshot, loadPullRequests };
84}
85
86const decodeShellSnapshotJson = Schema.decodeUnknownEffect(
87 Schema.toCodecJson(OrchestrationV2ShellSnapshot),
88);
89
90/** Decodes a JSON shell snapshot with its threads' pull request links deferred. */
91export const decodeShellSnapshotDeferringPullRequests = Effect.fnUntraced(function* (
92 json: unknown,
93) {
94 const rawLinks = detachPullRequests(json);
95 const snapshot = yield* decodeShellSnapshotJson(json);
96 return deferPullRequests(snapshot, rawLinks);
97});
packages/client-runtime/src/state/shellReducer.test.ts+42-2
...@@ -1,8 +1,12 @@...@@ -1,8 +1,12 @@
1import { ProjectId, ThreadId } from "@t3tools/contracts";1import { ProjectId, ThreadId, type ThreadPullRequestLink } from "@t3tools/contracts";
2import { describe, expect, it } from "vite-plus/test";2import { describe, expect, it } from "vite-plus/test";
33
4import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";4import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts";
5import { applyShellStreamEvent, mergeShellSnapshotProjects } from "./shellReducer.ts";5import {
6 applyShellStreamEvent,
7 mergeShellSnapshotProjects,
8 reuseUnchangedThreadShells,
9} from "./shellReducer.ts";
610
7const repositoryIdentity = {11const repositoryIdentity = {
8 canonicalKey: "github.com/example/repo",12 canonicalKey: "github.com/example/repo",
...@@ -463,3 +467,39 @@ describe("applyShellStreamEvent", () => {...@@ -463,3 +467,39 @@ describe("applyShellStreamEvent", () => {
463 expect(next).toBe(v2ShellSnapshot);467 expect(next).toBe(v2ShellSnapshot);
464 });468 });
465});469});
470
471describe("reuseUnchangedThreadShells", () => {
472 const link: ThreadPullRequestLink = {
473 host: "github.com",
474 repository: "pingdotgg/t3code",
475 number: 3,
476 url: "https://github.com/pingdotgg/t3code/pull/3",
477 source: "agent",
478 linkedAt: "2026-06-20T00:00:00.000Z",
479 snapshot: null,
480 stack: null,
481 };
482 const linked = { ...v2ThreadShell, pullRequests: [link] };
483 const other = { ...v2ThreadShell, id: ThreadId.make("thread-other"), title: "Other" };
484
485 it("keeps the previous object for an unchanged row, even without its deferred links", () => {
486 const previous = { ...v2ShellSnapshot, threads: [linked, other] };
487 const { pullRequests: _links, ...withoutLinks } = linked;
488 const next = reuseUnchangedThreadShells(previous, {
489 ...v2ShellSnapshot,
490 snapshotSequence: 5,
491 threads: [withoutLinks, { ...other, title: "Renamed" }],
492 });
493 expect(next.snapshotSequence).toBe(5);
494 expect(next.threads[0]).toBe(linked);
495 expect(next.threads[1]).not.toBe(other);
496 expect(next.threads[1]?.title).toBe("Renamed");
497 });
498
499 it("returns the incoming snapshot when no row is unchanged", () => {
500 const incoming = { ...v2ShellSnapshot, threads: [{ ...other, title: "Renamed" }] };
501 expect(reuseUnchangedThreadShells({ ...v2ShellSnapshot, threads: [other] }, incoming)).toBe(
502 incoming,
503 );
504 });
505});
packages/client-runtime/src/state/shellReducer.ts+36
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import {1import {
2 OrchestrationV2ThreadShell,2 OrchestrationV2ThreadShell,
3 ThreadPullRequestLink,
3 type OrchestrationProjectShell,4 type OrchestrationProjectShell,
4 type OrchestrationV2ShellSnapshot,5 type OrchestrationV2ShellSnapshot,
5 type OrchestrationV2ShellStreamItem,6 type OrchestrationV2ShellStreamItem,
...@@ -7,6 +8,41 @@ import {...@@ -7,6 +8,41 @@ import {
7import * as Schema from "effect/Schema";8import * as Schema from "effect/Schema";
89
9const sameThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell);10const sameThreadShell = Schema.toEquivalence(OrchestrationV2ThreadShell);
11const sameLinks = Schema.toEquivalence(Schema.UndefinedOr(Schema.Array(ThreadPullRequestLink)));
12
13/** Whether two rows' pull request links are equal, including both absent. */
14export function sameThreadPullRequests(
15 left: OrchestrationV2ThreadShell["pullRequests"],
16 right: OrchestrationV2ThreadShell["pullRequests"],
17): boolean {
18 return left === right || sameLinks(left, right);
19}
20
21/**
22 * Keep the previous object for each row a full snapshot left unchanged, so the list only
23 * re-renders rows that changed. Links are compared apart from the rest of the row because a
24 * deferred snapshot arrives without them; such a row keeps its links until they fill in.
25 */
26export function reuseUnchangedThreadShells(
27 previous: OrchestrationV2ShellSnapshot | null,
28 next: OrchestrationV2ShellSnapshot,
29): OrchestrationV2ShellSnapshot {
30 if (previous === null || previous.threads.length === 0) return next;
31 const previousById = new Map(previous.threads.map((thread) => [thread.id, thread] as const));
32 let reused = 0;
33 const threads = next.threads.map((thread) => {
34 const prior = previousById.get(thread.id);
35 if (prior === undefined) return thread;
36 const candidate =
37 thread.pullRequests === undefined && prior.pullRequests !== undefined
38 ? { ...thread, pullRequests: prior.pullRequests }
39 : thread;
40 if (!sameThreadShell(prior, candidate)) return thread;
41 reused += 1;
42 return prior;
43 });
44 return reused === 0 ? next : { ...next, threads };
45}
1046
11function upsertById<T extends { readonly id: unknown }>(47function upsertById<T extends { readonly id: unknown }>(
12 items: ReadonlyArray<T>,48 items: ReadonlyArray<T>,
packages/client-runtime/src/state/shellSnapshotHttp.ts+45-6
...@@ -1,10 +1,11 @@...@@ -1,10 +1,11 @@
1import type { OrchestrationV2ShellSnapshot } from "@t3tools/contracts";1import { EnvironmentHttpCommonError } from "@t3tools/contracts";
2import * as Cause from "effect/Cause";2import * as Cause from "effect/Cause";
3import * as Context from "effect/Context";3import * as Context from "effect/Context";
4import * as Effect from "effect/Effect";4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";5import * as Layer from "effect/Layer";
6import * as Option from "effect/Option";6import * as Option from "effect/Option";
7import { HttpClient } from "effect/http";7import * as Schema from "effect/Schema";
8import { HttpClient, HttpClientError, type HttpClientResponse } from "effect/http";
89
9import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";10import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
10import type { PreparedConnection } from "../connection/model.ts";11import type { PreparedConnection } from "../connection/model.ts";
...@@ -14,6 +15,10 @@ import {...@@ -14,6 +15,10 @@ import {
14 executeAuthenticatedEnvironmentHttpRequest,15 executeAuthenticatedEnvironmentHttpRequest,
15 withOrchestrationProtocolHeader,16 withOrchestrationProtocolHeader,
16} from "./environmentHttpAuth.ts";17} from "./environmentHttpAuth.ts";
18import {
19 decodeShellSnapshotDeferringPullRequests,
20 type DeferredShellSnapshot,
21} from "./shellPullRequests.ts";
1722
18// Long enough for a slow but alive server to finish. On timeout the socket asks23// Long enough for a slow but alive server to finish. On timeout the socket asks
19// the same server for the same full snapshot, so a short deadline only throws24// the same server for the same full snapshot, so a short deadline only throws
...@@ -45,11 +50,45 @@ export const fetchEnvironmentShellSnapshot = Effect.fn(...@@ -45,11 +50,45 @@ export const fetchEnvironmentShellSnapshot = Effect.fn(
45 method: "GET",50 method: "GET",
46 url: (httpBaseUrl) => environmentEndpointUrl(httpBaseUrl, "/api/orchestration/shell"),51 url: (httpBaseUrl) => environmentEndpointUrl(httpBaseUrl, "/api/orchestration/shell"),
47 timeoutMs: input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS,52 timeoutMs: input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS,
53 // The body is decoded here so each thread's pull request links can decode after its
54 // rows. Every declared error is an `EnvironmentHttpCommonError`, so decoding one keeps
55 // the error the credential retry checks for.
48 request: ({ client, headers }) =>56 request: ({ client, headers }) =>
49 client.shellSnapshot({ headers: withOrchestrationProtocolHeader(headers) }),57 client
58 .shellSnapshot({
59 headers: withOrchestrationProtocolHeader(headers),
60 responseMode: "response-only",
61 })
62 .pipe(Effect.flatMap(decodeShellSnapshotResponse)),
50 });63 });
51});64});
5265
66const decodeEnvironmentHttpCommonError = Schema.decodeUnknownEffect(
67 Schema.toCodecJson(EnvironmentHttpCommonError),
68);
69
70const decodeShellSnapshotResponse = (
71 response: HttpClientResponse.HttpClientResponse,
72): Effect.Effect<
73 DeferredShellSnapshot,
74 HttpClientError.HttpClientError | Schema.SchemaError | EnvironmentHttpCommonError
75> => {
76 if (response.status === 200) {
77 return response.json.pipe(Effect.flatMap(decodeShellSnapshotDeferringPullRequests));
78 }
79 // A body that is not a declared error keeps its status, as the generated decoder reported.
80 const statusError = new HttpClientError.HttpClientError({
81 reason: new HttpClientError.StatusCodeError({ request: response.request, response }),
82 });
83 return response.json.pipe(
84 Effect.flatMap(decodeEnvironmentHttpCommonError),
85 Effect.matchEffect({
86 onFailure: () => Effect.fail(statusError),
87 onSuccess: Effect.fail,
88 }),
89 );
90};
91
53/**92/**
54 * Loads the environment shell snapshot over HTTP, returning `Option.none()` when93 * Loads the environment shell snapshot over HTTP, returning `Option.none()` when
55 * it cannot be loaded (so the caller falls back to the socket-embedded snapshot).94 * it cannot be loaded (so the caller falls back to the socket-embedded snapshot).
...@@ -61,7 +100,7 @@ export class ShellSnapshotLoader extends Context.Service<...@@ -61,7 +100,7 @@ export class ShellSnapshotLoader extends Context.Service<
61 {100 {
62 readonly load: (101 readonly load: (
63 prepared: PreparedConnection,102 prepared: PreparedConnection,
64 ) => Effect.Effect<Option.Option<OrchestrationV2ShellSnapshot>>;103 ) => Effect.Effect<Option.Option<DeferredShellSnapshot>>;
65 }104 }
66>()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {}105>()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {}
67106
...@@ -78,14 +117,14 @@ export const layer: Layer.Layer<ShellSnapshotLoader, never, HttpClient.HttpClien...@@ -78,14 +117,14 @@ export const layer: Layer.Layer<ShellSnapshotLoader, never, HttpClient.HttpClien
78 return ShellSnapshotLoader.of({117 return ShellSnapshotLoader.of({
79 load: (prepared: PreparedConnection) =>118 load: (prepared: PreparedConnection) =>
80 fetchEnvironmentShellSnapshot({ prepared, signer, remoteAuthorization }).pipe(119 fetchEnvironmentShellSnapshot({ prepared, signer, remoteAuthorization }).pipe(
81 Effect.map(Option.some<OrchestrationV2ShellSnapshot>),120 Effect.map(Option.some<DeferredShellSnapshot>),
82 Effect.provideService(HttpClient.HttpClient, httpClient),121 Effect.provideService(HttpClient.HttpClient, httpClient),
83 Effect.catchCause((cause) =>122 Effect.catchCause((cause) =>
84 Effect.logWarning(123 Effect.logWarning(
85 "Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.",124 "Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.",
86 ).pipe(125 ).pipe(
87 Effect.annotateLogs({ cause: Cause.pretty(cause) }),126 Effect.annotateLogs({ cause: Cause.pretty(cause) }),
88 Effect.as(Option.none<OrchestrationV2ShellSnapshot>()),127 Effect.as(Option.none<DeferredShellSnapshot>()),
89 ),128 ),
90 ),129 ),
91 ),130 ),