1import type {
2 OrchestrationV2ProjectedTurnItem,
3 OrchestrationV2ThreadProjection,
4 OrchestrationV2TurnItem,
5 ThreadId,
6 TurnItemId,
7} from "@t3tools/contracts";
8import * as Schema from "effect/Schema";
9
10/**
11 * Every window stops at the byte budget. A window that holds a user turn also
12 * stops at `maxUserTurns` or `maxTurnItems` rows, which keeps ordinary turns
13 * whole (V1 kept 500 activities). Any other window, such as an agent-driven
14 * thread or the inside of one very long turn, stops at `maxItems` rows.
15 */
16export const THREAD_HISTORY_PAGE_POLICY = {
17 maxUserTurns: 10,
18 maxItems: 75,
19 maxTurnItems: 500,
20 maxEncodedBytes: 1_048_576,
21} as const;
22
23export const THREAD_HISTORY_INLINE_WORK_ITEM_LIMIT = 150;
24
25export const OLDER_THREAD_USER_TURN_LIMIT = 20;
26
27export function isConversationHistoryItem(item: Pick<OrchestrationV2TurnItem, "type">): boolean {
28 return !["command_execution", "reasoning", "file_change", "dynamic_tool"].includes(item.type);
29}
30export const THREAD_HISTORY_MAX_RAW_TURNS = 150;
31
32export const THREAD_HISTORY_SNAPSHOT_ROW_LIMIT = THREAD_HISTORY_PAGE_POLICY.maxItems + 2;
33/** The snapshot query's row limit once the history has a user turn. */
34export const THREAD_HISTORY_TURN_SNAPSHOT_ROW_LIMIT = THREAD_HISTORY_PAGE_POLICY.maxTurnItems + 2;
35
36const THREAD_HISTORY_CURSOR_MAX_LENGTH = 4_096;
37
38export type ThreadHistoryPagePolicy = {
39 readonly maxUserTurns?: number | undefined;
40 readonly maxItems: number;
41 /** Defaults to `maxItems`. */
42 readonly maxTurnItems?: number | undefined;
43 readonly maxEncodedBytes: number;
44};
45
46export type ThreadHistoryCursorPayload = {
47 readonly v: 1;
48 readonly seq: number;
49 readonly st: string;
50 readonly si: string;
51 readonly p: number;
52 readonly inclusive?: boolean;
53};
54
55type HistoryRow = Pick<
56 OrchestrationV2ProjectedTurnItem,
57 "position" | "visibility" | "sourceThreadId" | "sourceItemId"
58> & {
59 readonly item: Pick<OrchestrationV2TurnItem, "type"> & {
60 readonly id?: string;
61 readonly messageId?: string;
62 readonly inputIntent?: string;
63 readonly createdBy?: string;
64 readonly runId?: string | null;
65 readonly status?: OrchestrationV2TurnItem["status"];
66 readonly parentItemId?: string | null;
67 readonly ordinal?: number;
68 readonly runStatus?: string | undefined;
69 readonly attemptId?: string | null | undefined;
70 readonly attemptStatus?: string | null | undefined;
71 readonly persistentOutput?: boolean;
72 readonly deferredWork?: OrchestrationV2TurnItem["deferredWork"];
73 };
74};
75/** What a caller that passes only the newest part of a history knows about the rows it left out. */
76export type ThreadHistoryPrefix = {
77 readonly hasUserTurn: boolean;
78 /** Runs that have a user message or notification among the rows left out. */
79 readonly startedRuns: ReadonlySet<string>;
80};
81
82export type SelectTimelinePageResult<Row extends HistoryRow = OrchestrationV2ProjectedTurnItem> = {
83 readonly items: ReadonlyArray<Row>;
84 readonly nextCursor: string | null;
85 readonly hasMoreHistory: boolean;
86 /**
87 * Set with `prefix`: the oldest row the selection read, which can be older than `items[0]`.
88 * The page is the same as on the full history only when a row lies before this one.
89 */
90 readonly oldestScanned?: Row | undefined;
91};
92
93export type BoundedProjectionResult = {
94 readonly projection: OrchestrationV2ThreadProjection;
95 readonly historyCursor: string | null;
96 readonly hasMoreHistory: boolean;
97 readonly latestLocalTurnOrdinal: number | null;
98};
99
100export class InvalidThreadHistoryCursorError extends Schema.TaggedError<InvalidThreadHistoryCursorError>()(
101 "InvalidThreadHistoryCursorError",
102 {},
103) {
104 override get message(): string {
105 return "Invalid thread history cursor.";
106 }
107}
108
109function bytesOfJson(value: unknown): number {
110 return Buffer.byteLength(JSON.stringify(value), "utf8");
111}
112
113export function projectedRowEncodedBytes(row: HistoryRow): number {
114 return bytesOfJson(row);
115}
116
117/** Charge local items twice because bounded snapshots also include projection.turnItems; inherited rows appear only once. */
118export function projectedRowBoundedSnapshotEncodedBytes(
119 row: OrchestrationV2ProjectedTurnItem,
120 threadId: ThreadId | string,
121): number {
122 const rowBytes = projectedRowEncodedBytes(row);
123 if (row.visibility === "local" || String(row.sourceThreadId) === String(threadId)) {
124 return rowBytes + bytesOfJson(row.item);
125 }
126 return rowBytes;
127}
128
129function renumberPositions<Row extends HistoryRow>(items: ReadonlyArray<Row>): Row[] {
130 return items.map((row, position) => (row.position === position ? row : { ...row, position }));
131}
132
133function encodeCursorPayload(payload: ThreadHistoryCursorPayload): string {
134 return Buffer.from(JSON.stringify(payload), "utf8").toString("base64url");
135}
136
137export function encodeThreadHistoryCursor(input: {
138 readonly snapshotSequence: number;
139 readonly sourceThreadId: ThreadId | string;
140 readonly sourceItemId: TurnItemId | string;
141 readonly position: number;
142 readonly inclusive?: boolean;
143}): string {
144 return encodeCursorPayload({
145 v: 1,
146 seq: input.snapshotSequence,
147 st: String(input.sourceThreadId),
148 si: String(input.sourceItemId),
149 p: input.position,
150 ...(input.inclusive ? { inclusive: true } : {}),
151 });
152}
153
154export function decodeThreadHistoryCursor(cursor: string): ThreadHistoryCursorPayload {
155 if (cursor.length === 0 || cursor.length > THREAD_HISTORY_CURSOR_MAX_LENGTH) {
156 throw new InvalidThreadHistoryCursorError();
157 }
158 let parsed: unknown;
159 try {
160 const json = Buffer.from(cursor, "base64url").toString("utf8");
161 parsed = JSON.parse(json) as unknown;
162 } catch {
163 throw new InvalidThreadHistoryCursorError();
164 }
165 if (
166 typeof parsed !== "object" ||
167 parsed === null ||
168 !("v" in parsed) ||
169 parsed.v !== 1 ||
170 !("seq" in parsed) ||
171 typeof parsed.seq !== "number" ||
172 !Number.isInteger(parsed.seq) ||
173 parsed.seq < 0 ||
174 !("st" in parsed) ||
175 typeof parsed.st !== "string" ||
176 parsed.st.length === 0 ||
177 !("si" in parsed) ||
178 typeof parsed.si !== "string" ||
179 parsed.si.length === 0 ||
180 !("p" in parsed) ||
181 typeof parsed.p !== "number" ||
182 !Number.isInteger(parsed.p) ||
183 parsed.p < 0 ||
184 ("inclusive" in parsed && typeof parsed.inclusive !== "boolean")
185 ) {
186 throw new InvalidThreadHistoryCursorError();
187 }
188 return parsed as ThreadHistoryCursorPayload;
189}
190
191function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean {
192 return (
193 item.type === "user_message" &&
194 (item.inputIntent === "turn_start" || item.inputIntent === "queued_turn")
195 );
196}
197
198export const DEFERRED_WORK_ITEM_TYPES = [
199 "command_execution",
200 "reasoning",
201 "file_change",
202 "file_search",
203 "web_search",
204 "dynamic_tool",
205] as const;
206
207function isDeferrableWorkItem(item: HistoryRow["item"]): boolean {
208 return (
209 item.status === "completed" &&
210 item.persistentOutput !== true &&
211 DEFERRED_WORK_ITEM_TYPES.some((type) => type === item.type)
212 );
213}
214
215/** A row that folds when its run is not blocked and its group is large enough. */
216export function isFoldableWorkItem(item: HistoryRow["item"]): boolean {
217 return (
218 item.runId != null &&
219 (item.runStatus === "completed" || item.attemptStatus === "superseded") &&
220 isDeferrableWorkItem(item)
221 );
222}
223
224export function deferWorkHistoryRows<Row extends HistoryRow>(items: ReadonlyArray<Row>): Row[] {
225 const blocked = new Set(
226 items.flatMap(({ item }) =>
227 item.runId != null &&
228 (item.type === "run_interrupt_request" ||
229 item.type === "run_interrupt_result" ||
230 (item.type === "error" && item.status === "failed" && item.parentItemId == null))
231 ? [item.runId]
232 : [],
233 ),
234 );
235 const segments = new Map<string, number>();
236 const groups = new Map<string, Row[]>();
237 for (const row of items) {
238 const item = row.item;
239 if (item.runId == null) continue;
240 if (item.type === "user_message") segments.set(item.runId, (segments.get(item.runId) ?? 0) + 1);
241 if (blocked.has(item.runId) || !isFoldableWorkItem(item)) continue;
242 const key = `${row.sourceThreadId}:${item.runId}:${segments.get(item.runId) ?? 0}:${item.attemptId ?? ""}`;
243 const group = groups.get(key) ?? [];
244 group.push(row);
245 groups.set(key, group);
246 }
247 const omitted = new Set<Row>();
248 const anchors = new Map<Row, Row>();
249 for (const group of groups.values()) {
250 if (group.length <= THREAD_HISTORY_INLINE_WORK_ITEM_LIMIT) continue;
251 const first = group[0]!;
252 const last = group.at(-1)!;
253 for (const row of group.slice(1)) omitted.add(row);
254 anchors.set(first, {
255 ...first,
256 item: {
257 ...first.item,
258 deferredWork: {
259 cursor: encodeThreadHistoryCursor({
260 snapshotSequence: 0,
261 sourceThreadId: last.sourceThreadId,
262 sourceItemId: last.sourceItemId,
263 position: last.item.ordinal ?? last.position,
264 inclusive: true,
265 }),
266 itemCount: group.length,
267 lastOrdinal: last.item.ordinal ?? last.position,
268 },
269 },
270 });
271 }
272 return items.flatMap((row) => (omitted.has(row) ? [] : [anchors.get(row) ?? row]));
273}
274
275export function selectDeferredWorkPage<Row extends HistoryRow>(input: {
276 readonly items: ReadonlyArray<Row>;
277 readonly cursor: string;
278 readonly throughEntryId?: string | undefined;
279 readonly snapshotSequence: number;
280 readonly rowEncodedBytes?: ((row: Row) => number) | undefined;
281}): SelectTimelinePageResult<Row> {
282 const cursor = decodeThreadHistoryCursor(input.cursor);
283 const end = findCursorEndIndex(input.items, cursor);
284 const start = input.items.findIndex((row) => row.sourceItemId === input.throughEntryId);
285 if (start < 0 || start >= end) throw new InvalidThreadHistoryCursorError();
286 const anchor = input.items[start]!;
287 const items = input.items
288 .slice(start, end)
289 .filter(
290 (row) =>
291 row.sourceThreadId === anchor.sourceThreadId &&
292 row.item.runId === anchor.item.runId &&
293 row.item.attemptId === anchor.item.attemptId &&
294 isDeferrableWorkItem(row.item),
295 );
296 return selectOlderTimelinePage({
297 items,
298 exclusiveEndIndex: items.length,
299 snapshotSequence: input.snapshotSequence,
300 rowEncodedBytes: input.rowEncodedBytes,
301 policy: THREAD_HISTORY_PAGE_POLICY,
302 });
303}
304
305/** Notifications can start wake turns when they are the run's first input; match ProjectionStore run_starts. */
306function threadHistoryTurnStarts(
307 items: ReadonlyArray<HistoryRow["item"]>,
308 earlierRuns?: ReadonlySet<string>,
309): ReadonlyArray<boolean> {
310 const startedRuns = new Set<string>(earlierRuns);
311 return items.map((item) => {
312 if (item.type !== "user_message" && item.type !== "notification") return false;
313 const firstInput = item.runId != null && !startedRuns.has(item.runId);
314 if (item.runId != null) startedRuns.add(item.runId);
315 return item.type === "notification" ? firstInput : isThreadHistoryTurnStart(item);
316 });
317}
318
319export function isThreadHistoryUserTurn(item: HistoryRow["item"]): boolean {
320 return (
321 isThreadHistoryTurnStart(item) && item.type === "user_message" && item.createdBy === "user"
322 );
323}
324
325/**
326 * Always admit at least one row so oversized items cannot deadlock paging.
327 */
328function selectOlderTimelinePage<Row extends HistoryRow>(input: {
329 readonly items: ReadonlyArray<Row>;
330 readonly exclusiveEndIndex: number;
331 readonly snapshotSequence: number;
332 readonly policy?: ThreadHistoryPagePolicy | undefined;
333 readonly rowEncodedBytes?: ((row: Row) => number) | undefined;
334 readonly prefix?: ThreadHistoryPrefix | undefined;
335}): SelectTimelinePageResult<Row> {
336 const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY;
337 const rowCost = input.rowEncodedBytes ?? projectedRowEncodedBytes;
338 const end = Math.min(Math.max(input.exclusiveEndIndex, 0), input.items.length);
339 if (end <= 0) {
340 return { items: [], nextCursor: null, hasMoreHistory: false };
341 }
342
343 const selected: Row[] = [];
344 let encodedBytes = 0;
345 let userTurns = 0;
346 let rawTurns = 0;
347 const candidates = input.items.slice(0, end);
348 const turnStarts = threadHistoryTurnStarts(
349 candidates.map((row) => row.item),
350 input.prefix?.startedRuns,
351 );
352 const turnLimit =
353 input.prefix?.hasUserTurn === true ||
354 candidates.some((row) => isThreadHistoryUserTurn(row.item))
355 ? policy.maxUserTurns
356 : undefined;
357 const maxItems =
358 turnLimit === undefined ? policy.maxItems : (policy.maxTurnItems ?? policy.maxItems);
359 for (let index = end - 1; index >= 0; index -= 1) {
360 const row = input.items[index]!;
361 if (
362 selected.length > 0 &&
363 (selected.length >= maxItems ||
364 (turnLimit !== undefined &&
365 (userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS)))
366 ) {
367 break;
368 }
369 const rowBytes = rowCost(row);
370 if (selected.length > 0 && encodedBytes + rowBytes > policy.maxEncodedBytes) break;
371 selected.push(row);
372 encodedBytes += rowBytes;
373 if (isThreadHistoryUserTurn(row.item)) userTurns += 1;
374 if (turnStarts[index]) rawTurns += 1;
375 }
376 // Inside a turn longer than the window, no selected row starts a user turn.
377 // Later pages and the snapshot query see only such rows and use `maxItems`,
378 // so this page uses it too.
379 const oldestScanned = selected.at(-1);
380 if (userTurns === 0) selected.splice(policy.maxItems);
381 selected.reverse();
382
383 const oldest = selected[0];
384 const oldestIndex = oldest
385 ? input.items.findIndex(
386 (row) =>
387 row.sourceThreadId === oldest.sourceThreadId && row.sourceItemId === oldest.sourceItemId,
388 )
389 : -1;
390 const hasMoreHistory = oldestIndex > 0;
391 const nextCursor =
392 hasMoreHistory && oldest
393 ? encodeThreadHistoryCursor({
394 snapshotSequence: input.snapshotSequence,
395 sourceThreadId: oldest.sourceThreadId,
396 sourceItemId: oldest.sourceItemId,
397 position: oldest.position,
398 })
399 : null;
400
401 return {
402 items: renumberPositions(selected),
403 nextCursor,
404 hasMoreHistory,
405 ...(input.prefix === undefined ? {} : { oldestScanned }),
406 };
407}
408
409export function selectRecentTimelineWindow(input: {
410 readonly items: ReadonlyArray<OrchestrationV2ProjectedTurnItem>;
411 readonly snapshotSequence: number;
412 readonly policy?: ThreadHistoryPagePolicy | undefined;
413 readonly rowEncodedBytes?: ((row: OrchestrationV2ProjectedTurnItem) => number) | undefined;
414}): SelectTimelinePageResult {
415 return selectOlderTimelinePage({
416 items: input.items,
417 exclusiveEndIndex: input.items.length,
418 snapshotSequence: input.snapshotSequence,
419 ...(input.policy === undefined ? {} : { policy: input.policy }),
420 ...(input.rowEncodedBytes === undefined ? {} : { rowEncodedBytes: input.rowEncodedBytes }),
421 });
422}
423
424export function findCursorEndIndex(
425 items: ReadonlyArray<HistoryRow>,
426 cursor: ThreadHistoryCursorPayload,
427): number {
428 const byIdentity = items.findIndex(
429 (row) => String(row.sourceThreadId) === cursor.st && String(row.sourceItemId) === cursor.si,
430 );
431 if (byIdentity !== -1) {
432 return byIdentity + (cursor.inclusive ? 1 : 0);
433 }
434 // Inclusive deferred cursors carry source ordinals; ordinary cursors carry visible positions.
435 const byPosition = items.findIndex((row) =>
436 cursor.inclusive
437 ? String(row.sourceThreadId) === cursor.st && (row.item.ordinal ?? row.position) > cursor.p
438 : row.position >= cursor.p,
439 );
440 return byPosition === -1 ? items.length : byPosition;
441}
442
443/**
444 * Older pages allow twice the turns, and a targeted page widens further so it
445 * reaches `throughEntryId`. Callers that re-bound a page must reuse this policy
446 * or they drop the target again.
447 */
448export function olderHistoryPagePolicy<Row extends HistoryRow>(input: {
449 readonly items: ReadonlyArray<Row>;
450 readonly cursor: string;
451 readonly policy?: ThreadHistoryPagePolicy | undefined;
452 readonly throughEntryId?: string | undefined;
453}): ThreadHistoryPagePolicy {
454 const policy = input.policy ?? {
455 ...THREAD_HISTORY_PAGE_POLICY,
456 maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT,
457 };
458 if (input.throughEntryId === undefined) return policy;
459 const anchorIndex = findCursorEndIndex(input.items, decodeThreadHistoryCursor(input.cursor));
460 const targetIndex = input.items.findIndex(
461 (row) => (row.item.messageId ?? row.item.id) === input.throughEntryId,
462 );
463 if (targetIndex < 0 || targetIndex >= anchorIndex) return policy;
464 const turns = input.items
465 .slice(targetIndex, anchorIndex)
466 .filter((row) => isThreadHistoryUserTurn(row.item)).length;
467 return {
468 ...policy,
469 maxUserTurns: Math.min(
470 OLDER_THREAD_USER_TURN_LIMIT * 2,
471 Math.max(
472 OLDER_THREAD_USER_TURN_LIMIT,
473 turns + (isThreadHistoryUserTurn(input.items[targetIndex]!.item) ? 0 : 1),
474 ),
475 ),
476 maxItems: policy.maxItems * 2,
477 maxTurnItems: (policy.maxTurnItems ?? policy.maxItems) * 2,
478 };
479}
480
481export function selectHistoryPageFromCursor<Row extends HistoryRow>(input: {
482 readonly items: ReadonlyArray<Row>;
483 readonly cursor: string;
484 readonly snapshotSequence: number;
485 readonly policy?: ThreadHistoryPagePolicy | undefined;
486 readonly rowEncodedBytes?: ((row: Row) => number) | undefined;
487 readonly throughEntryId?: string | undefined;
488 /** Where the page ends in `items`. Set when `items` can lack the cursor's row. */
489 readonly exclusiveEndIndex?: number | undefined;
490 /** Set when `items` holds only the newest part of the history. */
491 readonly prefix?: ThreadHistoryPrefix | undefined;
492}): SelectTimelinePageResult<Row> {
493 const decoded = decodeThreadHistoryCursor(input.cursor);
494 return selectOlderTimelinePage({
495 items: input.items,
496 exclusiveEndIndex: input.exclusiveEndIndex ?? findCursorEndIndex(input.items, decoded),
497 snapshotSequence: input.snapshotSequence,
498 ...(input.rowEncodedBytes === undefined ? {} : { rowEncodedBytes: input.rowEncodedBytes }),
499 policy: olderHistoryPagePolicy(input),
500 prefix: input.prefix,
501 });
502}
503
504/**
505 * Row and byte budgets can split a turn between an interrupt result and its
506 * older request. Clients read a superseded result without a request as a
507 * legacy steer and hide it, so a page carries the requests its results need.
508 */
509export function missingInterruptRequests<Row extends HistoryRow>(
510 items: ReadonlyArray<Row>,
511 page: ReadonlyArray<Pick<HistoryRow, "item" | "sourceThreadId" | "sourceItemId" | "position">>,
512): Row[] {
513 const runs = new Set<string>();
514 for (const row of page) {
515 if (row.item.type === "run_interrupt_result" && row.item.runId != null)
516 runs.add(row.item.runId);
517 }
518 for (const row of page) {
519 if (row.item.type === "run_interrupt_request" && row.item.runId != null)
520 runs.delete(row.item.runId);
521 }
522 if (runs.size === 0) return [];
523 return items.filter(
524 (row) =>
525 row.item.type === "run_interrupt_request" &&
526 row.item.runId != null &&
527 runs.has(row.item.runId),
528 );
529}
530
531function isLocalProjectedRow(
532 projection: OrchestrationV2ThreadProjection,
533 row: OrchestrationV2ProjectedTurnItem,
534): boolean {
535 return row.visibility === "local" || String(row.sourceThreadId) === String(projection.thread.id);
536}
537
538function pendingRequestIdsOf(
539 projection: Pick<OrchestrationV2ThreadProjection, "runtimeRequests">,
540): ReadonlySet<string> {
541 return new Set(
542 projection.runtimeRequests
543 .filter((request) => request.status === "pending")
544 .map((request) => String(request.id)),
545 );
546}
547
548// Clients build the answer panel of an open question or approval from its row,
549// and a run that kept working pushes that row out of the newest turns.
550function isRetainedDependency(
551 item: OrchestrationV2TurnItem,
552 pendingRequestIds: ReadonlySet<string>,
553): boolean {
554 return (
555 item.type === "run_interrupt_request" ||
556 ((item.type === "user_input_request" || item.type === "approval_request") &&
557 pendingRequestIds.has(String(item.requestId)))
558 );
559}
560
561/** Retain small requests outside the visible window; later result pages still need them for attempt/run reducers. */
562function retainedInterruptRequestTurnItems(
563 projection: OrchestrationV2ThreadProjection,
564 visible: ReadonlyArray<OrchestrationV2ProjectedTurnItem>,
565): OrchestrationV2TurnItem[] {
566 const visibleLocalIds = new Set<string>();
567 for (const row of visible) {
568 if (isLocalProjectedRow(projection, row)) {
569 visibleLocalIds.add(String(row.sourceItemId));
570 }
571 }
572
573 const pendingRequestIds = pendingRequestIdsOf(projection);
574 const retained: OrchestrationV2TurnItem[] = [];
575 for (const item of projection.turnItems) {
576 if (!isRetainedDependency(item, pendingRequestIds)) {
577 continue;
578 }
579 if (visibleLocalIds.has(String(item.id))) {
580 continue;
581 }
582 retained.push(item);
583 }
584 return retained;
585}
586
587function localTurnItemsForVisibleWindow(
588 projection: OrchestrationV2ThreadProjection,
589 visible: ReadonlyArray<OrchestrationV2ProjectedTurnItem>,
590): OrchestrationV2TurnItem[] {
591 return visible.filter((row) => isLocalProjectedRow(projection, row)).map((row) => row.item);
592}
593
594function messagesForBoundedProjection(
595 projection: OrchestrationV2ThreadProjection,
596 turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
597): OrchestrationV2ThreadProjection["messages"] {
598 const retainedMessageIds = new Set<string>();
599 for (const item of turnItems) {
600 if ("messageId" in item) {
601 retainedMessageIds.add(String(item.messageId));
602 }
603 }
604
605 const latestRun = projection.runs.reduce<(typeof projection.runs)[number] | null>(
606 (latest, run) => (latest === null || run.ordinal > latest.ordinal ? run : latest),
607 null,
608 );
609 const retainedRunIds = new Set<string>();
610 for (const run of projection.runs) {
611 if (
612 run.id === latestRun?.id ||
613 run.status === "preparing" ||
614 run.status === "starting" ||
615 run.status === "running" ||
616 run.status === "waiting" ||
617 run.status === "queued"
618 ) {
619 retainedRunIds.add(String(run.id));
620 retainedMessageIds.add(String(run.userMessageId));
621 }
622 }
623
624 return projection.messages.filter(
625 (message) =>
626 retainedMessageIds.has(String(message.id)) ||
627 (message.runId !== null && retainedRunIds.has(String(message.runId))) ||
628 message.delegatedCompletion !== undefined ||
629 // Keep the workflow script outside the history window; clients use it to open the coordinator thread.
630 message.workflowScript === true,
631 );
632}
633
634/** Watermark prevents old missing updates from appending newest when bounded windows lack local rows. */
635export function computeLatestLocalTurnOrdinal(
636 turnItems: ReadonlyArray<{ readonly ordinal: number }>,
637): number | null {
638 let latest: number | null = null;
639 for (const item of turnItems) {
640 if (latest === null || item.ordinal > latest) {
641 latest = item.ordinal;
642 }
643 }
644 return latest;
645}
646
647export function boundedTimelineEncodedBytes(input: {
648 readonly visibleTurnItems: ReadonlyArray<OrchestrationV2ProjectedTurnItem>;
649 readonly turnItems: ReadonlyArray<OrchestrationV2TurnItem>;
650}): number {
651 let total = 0;
652 for (const row of input.visibleTurnItems) {
653 total += projectedRowEncodedBytes(row);
654 }
655 for (const item of input.turnItems) {
656 total += bytesOfJson(item);
657 }
658 return total;
659}
660
661export function buildBoundedThreadProjection(input: {
662 readonly projection: OrchestrationV2ThreadProjection;
663 readonly snapshotSequence: number;
664 readonly policy?: ThreadHistoryPagePolicy | undefined;
665}): BoundedProjectionResult {
666 const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY;
667 const threadId = input.projection.thread.id;
668 const runOrdinals = new Map(input.projection.runs.map((run) => [run.id, run.ordinal]));
669 let latestTodoList: (typeof input.projection.plans)[number] | undefined;
670 let latestTodoListOrdinal = -1;
671 for (const plan of input.projection.plans) {
672 const ordinal = plan.runId === null ? undefined : runOrdinals.get(plan.runId);
673 if (
674 plan.kind === "todo_list" &&
675 plan.status !== "superseded" &&
676 ordinal !== undefined &&
677 ordinal >= latestTodoListOrdinal
678 ) {
679 latestTodoList = plan;
680 latestTodoListOrdinal = ordinal;
681 }
682 }
683 const controlProjection = {
684 ...input.projection,
685 plans: input.projection.plans.filter(
686 (plan) => plan.status === "active" || plan.id === latestTodoList?.id,
687 ),
688 contextHandoffs: input.projection.contextHandoffs.filter(
689 (handoff) => handoff.status === "pending" || handoff.status === "ready",
690 ),
691 };
692 const latestLocalTurnOrdinal = computeLatestLocalTurnOrdinal(input.projection.turnItems);
693
694 const dependencyReserve = (() => {
695 let reserve = 0;
696 const pendingRequestIds = pendingRequestIdsOf(controlProjection);
697 for (const item of controlProjection.turnItems) {
698 if (isRetainedDependency(item, pendingRequestIds)) {
699 reserve += bytesOfJson(item);
700 }
701 }
702 return reserve;
703 })();
704 const controlBytes = bytesOfJson({
705 ...controlProjection,
706 messages: [],
707 turnItems: [],
708 visibleTurnItems: [],
709 });
710 // Control state cannot be paged, so the window keeps half the budget; dependency cards take their reserve from that half.
711 const windowBudget = Math.max(
712 Math.max(policy.maxEncodedBytes / 2 - dependencyReserve, 0),
713 policy.maxEncodedBytes - dependencyReserve - controlBytes - 1_024,
714 );
715 const windowPolicy: ThreadHistoryPagePolicy = { ...policy, maxEncodedBytes: windowBudget };
716
717 const window = selectRecentTimelineWindow({
718 items: controlProjection.visibleTurnItems,
719 snapshotSequence: input.snapshotSequence,
720 policy: windowPolicy,
721 rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId),
722 });
723 const visibleTurnItems = renumberPositions(window.items);
724 const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems);
725 const dependencyTurnItems = retainedInterruptRequestTurnItems(
726 controlProjection,
727 visibleTurnItems,
728 );
729 const turnItemById = new Map<string, OrchestrationV2TurnItem>();
730 for (const item of windowTurnItems) {
731 turnItemById.set(String(item.id), item);
732 }
733 for (const item of dependencyTurnItems) {
734 turnItemById.set(String(item.id), item);
735 }
736 const turnItems = [...turnItemById.values()];
737
738 const visiblePlanIds = new Set(
739 turnItems.flatMap((item) =>
740 item.type === "proposed_plan" || item.type === "todo_list" ? [String(item.planId)] : [],
741 ),
742 );
743 const visibleHandoffIds = new Set(
744 turnItems.flatMap((item) => (item.type === "handoff" ? [String(item.contextHandoffId)] : [])),
745 );
746 const plans = input.projection.plans
747 .filter(
748 (plan) =>
749 plan.status === "active" ||
750 plan.id === latestTodoList?.id ||
751 visiblePlanIds.has(String(plan.id)),
752 )
753 .map((plan) =>
754 plan.status === "active" || plan.id === latestTodoList?.id
755 ? plan
756 : plan.kind === "proposed_plan"
757 ? { ...plan, markdown: "", detailInTurnItem: true as const }
758 : { ...plan, explanation: undefined, detailInTurnItem: true as const },
759 );
760 const contextHandoffs = input.projection.contextHandoffs
761 .filter(
762 (handoff) =>
763 handoff.status === "pending" ||
764 handoff.status === "ready" ||
765 visibleHandoffIds.has(String(handoff.id)),
766 )
767 .map((handoff) =>
768 handoff.status === "pending" || handoff.status === "ready"
769 ? handoff
770 : { ...handoff, summaryText: "", detailInTurnItem: true as const },
771 );
772
773 const projection = {
774 ...controlProjection,
775 plans,
776 contextHandoffs,
777 messages: messagesForBoundedProjection(controlProjection, turnItems),
778 turnItems,
779 visibleTurnItems,
780 };
781 return {
782 projection,
783 historyCursor: window.nextCursor,
784 hasMoreHistory: window.hasMoreHistory,
785 latestLocalTurnOrdinal,
786 };
787}