Commit 482d56233dc3c27f614c843e0f6cc02b3f648a5f | the-knife

apps/mobile/src/connection/runtime.ts+6-1

...	...	@@ -1,4 +1,6 @@
1	1	import { Connection } from "@t3tools/client-runtime/connection";
	2	import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell";
	3	import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads";
2	4	import * as Layer from "effect/Layer";
3	5	import { Atom } from "effect/unstable/reactivity";
4	6
...	...	@@ -9,12 +11,15 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe(
9	11	Layer.provide(runtimeContextLayer),
10	12	);
11	13
	14	const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer);
	15
12	16	type ConnectionLayerSource =
13	17	\| typeof Connection.layer
	18	\| typeof snapshotLoaderLayer
14	19	\| typeof runtimeContextLayer
15	20	\| typeof connectionPlatformLayer;
16	21
17		const connectionLayer = Connection.layer.pipe(
	22	const connectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe(
18	23	Layer.provideMerge(Layer.mergeAll(runtimeContextLayer, providedConnectionPlatformLayer)),
19	24	);
20	25

apps/mobile/src/connection/storage.ts+11-8

...	...	@@ -16,8 +16,8 @@ import {
16	16	} from "@t3tools/client-runtime/connection";
17	17	import {
18	18	EnvironmentId,
19		OrchestrationThread,
20	19	OrchestrationShellSnapshot,
	20	OrchestrationThreadDetailSnapshot,
21	21	ThreadId,
22	22	} from "@t3tools/contracts";
23	23	import * as Context from "effect/Context";
...	...	@@ -32,7 +32,10 @@ import { makeCatalogStore, type SecureCatalogStorage } from "./catalog-store";
32	32	const SHELL_SNAPSHOT_CACHE_SCHEMA_VERSION = 1;
33	33	const SHELL_SNAPSHOT_CACHE_DIRECTORY = "connection-shell-snapshots";
34	34	const LEGACY_SHELL_SNAPSHOT_CACHE_DIRECTORY = "shell-snapshots";
35		const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 1;
	35	// v2 stores the snapshot sequence alongside the thread so a warm cache can
	36	// resume via `afterSequence` instead of re-downloading the full thread body.
	37	// Older v1 entries (no sequence) fail to decode and are treated as a cold cache.
	38	const THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION = 2;
36	39	const THREAD_SNAPSHOT_CACHE_DIRECTORY = "connection-thread-snapshots";
37	40
38	41	const StoredShellSnapshot = Schema.Struct({
...	...	@@ -45,7 +48,7 @@ const StoredThreadSnapshot = Schema.Struct({
45	48	schemaVersion: Schema.Literal(THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION),
46	49	environmentId: EnvironmentId,
47	50	threadId: ThreadId,
48		thread: OrchestrationThread,
	51	snapshot: OrchestrationThreadDetailSnapshot,
49	52	});
50	53
51	54	const LegacyStoredShellSnapshot = Schema.Struct({
...	...	@@ -355,18 +358,18 @@ export const connectionStorageLayer = Layer.effectContext(
355	358	Schema.decodeUnknownResult(StoredThreadSnapshot)(parsed),
356	359	).pipe(Effect.mapError((cause) => shellPersistenceError("load-thread", cause)));
357	360	return stored.environmentId === environmentId && stored.threadId === threadId
358		? Option.some(stored.thread)
	361	? Option.some(stored.snapshot)
359	362	: Option.none();
360	363	}),
361		saveThread: (environmentId, thread) =>
	364	saveThread: (environmentId, snapshot) =>
362	365	Effect.gen(function* () {
363		const file = yield* threadSnapshotFile(environmentId, thread.id, "save-thread");
	366	const file = yield* threadSnapshotFile(environmentId, snapshot.thread.id, "save-thread");
364	367	const encoded = yield* Effect.fromResult(
365	368	Schema.encodeUnknownResult(StoredThreadSnapshot)({
366	369	schemaVersion: THREAD_SNAPSHOT_CACHE_SCHEMA_VERSION,
367	370	environmentId,
368		threadId: thread.id,
369		thread,
	371	threadId: snapshot.thread.id,
	372	snapshot,
370	373	}),
371	374	).pipe(Effect.mapError((cause) => shellPersistenceError("save-thread", cause)));
372	375	yield* Effect.try({

apps/server/src/auth/http.ts+10

...	...	@@ -16,6 +16,8 @@ import {
16	16	EnvironmentOperationForbiddenError,
17	17	EnvironmentRequestInvalidError,
18	18	type EnvironmentRequestInvalidReason,
	19	EnvironmentResourceNotFoundError,
	20	type EnvironmentResourceNotFoundReason,
19	21	EnvironmentScopeRequiredError,
20	22	EnvironmentAuthenticatedAuth,
21	23	EnvironmentAuthenticatedPrincipal,
...	...	@@ -137,6 +139,14 @@ function failEnvironmentOperationForbidden(reason: "current_session_revoke_not_a
137	139	);
138	140	}
139	141
	142	export function failEnvironmentNotFound(reason: EnvironmentResourceNotFoundReason) {
	143	return currentEnvironmentTraceId.pipe(
	144	Effect.flatMap((traceId) =>
	145	Effect.fail(new EnvironmentResourceNotFoundError({ code: "not_found", reason, traceId })),
	146	),
	147	);
	148	}
	149
140	150	export function failEnvironmentInternal(reason: EnvironmentInternalErrorReason, error?: unknown) {
141	151	return Effect.gen(function* () {
142	152	const traceId = yield* currentEnvironmentTraceId;

apps/server/src/checkpointing/CheckpointDiffQuery.test.ts+5

...	...	@@ -107,6 +107,7 @@ describe("CheckpointDiffQuery.layer", () => {
107	107	}),
108	108	getThreadShellById: () => Effect.succeed(Option.none()),
109	109	getThreadDetailById: () => Effect.succeed(Option.none()),
	110	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
110	111	}),
111	112	),
112	113	);
...	...	@@ -199,6 +200,7 @@ describe("CheckpointDiffQuery.layer", () => {
199	200	getFullThreadDiffContext: () => Effect.die("unused"),
200	201	getThreadShellById: () => Effect.succeed(Option.none()),
201	202	getThreadDetailById: () => Effect.succeed(Option.none()),
	203	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
202	204	}),
203	205	),
204	206	);
...	...	@@ -281,6 +283,7 @@ describe("CheckpointDiffQuery.layer", () => {
281	283	getFullThreadDiffContext: () => Effect.die("unused"),
282	284	getThreadShellById: () => Effect.succeed(Option.none()),
283	285	getThreadDetailById: () => Effect.succeed(Option.none()),
	286	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
284	287	}),
285	288	),
286	289	);
...	...	@@ -348,6 +351,7 @@ describe("CheckpointDiffQuery.layer", () => {
348	351	getFullThreadDiffContext: () => Effect.die("unused"),
349	352	getThreadShellById: () => Effect.succeed(Option.none()),
350	353	getThreadDetailById: () => Effect.succeed(Option.none()),
	354	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
351	355	}),
352	356	),
353	357	);
...	...	@@ -400,6 +404,7 @@ describe("CheckpointDiffQuery.layer", () => {
400	404	getFullThreadDiffContext: () => Effect.succeed(Option.none()),
401	405	getThreadShellById: () => Effect.succeed(Option.none()),
402	406	getThreadDetailById: () => Effect.succeed(Option.none()),
	407	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
403	408	}),
404	409	),
405	410	);

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+1

...	...	@@ -200,6 +200,7 @@ describe("OrchestrationEngine", () => {
200	200	getFullThreadDiffContext: () => Effect.succeed(Option.none()),
201	201	getThreadShellById: () => Effect.succeed(Option.none()),
202	202	getThreadDetailById: () => Effect.succeed(Option.none()),
	203	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
203	204	}),
204	205	),
205	206	Layer.provide(

apps/server/src/orchestration/Layers/OrchestrationEngine.ts+2-2

...	...	@@ -306,8 +306,8 @@ const makeOrchestrationEngine = Effect.gen(function* () {
306	306	Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }),
307	307	);
308	308
309		const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) =>
310		eventStore.readFromSequence(fromSequenceExclusive);
	309	const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
	310	eventStore.readFromSequence(fromSequenceExclusive, limit);
311	311
312	312	const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
313	313	Effect.gen(function* () {

apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts+31

...	...	@@ -9,6 +9,7 @@ import {
9	9	OrchestrationReadModel,
10	10	OrchestrationShellSnapshot,
11	11	OrchestrationThread,
	12	OrchestrationThreadDetailSnapshot,
12	13	ProjectScript,
13	14	TurnId,
14	15	type OrchestrationCheckpointSummary,
...	...	@@ -2033,6 +2034,35 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
2033	2034	);
2034	2035	});
2035	2036
	2037	const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = (
	2038	threadId,
	2039	) =>
	2040	// Read the thread detail and the snapshot sequence within a single
	2041	// transaction so the sequence is consistent with the returned state; a
	2042	// projector update landing between two separate reads could otherwise return
	2043	// a sequence ahead of the thread detail, causing the client to resume from
	2044	// too far and drop events.
	2045	sql
	2046	.withTransaction(
	2047	Effect.gen(function* () {
	2048	const thread = yield* getThreadDetailById(threadId);
	2049	if (Option.isNone(thread)) {
	2050	return Option.none<OrchestrationThreadDetailSnapshot>();
	2051	}
	2052	const { snapshotSequence } = yield* getSnapshotSequence();
	2053	return Option.some({ snapshotSequence, thread: thread.value });
	2054	}),
	2055	)
	2056	.pipe(
	2057	Effect.mapError((error) =>
	2058	isPersistenceError(error)
	2059	? error
	2060	: toPersistenceSqlError("ProjectionSnapshotQuery.getThreadDetailSnapshot:transaction")(
	2061	error,
	2062	),
	2063	),
	2064	);
	2065
2036	2066	return {
2037	2067	getCommandReadModel,
2038	2068	getSnapshot,
...	...	@@ -2047,6 +2077,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () {
2047	2077	getFullThreadDiffContext,
2048	2078	getThreadShellById,
2049	2079	getThreadDetailById,
	2080	getThreadDetailSnapshot,
2050	2081	} satisfies ProjectionSnapshotQueryShape;
2051	2082	});
2052	2083

apps/server/src/orchestration/Services/OrchestrationEngine.ts+5

...	...	@@ -26,10 +26,15 @@ export interface OrchestrationEngineShape {
26	26	* Replay persisted orchestration events from an exclusive sequence cursor.
27	27	*
28	28	* @param fromSequenceExclusive - Sequence cursor (exclusive).
	29	* @param limit - Maximum number of events to read. Defaults to the event
	30	* store's page-bounded default; pass a higher value when the caller must
	31	* read every event after the cursor (e.g. per-thread catch-up that filters
	32	* a small subset out of a potentially larger global range).
29	33	* @returns Stream containing ordered events.
30	34	*/
31	35	readonly readEvents: (
32	36	fromSequenceExclusive: number,
	37	limit?: number,
33	38	) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
34	39
35	40	/**

apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts+11

...	...	@@ -14,6 +14,7 @@ import type {
14	14	OrchestrationReadModel,
15	15	OrchestrationShellSnapshot,
16	16	OrchestrationThread,
	17	OrchestrationThreadDetailSnapshot,
17	18	OrchestrationThreadShell,
18	19	ProjectId,
19	20	ThreadId,
...	...	@@ -157,6 +158,16 @@ export interface ProjectionSnapshotQueryShape {
157	158	readonly getThreadDetailById: (
158	159	threadId: ThreadId,
159	160	) => Effect.Effect<Option.Option<OrchestrationThread>, ProjectionRepositoryError>;
	161
	162	/**
	163	* Read a single active thread detail together with the projection snapshot
	164	* sequence in one consistent transaction, so the returned `snapshotSequence`
	165	* exactly matches the state reflected in `thread` (no interleaving projector
	166	* update between the two reads).
	167	*/
	168	readonly getThreadDetailSnapshot: (
	169	threadId: ThreadId,
	170	) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>, ProjectionRepositoryError>;
160	171	}
161	172
162	173	/**

apps/server/src/orchestration/http.ts+34

...	...	@@ -4,6 +4,7 @@ import {
4	4	EnvironmentHttpApi,
5	5	} from "@t3tools/contracts";
6	6	import * as Effect from "effect/Effect";
	7	import * as Option from "effect/Option";
7	8	import * as HttpApiBuilder from "effect/unstable/httpapi/HttpApiBuilder";
8	9
9	10	import { normalizeDispatchCommand } from "./Normalizer.ts";
...	...	@@ -11,6 +12,7 @@ import {
11	12	annotateEnvironmentRequest,
12	13	failEnvironmentInternal,
13	14	failEnvironmentInvalidRequest,
	15	failEnvironmentNotFound,
14	16	requireEnvironmentScope,
15	17	} from "../auth/http.ts";
16	18	import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts";
...	...	@@ -38,6 +40,38 @@ export const orchestrationHttpApiLayer = HttpApiBuilder.group(
38	40	);
39	41	}),
40	42	)
	43	.handle(
	44	"shellSnapshot",
	45	Effect.fn("environment.orchestration.shellSnapshot")(function* (args) {
	46	yield* annotateEnvironmentRequest(args.endpoint.name);
	47	yield* requireEnvironmentScope(AuthOrchestrationReadScope);
	48	return yield* projectionSnapshotQuery
	49	.getShellSnapshot()
	50	.pipe(
	51	Effect.catch((cause) =>
	52	failEnvironmentInternal("orchestration_snapshot_failed", cause),
	53	),
	54	);
	55	}),
	56	)
	57	.handle(
	58	"threadSnapshot",
	59	Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
	60	yield* annotateEnvironmentRequest(args.endpoint.name);
	61	yield* requireEnvironmentScope(AuthOrchestrationReadScope);
	62	const snapshot = yield* projectionSnapshotQuery
	63	.getThreadDetailSnapshot(args.params.threadId)
	64	.pipe(
	65	Effect.catch((cause) =>
	66	failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
	67	),
	68	);
	69	if (Option.isNone(snapshot)) {
	70	return yield* failEnvironmentNotFound("thread_not_found");
	71	}
	72	return snapshot.value;
	73	}),
	74	)
41	75	.handle(
42	76	"dispatch",
43	77	Effect.fn("environment.orchestration.dispatch")(function* (args) {

apps/server/src/project/ProjectSetupScriptRunner.test.ts+1

...	...	@@ -43,6 +43,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) =>
43	43	getFullThreadDiffContext: () => Effect.die("unused"),
44	44	getThreadShellById: () => Effect.die("unused"),
45	45	getThreadDetailById: () => Effect.die("unused"),
	46	getThreadDetailSnapshot: () => Effect.die("unused"),
46	47	});
47	48
48	49	const makeTerminalManagerLayer = (

apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+1

...	...	@@ -209,6 +209,7 @@ describe("ProviderSessionReaper", () => {
209	209	: Option.none(),
210	210	),
211	211	getThreadDetailById: () => Effect.die("unused"),
	212	getThreadDetailSnapshot: () => Effect.die("unused"),
212	213	}),
213	214	),
214	215	Layer.provideMerge(NodeServices.layer),

apps/server/src/server.test.ts+1

...	...	@@ -702,6 +702,7 @@ const buildAppUnderTest = (options?: {
702	702	getProjectShellById: () => Effect.succeed(Option.none()),
703	703	getThreadShellById: () => Effect.succeed(Option.none()),
704	704	getThreadDetailById: () => Effect.succeed(Option.none()),
	705	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
705	706	getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }),
706	707	getActiveProjectByWorkspaceRoot: () => Effect.succeed(Option.none()),
707	708	getFirstActiveThreadIdByProjectId: () => Effect.succeed(Option.none()),

apps/server/src/serverRuntimeStartup.test.ts+4

...	...	@@ -96,6 +96,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa
96	96	getFullThreadDiffContext: () => Effect.succeed(Option.none()),
97	97	getThreadShellById: () => Effect.succeed(Option.none()),
98	98	getThreadDetailById: () => Effect.succeed(Option.none()),
	99	getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
99	100	}),
100	101	Effect.provideService(AnalyticsService.AnalyticsService, {
101	102	record: () => Effect.void,
...	...	@@ -158,6 +159,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa
158	159	getFullThreadDiffContext: () => Effect.succeed(Option.none()),
159	160	getThreadShellById: () => Effect.die("unused"),
160	161	getThreadDetailById: () => Effect.die("unused"),
	162	getThreadDetailSnapshot: () => Effect.die("unused"),
161	163	}),
162	164	Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
163	165	readEvents: () => Stream.empty,
...	...	@@ -200,6 +202,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when
200	202	getFullThreadDiffContext: () => Effect.succeed(Option.none()),
201	203	getThreadShellById: () => Effect.die("unused"),
202	204	getThreadDetailById: () => Effect.die("unused"),
	205	getThreadDetailSnapshot: () => Effect.die("unused"),
203	206	}),
204	207	Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
205	208	readEvents: () => Stream.empty,
...	...	@@ -248,6 +251,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa
248	251	getFullThreadDiffContext: () => Effect.succeed(Option.none()),
249	252	getThreadShellById: () => Effect.die("unused"),
250	253	getThreadDetailById: () => Effect.die("unused"),
	254	getThreadDetailSnapshot: () => Effect.die("unused"),
251	255	}),
252	256	Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
253	257	readEvents: () => Stream.empty,

apps/server/src/ws.ts+109-40

...	...	@@ -30,6 +30,8 @@ import {
30	30	OrchestrationDispatchCommandError,
31	31	type OrchestrationEvent,
32	32	type OrchestrationShellStreamEvent,
	33	type OrchestrationShellStreamItem,
	34	type OrchestrationThreadStreamItem,
33	35	OrchestrationGetFullThreadDiffError,
34	36	OrchestrationGetSnapshotError,
35	37	OrchestrationGetTurnDiffError,
...	...	@@ -1059,10 +1061,54 @@ const makeWsRpcLayer = (
1059	1061	),
1060	1062	{ "rpc.aggregate": "orchestration" },
1061	1063	),
1062		[ORCHESTRATION_WS_METHODS.subscribeShell]: (_input) =>
	1064	[ORCHESTRATION_WS_METHODS.subscribeShell]: (input) =>
1063	1065	observeRpcStreamEffect(
1064	1066	ORCHESTRATION_WS_METHODS.subscribeShell,
1065	1067	Effect.gen(function* () {
	1068	const liveStream = orchestrationEngine.streamDomainEvents.pipe(
	1069	Stream.mapEffect(toShellStreamEvent),
	1070	Stream.flatMap((event) =>
	1071	Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty,
	1072	),
	1073	);
	1074
	1075	// When the client already holds a shell snapshot (cached, or loaded
	1076	// over HTTP) it passes that snapshot's sequence, and we resume by
	1077	// replaying shell events after it instead of re-sending the whole
	1078	// projects/threads list over the socket. As in the thread path, the
	1079	// live subscription is attached (into a scope-bound buffer) before
	1080	// draining the catch-up replay so no event published during the
	1081	// replay window is lost; overlapping events are deduped by sequence
	1082	// on the client. The full range is read (not the store's default
	1083	// page limit) since the shell filter runs after reading.
	1084	if (input.afterSequence !== undefined) {
	1085	const afterSequence = input.afterSequence;
	1086	return Stream.unwrap(
	1087	Effect.gen(function* () {
	1088	const liveBuffer = yield* Queue.unbounded<OrchestrationShellStreamItem>();
	1089	yield* Effect.forkScoped(
	1090	liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
	1091	);
	1092	const catchUpStream = orchestrationEngine
	1093	.readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
	1094	.pipe(
	1095	Stream.mapEffect(toShellStreamEvent),
	1096	Stream.flatMap((event) =>
	1097	Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty,
	1098	),
	1099	Stream.mapError(
	1100	(cause) =>
	1101	new OrchestrationGetSnapshotError({
	1102	message: "Failed to replay orchestration shell events",
	1103	cause,
	1104	}),
	1105	),
	1106	);
	1107	return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer));
	1108	}),
	1109	);
	1110	}
	1111
1066	1112	const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe(
1067	1113	Effect.tapError((cause) =>
1068	1114	Effect.logError("orchestration shell snapshot load failed", { cause }),
...	...	@@ -1076,13 +1122,6 @@ const makeWsRpcLayer = (
1076	1122	),
1077	1123	);
1078	1124
1079		const liveStream = orchestrationEngine.streamDomainEvents.pipe(
1080		Stream.mapEffect(toShellStreamEvent),
1081		Stream.flatMap((event) =>
1082		Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty,
1083		),
1084		);
1085
1086	1125	return Stream.concat(
1087	1126	Stream.make({
1088	1127	kind: "snapshot" as const,
...	...	@@ -1114,8 +1153,65 @@ const makeWsRpcLayer = (
1114	1153	observeRpcStreamEffect(
1115	1154	ORCHESTRATION_WS_METHODS.subscribeThread,
1116	1155	Effect.gen(function* () {
1117		const [threadDetail, snapshotSequence] = yield* Effect.all([
1118		projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe(
	1156	const isThisThreadDetailEvent = (event: OrchestrationEvent) =>
	1157	event.aggregateKind === "thread" &&
	1158	event.aggregateId === input.threadId &&
	1159	isThreadDetailEvent(event);
	1160
	1161	const liveStream = orchestrationEngine.streamDomainEvents.pipe(
	1162	Stream.filter(isThisThreadDetailEvent),
	1163	Stream.map((event) => ({
	1164	kind: "event" as const,
	1165	event,
	1166	})),
	1167	);
	1168
	1169	// When the client already loaded the snapshot over HTTP it passes
	1170	// that snapshot's sequence, and we resume the live subscription by
	1171	// replaying persisted events after it instead of re-sending the
	1172	// (potentially multi-KB) snapshot frame over the socket.
	1173	//
	1174	// The live PubSub subscription must be attached before draining
	1175	// the catch-up replay, otherwise events published during the replay
	1176	// window are dropped (they are past the persisted tail the replay
	1177	// read, but the live stream is not yet subscribed). So fork the
	1178	// live stream into a buffer bound to this stream's scope, then emit
	1179	// catch-up followed by the buffered/ongoing live events. Overlapping
	1180	// events are deduped by sequence on the client.
	1181	//
	1182	// Read the full range after the cursor (not the store's default
	1183	// page-bounded limit): the range is normally tiny (a fresh HTTP
	1184	// snapshot sequence) and the per-thread filter runs after reading,
	1185	// so a global cap could otherwise omit this thread's events.
	1186	if (input.afterSequence !== undefined) {
	1187	const afterSequence = input.afterSequence;
	1188	return Stream.unwrap(
	1189	Effect.gen(function* () {
	1190	const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
	1191	yield* Effect.forkScoped(
	1192	liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
	1193	);
	1194	const catchUpStream = orchestrationEngine
	1195	.readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
	1196	.pipe(
	1197	Stream.filter(isThisThreadDetailEvent),
	1198	Stream.map((event) => ({ kind: "event" as const, event })),
	1199	Stream.mapError(
	1200	(cause) =>
	1201	new OrchestrationGetSnapshotError({
	1202	message: `Failed to replay thread ${input.threadId} events`,
	1203	cause,
	1204	}),
	1205	),
	1206	);
	1207	return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer));
	1208	}),
	1209	);
	1210	}
	1211
	1212	const snapshot = yield* projectionSnapshotQuery
	1213	.getThreadDetailSnapshot(input.threadId)
	1214	.pipe(
1119	1215	Effect.mapError(
1120	1216	(cause) =>
1121	1217	new OrchestrationGetSnapshotError({
...	...	@@ -1123,46 +1219,19 @@ const makeWsRpcLayer = (
1123	1219	cause,
1124	1220	}),
1125	1221	),
1126		),
1127		projectionSnapshotQuery.getSnapshotSequence().pipe(
1128		Effect.map(({ snapshotSequence }) => snapshotSequence),
1129		Effect.mapError(
1130		(cause) =>
1131		new OrchestrationGetSnapshotError({
1132		message: "Failed to load orchestration snapshot sequence",
1133		cause,
1134		}),
1135		),
1136		),
1137		]);
	1222	);
1138	1223
1139		if (Option.isNone(threadDetail)) {
	1224	if (Option.isNone(snapshot)) {
1140	1225	return yield* new OrchestrationGetSnapshotError({
1141	1226	message: `Thread ${input.threadId} was not found`,
1142	1227	cause: input.threadId,
1143	1228	});
1144	1229	}
1145	1230
1146		const liveStream = orchestrationEngine.streamDomainEvents.pipe(
1147		Stream.filter(
1148		(event) =>
1149		event.aggregateKind === "thread" &&
1150		event.aggregateId === input.threadId &&
1151		isThreadDetailEvent(event),
1152		),
1153		Stream.map((event) => ({
1154		kind: "event" as const,
1155		event,
1156		})),
1157		);
1158
1159	1231	return Stream.concat(
1160	1232	Stream.make({
1161	1233	kind: "snapshot" as const,
1162		snapshot: {
1163		snapshotSequence,
1164		thread: threadDetail.value,
1165		},
	1234	snapshot: snapshot.value,
1166	1235	}),
1167	1236	liveStream,
1168	1237	);

apps/web/src/connection/runtime.ts+6-1

...	...	@@ -1,4 +1,6 @@
1	1	import { Connection } from "@t3tools/client-runtime/connection";
	2	import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell";
	3	import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads";
2	4	import * as Layer from "effect/Layer";
3	5	import { Atom } from "effect/unstable/reactivity";
4	6
...	...	@@ -9,12 +11,15 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe(
9	11	Layer.provide(runtimeContextLayer),
10	12	);
11	13
	14	const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer);
	15
12	16	type ConnectionLayerSource =
13	17	\| typeof Connection.layer
	18	\| typeof snapshotLoaderLayer
14	19	\| typeof runtimeContextLayer
15	20	\| typeof connectionPlatformLayer;
16	21
17		const connectionLayer = Connection.layer.pipe(
	22	const connectionLayer = Layer.merge(Connection.layer, snapshotLoaderLayer).pipe(
18	23	Layer.provideMerge(Layer.mergeAll(runtimeContextLayer, providedConnectionPlatformLayer)),
19	24	);
20	25

apps/web/src/connection/storage.ts+12-9

...	...	@@ -20,7 +20,7 @@ import {
20	20	import {
21	21	EnvironmentId,
22	22	OrchestrationShellSnapshot,
23		OrchestrationThread,
	23	OrchestrationThreadDetailSnapshot,
24	24	ThreadId,
25	25	} from "@t3tools/contracts";
26	26	import * as Context from "effect/Context";
...	...	@@ -45,11 +45,14 @@ const StoredShellSnapshot = Schema.Struct({
45	45	snapshot: OrchestrationShellSnapshot,
46	46	});
47	47	const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot);
	48	// v2 stores the snapshot sequence alongside the thread so a warm cache can
	49	// resume via `afterSequence` instead of re-downloading the full thread body.
	50	// Older v1 entries (no sequence) fail to decode and are treated as a cold cache.
48	51	const StoredThreadSnapshot = Schema.Struct({
49		schemaVersion: Schema.Literal(1),
	52	schemaVersion: Schema.Literal(2),
50	53	environmentId: EnvironmentId,
51	54	threadId: ThreadId,
52		thread: OrchestrationThread,
	55	snapshot: OrchestrationThreadDetailSnapshot,
53	56	});
54	57	const StoredThreadSnapshotJson = Schema.fromJsonString(StoredThreadSnapshot);
55	58	const ConnectionCatalogDocumentJson = Schema.fromJsonString(ConnectionCatalogDocument);
...	...	@@ -473,7 +476,7 @@ export const connectionStorageLayer = Layer.effectContext(
473	476	Effect.mapError((cause) => persistenceError("load-thread", cause)),
474	477	Effect.map((stored) =>
475	478	stored.environmentId === environmentId && stored.threadId === threadId
476		? Option.some(stored.thread)
	479	? Option.some(stored.snapshot)
477	480	: Option.none(),
478	481	),
479	482	);
...	...	@@ -484,18 +487,18 @@ export const connectionStorageLayer = Layer.effectContext(
484	487	: persistenceError("load-thread", cause),
485	488	),
486	489	),
487		saveThread: (environmentId, thread) =>
	490	saveThread: (environmentId, snapshot) =>
488	491	Effect.gen(function* () {
489	492	const encoded = yield* encodeStoredThreadSnapshot({
490		schemaVersion: 1,
	493	schemaVersion: 2,
491	494	environmentId,
492		threadId: thread.id,
493		thread,
	495	threadId: snapshot.thread.id,
	496	snapshot,
494	497	}).pipe(Effect.mapError((cause) => persistenceError("save-thread", cause)));
495	498	yield* writeDatabaseValue(
496	499	database,
497	500	THREAD_STORE_NAME,
498		threadCacheKey(environmentId, thread.id),
	501	threadCacheKey(environmentId, snapshot.thread.id),
499	502	encoded,
500	503	);
501	504	}).pipe(

apps/web/src/environments/primary/auth.ts+2

...	...	@@ -221,6 +221,8 @@ function readEnvironmentHttpErrorStatus(error: EnvironmentHttpCommonErrorType):
221	221	case "EnvironmentScopeRequiredError":
222	222	case "EnvironmentOperationForbiddenError":
223	223	return 403;
	224	case "EnvironmentResourceNotFoundError":
	225	return 404;
224	226	case "EnvironmentInternalError":
225	227	return 500;
226	228	}

packages/client-runtime/src/connection/errors.ts+9

...	...	@@ -132,6 +132,15 @@ export function mapRemoteEnvironmentError(
132	132	detail: "The environment rejected the authentication request.",
133	133	traceId: error.traceId,
134	134	});
	135	case "EnvironmentResourceNotFoundError":
	136	// Not expected during connection authorization, but the shared request
	137	// error type now includes it (used by resource fetches like the thread
	138	// snapshot). Treat it as a configuration issue with the endpoint.
	139	return new ConnectionBlockedError({
	140	reason: "configuration",
	141	detail: "The environment endpoint could not be found.",
	142	traceId: error.traceId,
	143	});
135	144	case "RemoteEnvironmentAuthTimeoutError":
136	145	return new ConnectionTransientError({
137	146	reason: "timeout",

packages/client-runtime/src/platform/persistence.ts+6-3

...	...	@@ -1,7 +1,7 @@
1	1	import {
2	2	type EnvironmentId,
3		type OrchestrationThread,
4	3	type OrchestrationShellSnapshot,
	4	type OrchestrationThreadDetailSnapshot,
5	5	type ThreadId,
6	6	} from "@t3tools/contracts";
7	7	import * as Context from "effect/Context";
...	...	@@ -60,10 +60,13 @@ export class EnvironmentCacheStore extends Context.Service<
60	60	readonly loadThread: (
61	61	environmentId: EnvironmentId,
62	62	threadId: ThreadId,
63		) => Effect.Effect<Option.Option<OrchestrationThread>, ConnectionPersistenceError>;
	63	) => Effect.Effect<
	64	Option.Option<OrchestrationThreadDetailSnapshot>,
	65	ConnectionPersistenceError
	66	>;
64	67	readonly saveThread: (
65	68	environmentId: EnvironmentId,
66		thread: OrchestrationThread,
	69	snapshot: OrchestrationThreadDetailSnapshot,
67	70	) => Effect.Effect<void, ConnectionPersistenceError>;
68	71	readonly removeThread: (
69	72	environmentId: EnvironmentId,

packages/client-runtime/src/rpc/http.ts+2

...	...	@@ -5,6 +5,7 @@ import {
5	5	type EnvironmentInternalError,
6	6	type EnvironmentOperationForbiddenError,
7	7	type EnvironmentRequestInvalidError,
	8	type EnvironmentResourceNotFoundError,
8	9	type EnvironmentScopeRequiredError,
9	10	} from "@t3tools/contracts";
10	11	import { httpHeaderRedactionLayer } from "@t3tools/shared/httpObservability";
...	...	@@ -70,6 +71,7 @@ export type RemoteEnvironmentRequestError =
70	71	\| EnvironmentAuthInvalidError
71	72	\| EnvironmentScopeRequiredError
72	73	\| EnvironmentOperationForbiddenError
	74	\| EnvironmentResourceNotFoundError
73	75	\| EnvironmentInternalError
74	76	\| RemoteEnvironmentAuthFetchError
75	77	\| RemoteEnvironmentAuthInvalidJsonError

packages/client-runtime/src/state/environmentHttpAuth.ts created+73

...	...	@@ -0,0 +1,73 @@
	1	import * as Effect from "effect/Effect";
	2	import * as Option from "effect/Option";
	3	import { FetchHttpClient, type HttpMethod } from "effect/unstable/http";
	4
	5	import type { PreparedHttpAuthorization } from "../connection/model.ts";
	6	import type { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";
	7	import { RemoteEnvironmentAuthFetchError } from "../rpc/http.ts";
	8
	9	export interface EnvironmentHttpAuthHeaders {
	10	readonly authorization?: string;
	11	readonly dpop?: string;
	12	}
	13
	14	/**
	15	* Primary/local environments with no bearer or DPoP credential authenticate the
	16	* browser via a session cookie. A cross-origin `fetch` does not send cookies by
	17	* default, so those requests must opt into credentialed mode; bearer/DPoP
	18	* connections carry their credential in a header and need no cookies. Applied
	19	* per-request via `FetchHttpClient.RequestInit`, which the fetch client reads
	20	* from the fiber context at request time.
	21	*/
	22	export const withEnvironmentCredentials = <A, E, R>(
	23	authorization: PreparedHttpAuthorization \| null,
	24	request: Effect.Effect<A, E, R>,
	25	): Effect.Effect<A, E, R> =>
	26	authorization === null
	27	? request.pipe(Effect.provideService(FetchHttpClient.RequestInit, { credentials: "include" }))
	28	: request;
	29
	30	/**
	31	* Build the authorization headers for an authenticated environment HTTP
	32	* request, matching the credential the connection was prepared with:
	33	* - primary/local connections carry no credential,
	34	* - bearer connections send a static `Bearer` token,
	35	* - relay connections send a `DPoP` access token with a freshly signed proof
	36	* bound to this request's method and URL.
	37	*
	38	* The DPoP signer is passed in (not resolved from context) and is only required
	39	* for relay/DPoP connections, so bearer/primary connections work even when no
	40	* signer is available.
	41	*/
	42	export const buildEnvironmentAuthHeaders = (
	43	authorization: PreparedHttpAuthorization \| null,
	44	method: HttpMethod.HttpMethod,
	45	url: string,
	46	signer: Option.Option<ManagedRelayDpopSigner["Service"]>,
	47	): Effect.Effect<EnvironmentHttpAuthHeaders, RemoteEnvironmentAuthFetchError> =>
	48	Effect.gen(function* () {
	49	if (authorization === null) {
	50	return {};
	51	}
	52	if (authorization._tag === "Bearer") {
	53	return { authorization: `Bearer ${authorization.token}` };
	54	}
	55	if (Option.isNone(signer)) {
	56	return yield* new RemoteEnvironmentAuthFetchError({
	57	message: "No DPoP signer is available to authorize the environment request.",
	58	cause: authorization._tag,
	59	});
	60	}
	61	const proof = yield* signer.value
	62	.createProof({ method, url, accessToken: authorization.accessToken })
	63	.pipe(
	64	Effect.mapError(
	65	(cause) =>
	66	new RemoteEnvironmentAuthFetchError({
	67	message: "Could not create the environment request authorization proof.",
	68	cause,
	69	}),
	70	),
	71	);
	72	return { authorization: `DPoP ${authorization.accessToken}`, dpop: proof };
	73	});

packages/client-runtime/src/state/shell-sync.test.ts+78-2

...	...	@@ -20,7 +20,7 @@ import * as EnvironmentSupervisor from "../connection/supervisor.ts";
20	20	import * as Persistence from "../platform/persistence.ts";
21	21	import * as RpcSession from "../rpc/session.ts";
22	22	import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
23		import { makeEnvironmentShellState } from "./shell.ts";
	23	import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts";
24	24
25	25	const TARGET = new PrimaryConnectionTarget({
26	26	environmentId: EnvironmentId.make("environment-1"),
...	...	@@ -29,6 +29,15 @@ const TARGET = new PrimaryConnectionTarget({
29	29	wsBaseUrl: "wss://environment.example.test",
30	30	});
31	31
	32	const PREPARED: PreparedConnection = {
	33	environmentId: TARGET.environmentId,
	34	label: TARGET.label,
	35	httpBaseUrl: TARGET.httpBaseUrl,
	36	socketUrl: TARGET.wsBaseUrl,
	37	httpAuthorization: null,
	38	target: TARGET,
	39	};
	40
32	41	const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = {
33	42	snapshotSequence: 1,
34	43	projects: [],
...	...	@@ -61,7 +70,7 @@ describe("environment shell synchronization", () => {
61	70	target: TARGET,
62	71	state: supervisorState,
63	72	session: activeSession,
64		prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),
	73	prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
65	74	connect: Effect.void,
66	75	disconnect: Effect.void,
67	76	retryNow: Effect.void,
...	...	@@ -74,9 +83,15 @@ describe("environment shell synchronization", () => {
74	83	removeThread: () => Effect.void,
75	84	clear: () => Effect.void,
76	85	});
	86	// Cold cache with no HTTP snapshot available → falls back to the
	87	// socket-embedded snapshot.
	88	const snapshotLoader = ShellSnapshotLoader.of({
	89	load: () => Effect.succeed(Option.none()),
	90	});
77	91	const shellState = yield* makeEnvironmentShellState().pipe(
78	92	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
79	93	Effect.provideService(Persistence.EnvironmentCacheStore, cache),
	94	Effect.provideService(ShellSnapshotLoader, snapshotLoader),
80	95	);
81	96
82	97	yield* SubscriptionRef.set(supervisorState, {
...	...	@@ -117,4 +132,65 @@ describe("environment shell synchronization", () => {
117	132	expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT);
118	133	}),
119	134	);
	135
	136	it.effect("resumes a warm shell cache via afterSequence without an HTTP fetch", () =>
	137	Effect.gen(function* () {
	138	const cachedSnapshot: OrchestrationShellSnapshot = {
	139	snapshotSequence: 5,
	140	projects: [],
	141	threads: [],
	142	updatedAt: "2026-06-06T00:00:00.000Z",
	143	};
	144	const events = yield* Queue.unbounded<OrchestrationShellStreamItem>();
	145	const capturedAfterSequence = yield* SubscriptionRef.make<number \| undefined>(undefined);
	146	const loaderCalls = yield* SubscriptionRef.make(0);
	147	const client = {
	148	[ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) =>
	149	Stream.unwrap(
	150	SubscriptionRef.set(capturedAfterSequence, input.afterSequence).pipe(
	151	Effect.as(Stream.fromQueue(events)),
	152	),
	153	),
	154	} as unknown as WsRpcProtocolClient;
	155	const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);
	156	const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
	157	Option.some(session(client)),
	158	);
	159	const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
	160	target: TARGET,
	161	state: supervisorState,
	162	session: activeSession,
	163	prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
	164	connect: Effect.void,
	165	disconnect: Effect.void,
	166	retryNow: Effect.void,
	167	} satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
	168	const cache = Persistence.EnvironmentCacheStore.of({
	169	loadShell: () => Effect.succeed(Option.some(cachedSnapshot)),
	170	saveShell: () => Effect.void,
	171	loadThread: () => Effect.succeed(Option.none()),
	172	saveThread: () => Effect.void,
	173	removeThread: () => Effect.void,
	174	clear: () => Effect.void,
	175	});
	176	const snapshotLoader = ShellSnapshotLoader.of({
	177	load: () =>
	178	SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(Effect.as(Option.none())),
	179	});
	180	yield* makeEnvironmentShellState().pipe(
	181	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
	182	Effect.provideService(Persistence.EnvironmentCacheStore, cache),
	183	Effect.provideService(ShellSnapshotLoader, snapshotLoader),
	184	);
	185
	186	// Wait until the subscription is established from the warm cache.
	187	yield* SubscriptionRef.changes(capturedAfterSequence).pipe(
	188	Stream.filter((value) => value !== undefined),
	189	Stream.runHead,
	190	);
	191
	192	expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(5);
	193	expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0);
	194	}),
	195	);
120	196	});

packages/client-runtime/src/state/shell.ts+46-8

...	...	@@ -19,6 +19,7 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts";
19	19	import { safeErrorLogAttributes } from "../errors/safeLog.ts";
20	20	import { EnvironmentCacheStore } from "../platform/persistence.ts";
21	21	import { subscribe } from "../rpc/client.ts";
	22	import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";
22	23	import { applyShellStreamEvent } from "./shellReducer.ts";
23	24	import type { EnvironmentCatalogState } from "./connections.ts";
24	25	import { followStreamInEnvironment } from "./runtime.ts";
...	...	@@ -48,6 +49,7 @@ const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment d
48	49	export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {
49	50	const supervisor = yield* EnvironmentSupervisor;
50	51	const cache = yield* EnvironmentCacheStore;
	52	const snapshotLoader = yield* ShellSnapshotLoader;
51	53	const environmentId = supervisor.target.environmentId;
52	54	const cachedSnapshot = yield* cache.loadShell(environmentId).pipe(
53	55	Effect.catch((error) =>
...	...	@@ -147,13 +149,45 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
147	149	yield* Queue.offer(persistence, nextSnapshot);
148	150	});
149	151
150		yield* subscribe(
151		ORCHESTRATION_WS_METHODS.subscribeShell,
152		{},
153		{
154		onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
155		},
156		).pipe(Stream.runForEach(applyItem), Effect.forkScoped);
	152	yield* Effect.forkScoped(
	153	Effect.gen(function* () {
	154	// Establish the base shell snapshot to resume from, minimizing bytes over
	155	// the wire:
	156	// - Warm cache: reuse the cached snapshot (zero network) and resume via
	157	// `afterSequence` so we only receive shell events since the cached
	158	// sequence.
	159	// - Cold cache: load the full shell snapshot over HTTP (gzip-compressible,
	160	// and off the socket), then resume via `afterSequence`.
	161	// If no base can be established we fall back to the socket-embedded
	162	// snapshot so the shell still synchronizes. Overlapping/replayed events are
	163	// deduped by sequence in applyItem.
	164	const base = Option.isSome(cachedSnapshot)
	165	? cachedSnapshot
	166	: yield* Effect.gen(function* () {
	167	const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe(
	168	Stream.filter(Option.isSome),
	169	Stream.map((current) => current.value),
	170	Stream.runHead,
	171	);
	172	return Option.isSome(prepared)
	173	? yield* snapshotLoader.load(prepared.value)
	174	: Option.none<OrchestrationShellSnapshot>();
	175	});
	176
	177	if (Option.isSome(base)) {
	178	yield* applyItem({ kind: "snapshot", snapshot: base.value });
	179	}
	180
	181	const subscribeInput = Option.match(base, {
	182	onNone: () => ({}),
	183	onSome: (snapshot) => ({ afterSequence: snapshot.snapshotSequence }),
	184	});
	185
	186	yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, {
	187	onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
	188	}).pipe(Stream.runForEach(applyItem));
	189	}),
	190	);
157	191	yield* SubscriptionRef.changes(supervisor.state).pipe(
158	192	Stream.runForEach((connectionState) => {
159	193	switch (connectionProjectionPhase(connectionState)) {
...	...	@@ -293,7 +327,10 @@ export function createEnvironmentServerConfigsAtom(input: {
293	327	}
294	328
295	329	export function createEnvironmentShellAtoms<R, E>(
296		runtime: Atom.AtomRuntime<EnvironmentRegistry \| EnvironmentCacheStore \| R, E>,
	330	runtime: Atom.AtomRuntime<
	331	EnvironmentRegistry \| EnvironmentCacheStore \| ShellSnapshotLoader \| R,
	332	E
	333	>,
297	334	) {
298	335	const stateAtom = Atom.family((environmentId: EnvironmentId) =>
299	336	runtime.atom(shellStateChanges(environmentId), {
...	...	@@ -316,4 +353,5 @@ export function createEnvironmentShellAtoms<R, E>(
316	353	export * from "./models.ts";
317	354	export * from "./shellCommands.ts";
318	355	export * from "./shellReducer.ts";
	356	export * from "./shellSnapshotHttp.ts";
319	357	export * from "./snapshots.ts";

packages/client-runtime/src/state/shellSnapshotHttp.ts created+92

...	...	@@ -0,0 +1,92 @@
	1	import type { OrchestrationShellSnapshot } from "@t3tools/contracts";
	2	import * as Cause from "effect/Cause";
	3	import * as Context from "effect/Context";
	4	import * as Effect from "effect/Effect";
	5	import * as Layer from "effect/Layer";
	6	import * as Option from "effect/Option";
	7	import { HttpClient } from "effect/unstable/http";
	8
	9	import type { PreparedConnection } from "../connection/model.ts";
	10	import { environmentEndpointUrl } from "../environment/endpoint.ts";
	11	import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";
	12	import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiClient } from "../rpc/http.ts";
	13	import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts";
	14
	15	// Bounded so a pathologically slow endpoint cannot block the (cheaper) socket
	16	// fallback for long. The cached shell renders while this runs.
	17	const DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS = 6_000;
	18
	19	/**
	20	* Load the environment shell snapshot (projects + thread shells) over HTTP
	21	* instead of as the WebSocket subscription's first frame. The response is
	22	* gzip-compressible by the transport and keeps the (potentially large) list off
	23	* the socket.
	24	*/
	25	export const fetchEnvironmentShellSnapshot = Effect.fn(
	26	"clientRuntime.state.fetchEnvironmentShellSnapshot",
	27	)(function* (input: {
	28	readonly prepared: PreparedConnection;
	29	readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;
	30	readonly timeoutMs?: number;
	31	}) {
	32	const requestUrl = environmentEndpointUrl(input.prepared.httpBaseUrl, "/api/orchestration/shell");
	33	const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl);
	34	const headers = yield* buildEnvironmentAuthHeaders(
	35	input.prepared.httpAuthorization,
	36	"GET",
	37	requestUrl,
	38	input.signer,
	39	);
	40	return yield* executeEnvironmentHttpRequest(
	41	requestUrl,
	42	input.timeoutMs ?? DEFAULT_SHELL_SNAPSHOT_TIMEOUT_MS,
	43	withEnvironmentCredentials(
	44	input.prepared.httpAuthorization,
	45	client.orchestration.shellSnapshot({ headers }),
	46	),
	47	);
	48	});
	49
	50	/**
	51	* Loads the environment shell snapshot over HTTP, returning `Option.none()` when
	52	* it cannot be loaded (so the caller falls back to the socket-embedded snapshot).
	53	* Decouples the shell state machine from the underlying HTTP + DPoP details and
	54	* keeps them out of test contexts.
	55	*/
	56	export class ShellSnapshotLoader extends Context.Service<
	57	ShellSnapshotLoader,
	58	{
	59	readonly load: (
	60	prepared: PreparedConnection,
	61	) => Effect.Effect<Option.Option<OrchestrationShellSnapshot>>;
	62	}
	63	>()("@t3tools/client-runtime/state/shellSnapshotHttp/ShellSnapshotLoader") {}
	64
	65	export const shellSnapshotLoaderLayer: Layer.Layer<
	66	ShellSnapshotLoader,
	67	never,
	68	HttpClient.HttpClient
	69	> = Layer.effect(
	70	ShellSnapshotLoader,
	71	Effect.gen(function* () {
	72	const httpClient = yield* HttpClient.HttpClient;
	73	// Resolve the DPoP signer optionally: it is only needed for relay/DPoP
	74	// connections, so the loader must not hard-require it.
	75	const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner);
	76	return ShellSnapshotLoader.of({
	77	load: (prepared: PreparedConnection) =>
	78	fetchEnvironmentShellSnapshot({ prepared, signer }).pipe(
	79	Effect.map(Option.some<OrchestrationShellSnapshot>),
	80	Effect.provideService(HttpClient.HttpClient, httpClient),
	81	Effect.catchCause((cause) =>
	82	Effect.logWarning(
	83	"Could not load the environment shell snapshot over HTTP; using the socket snapshot instead.",
	84	).pipe(
	85	Effect.annotateLogs({ cause: Cause.pretty(cause) }),
	86	Effect.as(Option.none<OrchestrationShellSnapshot>()),
	87	),
	88	),
	89	),
	90	});
	91	}),
	92	);

packages/client-runtime/src/state/threadSnapshotHttp.ts created+120

...	...	@@ -0,0 +1,120 @@
	1	import type { OrchestrationThreadDetailSnapshot, ThreadId } from "@t3tools/contracts";
	2	import * as Cause from "effect/Cause";
	3	import * as Context from "effect/Context";
	4	import * as Effect from "effect/Effect";
	5	import * as Layer from "effect/Layer";
	6	import * as Option from "effect/Option";
	7	import { HttpClient } from "effect/unstable/http";
	8
	9	import type { PreparedConnection } from "../connection/model.ts";
	10	import { environmentEndpointUrl } from "../environment/endpoint.ts";
	11	import { ManagedRelayDpopSigner } from "../relay/managedRelay.ts";
	12	import {
	13	executeEnvironmentHttpRequest,
	14	makeEnvironmentHttpApiClient,
	15	type RemoteEnvironmentRequestError,
	16	} from "../rpc/http.ts";
	17	import { buildEnvironmentAuthHeaders, withEnvironmentCredentials } from "./environmentHttpAuth.ts";
	18
	19	// Bounded so a pathologically slow endpoint cannot block the (cheaper) socket
	20	// fallback for long. The cached thread renders while this runs, so the wait only
	21	// delays the transition to live data on the first open, not the initial paint.
	22	const DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000;
	23
	24	/**
	25	* Load a thread's detail snapshot over HTTP instead of embedding it in the
	26	* WebSocket subscription's first frame. The response is gzip-compressible by
	27	* the transport and keeps the (potentially multi-KB) snapshot off the socket.
	28	*/
	29	export const fetchEnvironmentThreadSnapshot = Effect.fn(
	30	"clientRuntime.state.fetchEnvironmentThreadSnapshot",
	31	)(function* (input: {
	32	readonly prepared: PreparedConnection;
	33	readonly threadId: ThreadId;
	34	readonly signer: Option.Option<ManagedRelayDpopSigner["Service"]>;
	35	readonly timeoutMs?: number;
	36	}) {
	37	const requestUrl = environmentEndpointUrl(
	38	input.prepared.httpBaseUrl,
	39	`/api/orchestration/threads/${input.threadId}`,
	40	);
	41	const client = yield* makeEnvironmentHttpApiClient(input.prepared.httpBaseUrl);
	42	const headers = yield* buildEnvironmentAuthHeaders(
	43	input.prepared.httpAuthorization,
	44	"GET",
	45	requestUrl,
	46	input.signer,
	47	);
	48	return yield* executeEnvironmentHttpRequest(
	49	requestUrl,
	50	input.timeoutMs ?? DEFAULT_THREAD_SNAPSHOT_TIMEOUT_MS,
	51	withEnvironmentCredentials(
	52	input.prepared.httpAuthorization,
	53	client.orchestration.threadSnapshot({
	54	params: { threadId: input.threadId },
	55	headers,
	56	}),
	57	),
	58	);
	59	});
	60
	61	export type FetchEnvironmentThreadSnapshotError = RemoteEnvironmentRequestError;
	62
	63	/**
	64	* Loads a thread's detail snapshot over HTTP, returning `Option.none()` when it
	65	* cannot be loaded (so the caller falls back to the socket-embedded snapshot).
	66	* Decouples the thread state machine from the underlying HTTP + DPoP details and
	67	* keeps them out of test contexts.
	68	*/
	69	export class ThreadSnapshotLoader extends Context.Service<
	70	ThreadSnapshotLoader,
	71	{
	72	readonly load: (
	73	prepared: PreparedConnection,
	74	threadId: ThreadId,
	75	) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>>;
	76	}
	77	>()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {}
	78
	79	export const threadSnapshotLoaderLayer: Layer.Layer<
	80	ThreadSnapshotLoader,
	81	never,
	82	HttpClient.HttpClient
	83	> = Layer.effect(
	84	ThreadSnapshotLoader,
	85	Effect.gen(function* () {
	86	const httpClient = yield* HttpClient.HttpClient;
	87	// Resolve the DPoP signer optionally: it is only needed for relay/DPoP
	88	// connections, so the loader must not hard-require it (bearer/primary
	89	// connections work without one).
	90	const signer = yield* Effect.serviceOption(ManagedRelayDpopSigner);
	91	return ThreadSnapshotLoader.of({
	92	load: (prepared: PreparedConnection, threadId: ThreadId) =>
	93	fetchEnvironmentThreadSnapshot({ prepared, threadId, signer }).pipe(
	94	Effect.map(Option.some<OrchestrationThreadDetailSnapshot>),
	95	Effect.provideService(HttpClient.HttpClient, httpClient),
	96	// A genuinely missing thread (404) is expected — the socket
	97	// subscription is the source of truth for thread existence and will
	98	// surface the deletion — so don't treat it as an error worth warning
	99	// about; just defer to the socket path.
	100	Effect.catchTags({
	101	EnvironmentResourceNotFoundError: () =>
	102	Effect.logDebug(
	103	"Thread snapshot not found over HTTP; deferring to the socket subscription.",
	104	).pipe(
	105	Effect.annotateLogs({ threadId }),
	106	Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
	107	),
	108	}),
	109	Effect.catchCause((cause) =>
	110	Effect.logWarning(
	111	"Could not load the thread snapshot over HTTP; using the socket snapshot instead.",
	112	).pipe(
	113	Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }),
	114	Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
	115	),
	116	),
	117	),
	118	});
	119	}),
	120	);

packages/client-runtime/src/state/threads-atoms.test.ts+2-2

...	...	@@ -6,12 +6,12 @@ import { Atom } from "effect/unstable/reactivity";
6	6	import type { EnvironmentRegistry } from "../connection/registry.ts";
7	7	import type { EnvironmentCacheStore } from "../platform/persistence.ts";
8	8	import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
9		import { createEnvironmentThreadStateAtoms } from "./threads.ts";
	9	import { createEnvironmentThreadStateAtoms, type ThreadSnapshotLoader } from "./threads.ts";
10	10
11	11	describe("createEnvironmentThreadStateAtoms", () => {
12	12	it("retains thread state across short subscriber gaps", () => {
13	13	const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime<
14		EnvironmentRegistry \| EnvironmentCacheStore,
	14	EnvironmentRegistry \| EnvironmentCacheStore \| ThreadSnapshotLoader,
15	15	never
16	16	>;
17	17	const threads = createEnvironmentThreadStateAtoms(runtime);

packages/client-runtime/src/state/threads-sync.test.ts+90-11

...	...	@@ -6,6 +6,7 @@ import {
6	6	ProviderInstanceId,
7	7	ThreadId,
8	8	type OrchestrationThread,
	9	type OrchestrationThreadDetailSnapshot,
9	10	type OrchestrationThreadStreamItem,
10	11	} from "@t3tools/contracts";
11	12	import { describe, expect, it } from "@effect/vitest";
...	...	@@ -30,6 +31,7 @@ import * as RpcSession from "../rpc/session.ts";
30	31	import {
31	32	EMPTY_ENVIRONMENT_THREAD_STATE,
32	33	makeEnvironmentThreadState,
	34	ThreadSnapshotLoader,
33	35	type EnvironmentThreadState,
34	36	} from "./threads.ts";
35	37
...	...	@@ -40,6 +42,15 @@ const TARGET = new PrimaryConnectionTarget({
40	42	wsBaseUrl: "wss://environment.example.test",
41	43	});
42	44	const THREAD_ID = ThreadId.make("thread-1");
	45	const CACHED_SNAPSHOT_SEQUENCE = 7;
	46	const PREPARED: PreparedConnection = {
	47	environmentId: TARGET.environmentId,
	48	label: TARGET.label,
	49	httpBaseUrl: TARGET.httpBaseUrl,
	50	socketUrl: TARGET.wsBaseUrl,
	51	httpAuthorization: null,
	52	target: TARGET,
	53	};
43	54	const BASE_THREAD: OrchestrationThread = {
44	55	id: THREAD_ID,
45	56	projectId: ProjectId.make("project-1"),
...	...	@@ -89,13 +100,16 @@ function awaitThreadState(
89	100
90	101	const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: {
91	102	readonly cached?: OrchestrationThread;
	103	readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;
92	104	}) {
93	105	const inputs = yield* Queue.unbounded<TestThreadInput>();
94	106	const observed = yield* Queue.unbounded<EnvironmentThreadState>();
95	107	const latest = yield* Ref.make<EnvironmentThreadState>(EMPTY_ENVIRONMENT_THREAD_STATE);
96	108	const retryCount = yield* Ref.make(0);
97	109	const subscriptionCount = yield* Ref.make(0);
98		const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThread>>([]);
	110	const loaderCalls = yield* Ref.make(0);
	111	const lastSubscribeAfterSequence = yield* Ref.make<number \| undefined>(undefined);
	112	const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationThreadDetailSnapshot>>([]);
99	113	const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
100	114	const supervisorState = yield* SubscriptionRef.make<SupervisorConnectionState>(
101	115	AVAILABLE_CONNECTION_STATE,
...	...	@@ -107,17 +121,30 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
107	121	),
108	122	);
109	123	const client = {
110		[ORCHESTRATION_WS_METHODS.subscribeThread]: () =>
	124	[ORCHESTRATION_WS_METHODS.subscribeThread]: (input: { readonly afterSequence?: number }) =>
111	125	Stream.unwrap(
112	126	Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(
113		Effect.map(() => streamFrom(inputs)),
	127	Effect.andThen(Ref.set(lastSubscribeAfterSequence, input.afterSequence)),
	128	Effect.as(streamFrom(inputs)),
114	129	),
115	130	),
116	131	} as unknown as WsRpcProtocolClient;
117	132	const supervisorSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(
118	133	Option.some(testSession(client)),
119	134	);
120		const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none());
	135	const prepared = yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(
	136	Option.some(PREPARED),
	137	);
	138	const snapshotLoader = ThreadSnapshotLoader.of({
	139	load: (_prepared, threadId) =>
	140	Ref.update(loaderCalls, (count) => count + 1).pipe(
	141	Effect.as(
	142	threadId === THREAD_ID
	143	? (options?.httpSnapshot ?? Option.none<OrchestrationThreadDetailSnapshot>())
	144	: Option.none<OrchestrationThreadDetailSnapshot>(),
	145	),
	146	),
	147	});
121	148	const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
122	149	target: TARGET,
123	150	state: supervisorState,
...	...	@@ -133,7 +160,10 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
133	160	loadThread: (_environmentId, threadId) =>
134	161	Effect.succeed(
135	162	threadId === THREAD_ID && options?.cached !== undefined
136		? Option.some(options.cached)
	163	? Option.some({
	164	snapshotSequence: CACHED_SNAPSHOT_SEQUENCE,
	165	thread: options.cached,
	166	})
137	167	: Option.none(),
138	168	),
139	169	saveThread: (_environmentId, thread) =>
...	...	@@ -145,6 +175,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
145	175	const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe(
146	176	Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
147	177	Effect.provideService(Persistence.EnvironmentCacheStore, cache),
	178	Effect.provideService(ThreadSnapshotLoader, snapshotLoader),
148	179	);
149	180	yield* SubscriptionRef.changes(threadState).pipe(
150	181	Stream.runForEach((state) =>
...	...	@@ -159,6 +190,8 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
159	190	latest,
160	191	retryCount,
161	192	subscriptionCount,
	193	loaderCalls,
	194	lastSubscribeAfterSequence,
162	195	supervisorState,
163	196	supervisorSession,
164	197	savedThreads,
...	...	@@ -217,19 +250,38 @@ const deleted = (): OrchestrationThreadStreamItem => ({
217	250	});
218	251
219	252	describe("EnvironmentThreads", () => {
220		it.effect("publishes cached data before a live snapshot arrives", () =>
	253	it.effect("publishes cached data immediately from a warm cache", () =>
221	254	Effect.gen(function* () {
222	255	const harness = yield* makeHarness({ cached: BASE_THREAD });
223		const state = yield* awaitThreadState(
224		harness.observed,
225		(value) => value.status === "cached" && Option.isSome(value.data),
226		);
	256	const state = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.data));
227	257
228	258	expect(Option.getOrThrow(state.data)).toEqual(BASE_THREAD);
229	259	expect(Option.isNone(state.error)).toBe(true);
230	260	}),
231	261	);
232	262
	263	it.effect("resumes a warm cache via afterSequence without an HTTP fetch", () =>
	264	Effect.gen(function* () {
	265	const harness = yield* makeHarness({ cached: BASE_THREAD });
	266
	267	// The warm cache reaches live from the cached data, and a live event
	268	// applies on top of it.
	269	yield* Queue.offer(harness.inputs, titleUpdated("Live title", CACHED_SNAPSHOT_SEQUENCE + 1));
	270	yield* awaitThreadState(
	271	harness.observed,
	272	(value) =>
	273	value.status === "live" &&
	274	Option.isSome(value.data) &&
	275	value.data.value.title === "Live title",
	276	);
	277
	278	// The subscription resumed from the cached sequence and never fetched the
	279	// full snapshot over HTTP.
	280	expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(CACHED_SNAPSHOT_SEQUENCE);
	281	expect(yield* Ref.get(harness.loaderCalls)).toBe(0);
	282	}),
	283	);
	284
233	285	it.effect("reduces live events and persists the latest thread", () =>
234	286	Effect.gen(function* () {
235	287	const harness = yield* makeHarness({ cached: BASE_THREAD });
...	...	@@ -247,7 +299,34 @@ describe("EnvironmentThreads", () => {
247	299	yield* Effect.yieldNow;
248	300
249	301	expect(Option.getOrThrow(state.data).title).toBe("Live title");
250		expect((yield* Ref.get(harness.savedThreads)).at(-1)?.title).toBe("Live title");
	302	expect((yield* Ref.get(harness.savedThreads)).at(-1)?.thread.title).toBe("Live title");
	303	expect((yield* Ref.get(harness.savedThreads)).at(-1)?.snapshotSequence).toBe(2);
	304	}),
	305	);
	306
	307	it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>
	308	Effect.gen(function* () {
	309	const httpThread: OrchestrationThread = { ...BASE_THREAD, title: "HTTP title" };
	310	const harness = yield* makeHarness({
	311	httpSnapshot: Option.some({ snapshotSequence: 1, thread: httpThread }),
	312	});
	313	// No socket snapshot is pushed; only a live event arrives over the socket.
	314	// It can only be applied if the HTTP snapshot already seeded the thread.
	315	yield* Queue.offer(harness.inputs, titleUpdated("Live title", 2));
	316
	317	const state = yield* awaitThreadState(
	318	harness.observed,
	319	(value) =>
	320	value.status === "live" &&
	321	Option.isSome(value.data) &&
	322	value.data.value.title === "Live title",
	323	);
	324
	325	expect(Option.getOrThrow(state.data).title).toBe("Live title");
	326	// Cold cache: the full snapshot was loaded over HTTP and the socket
	327	// resumed from that snapshot's sequence.
	328	expect(yield* Ref.get(harness.loaderCalls)).toBeGreaterThanOrEqual(1);
	329	expect(yield* Ref.get(harness.lastSubscribeAfterSequence)).toBe(1);
251	330	}),
252	331	);
253	332

packages/client-runtime/src/state/threads.ts+69-20

...	...	@@ -2,6 +2,7 @@ import {
2	2	ORCHESTRATION_WS_METHODS,
3	3	type EnvironmentId as EnvironmentIdType,
4	4	type OrchestrationThread,
	5	type OrchestrationThreadDetailSnapshot,
5	6	type OrchestrationThreadStreamItem,
6	7	type ThreadId as ThreadIdType,
7	8	} from "@t3tools/contracts";
...	...	@@ -18,6 +19,7 @@ import { connectionProjectionPhase } from "../connection/model.ts";
18	19	import { EnvironmentSupervisor } from "../connection/supervisor.ts";
19	20	import { EnvironmentCacheStore } from "../platform/persistence.ts";
20	21	import { subscribe } from "../rpc/client.ts";
	22	import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts";
21	23	import { parseThreadKey, threadKey } from "./entities.ts";
22	24	import { applyThreadDetailEvent } from "./threadReducer.ts";
23	25	import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts";
...	...	@@ -44,6 +46,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
44	46	) {
45	47	const supervisor = yield* EnvironmentSupervisor;
46	48	const cache = yield* EnvironmentCacheStore;
	49	const snapshotLoader = yield* ThreadSnapshotLoader;
47	50	const environmentId = supervisor.target.environmentId;
48	51	const cached = yield* cache.loadThread(environmentId, threadId).pipe(
49	52	Effect.catch((error) =>
...	...	@@ -53,22 +56,27 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
53	56	threadId,
54	57	error: error.message,
55	58	}),
56		Effect.as(Option.none<OrchestrationThread>()),
	59	Effect.as(Option.none<OrchestrationThreadDetailSnapshot>()),
57	60	),
58	61	),
59	62	);
	63	const cachedThread = Option.map(cached, (snapshot) => snapshot.thread);
60	64	const state = yield* SubscriptionRef.make<EnvironmentThreadState>({
61		data: cached,
62		status: statusWithoutLiveData(cached),
	65	data: cachedThread,
	66	status: statusWithoutLiveData(cachedThread),
63	67	error: Option.none(),
64	68	});
65		const lastSequence = yield* SubscriptionRef.make(0);
66		const persistence = yield* Queue.sliding<OrchestrationThread>(1);
	69	// Seed the resume cursor from the cached snapshot so a warm cache can catch up
	70	// via `afterSequence` instead of re-downloading the full thread body.
	71	const lastSequence = yield* SubscriptionRef.make(
	72	Option.match(cached, { onNone: () => 0, onSome: (snapshot) => snapshot.snapshotSequence }),
	73	);
	74	const persistence = yield* Queue.sliding<OrchestrationThreadDetailSnapshot>(1);
67	75
68	76	const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
69		thread: OrchestrationThread,
	77	snapshot: OrchestrationThreadDetailSnapshot,
70	78	) {
71		yield* cache.saveThread(environmentId, thread).pipe(
	79	yield* cache.saveThread(environmentId, snapshot).pipe(
72	80	Effect.catch((error) =>
73	81	Effect.logWarning("Could not persist the thread cache.").pipe(
74	82	Effect.annotateLogs({
...	...	@@ -120,7 +128,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
120	128	status: "live",
121	129	error: Option.none(),
122	130	});
123		yield* Queue.offer(persistence, thread);
	131	// Persist the thread together with the sequence it reflects so the next warm
	132	// cache can resume from exactly here.
	133	const snapshotSequence = yield* SubscriptionRef.get(lastSequence);
	134	yield* Queue.offer(persistence, { snapshotSequence, thread });
124	135	});
125	136
126	137	const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () {
...	...	@@ -187,21 +198,55 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
187	198	);
188	199
189	200	yield* setSynchronizing;
190		yield* subscribe(
191		ORCHESTRATION_WS_METHODS.subscribeThread,
192		{ threadId },
193		{
194		onExpectedFailure: setStreamError,
195		retryExpectedFailureAfter: "250 millis",
196		},
197		).pipe(Stream.runForEach(applyItem), Effect.forkScoped);
	201	yield* Effect.forkScoped(
	202	Effect.gen(function* () {
	203	// Establish the base snapshot to resume from, minimizing bytes over the
	204	// wire:
	205	// - Warm cache: reuse the cached snapshot (zero network) and resume via
	206	// `afterSequence` so we only receive events since the cached sequence.
	207	// - Cold cache: load the full snapshot over HTTP (gzip-compressible, and
	208	// off the socket), then resume via `afterSequence`.
	209	// If no base can be established we fall back to the socket-embedded
	210	// snapshot so the thread still synchronizes. Overlapping/replayed events
	211	// are deduped by sequence in applyItem.
	212	const base = Option.isSome(cached)
	213	? cached
	214	: yield* Effect.gen(function* () {
	215	// Cold cache only: wait for a prepared connection so we can
	216	// authenticate the HTTP request; this mirrors the socket path, which
	217	// likewise waits for a live session.
	218	const prepared = yield* SubscriptionRef.changes(supervisor.prepared).pipe(
	219	Stream.filter(Option.isSome),
	220	Stream.map((current) => current.value),
	221	Stream.runHead,
	222	);
	223	return Option.isSome(prepared)
	224	? yield* snapshotLoader.load(prepared.value, threadId)
	225	: Option.none<OrchestrationThreadDetailSnapshot>();
	226	});
	227
	228	if (Option.isSome(base)) {
	229	yield* applyItem({ kind: "snapshot", snapshot: base.value });
	230	}
	231
	232	const subscribeInput = Option.match(base, {
	233	onNone: () => ({ threadId }),
	234	onSome: (snapshot) => ({ threadId, afterSequence: snapshot.snapshotSequence }),
	235	});
	236
	237	yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeThread, subscribeInput, {
	238	onExpectedFailure: setStreamError,
	239	retryExpectedFailureAfter: "250 millis",
	240	}).pipe(Stream.runForEach(applyItem));
	241	}),
	242	);
198	243
199	244	yield* Effect.addFinalizer(() =>
200		SubscriptionRef.get(state).pipe(
201		Effect.flatMap((current) =>
	245	Effect.all([SubscriptionRef.get(state), SubscriptionRef.get(lastSequence)]).pipe(
	246	Effect.flatMap(([current, snapshotSequence]) =>
202	247	Option.match(current.data, {
203	248	onNone: () => Effect.void,
204		onSome: persist,
	249	onSome: (thread) => persist({ snapshotSequence, thread }),
205	250	}),
206	251	),
207	252	),
...	...	@@ -218,7 +263,10 @@ export function threadStateChanges(environmentId: EnvironmentIdType, threadId: T
218	263	}
219	264
220	265	export function createEnvironmentThreadStateAtoms<R, E>(
221		runtime: Atom.AtomRuntime<EnvironmentRegistry \| EnvironmentCacheStore \| R, E>,
	266	runtime: Atom.AtomRuntime<
	267	EnvironmentRegistry \| EnvironmentCacheStore \| ThreadSnapshotLoader \| R,
	268	E
	269	>,
222	270	) {
223	271	const family = Atom.family((key: string) => {
224	272	const { environmentId, threadId } = parseThreadKey(key);
...	...	@@ -240,6 +288,7 @@ export function createEnvironmentThreadStateAtoms<R, E>(
240	288
241	289	export * from "./archivedThreads.ts";
242	290	export * from "./checkpointDiff.ts";
	291	export * from "./threadSnapshotHttp.ts";
243	292	export * from "./composerPathSearch.ts";
244	293	export * from "./threadCommands.ts";
245	294	export * from "./threadDetail.ts";

packages/contracts/src/environmentHttp.ts+46-1

...	...	@@ -24,12 +24,14 @@ import {
24	24	AuthWebSocketTicketResult,
25	25	ServerAuthSessionMethod,
26	26	} from "./auth.ts";
27		import { AuthSessionId, TrimmedNonEmptyString } from "./baseSchemas.ts";
	27	import { AuthSessionId, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts";
28	28	import { ExecutionEnvironmentDescriptor } from "./environment.ts";
29	29	import {
30	30	ClientOrchestrationCommand,
31	31	DispatchResult,
32	32	OrchestrationReadModel,
	33	OrchestrationShellSnapshot,
	34	OrchestrationThreadDetailSnapshot,
33	35	} from "./orchestration.ts";
34	36	import {
35	37	RelayCloudEnvironmentHealthRequest,
...	...	@@ -80,6 +82,7 @@ export const EnvironmentInternalErrorReason = Schema.Literals([
80	82	"client_sessions_load_failed",
81	83	"client_session_revoke_failed",
82	84	"orchestration_snapshot_failed",
	85	"orchestration_thread_snapshot_failed",
83	86	"orchestration_dispatch_failed",
84	87	"internal_error",
85	88	]);
...	...	@@ -155,11 +158,29 @@ export class EnvironmentInternalError extends Schema.TaggedErrorClass<Environmen
155	158	}
156	159	}
157	160
	161	export const EnvironmentResourceNotFoundReason = Schema.Literals(["thread_not_found"]);
	162	export type EnvironmentResourceNotFoundReason = typeof EnvironmentResourceNotFoundReason.Type;
	163
	164	export class EnvironmentResourceNotFoundError extends Schema.TaggedErrorClass<EnvironmentResourceNotFoundError>()(
	165	"EnvironmentResourceNotFoundError",
	166	{
	167	code: Schema.Literal("not_found"),
	168	reason: EnvironmentResourceNotFoundReason,
	169	traceId: TrimmedNonEmptyString,
	170	},
	171	{ httpApiStatus: 404 },
	172	) {
	173	[HttpServerRespondable.symbol]() {
	174	return HttpServerResponse.schemaJson(EnvironmentResourceNotFoundError)(this, { status: 404 });
	175	}
	176	}
	177
158	178	export const EnvironmentHttpCommonError = Schema.Union([
159	179	EnvironmentRequestInvalidError,
160	180	EnvironmentAuthInvalidError,
161	181	EnvironmentScopeRequiredError,
162	182	EnvironmentOperationForbiddenError,
	183	EnvironmentResourceNotFoundError,
163	184	EnvironmentInternalError,
164	185	]);
165	186	export type EnvironmentHttpCommonError = typeof EnvironmentHttpCommonError.Type;
...	...	@@ -269,6 +290,11 @@ const EnvironmentOrchestrationSnapshotErrors = [
269	290	EnvironmentScopeRequiredError,
270	291	EnvironmentInternalError,
271	292	] as const;
	293	const EnvironmentOrchestrationThreadSnapshotErrors = [
	294	EnvironmentScopeRequiredError,
	295	EnvironmentResourceNotFoundError,
	296	EnvironmentInternalError,
	297	] as const;
272	298	const EnvironmentOrchestrationDispatchErrors = [
273	299	EnvironmentRequestInvalidError,
274	300	EnvironmentScopeRequiredError,
...	...	@@ -422,6 +448,10 @@ export class EnvironmentAuthHttpApi extends HttpApiGroup.make("auth")
422	448	}).middleware(EnvironmentAuthenticatedAuth),
423	449	) {}
424	450
	451	const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({
	452	threadId: ThreadId,
	453	});
	454
425	455	export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration")
426	456	.add(
427	457	HttpApiEndpoint.get("snapshot", "/api/orchestration/snapshot", {
...	...	@@ -430,6 +460,21 @@ export class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestr
430	460	error: EnvironmentOrchestrationSnapshotErrors,
431	461	}).middleware(EnvironmentAuthenticatedAuth),
432	462	)
	463	.add(
	464	HttpApiEndpoint.get("shellSnapshot", "/api/orchestration/shell", {
	465	headers: OptionalBearerHeaders,
	466	success: OrchestrationShellSnapshot,
	467	error: EnvironmentOrchestrationSnapshotErrors,
	468	}).middleware(EnvironmentAuthenticatedAuth),
	469	)
	470	.add(
	471	HttpApiEndpoint.get("threadSnapshot", "/api/orchestration/threads/:threadId", {
	472	headers: OptionalBearerHeaders,
	473	params: EnvironmentOrchestrationThreadSnapshotParams,
	474	success: OrchestrationThreadDetailSnapshot,
	475	error: EnvironmentOrchestrationThreadSnapshotErrors,
	476	}).middleware(EnvironmentAuthenticatedAuth),
	477	)
433	478	.add(
434	479	HttpApiEndpoint.post("dispatch", "/api/orchestration/dispatch", {
435	480	headers: OptionalBearerHeaders,

packages/contracts/src/orchestration.ts+22-1

...	...	@@ -451,8 +451,29 @@ export const OrchestrationShellStreamItem = Schema.Union([
451	451	]);
452	452	export type OrchestrationShellStreamItem = typeof OrchestrationShellStreamItem.Type;
453	453
	454	export const OrchestrationSubscribeShellInput = Schema.Struct({
	455	/**
	456	* When provided, the server skips the initial full shell snapshot and instead
	457	* replays shell events after this sequence before streaming live events.
	458	* Clients that already hold a cached (or HTTP-loaded) shell snapshot pass its
	459	* sequence here so the subscription resumes without re-sending the entire
	460	* projects/threads list (overlapping events are deduped by sequence on the
	461	* client).
	462	*/
	463	afterSequence: Schema.optionalKey(NonNegativeInt),
	464	});
	465	export type OrchestrationSubscribeShellInput = typeof OrchestrationSubscribeShellInput.Type;
	466
454	467	export const OrchestrationSubscribeThreadInput = Schema.Struct({
455	468	threadId: ThreadId,
	469	/**
	470	* When provided, the server skips the initial snapshot frame and instead
	471	* replays events after this sequence before streaming live events. Clients
	472	* that load the snapshot over HTTP pass the snapshot's sequence here so the
	473	* live subscription resumes without a gap (overlapping events are deduped by
	474	* sequence on the client).
	475	*/
	476	afterSequence: Schema.optionalKey(NonNegativeInt),
456	477	});
457	478	export type OrchestrationSubscribeThreadInput = typeof OrchestrationSubscribeThreadInput.Type;
458	479
...	...	@@ -1244,7 +1265,7 @@ export const OrchestrationRpcSchemas = {
1244	1265	output: OrchestrationThreadStreamItem,
1245	1266	},
1246	1267	subscribeShell: {
1247		input: Schema.Struct({}),
	1268	input: OrchestrationSubscribeShellInput,
1248	1269	output: OrchestrationShellStreamItem,
1249	1270	},
1250	1271	} as const;

author	Julius Marminge <julius0216@outlook.com> 2026-07-05 22:00:13-07:00
committer	GitHub <noreply@github.com> 2026-07-05 22:00:13-07:00
log	482d56233dc3c27f614c843e0f6cc02b3f648a5f
tree	1554867b625abf454049dd635e0b6402b624d5c8
parent	6009720844f053e048e18ca1abe4f429559ab9fc
signature	Signed by PGP key B5690EEEBB952194

Load thread snapshots over HTTP before live sync (#3719)