Open History to see the last commit
1import type {
2 OrchestrationV2ThreadProjection,
3 OrchestrationV2ThreadStreamItem,
4} from "@t3tools/contracts";
5
6import { omitLocalVisibleTurnItems } from "@t3tools/shared/orchestrationV2BoundedSnapshot";
7import { dedupeSnapshotBodies } from "@t3tools/shared/orchestrationV2DedupedBodies";
8
9import {
10 buildBoundedThreadProjection,
11 type BoundedProjectionResult,
12} from "./threadHistoryPaging.ts";
13import {
14 capProjectionCheckpointFilesForWire,
15 projectThreadProjectionForWire,
16} from "./WireProjection.ts";
17
18export const THREAD_RESUME_MAX_REPLAY_EVENTS = 128;
19
20export const THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES = 1_048_576;
21
22/** Decoded objects and temporary strings exceed stored JSON size, even before wire slimming. */
23export const THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES = 1_048_576;
24
25export function isThreadReplayRawPayloadSafe(rawPayloadBytes: number): boolean {
26 return rawPayloadBytes <= THREAD_RESUME_MAX_RAW_PAYLOAD_BYTES;
27}
28
29export function threadReplayEncodedBytes(
30 items: ReadonlyArray<unknown>,
31 limit: number = THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES,
32): number {
33 let total = 0;
34 for (const item of items) {
35 const encoded = JSON.stringify(item);
36 total += Buffer.byteLength(encoded ?? "", "utf8");
37 if (total > limit) break;
38 }
39 return total;
40}
41
42type ThreadSnapshotStreamItem = Extract<
43 OrchestrationV2ThreadStreamItem,
44 { readonly kind: "snapshot" }
45>;
46
47export function boundedSnapshotResponseFields(input: {
48 readonly bounded: BoundedProjectionResult;
49 readonly compactTurnItems: boolean;
50 readonly dedupeBodies: boolean;
51 readonly compactIds: boolean;
52}) {
53 const { bounded } = input;
54 const capped = capProjectionCheckpointFilesForWire(bounded.projection);
55 // Dedupe keeps shared items one object, which the compaction's identity check needs.
56 const projection = input.dedupeBodies ? dedupeSnapshotBodies(capped) : capped;
57 const compactProjection = input.compactTurnItems ? omitLocalVisibleTurnItems(projection) : null;
58 return {
59 projection: compactProjection ?? projection,
60 historyCursor: bounded.historyCursor,
61 hasMoreHistory: bounded.hasMoreHistory,
62 latestLocalTurnOrdinal: bounded.latestLocalTurnOrdinal,
63 ...(compactProjection === null ? {} : { turnItemsOmitLocalVisible: true as const }),
64 ...(input.compactIds ? { idTable: true as const } : {}),
65 };
66}
67
68export function buildBoundedThreadStreamSnapshot(input: {
69 readonly snapshotSequence: number;
70 readonly projection: OrchestrationV2ThreadProjection;
71 readonly compactTurnItems?: boolean;
72 readonly dedupeBodies?: boolean;
73 readonly compactIds?: boolean;
74}): ThreadSnapshotStreamItem {
75 const bounded = buildBoundedThreadProjection({
76 snapshotSequence: input.snapshotSequence,
77 projection: projectThreadProjectionForWire(input.projection),
78 });
79 return {
80 kind: "snapshot",
81 snapshotSequence: input.snapshotSequence,
82 ...boundedSnapshotResponseFields({
83 bounded,
84 compactTurnItems: input.compactTurnItems === true,
85 dedupeBodies: input.dedupeBodies === true,
86 compactIds: input.compactIds === true,
87 }),
88 };
89}
90
91export type ThreadResumePlan =
92 | {
93 readonly mode: "replay";
94 readonly afterSequence: number;
95 readonly throughSequence: number;
96 }
97 | { readonly mode: "snapshot" };
98
99/** Bound event count and encoded bytes; replace excessive replay with one current snapshot. */
100export function decideThreadResume(input: {
101 readonly afterSequence: number;
102 readonly highWater: number;
103 readonly replayEventCount: number;
104 readonly replayEncodedBytes: number;
105}): ThreadResumePlan {
106 if (
107 input.afterSequence > input.highWater ||
108 input.replayEventCount > THREAD_RESUME_MAX_REPLAY_EVENTS ||
109 input.replayEncodedBytes > THREAD_RESUME_MAX_REPLAY_ENCODED_BYTES
110 ) {
111 return { mode: "snapshot" };
112 }
113 return {
114 mode: "replay",
115 afterSequence: input.afterSequence,
116 throughSequence: input.highWater,
117 };
118}