authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-07 22:30:47-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-07 22:30:47-07:00
log28f5516a9acfcfc1e203487e78869a406fd23b76
tree4c1b542a9af9bd8676a6bb8fb4d2f18bd7943c50
parent6053a7878d24e5fd596d2b417ed94964a4b9c1c6
signature Signed by PGP key B5690EEEBB952194

perf(mobile): omit duplicated turn items from bounded thread snapshots (#15385)

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

15 files changed, 1038 insertions(+), 16 deletions(-)

apps/server/src/orchestration-v2/ThreadStream.test.ts+103
......@@ -1,5 +1,6 @@
11import {
22 EventId,
3 OrchestrationV2ThreadStreamItem,
34 ThreadId,
45 TurnItemId,
56 type OrchestrationV2ProjectedTurnItem,
......@@ -8,6 +9,8 @@ import {
89} from "@t3tools/contracts";
910import { describe, expect, it } from "vite-plus/test";
1011import * as DateTime from "effect/DateTime";
12import * as Schema from "effect/Schema";
13import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
1114
1215import {
1316 buildBoundedThreadStreamSnapshot,
......@@ -262,3 +265,103 @@ describe("decideThreadResume", () => {
262265 ).toEqual({ mode: "replay", afterSequence: 9, throughSequence: 10 });
263266 });
264267});
268
269describe("compact bounded snapshots", () => {
270 const FORK_PARENT = ThreadId.make("thread-stream-parent");
271 const decodeStreamItem = Schema.decodeUnknownSync(
272 Schema.toCodecJson(OrchestrationV2ThreadStreamItem),
273 );
274 const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem));
275
276 /** Fork child: inherited parent rows and marker, then local rows, plus an older interrupt request. */
277 function forkProjection(): OrchestrationV2ThreadProjection {
278 const base = timelineProjection(6);
279 const local = base.visibleTurnItems.slice(3).map((row, index) => ({ ...row, position: index }));
280 const inherited = base.visibleTurnItems.slice(0, 3).map((row, index) => {
281 const item = { ...row.item, id: TurnItemId.make(`parent-${index}`), threadId: FORK_PARENT };
282 return {
283 position: index,
284 visibility: "inherited" as const,
285 sourceThreadId: FORK_PARENT,
286 sourceItemId: item.id,
287 item,
288 };
289 });
290 const request = {
291 ...base.turnItems[0]!,
292 id: TurnItemId.make("interrupt-request"),
293 type: "run_interrupt_request",
294 ordinal: 0,
295 message: "Stop",
296 } as unknown as OrchestrationV2TurnItem;
297 return {
298 ...base,
299 turnItems: [request, ...local.map((row) => row.item)],
300 visibleTurnItems: [...inherited, ...local].map((row, position) => ({ ...row, position })),
301 };
302 }
303
304 it.each([
305 ["linear", () => timelineProjection(80)],
306 ["fork with retained interrupt request", forkProjection],
307 ])("restores the exact opted-out snapshot for %s threads", (_, makeProjection) => {
308 const base = makeProjection();
309 // The shared fixture omits a model; the wire codec requires one.
310 const projection = {
311 ...base,
312 thread: { ...base.thread, modelSelection: { instanceId: "codex", model: "gpt-5" } },
313 } as OrchestrationV2ThreadProjection;
314 const full = buildBoundedThreadStreamSnapshot({ snapshotSequence: 44, projection });
315 const compact = buildBoundedThreadStreamSnapshot({
316 snapshotSequence: 44,
317 projection,
318 compactTurnItems: true,
319 });
320
321 // Opted-out clients get exactly the previous representation.
322 expect("turnItemsOmitLocalVisible" in full).toBe(false);
323 expect(compact.turnItemsOmitLocalVisible).toBe(true);
324 expect(compact.projection.turnItems.length).toBeLessThan(full.projection.turnItems.length);
325 const { turnItemsOmitLocalVisible: _marker, ...compactFields } = compact;
326 expect({ ...compactFields, projection: full.projection }).toEqual(full);
327
328 // Decode the wire form like a client, then restore before use.
329 const decodedFull = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(full))));
330 const decodedCompact = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(compact))));
331 if (decodedFull.kind !== "snapshot" || decodedCompact.kind !== "snapshot") {
332 throw new Error("Expected snapshots");
333 }
334 expect(boundedSnapshotProjection(decodedCompact)).toEqual(decodedFull.projection);
335 });
336
337 it("keeps the retained interrupt request outside the visible window", () => {
338 const compact = buildBoundedThreadStreamSnapshot({
339 snapshotSequence: 1,
340 projection: forkProjection(),
341 compactTurnItems: true,
342 });
343 expect(compact.projection.turnItems.map((item) => item.id)).toEqual(["interrupt-request"]);
344 expect(boundedSnapshotProjection(compact).turnItems.map((item) => String(item.id))).toEqual([
345 "item-3",
346 "item-4",
347 "item-5",
348 "interrupt-request",
349 ]);
350 });
351
352 it("omits the marker when the window has no local rows", () => {
353 const fork = forkProjection();
354 const inheritedOnly = {
355 ...fork,
356 turnItems: fork.turnItems.slice(0, 1),
357 visibleTurnItems: fork.visibleTurnItems.slice(0, 3),
358 };
359 const compact = buildBoundedThreadStreamSnapshot({
360 snapshotSequence: 1,
361 projection: inheritedOnly,
362 compactTurnItems: true,
363 });
364 expect("turnItemsOmitLocalVisible" in compact).toBe(false);
365 expect(compact.projection.turnItems.map((item) => item.id)).toEqual(["interrupt-request"]);
366 });
367});
apps/server/src/orchestration-v2/ThreadStream.ts+34-6
......@@ -3,7 +3,12 @@ import type {
33 OrchestrationV2ThreadStreamItem,
44} from "@t3tools/contracts";
55
6import { buildBoundedThreadProjection } from "./threadHistoryPaging.ts";
6import { omitLocalVisibleTurnItems } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
7
8import {
9 buildBoundedThreadProjection,
10 type BoundedProjectionResult,
11} from "./threadHistoryPaging.ts";
712import { projectThreadProjectionForWire } from "./WireProjection.ts";
813
914/** Maximum number of reducer applications allowed during a thread resume. */
......@@ -39,10 +44,34 @@ type ThreadSnapshotStreamItem = Extract<
3944 { readonly kind: "snapshot" }
4045>;
4146
47/**
48 * Bounded snapshot fields shared by the HTTP route and socket fallbacks. Only
49 * clients that opted in get compact `turnItems` and the marker; everyone else
50 * gets the unchanged representation.
51 */
52export function boundedSnapshotResponseFields(input: {
53 readonly bounded: BoundedProjectionResult;
54 readonly compactTurnItems: boolean;
55}) {
56 const { bounded } = input;
57 const compactProjection = input.compactTurnItems
58 ? omitLocalVisibleTurnItems(bounded.projection)
59 : null;
60 return {
61 projection: compactProjection ?? bounded.projection,
62 historyCursor: bounded.historyCursor,
63 hasMoreHistory: bounded.hasMoreHistory,
64 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
65 payloadBudgetExceeded: bounded.payloadBudgetExceeded,
66 ...(compactProjection === null ? {} : { turnItemsOmitLocalVisible: true as const }),
67 };
68}
69
4270/** Build the same bounded, wire-projected snapshot for every socket fallback path. */
4371export function buildBoundedThreadStreamSnapshot(input: {
4472 readonly snapshotSequence: number;
4573 readonly projection: OrchestrationV2ThreadProjection;
74 readonly compactTurnItems?: boolean;
4675}): ThreadSnapshotStreamItem {
4776 const bounded = buildBoundedThreadProjection({
4877 snapshotSequence: input.snapshotSequence,
......@@ -51,11 +80,10 @@ export function buildBoundedThreadStreamSnapshot(input: {
5180 return {
5281 kind: "snapshot",
5382 snapshotSequence: input.snapshotSequence,
54 projection: bounded.projection,
55 historyCursor: bounded.historyCursor,
56 hasMoreHistory: bounded.hasMoreHistory,
57 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
58 payloadBudgetExceeded: bounded.payloadBudgetExceeded,
83 ...boundedSnapshotResponseFields({
84 bounded,
85 compactTurnItems: input.compactTurnItems === true,
86 }),
5987 };
6088}
6189
apps/server/src/orchestration-v2/boundedSnapshotTransport.test.ts created+370
......@@ -0,0 +1,370 @@
1import { assert, describe, it } from "@effect/vitest";
2import {
3 AuthOrchestrationReadScope,
4 AuthSessionId,
5 EnvironmentAuthenticatedAuth,
6 EnvironmentAuthenticatedPrincipal,
7 EnvironmentHttpApi,
8 EventId,
9 MessageId,
10 NodeId,
11 ORCHESTRATION_PROTOCOL_HEADER,
12 ORCHESTRATION_PROTOCOL_VERSION,
13 OrchestrationV2ThreadBoundedSnapshot,
14 OrchestrationV2ThreadStreamItem,
15 ProjectId,
16 ProviderDriverKind,
17 ProviderInstanceId,
18 RunId,
19 ThreadId,
20 TurnItemId,
21 type OrchestrationV2DomainEvent,
22 type OrchestrationV2ThreadProjection,
23 type OrchestrationV2TurnItem,
24} from "@t3tools/contracts";
25import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
26import * as NodeHttpPlatform from "@effect/platform-node/NodeHttpPlatform";
27import * as NodeServices from "@effect/platform-node/NodeServices";
28import * as DateTime from "effect/DateTime";
29import * as Effect from "effect/Effect";
30import * as Layer from "effect/Layer";
31import * as Schema from "effect/Schema";
32import * as Stream from "effect/Stream";
33import * as HttpApi from "effect/http-api/HttpApi";
34import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
35import { Etag, HttpRouter } from "effect/http";
36
37import * as SqlitePersistence from "../persistence/Sqlite.ts";
38import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
39import * as ProjectEnrichmentService from "../project/ProjectEnrichmentService.ts";
40import { subscribeOrchestrationV2Thread } from "../ws.ts";
41import * as OrchestrationHttp from "./http.ts";
42import * as ProjectionStore from "./ProjectionStore.ts";
43import * as ProjectStore from "./ProjectStore.ts";
44import * as ThreadManagementService from "./ThreadManagementService.ts";
45
46const decodeBounded = Schema.decodeUnknownSync(
47 Schema.fromJsonString(Schema.toCodecJson(OrchestrationV2ThreadBoundedSnapshot)),
48);
49const decodeStreamItem = Schema.decodeUnknownSync(
50 Schema.toCodecJson(OrchestrationV2ThreadStreamItem),
51);
52const encodeStreamItem = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadStreamItem));
53
54const provider = ProviderInstanceId.make("codex");
55const modelSelection = { instanceId: provider, model: "gpt-5.4" };
56const projectId = ProjectId.make("project:compact-transport");
57const PARENT = ThreadId.make("thread:compact-transport:parent");
58const FORK = ThreadId.make("thread:compact-transport:fork");
59const LONG = ThreadId.make("thread:compact-transport:long");
60const parentRun = RunId.make("run:compact-transport:parent");
61const longRun = RunId.make("run:compact-transport:long");
62
63const store = Layer.mergeAll(
64 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
65 SqlitePersistence.layerMemory,
66);
67const management = Layer.unwrap(
68 Effect.gen(function* () {
69 const projections = yield* ProjectionStore.ProjectionStoreV2;
70 return Layer.mock(ThreadManagementService.ThreadManagementService)({
71 ensureLegacyTranscript: () => Effect.void,
72 getThreadSnapshot: (id) => projections.getThreadSnapshot(id).pipe(Effect.orDie),
73 getThreadSnapshotWindow: (id, options) =>
74 projections.getThreadSnapshotWindow(id, options).pipe(Effect.orDie),
75 streamStoredEventsFrom: () => Stream.empty,
76 });
77 }),
78);
79
80class OrchestrationApi extends HttpApi.make("environment").add(
81 EnvironmentHttpApi.groups.orchestration,
82) {}
83const auth = Layer.succeed(EnvironmentAuthenticatedAuth, (effect) =>
84 effect.pipe(
85 Effect.provideService(EnvironmentAuthenticatedPrincipal, {
86 sessionId: AuthSessionId.make("compact-transport"),
87 subject: "compact-transport",
88 method: "browser-session-cookie",
89 scopes: new Set([AuthOrchestrationReadScope]),
90 }),
91 ),
92);
93
94function threadCreated(id: ThreadId, at: DateTime.Utc, forkOf?: RunId): OrchestrationV2DomainEvent {
95 return {
96 id: EventId.make(`event:${id}:created`),
97 type: "thread.created",
98 threadId: id,
99 occurredAt: at,
100 payload: {
101 createdBy: "user",
102 creationSource: "web",
103 id,
104 projectId,
105 title: String(id),
106 providerInstanceId: provider,
107 modelSelection,
108 runtimeMode: "full-access",
109 interactionMode: "default",
110 branch: null,
111 worktreePath: null,
112 activeProviderThreadId: null,
113 lineage:
114 forkOf === undefined
115 ? { parentThreadId: null, relationshipToParent: null, rootThreadId: id }
116 : { parentThreadId: PARENT, relationshipToParent: "fork", rootThreadId: PARENT },
117 forkedFrom: forkOf === undefined ? null : { type: "run", threadId: PARENT, runId: forkOf },
118 createdAt: at,
119 updatedAt: at,
120 archivedAt: null,
121 settledOverride: null,
122 settledAt: null,
123 lastVisitedAt: null,
124 deletedAt: null,
125 },
126 };
127}
128
129function runCreated(
130 threadId: ThreadId,
131 runId: RunId,
132 at: DateTime.Utc,
133): OrchestrationV2DomainEvent {
134 const rootNodeId = NodeId.make(`node:${runId}`);
135 return {
136 id: EventId.make(`event:${runId}:created`),
137 type: "run.created",
138 threadId,
139 runId,
140 nodeId: rootNodeId,
141 driver: ProviderDriverKind.make("codex"),
142 occurredAt: at,
143 payload: {
144 id: runId,
145 threadId,
146 ordinal: 1,
147 providerInstanceId: provider,
148 modelSelection,
149 providerThreadId: null,
150 userMessageId: MessageId.make(`message:${runId}`),
151 rootNodeId,
152 activeAttemptId: null,
153 status: "completed",
154 requestedAt: at,
155 startedAt: at,
156 completedAt: at,
157 checkpointId: null,
158 contextHandoffId: null,
159 },
160 };
161}
162
163function itemUpdated(
164 threadId: ThreadId,
165 runId: RunId | null,
166 key: string,
167 ordinal: number,
168 at: DateTime.Utc,
169 type:
170 | "command_execution"
171 | "run_interrupt_request"
172 | "run_interrupt_result" = "command_execution",
173): OrchestrationV2DomainEvent {
174 const base = {
175 id: TurnItemId.make(`turn-item:${threadId}:${key}`),
176 threadId,
177 runId,
178 nodeId: null,
179 providerThreadId: null,
180 providerTurnId: null,
181 nativeItemRef: null,
182 parentItemId: null,
183 ordinal,
184 status: "completed" as const,
185 title: key,
186 startedAt: at,
187 completedAt: at,
188 updatedAt: at,
189 };
190 const payload: OrchestrationV2TurnItem =
191 type === "command_execution"
192 ? { ...base, type, input: `echo ${key}`, output: `output ${key}`, exitCode: 0 }
193 : { ...base, type, message: key };
194 return {
195 id: EventId.make(`event:${threadId}:${key}`),
196 type: "turn-item.updated",
197 threadId,
198 ...(runId === null ? {} : { runId }),
199 occurredAt: at,
200 payload,
201 };
202}
203
204const seed = Effect.gen(function* () {
205 const projections = yield* ProjectionStore.ProjectionStoreV2;
206 const at = yield* DateTime.now;
207 const events: OrchestrationV2DomainEvent[] = [
208 threadCreated(PARENT, at),
209 runCreated(PARENT, parentRun, at),
210 ...[1, 2, 3].map((n) => itemUpdated(PARENT, parentRun, `parent-${n}`, n, at)),
211 threadCreated(FORK, at, parentRun),
212 ...[1, 2, 3, 4].map((n) => itemUpdated(FORK, null, `fork-${n}`, n, at)),
213 threadCreated(LONG, at),
214 runCreated(LONG, longRun, at),
215 // The request sits far outside the recent window; its result is inside it.
216 itemUpdated(LONG, longRun, "request", 1, at, "run_interrupt_request"),
217 ...Array.from({ length: 90 }, (_, index) =>
218 itemUpdated(LONG, longRun, `tool-${index}`, index + 2, at),
219 ),
220 itemUpdated(LONG, longRun, "result", 92, at, "run_interrupt_result"),
221 ];
222 for (const event of events) yield* projections.apply(event);
223});
224
225// Bounded subscriptions without afterSequence never read the event store, and
226// thread routes never touch projects.
227const TestLayer = Layer.mergeAll(
228 management,
229 Layer.effectDiscard(seed),
230 Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}),
231 Layer.mock(ProjectStore.ProjectStoreV2)({}),
232 Layer.mock(ProjectEnrichmentService.ProjectEnrichmentService)({}),
233).pipe(Layer.provideMerge(store));
234
235const withHttp = <A>(use: (get: (path: string) => Promise<Response>) => Promise<A>) =>
236 Effect.gen(function* () {
237 const context = yield* Effect.context<Layer.Success<typeof TestLayer>>();
238 const appLayer = HttpApiBuilder.layer(OrchestrationApi).pipe(
239 Layer.provide(OrchestrationHttp.layer),
240 Layer.provide(auth),
241 Layer.provide(Layer.succeedContext(context)),
242 Layer.provide(NodeHttpPlatform.layer),
243 Layer.provide(Etag.layerWeak),
244 Layer.provide(NodeServices.layer),
245 );
246 return yield* Effect.acquireUseRelease(
247 Effect.sync(() => HttpRouter.toWebHandler(appLayer, { disableLogger: true })),
248 ({ handler }) =>
249 Effect.promise(() =>
250 use((path) =>
251 handler(
252 new Request(`http://environment.test${path}`, {
253 headers: {
254 [ORCHESTRATION_PROTOCOL_HEADER]: String(ORCHESTRATION_PROTOCOL_VERSION),
255 },
256 }),
257 ),
258 ),
259 ),
260 ({ dispose }) => Effect.promise(() => dispose()),
261 );
262 });
263
264const firstSnapshot = (input: Parameters<typeof subscribeOrchestrationV2Thread>[0]) =>
265 subscribeOrchestrationV2Thread(input).pipe(
266 Effect.flatMap((stream) => Stream.runHead(stream)),
267 Effect.map((head) => {
268 if (head._tag !== "Some" || head.value.kind !== "snapshot") throw new Error("No snapshot");
269 // Through the wire codec, as a client receives it.
270 const decoded = decodeStreamItem(JSON.parse(JSON.stringify(encodeStreamItem(head.value))));
271 if (decoded.kind !== "snapshot") throw new Error("No snapshot");
272 return { raw: head.value, decoded };
273 }),
274 );
275
276function itemIds(projection: OrchestrationV2ThreadProjection): string[] {
277 return projection.turnItems.map((item) => String(item.id));
278}
279
280it.layer(TestLayer)("compact bounded snapshot transport", (it) => {
281 describe.each([
282 ["fork", FORK],
283 ["long run with a retained interrupt request", LONG],
284 ])("%s", (_, threadId) => {
285 it.effect("HTTP: legacy, ignored and opted-in requests restore to the same snapshot", () =>
286 withHttp(async (get) => {
287 const base = `/api/orchestration/threads/${encodeURIComponent(threadId)}/bounded`;
288 const legacyBody = await (await get(base)).text();
289 const ignoredBody = await (await get(`${base}?compactTurnItems=yes&unknown=1`)).text();
290 const compactResponse = await get(`${base}?compactTurnItems=1`);
291 assert.strictEqual(compactResponse.status, 200);
292 const compactBody = await compactResponse.text();
293
294 // Requests without the exact opt-in get the unchanged representation.
295 assert.notInclude(legacyBody, "turnItemsOmitLocalVisible");
296 assert.strictEqual(ignoredBody, legacyBody);
297 assert.isBelow(compactBody.length, legacyBody.length);
298
299 const legacy = decodeBounded(legacyBody);
300 const compact = decodeBounded(compactBody);
301 assert.isTrue(compact.turnItemsOmitLocalVisible);
302 const { turnItemsOmitLocalVisible: _marker, projection: _p, ...compactRest } = compact;
303 const { projection: _legacyProjection, ...legacyRest } = legacy;
304 assert.deepStrictEqual(compactRest, legacyRest);
305 assert.deepStrictEqual(boundedSnapshotProjection(compact), legacy.projection);
306 return { legacy, compact };
307 }).pipe(
308 Effect.tap(({ legacy }) =>
309 Effect.sync(() => {
310 if (threadId === FORK) {
311 const visibility = legacy.projection.visibleTurnItems.map((row) => row.visibility);
312 assert.deepStrictEqual(visibility, [
313 "inherited",
314 "inherited",
315 "inherited",
316 "synthetic",
317 "local",
318 "local",
319 "local",
320 "local",
321 ]);
322 } else {
323 const ids = itemIds(legacy.projection);
324 assert.isTrue(legacy.hasMoreHistory);
325 // Window rows first, then the retained request outside the window.
326 assert.strictEqual(ids.at(-1), `turn-item:${LONG}:request`);
327 assert.isFalse(
328 legacy.projection.visibleTurnItems.some((row) => row.sourceItemId === ids.at(-1)),
329 );
330 }
331 }),
332 ),
333 ),
334 );
335
336 it.effect("WS: bounded snapshots stay unchanged unless the client opts in", () =>
337 Effect.gen(function* () {
338 const legacy = yield* firstSnapshot({ threadId, acceptBoundedSnapshot: true });
339 const compact = yield* firstSnapshot({
340 threadId,
341 acceptBoundedSnapshot: true,
342 acceptCompactTurnItems: true,
343 });
344 // Full snapshots ignore the compact opt-in entirely.
345 const full = yield* firstSnapshot({ threadId, acceptCompactTurnItems: true });
346
347 assert.isUndefined(legacy.decoded.turnItemsOmitLocalVisible);
348 assert.isUndefined(full.decoded.turnItemsOmitLocalVisible);
349 assert.isUndefined(full.decoded.historyCursor);
350 assert.isTrue(compact.decoded.turnItemsOmitLocalVisible);
351 assert.isBelow(
352 compact.decoded.projection.turnItems.length,
353 legacy.decoded.projection.turnItems.length,
354 );
355 assert.deepStrictEqual(
356 boundedSnapshotProjection(compact.decoded),
357 legacy.decoded.projection,
358 );
359 assert.deepStrictEqual(
360 {
361 ...compact.decoded,
362 projection: legacy.decoded.projection,
363 turnItemsOmitLocalVisible: undefined,
364 },
365 { ...legacy.decoded, turnItemsOmitLocalVisible: undefined },
366 );
367 }),
368 );
369 });
370});
apps/server/src/orchestration-v2/http.ts+5-5
......@@ -32,6 +32,7 @@ import {
3232import * as ThreadManagementService from "./ThreadManagementService.ts";
3333import * as ProjectStore from "./ProjectStore.ts";
3434import { buildActiveShellSnapshot } from "./ShellStream.ts";
35import { boundedSnapshotResponseFields } from "./ThreadStream.ts";
3536import { projectThreadProjectionForWire } from "./WireProjection.ts";
3637
3738function isThreadNotFound(error: unknown): boolean {
......@@ -211,11 +212,10 @@ export const layer = HttpApiBuilder.group(
211212 });
212213 return {
213214 snapshotSequence: snapshot.snapshotSequence,
214 projection: bounded.projection,
215 historyCursor: bounded.historyCursor,
216 hasMoreHistory: bounded.hasMoreHistory,
217 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
218 payloadBudgetExceeded: bounded.payloadBudgetExceeded,
215 ...boundedSnapshotResponseFields({
216 bounded,
217 compactTurnItems: args.query.compactTurnItems === "1",
218 }),
219219 };
220220 }),
221221 )
apps/server/src/ws.ts+5-1
......@@ -748,6 +748,7 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs
748748 readonly afterSequence?: number;
749749 readonly requestCompletionMarker?: boolean;
750750 readonly acceptBoundedSnapshot?: boolean;
751 readonly acceptCompactTurnItems?: boolean;
751752 }) {
752753 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
753754 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
......@@ -841,7 +842,10 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs
841842 );
842843 const { snapshotSequence } = snapshot;
843844 const snapshotItem = useBoundedSnapshot
844 ? buildBoundedThreadStreamSnapshot(snapshot)
845 ? buildBoundedThreadStreamSnapshot({
846 ...snapshot,
847 compactTurnItems: input.acceptCompactTurnItems === true,
848 })
845849 : {
846850 kind: "snapshot" as const,
847851 snapshotSequence,
packages/client-runtime/src/state/boundedThreadSnapshotHttp.test.ts+77
......@@ -76,7 +76,84 @@ const FULL_SNAPSHOT_BODY = {
7676 },
7777};
7878
79const AT = "2026-06-20T00:00:00.000Z";
80const command = (id: string, ordinal: number) => ({
81 id,
82 type: "command_execution",
83 threadId: String(THREAD_ID),
84 runId: null,
85 nodeId: null,
86 providerThreadId: null,
87 providerTurnId: null,
88 nativeItemRef: null,
89 parentItemId: null,
90 ordinal,
91 status: "completed",
92 title: null,
93 input: id,
94 exitCode: 0,
95 startedAt: AT,
96 completedAt: AT,
97 updatedAt: AT,
98});
99const LOCAL_ITEMS = [command("local-1", 1), command("local-2", 2)];
100const DEPENDENCY = command("older-dependency", 0);
101const VISIBLE = LOCAL_ITEMS.map((item, position) => ({
102 position,
103 visibility: "local",
104 sourceThreadId: String(THREAD_ID),
105 sourceItemId: item.id,
106 item,
107}));
108const boundedBody = (compact: boolean) => ({
109 snapshotSequence: 12,
110 projection: {
111 ...FULL_SNAPSHOT_BODY.projection,
112 turnItems: compact ? [DEPENDENCY] : [...LOCAL_ITEMS, DEPENDENCY],
113 visibleTurnItems: VISIBLE,
114 },
115 historyCursor: "bounded-cursor",
116 hasMoreHistory: true,
117 latestLocalTurnOrdinal: 2,
118 payloadBudgetExceeded: false,
119 ...(compact ? { turnItemsOmitLocalVisible: true } : {}),
120});
121
79122describe("boundedThreadSnapshotLoader", () => {
123 it.effect.each([
124 ["an older server that ignores the opt-in", false],
125 ["a server that sends compact turnItems", true],
126 ] as const)("requests compact turnItems and loads %s", ([, compact]) => {
127 const urls: string[] = [];
128 const fetchFn = ((input: RequestInfo | URL) => {
129 urls.push(String(input));
130 return Promise.resolve(Response.json(boundedBody(compact)));
131 }) satisfies typeof fetch;
132
133 return Effect.gen(function* () {
134 const loader = yield* ThreadSnapshotLoader.ThreadSnapshotLoader;
135 const result = yield* loader.load(PREPARED, THREAD_ID);
136 expect(new URL(urls[0]!).searchParams.get("compactTurnItems")).toBe("1");
137 expect(result._tag).toBe("present");
138 if (result._tag !== "present") return;
139 expect(result.snapshot.projection.turnItems.map((item) => String(item.id))).toEqual([
140 "local-1",
141 "local-2",
142 "older-dependency",
143 ]);
144 expect("turnItemsOmitLocalVisible" in result.snapshot).toBe(false);
145 expect(result.history).toEqual({
146 historyCursor: "bounded-cursor",
147 hasMoreHistory: true,
148 latestLocalTurnOrdinal: 2,
149 });
150 }).pipe(
151 Effect.provide(
152 Layer.provide(BoundedThreadSnapshotHttp.layer, RpcHttp.layerRemoteHttpClient(fetchFn)),
153 ),
154 );
155 });
156
80157 it.effect("falls back to full HTTP snapshot when bounded returns a plain route 404", () => {
81158 const fetchFn = ((input: RequestInfo | URL) => {
82159 const url = String(input);
packages/client-runtime/src/state/boundedThreadSnapshotHttp.ts+14-2
......@@ -1,4 +1,5 @@
11import type { ThreadId } from "@t3tools/contracts";
2import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
23import * as Cause from "effect/Cause";
34import * as Effect from "effect/Effect";
45import * as Layer from "effect/Layer";
......@@ -18,7 +19,11 @@ import * as ThreadSnapshotLoader from "./threadSnapshotHttp.ts";
1819// Same cold-open budget as the full snapshot path; bounded payloads should fit.
1920const DEFAULT_BOUNDED_THREAD_SNAPSHOT_TIMEOUT_MS = 6_000;
2021
21/** Load a bounded recent-window thread snapshot over HTTP. */
22/**
23 * Load a bounded recent-window thread snapshot over HTTP. Opts into compact
24 * turnItems and restores them, so callers always see the full bounded shape.
25 * Older servers ignore the query and send the full shape.
26 */
2227export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn(
2328 "clientRuntime.state.fetchEnvironmentBoundedThreadSnapshot",
2429)(function* (input: {
......@@ -40,9 +45,16 @@ export const fetchEnvironmentBoundedThreadSnapshot = Effect.fn(
4045 request: ({ client, headers }) =>
4146 client.threadBoundedSnapshot({
4247 params: { threadId: input.threadId },
48 query: { compactTurnItems: "1" },
4349 headers: withOrchestrationProtocolHeader(headers),
4450 }),
45 });
51 }).pipe(
52 // Drop the marker with the restore so nothing can restore twice.
53 Effect.map(({ turnItemsOmitLocalVisible, ...snapshot }) => ({
54 ...snapshot,
55 projection: boundedSnapshotProjection({ ...snapshot, turnItemsOmitLocalVisible }),
56 })),
57 );
4658});
4759
4860/**
packages/client-runtime/src/state/threads-sync.test.ts+111
......@@ -122,6 +122,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
122122 const lastSubscribeAfterSequence = yield* Ref.make<number | undefined>(undefined);
123123 const lastRequestCompletionMarker = yield* Ref.make(false);
124124 const lastAcceptBoundedSnapshot = yield* Ref.make<true | undefined>(undefined);
125 const lastAcceptCompactTurnItems = yield* Ref.make<true | undefined>(undefined);
125126 const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();
126127 const savedThreads = yield* Ref.make<ReadonlyArray<OrchestrationV2ThreadDetailSnapshot>>([]);
127128 const removedThreads = yield* Ref.make<ReadonlyArray<ThreadId>>([]);
......@@ -139,6 +140,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
139140 readonly afterSequence?: number;
140141 readonly requestCompletionMarker?: true;
141142 readonly acceptBoundedSnapshot?: true;
143 readonly acceptCompactTurnItems?: true;
142144 }) =>
143145 Stream.unwrap(
144146 Ref.updateAndGet(subscriptionCount, (count) => count + 1).pipe(
......@@ -147,6 +149,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
147149 Ref.set(lastRequestCompletionMarker, input.requestCompletionMarker === true),
148150 ),
149151 Effect.andThen(Ref.set(lastAcceptBoundedSnapshot, input.acceptBoundedSnapshot)),
152 Effect.andThen(Ref.set(lastAcceptCompactTurnItems, input.acceptCompactTurnItems)),
150153 Effect.as(streamFrom(inputs)),
151154 ),
152155 ),
......@@ -261,6 +264,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
261264 lastSubscribeAfterSequence,
262265 lastRequestCompletionMarker,
263266 lastAcceptBoundedSnapshot,
267 lastAcceptCompactTurnItems,
264268 supervisorState,
265269 supervisorSession,
266270 savedThreads,
......@@ -842,6 +846,113 @@ describe("EnvironmentThreads", () => {
842846 }),
843847 );
844848
849 it.effect("restores compact socket snapshot turnItems before reducing and caching", () =>
850 Effect.gen(function* () {
851 const at = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z");
852 const command = (id: string, ordinal: number, threadId = THREAD_ID) =>
853 ({
854 id: TurnItemId.make(id),
855 threadId,
856 runId: null,
857 nodeId: null,
858 providerThreadId: null,
859 providerTurnId: null,
860 nativeItemRef: null,
861 parentItemId: null,
862 ordinal,
863 status: "completed" as const,
864 title: null,
865 startedAt: at,
866 completedAt: at,
867 updatedAt: at,
868 type: "command_execution" as const,
869 input: id,
870 output: id,
871 exitCode: 0,
872 }) satisfies OrchestrationV2TurnItem;
873 const parentThread = ThreadId.make("thread:compact-parent");
874 const inherited = command("parent-item", 1, parentThread);
875 const local = [command("local-1", 1), command("local-2", 2)];
876 const retainedRequest = {
877 ...command("retained-request", 0),
878 type: "run_interrupt_request" as const,
879 message: "Stop",
880 } as unknown as OrchestrationV2TurnItem;
881 const visibleTurnItems = [
882 {
883 position: 0,
884 visibility: "inherited" as const,
885 sourceThreadId: parentThread,
886 sourceItemId: inherited.id,
887 item: inherited,
888 },
889 ...local.map((item, index) => ({
890 position: index + 1,
891 visibility: "local" as const,
892 sourceThreadId: THREAD_ID,
893 sourceItemId: item.id,
894 item,
895 })),
896 ];
897 const harness = yield* makeHarness();
898 yield* Queue.offer(harness.inputs, {
899 kind: "snapshot",
900 snapshotSequence: 5,
901 // Only the dependency that is not a visible local row is on the wire.
902 projection: { ...BASE_PROJECTION, turnItems: [retainedRequest], visibleTurnItems },
903 historyCursor: "compact-cursor",
904 hasMoreHistory: true,
905 latestLocalTurnOrdinal: 2,
906 payloadBudgetExceeded: false,
907 turnItemsOmitLocalVisible: true,
908 });
909
910 const seeded = yield* awaitThreadState(
911 harness.observed,
912 (value) => value.history.historyCursor === "compact-cursor",
913 );
914 expect(Option.getOrThrow(seeded.data).turnItems).toEqual([...local, retainedRequest]);
915 expect(yield* Ref.get(harness.lastAcceptBoundedSnapshot)).toBe(true);
916 expect(yield* Ref.get(harness.lastAcceptCompactTurnItems)).toBe(true);
917
918 // Live updates reduce against the restored list, not the compact one.
919 const updated = { ...local[1]!, output: "updated", ordinal: 2 };
920 yield* Queue.offer(harness.inputs, {
921 kind: "event",
922 sequence: 6,
923 event: {
924 id: EventId.make("event-compact-update"),
925 type: "turn-item.updated",
926 threadId: THREAD_ID,
927 occurredAt: at,
928 payload: updated,
929 },
930 });
931 const live = yield* awaitThreadState(harness.observed, (value) =>
932 Option.exists(value.data, (data) =>
933 data.turnItems.some(
934 (item) => item.type === "command_execution" && item.output === "updated",
935 ),
936 ),
937 );
938 expect(Option.getOrThrow(live.data).turnItems.map((item) => String(item.id))).toEqual([
939 "local-1",
940 "local-2",
941 "retained-request",
942 ]);
943
944 yield* TestClock.adjust("500 millis");
945 yield* Effect.yieldNow;
946 const saved = (yield* Ref.get(harness.savedThreads)).at(-1);
947 expect(saved?.projection.turnItems.map((item) => String(item.id))).toEqual([
948 "local-1",
949 "local-2",
950 "retained-request",
951 ]);
952 expect(saved && "turnItemsOmitLocalVisible" in saved).toBe(false);
953 }),
954 );
955
845956 it.effect("seeds the thread from the HTTP snapshot and resumes live events", () =>
846957 Effect.gen(function* () {
847958 const httpProjection: OrchestrationV2ThreadProjection = {
packages/client-runtime/src/state/threads.ts+5-2
......@@ -6,6 +6,7 @@ import {
66 type OrchestrationV2ThreadStreamItem,
77 type ThreadId as ThreadIdType,
88} from "@t3tools/contracts";
9import { boundedSnapshotProjection } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
910import * as Cause from "effect/Cause";
1011import * as Effect from "effect/Effect";
1112import * as Fiber from "effect/Fiber";
......@@ -589,7 +590,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
589590 // Bounded socket fallbacks carry their cursor. Legacy-compatible full
590591 // snapshots omit these fields and still replace progressive state.
591592 yield* setThread(
592 item.projection,
593 boundedSnapshotProjection(item),
593594 hasProgressiveHistory
594595 ? {
595596 history: {
......@@ -919,7 +920,9 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
919920 threadId,
920921 ...(canResume ? { afterSequence: sequence } : {}),
921922 ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
922 ...(acceptBoundedSnapshot ? { acceptBoundedSnapshot: true as const } : {}),
923 ...(acceptBoundedSnapshot
924 ? { acceptBoundedSnapshot: true as const, acceptCompactTurnItems: true as const }
925 : {}),
923926 };
924927 }),
925928 {
packages/contracts/src/environmentHttp.ts+9
......@@ -584,6 +584,14 @@ const EnvironmentOrchestrationThreadSnapshotParams = Schema.Struct({
584584 threadId: ThreadId,
585585});
586586
587/**
588 * `compactTurnItems=1` opts into `turnItemsOmitLocalVisible`. Other values are
589 * ignored, and older servers ignore the whole query.
590 */
591const EnvironmentOrchestrationThreadBoundedSnapshotQuery = Schema.Struct({
592 compactTurnItems: Schema.optionalKey(Schema.String),
593});
594
587595const EnvironmentOrchestrationThreadHistoryQuery = Schema.Struct({
588596 cursor: TrimmedNonEmptyString,
589597});
......@@ -615,6 +623,7 @@ class EnvironmentOrchestrationHttpApi extends HttpApiGroup.make("orchestration")
615623 HttpApiEndpoint.get("threadBoundedSnapshot", "/api/orchestration/threads/:threadId/bounded", {
616624 headers: OrchestrationProtocolHeaders,
617625 params: EnvironmentOrchestrationThreadSnapshotParams,
626 query: EnvironmentOrchestrationThreadBoundedSnapshotQuery,
618627 success: OrchestrationV2ThreadBoundedSnapshot,
619628 error: EnvironmentOrchestrationThreadSnapshotErrors,
620629 }).middleware(EnvironmentAuthenticatedAuth),
packages/contracts/src/orchestrationV2.test.ts+83
......@@ -45,6 +45,48 @@ import {
4545} from "./orchestrationV2.ts";
4646
4747const now = DateTime.makeUnsafe("2026-04-20T00:00:00.000Z");
48function emptyThreadProjection() {
49 return {
50 thread: {
51 createdBy: "user",
52 creationSource: "web",
53 id: "thread-1",
54 projectId: "project-1",
55 title: "Thread",
56 providerInstanceId: "codex",
57 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex" },
58 runtimeMode: "full-access",
59 interactionMode: "default",
60 branch: null,
61 worktreePath: null,
62 activeProviderThreadId: null,
63 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: "thread-1" },
64 forkedFrom: null,
65 createdAt: now,
66 updatedAt: now,
67 archivedAt: null,
68 deletedAt: null,
69 },
70 runs: [],
71 attempts: [],
72 nodes: [],
73 subagents: [],
74 providerSessions: [],
75 providerThreads: [],
76 providerTurns: [],
77 runtimeRequests: [],
78 messages: [],
79 plans: [],
80 turnItems: [],
81 checkpointScopes: [],
82 checkpoints: [],
83 contextHandoffs: [],
84 contextTransfers: [],
85 visibleTurnItems: [],
86 updatedAt: now,
87 };
88}
89
4890const LegacyShellStreamItem = Schema.Union([
4991 Schema.Struct({ kind: Schema.Literal("synchronized") }),
5092 Schema.Struct({
......@@ -326,6 +368,47 @@ describe("orchestration V2 contracts", () => {
326368 expect("acceptBoundedSnapshot" in legacyDecoded).toBe(false);
327369 });
328370
371 it("negotiates compact bounded turnItems without disturbing older peers", () => {
372 // Older servers strip the unknown opt-in, so they keep sending full turnItems.
373 const legacyDecoded = decodeLegacySubscribeThreadInput({
374 threadId: "thread-1",
375 acceptBoundedSnapshot: true,
376 acceptCompactTurnItems: true,
377 });
378 expect("acceptCompactTurnItems" in legacyDecoded).toBe(false);
379 expect(
380 decodeOrchestrationV2SubscribeThreadInput({
381 threadId: "thread-1",
382 acceptCompactTurnItems: true,
383 }).acceptCompactTurnItems,
384 ).toBe(true);
385
386 const decodeStreamItem = Schema.decodeUnknownSync(OrchestrationV2ThreadStreamItem);
387 const projection = decodeStreamItem({
388 kind: "snapshot",
389 snapshotSequence: 1,
390 projection: emptyThreadProjection(),
391 });
392 // Older servers never send the marker; newer clients treat absence as full turnItems.
393 expect(projection.kind === "snapshot" && projection.turnItemsOmitLocalVisible).toBe(undefined);
394 const marked = decodeStreamItem({
395 kind: "snapshot",
396 snapshotSequence: 1,
397 projection: emptyThreadProjection(),
398 turnItemsOmitLocalVisible: true,
399 });
400 expect(marked.kind === "snapshot" && marked.turnItemsOmitLocalVisible).toBe(true);
401 // The marker only means "omitted"; any other value is a protocol error.
402 expect(() =>
403 decodeStreamItem({
404 kind: "snapshot",
405 snapshotSequence: 1,
406 projection: emptyThreadProjection(),
407 turnItemsOmitLocalVisible: false,
408 }),
409 ).toThrow();
410 });
411
329412 it("decodes persisted capability snapshots that predate runtimePolicy", () => {
330413 // Events written before the field existed must replay; absent decodes to
331414 // the weaker client-boundary guarantee so history never overclaims.
packages/contracts/src/orchestrationV2.ts+13
......@@ -3244,6 +3244,11 @@ export const OrchestrationV2SubscribeThreadInput = Schema.Struct({
32443244 requestCompletionMarker: Schema.optionalKey(Schema.Boolean),
32453245 /** Allows snapshot fallbacks to contain a bounded, pageable history window. */
32463246 acceptBoundedSnapshot: Schema.optionalKey(Schema.Boolean),
3247 /**
3248 * Allows bounded snapshot fallbacks to omit `projection.turnItems` entries
3249 * that repeat local visible rows. See `turnItemsOmitLocalVisible`.
3250 */
3251 acceptCompactTurnItems: Schema.optionalKey(Schema.Boolean),
32473252});
32483253export type OrchestrationV2SubscribeThreadInput = typeof OrchestrationV2SubscribeThreadInput.Type;
32493254
......@@ -3283,6 +3288,12 @@ export const OrchestrationV2ThreadBoundedSnapshot = Schema.Struct({
32833288 latestLocalTurnOrdinal: Schema.NullOr(NonNegativeInt),
32843289 /** True when complete turns or required live control state exceed the usual byte budget. */
32853290 payloadBudgetExceeded: Schema.optional(Schema.Boolean),
3291 /**
3292 * Set only for clients that opted in: `projection.turnItems` omits the items
3293 * of local visible rows, which lead the full list. Clients must restore them
3294 * with `boundedSnapshotProjection` before using the projection.
3295 */
3296 turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)),
32863297});
32873298export type OrchestrationV2ThreadBoundedSnapshot = typeof OrchestrationV2ThreadBoundedSnapshot.Type;
32883299
......@@ -3363,6 +3374,8 @@ export const OrchestrationV2ThreadStreamItem = Schema.Union([
33633374 hasMoreHistory: Schema.optionalKey(Schema.Boolean),
33643375 latestLocalTurnOrdinal: Schema.optionalKey(Schema.NullOr(NonNegativeInt)),
33653376 payloadBudgetExceeded: Schema.optionalKey(Schema.Boolean),
3377 /** Same meaning as on `OrchestrationV2ThreadBoundedSnapshot`. */
3378 turnItemsOmitLocalVisible: Schema.optionalKey(Schema.Literal(true)),
33663379 }),
33673380 Schema.Struct({
33683381 kind: Schema.Literal("event"),
packages/shared/package.json+4
......@@ -359,6 +359,10 @@
359359 "types": "./src/nodeSqliteClient.ts",
360360 "import": "./src/nodeSqliteClient.ts"
361361 },
362 "./orchestrationV2BoundedSnapshot": {
363 "types": "./src/orchestrationV2BoundedSnapshot.ts",
364 "import": "./src/orchestrationV2BoundedSnapshot.ts"
365 },
362366 "./orchestrationV2Timeline": {
363367 "types": "./src/orchestrationV2Timeline.ts",
364368 "import": "./src/orchestrationV2Timeline.ts"
packages/shared/src/orchestrationV2BoundedSnapshot.test.ts created+147
......@@ -0,0 +1,147 @@
1import {
2 RunId,
3 ThreadId,
4 TurnItemId,
5 type OrchestrationV2ProjectedTurnItem,
6 type OrchestrationV2ThreadProjection,
7 type OrchestrationV2TurnItem,
8} from "@t3tools/contracts";
9import { describe, expect, it } from "vite-plus/test";
10
11import {
12 boundedSnapshotProjection,
13 omitLocalVisibleTurnItems,
14 restoreLocalVisibleTurnItems,
15} from "./orchestrationV2BoundedSnapshot.ts";
16
17const PARENT = ThreadId.make("thread:compact-parent");
18const CHILD = ThreadId.make("thread:compact-child");
19const RUN = RunId.make("run:compact");
20
21function item(
22 threadId: ThreadId,
23 id: string,
24 ordinal: number,
25 type: "command_execution" | "run_interrupt_request" | "fork" = "command_execution",
26): OrchestrationV2TurnItem {
27 return {
28 id: TurnItemId.make(id),
29 type,
30 threadId,
31 runId: type === "command_execution" ? null : RUN,
32 nodeId: null,
33 providerThreadId: null,
34 providerTurnId: null,
35 nativeItemRef: null,
36 parentItemId: null,
37 ordinal,
38 status: "completed",
39 title: id,
40 input: id,
41 exitCode: 0,
42 startedAt: null,
43 completedAt: null,
44 updatedAt: "2026-10-01T00:00:00.000Z",
45 } as unknown as OrchestrationV2TurnItem;
46}
47
48function row(
49 visibility: OrchestrationV2ProjectedTurnItem["visibility"],
50 sourceThreadId: ThreadId,
51 value: OrchestrationV2TurnItem,
52): Omit<OrchestrationV2ProjectedTurnItem, "position"> {
53 return { visibility, sourceThreadId, sourceItemId: value.id, item: value };
54}
55
56function projection(input: {
57 readonly visible: ReadonlyArray<Omit<OrchestrationV2ProjectedTurnItem, "position">>;
58 readonly turnItems: ReadonlyArray<OrchestrationV2TurnItem>;
59}): OrchestrationV2ThreadProjection {
60 return {
61 thread: { id: CHILD },
62 turnItems: [...input.turnItems],
63 visibleTurnItems: input.visible.map((value, position) => ({ ...value, position })),
64 } as unknown as OrchestrationV2ThreadProjection;
65}
66
67/** Fork child: inherited parent rows, synthetic marker, local rows, retained interrupt request. */
68function forkWindow() {
69 const inherited = [item(PARENT, "parent-1", 1), item(PARENT, "parent-2", 2)];
70 const marker = item(PARENT, "fork-marker", 3, "fork");
71 const local = [item(CHILD, "child-1", 1), item(CHILD, "child-2", 2), item(CHILD, "child-3", 3)];
72 const retainedRequest = item(CHILD, "interrupt-request", 0, "run_interrupt_request");
73 return {
74 local,
75 retainedRequest,
76 projection: projection({
77 visible: [
78 ...inherited.map((value) => row("inherited", PARENT, value)),
79 row("synthetic", PARENT, marker),
80 ...local.map((value) => row("local", CHILD, value)),
81 ],
82 turnItems: [...local, retainedRequest],
83 }),
84 };
85}
86
87describe("compact bounded snapshot turnItems", () => {
88 it("omits only local visible items and restores the exact list and order", () => {
89 const { projection: full, local, retainedRequest } = forkWindow();
90 const compact = omitLocalVisibleTurnItems(full);
91
92 expect(compact).not.toBeNull();
93 // Inherited and synthetic rows never had turnItems; the retained request stays.
94 expect(compact!.turnItems).toEqual([retainedRequest]);
95 expect(compact!.visibleTurnItems).toBe(full.visibleTurnItems);
96
97 const restored = restoreLocalVisibleTurnItems(compact!);
98 expect(restored.turnItems).toEqual([...local, retainedRequest]);
99 expect(restored).toEqual(full);
100 });
101
102 it("round-trips through JSON as a separately decoded client would see it", () => {
103 const { projection: full } = forkWindow();
104 const wire = JSON.parse(JSON.stringify(omitLocalVisibleTurnItems(full)));
105 expect(
106 boundedSnapshotProjection({ projection: wire, turnItemsOmitLocalVisible: true }),
107 ).toEqual(JSON.parse(JSON.stringify(full)));
108 });
109
110 it("treats rows sourced from the thread itself as local", () => {
111 const own = item(CHILD, "own", 1);
112 const full = projection({ visible: [row("inherited", CHILD, own)], turnItems: [own] });
113 const compact = omitLocalVisibleTurnItems(full);
114 expect(compact?.turnItems).toEqual([]);
115 expect(restoreLocalVisibleTurnItems(compact!)).toEqual(full);
116 });
117
118 it("refuses shapes it cannot restore exactly", () => {
119 const [first, second] = [item(CHILD, "a", 1), item(CHILD, "b", 2)];
120 const visible = [row("local", CHILD, first), row("local", CHILD, second)];
121 // Different order than the visible rows.
122 expect(omitLocalVisibleTurnItems(projection({ visible, turnItems: [second, first] }))).toBe(
123 null,
124 );
125 // Equal content but a different object: restoring would not be provably exact.
126 expect(
127 omitLocalVisibleTurnItems(projection({ visible, turnItems: [{ ...first }, second] })),
128 ).toBe(null);
129 // A local row without its turn item.
130 expect(omitLocalVisibleTurnItems(projection({ visible, turnItems: [first] }))).toBe(null);
131 });
132
133 it("sends nothing compact when no local rows are visible", () => {
134 const inherited = item(PARENT, "parent", 1);
135 const request = item(CHILD, "interrupt-request", 0, "run_interrupt_request");
136 expect(
137 omitLocalVisibleTurnItems(
138 projection({ visible: [row("inherited", PARENT, inherited)], turnItems: [request] }),
139 ),
140 ).toBe(null);
141 });
142
143 it("leaves unmarked snapshots untouched", () => {
144 const { projection: full } = forkWindow();
145 expect(boundedSnapshotProjection({ projection: full })).toBe(full);
146 });
147});
packages/shared/src/orchestrationV2BoundedSnapshot.ts created+58
......@@ -0,0 +1,58 @@
1import type {
2 OrchestrationV2ProjectedTurnItem,
3 OrchestrationV2ThreadProjection,
4} from "@t3tools/contracts";
5
6/** Local timeline rows also carry their item in `projection.turnItems`. */
7function isLocalTimelineRow(
8 projection: Pick<OrchestrationV2ThreadProjection, "thread">,
9 row: OrchestrationV2ProjectedTurnItem,
10): boolean {
11 return row.visibility === "local" || row.sourceThreadId === projection.thread.id;
12}
13
14/**
15 * Server half of the opt-in compact bounded snapshot. Bounded `turnItems` start
16 * with the items of the window's local visible rows, so those entries can be
17 * dropped and rebuilt by `restoreLocalVisibleTurnItems`. Returns null when the
18 * projection does not have that exact shape, or when nothing would be omitted;
19 * callers then send the unchanged projection without a marker.
20 */
21export function omitLocalVisibleTurnItems(
22 projection: OrchestrationV2ThreadProjection,
23): OrchestrationV2ThreadProjection | null {
24 let omitted = 0;
25 for (const row of projection.visibleTurnItems) {
26 if (!isLocalTimelineRow(projection, row)) continue;
27 // Same object, not merely the same id: restoring from the row must
28 // reproduce the original entry exactly.
29 if (projection.turnItems[omitted] !== row.item) return null;
30 omitted += 1;
31 }
32 if (omitted === 0) return null;
33 return { ...projection, turnItems: projection.turnItems.slice(omitted) };
34}
35
36/**
37 * Client half: call on a decoded snapshot that carries the
38 * `turnItemsOmitLocalVisible` marker, before reducers or caches see it.
39 */
40export function restoreLocalVisibleTurnItems(
41 projection: OrchestrationV2ThreadProjection,
42): OrchestrationV2ThreadProjection {
43 const local = projection.visibleTurnItems
44 .filter((row) => isLocalTimelineRow(projection, row))
45 .map((row) => row.item);
46 if (local.length === 0) return projection;
47 return { ...projection, turnItems: [...local, ...projection.turnItems] };
48}
49
50/** The usable projection of a decoded bounded snapshot, compact or not. */
51export function boundedSnapshotProjection(snapshot: {
52 readonly projection: OrchestrationV2ThreadProjection;
53 readonly turnItemsOmitLocalVisible?: true | undefined;
54}): OrchestrationV2ThreadProjection {
55 return snapshot.turnItemsOmitLocalVisible === true
56 ? restoreLocalVisibleTurnItems(snapshot.projection)
57 : snapshot.projection;
58}