clover's git

the-knife

1	import type {
2	OrchestrationV2ProjectedTurnItem,
3	OrchestrationV2ThreadProjection,
4	OrchestrationV2TurnItem,
5	ThreadId,
6	TurnItemId,
7	} from "@t3tools/contracts";
8
9	/**
10	* Match the V1 conversation windows. Item/byte budgets only apply to histories
11	* without turn starts; tool activity must not split an ordinary conversation turn.
12	*/
13	export const THREAD_HISTORY_PAGE_POLICY = {
14	maxUserTurns: 10,
15	maxItems: 75,
16	maxEncodedBytes: 1_048_576,
17	} as const;
18
19	export const OLDER_THREAD_USER_TURN_LIMIT = 20;
20
21	/** Load conversation rows before the bulk of a turn's folded tool activity. */
22	export function isConversationHistoryItem(item: Pick<OrchestrationV2TurnItem, "type">): boolean {
23	return !["command_execution", "reasoning", "file_change", "dynamic_tool"].includes(item.type);
24	}
25	export const THREAD_HISTORY_MAX_RAW_TURNS = 150;
26
27	/** Extra rows let the projection query retain an inclusive cursor and prove another page exists. */
28	export const THREAD_HISTORY_SNAPSHOT_ROW_LIMIT = THREAD_HISTORY_PAGE_POLICY.maxItems + 2;
29
30	/** Reject absurd cursors before base64 work or JSON parse. */
31	export const THREAD_HISTORY_CURSOR_MAX_LENGTH = 4_096;
32
33	export type ThreadHistoryPagePolicy = {
34	readonly maxUserTurns?: number | undefined;
35	readonly maxItems: number;
36	readonly maxEncodedBytes: number;
37	};
38
39	export type ThreadHistoryCursorPayload = {
40	readonly v: 1;
41	readonly seq: number;
42	readonly st: string;
43	readonly si: string;
44	readonly p: number;
45	};
46
47	type HistoryRow = Pick<
48	OrchestrationV2ProjectedTurnItem,
49	"position" | "visibility" | "sourceThreadId" | "sourceItemId"
50	> & {
51	readonly item: Pick<OrchestrationV2TurnItem, "type"> & {
52	readonly id?: string;
53	readonly messageId?: string;
54	readonly inputIntent?: string;
55	readonly createdBy?: string;
56	readonly runId?: string | null;
57	};
58	};
59	export type SelectTimelinePageResult<Row extends HistoryRow = OrchestrationV2ProjectedTurnItem> = {
60	readonly items: ReadonlyArray<Row>;
61	readonly nextCursor: string | null;
62	readonly hasMoreHistory: boolean;
63	};
64
65	export type BoundedProjectionResult = {
66	readonly projection: OrchestrationV2ThreadProjection;
67	readonly historyCursor: string | null;
68	readonly hasMoreHistory: boolean;
69	/** Max ordinal across the authoritative full projection's local turnItems. */
70	readonly latestLocalTurnOrdinal: number | null;
71	readonly payloadBudgetExceeded: boolean;
72	};
73
74	export class InvalidThreadHistoryCursorError extends Error {
75	readonly _tag = "InvalidThreadHistoryCursorError";
76	constructor(message = "Invalid thread history cursor.") {
77	super(message);
78	this.name = "InvalidThreadHistoryCursorError";
79	}
80	}
81
82	function bytesOfJson(value: unknown): number {
83	return Buffer.byteLength(JSON.stringify(value), "utf8");
84	}
85
86	/** Ordinary history-page cost: one projected row (item is nested once). */
87	export function projectedRowEncodedBytes(row: HistoryRow): number {
88	return bytesOfJson(row);
89	}
90
91	/**
92	* Bounded-snapshot cost: local rows also land in `projection.turnItems`, so
93	* charge the nested item a second time. Inherited rows only appear in the
94	* visible list.
95	*/
96	export function projectedRowBoundedSnapshotEncodedBytes(
97	row: OrchestrationV2ProjectedTurnItem,
98	threadId: ThreadId | string,
99	): number {
100	const rowBytes = projectedRowEncodedBytes(row);
101	if (row.visibility === "local" || String(row.sourceThreadId) === String(threadId)) {
102	return rowBytes + bytesOfJson(row.item);
103	}
104	return rowBytes;
105	}
106
107	function renumberPositions<Row extends HistoryRow>(items: ReadonlyArray<Row>): Row[] {
108	return items.map((row, position) => (row.position === position ? row : { ...row, position }));
109	}
110
111	function encodeCursorPayload(payload: ThreadHistoryCursorPayload): string {
112	return Buffer.from(JSON.stringify(payload), "utf8").toString("base64url");
113	}
114
115	export function encodeThreadHistoryCursor(input: {
116	readonly snapshotSequence: number;
117	readonly sourceThreadId: ThreadId | string;
118	readonly sourceItemId: TurnItemId | string;
119	readonly position: number;
120	}): string {
121	return encodeCursorPayload({
122	v: 1,
123	seq: input.snapshotSequence,
124	st: String(input.sourceThreadId),
125	si: String(input.sourceItemId),
126	p: input.position,
127	});
128	}
129
130	export function decodeThreadHistoryCursor(cursor: string): ThreadHistoryCursorPayload {
131	if (cursor.length === 0 || cursor.length > THREAD_HISTORY_CURSOR_MAX_LENGTH) {
132	throw new InvalidThreadHistoryCursorError();
133	}
134	let parsed: unknown;
135	try {
136	const json = Buffer.from(cursor, "base64url").toString("utf8");
137	parsed = JSON.parse(json) as unknown;
138	} catch {
139	throw new InvalidThreadHistoryCursorError();
140	}
141	if (
142	typeof parsed !== "object" ||
143	parsed === null ||
144	!("v" in parsed) ||
145	(parsed as { v: unknown }).v !== 1 ||
146	!("seq" in parsed) ||
147	typeof (parsed as { seq: unknown }).seq !== "number" ||
148	!Number.isInteger((parsed as { seq: number }).seq) ||
149	(parsed as { seq: number }).seq < 0 ||
150	!("st" in parsed) ||
151	typeof (parsed as { st: unknown }).st !== "string" ||
152	(parsed as { st: string }).st.length === 0 ||
153	!("si" in parsed) ||
154	typeof (parsed as { si: unknown }).si !== "string" ||
155	(parsed as { si: string }).si.length === 0 ||
156	!("p" in parsed) ||
157	typeof (parsed as { p: unknown }).p !== "number" ||
158	!Number.isInteger((parsed as { p: number }).p) ||
159	(parsed as { p: number }).p < 0
160	) {
161	throw new InvalidThreadHistoryCursorError();
162	}
163	return parsed as ThreadHistoryCursorPayload;
164	}
165
166	function isThreadHistoryTurnStart(item: HistoryRow["item"]): boolean {
167	return (
168	item.type === "user_message" &&
169	(item.inputIntent === "turn_start" || item.inputIntent === "queued_turn")
170	);
171	}
172
173	/**
174	* Marks the turn starts of a chronological timeline. A wake (PR watch, task
175	* completion) starts its run with a notification instead of a user message, so
176	* a notification starts a turn when it is its run's first input. One steered
177	* into a running turn comes later. ProjectionStore's `run_starts` window query
178	* applies the same rule.
179	*/
180	export function threadHistoryTurnStarts(
181	items: ReadonlyArray<HistoryRow["item"]>,
182	): ReadonlyArray<boolean> {
183	const startedRuns = new Set<string>();
184	return items.map((item) => {
185	if (item.type !== "user_message" && item.type !== "notification") return false;
186	const firstInput = item.runId != null && !startedRuns.has(item.runId);
187	if (item.runId != null) startedRuns.add(item.runId);
188	return item.type === "notification" ? firstInput : isThreadHistoryTurnStart(item);
189	});
190	}
191
192	/** Steering belongs to its existing turn and must not consume another page slot. */
193	export function isThreadHistoryUserTurn(item: HistoryRow["item"]): boolean {
194	return (
195	isThreadHistoryTurnStart(item) && item.type === "user_message" && item.createdBy === "user"
196	);
197	}
198
199	/**
200	* Walk a chronological timeline backward from the exclusive end, collecting rows
201	* through complete user turns. Histories without turn starts use row budgets
202	* and always admit at least one row so oversized items cannot deadlock paging.
203	*/
204	function selectOlderTimelinePage<Row extends HistoryRow>(input: {
205	readonly items: ReadonlyArray<Row>;
206	readonly exclusiveEndIndex: number;
207	readonly snapshotSequence: number;
208	readonly policy?: ThreadHistoryPagePolicy | undefined;
209	readonly rowEncodedBytes?: ((row: Row) => number) | undefined;
210	}): SelectTimelinePageResult<Row> {
211	const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY;
212	const rowCost = input.rowEncodedBytes ?? projectedRowEncodedBytes;
213	const end = Math.min(Math.max(input.exclusiveEndIndex, 0), input.items.length);
214	if (end <= 0) {
215	return { items: [], nextCursor: null, hasMoreHistory: false };
216	}
217
218	const selected: Row[] = [];
219	let encodedBytes = 0;
220	let userTurns = 0;
221	let rawTurns = 0;
222	const candidates = input.items.slice(0, end);
223	const turnStarts = threadHistoryTurnStarts(candidates.map((row) => row.item));
224	const turnLimit = candidates.some((row) => isThreadHistoryUserTurn(row.item))
225	? policy.maxUserTurns
226	: undefined;
227	for (let index = end - 1; index >= 0; index -= 1) {
228	const row = input.items[index]!;
229	const rowBytes = turnLimit === undefined ? rowCost(row) : 0;
230	if (
231	selected.length > 0 &&
232	(turnLimit === undefined
233	? selected.length >= policy.maxItems || encodedBytes + rowBytes > policy.maxEncodedBytes
234	: userTurns >= turnLimit || rawTurns >= THREAD_HISTORY_MAX_RAW_TURNS)
235	) {
236	break;
237	}
238	selected.push(row);
239	encodedBytes += rowBytes;
240	if (isThreadHistoryUserTurn(row.item)) userTurns += 1;
241	if (turnStarts[index]) rawTurns += 1;
242	}
243	selected.reverse();
244
245	const oldest = selected[0];
246	const oldestIndex = oldest
247	? input.items.findIndex(
248	(row) =>
249	row.sourceThreadId === oldest.sourceThreadId && row.sourceItemId === oldest.sourceItemId,
250	)
251	: -1;
252	const hasMoreHistory = oldestIndex > 0;
253	const nextCursor =
254	hasMoreHistory && oldest
255	? encodeThreadHistoryCursor({
256	snapshotSequence: input.snapshotSequence,
257	sourceThreadId: oldest.sourceThreadId,
258	sourceItemId: oldest.sourceItemId,
259	position: oldest.position,
260	})
261	: null;
262
263	return {
264	items: renumberPositions(selected),
265	nextCursor,
266	hasMoreHistory,
267	};
268	}
269
270	export function selectRecentTimelineWindow(input: {
271	readonly items: ReadonlyArray<OrchestrationV2ProjectedTurnItem>;
272	readonly snapshotSequence: number;
273	readonly policy?: ThreadHistoryPagePolicy | undefined;
274	readonly rowEncodedBytes?: ((row: OrchestrationV2ProjectedTurnItem) => number) | undefined;
275	}): SelectTimelinePageResult {
276	return selectOlderTimelinePage({
277	items: input.items,
278	exclusiveEndIndex: input.items.length,
279	snapshotSequence: input.snapshotSequence,
280	...(input.policy === undefined ? {} : { policy: input.policy }),
281	...(input.rowEncodedBytes === undefined ? {} : { rowEncodedBytes: input.rowEncodedBytes }),
282	});
283	}
284
285	function findCursorIndex(
286	items: ReadonlyArray<Pick<HistoryRow, "sourceThreadId" | "sourceItemId">>,
287	cursor: ThreadHistoryCursorPayload,
288	): number {
289	const byIdentity = items.findIndex(
290	(row) => String(row.sourceThreadId) === cursor.st && String(row.sourceItemId) === cursor.si,
291	);
292	if (byIdentity !== -1) {
293	return byIdentity;
294	}
295	// Identity miss can happen if the cursor item was deleted; fall back to the
296	// recorded position clamped into the current timeline so pagination can
297	// continue from a stable-ish anchor rather than failing hard. Clamp to
298	// items.length (exclusive end) so a shrunken timeline never 400s forever.
299	// decodeThreadHistoryCursor already guarantees p is a non-negative integer.
300	return Math.min(cursor.p, items.length);
301	}
302
303	export function selectHistoryPageFromCursor<Row extends HistoryRow>(input: {
304	readonly items: ReadonlyArray<Row>;
305	readonly cursor: string;
306	readonly snapshotSequence: number;
307	readonly policy?: ThreadHistoryPagePolicy | undefined;
308	readonly rowEncodedBytes?: ((row: Row) => number) | undefined;
309	readonly throughEntryId?: string | undefined;
310	}): SelectTimelinePageResult<Row> {
311	const decoded = decodeThreadHistoryCursor(input.cursor);
312	const anchorIndex = findCursorIndex(input.items, decoded);
313	let policy = input.policy ?? {
314	...THREAD_HISTORY_PAGE_POLICY,
315	maxUserTurns: OLDER_THREAD_USER_TURN_LIMIT,
316	};
317	if (input.throughEntryId !== undefined) {
318	const targetIndex = input.items.findIndex(
319	(row) => (row.item.messageId ?? row.item.id) === input.throughEntryId,
320	);
321	if (targetIndex >= 0 && targetIndex < anchorIndex) {
322	// A search jump can combine two pages, but never grow into an unbounded read.
323	const turns = input.items
324	.slice(targetIndex, anchorIndex)
325	.filter((row) => isThreadHistoryUserTurn(row.item)).length;
326	policy = {
327	...policy,
328	maxUserTurns: Math.min(
329	OLDER_THREAD_USER_TURN_LIMIT * 2,
330	Math.max(
331	OLDER_THREAD_USER_TURN_LIMIT,
332	turns + (isThreadHistoryUserTurn(input.items[targetIndex]!.item) ? 0 : 1),
333	),
334	),
335	maxItems: policy.maxItems * 2,
336	};
337	}
338	}
339	return selectOlderTimelinePage({
340	items: input.items,
341	exclusiveEndIndex: anchorIndex,
342	snapshotSequence: input.snapshotSequence,
343	...(input.rowEncodedBytes === undefined ? {} : { rowEncodedBytes: input.rowEncodedBytes }),
344	policy,
345	});
346	}
347
348	function isLocalProjectedRow(
349	projection: OrchestrationV2ThreadProjection,
350	row: OrchestrationV2ProjectedTurnItem,
351	): boolean {
352	return row.visibility === "local" || String(row.sourceThreadId) === String(projection.thread.id);
353	}
354
355	/**
356	* Visibility for superseded `run_interrupt_result` rows depends on a matching
357	* `run_interrupt_request` in `turnItems`. Keep every small request item from the
358	* full projection even when it sits outside the recent visible window, so a
359	* later history page that introduces the matching result still has the request
360	* available for live attempt/run reducers.
361	*/
362	function retainedInterruptRequestTurnItems(
363	projection: OrchestrationV2ThreadProjection,
364	visible: ReadonlyArray<OrchestrationV2ProjectedTurnItem>,
365	): OrchestrationV2TurnItem[] {
366	const visibleLocalIds = new Set<string>();
367	for (const row of visible) {
368	if (isLocalProjectedRow(projection, row)) {
369	visibleLocalIds.add(String(row.sourceItemId));
370	}
371	}
372
373	const retained: OrchestrationV2TurnItem[] = [];
374	for (const item of projection.turnItems) {
375	if (item.type !== "run_interrupt_request") {
376	continue;
377	}
378	// Already covered by local turnItems for the visible window.
379	if (visibleLocalIds.has(String(item.id))) {
380	continue;
381	}
382	retained.push(item);
383	}
384	return retained;
385	}
386
387	function localTurnItemsForVisibleWindow(
388	projection: OrchestrationV2ThreadProjection,
389	visible: ReadonlyArray<OrchestrationV2ProjectedTurnItem>,
390	): OrchestrationV2TurnItem[] {
391	const localIds = new Set<string>();
392	for (const row of visible) {
393	if (isLocalProjectedRow(projection, row)) {
394	localIds.add(String(row.sourceItemId));
395	}
396	}
397	if (localIds.size === 0) {
398	return [];
399	}
400	return projection.turnItems.filter((item) => localIds.has(String(item.id)));
401	}
402
403	function messagesForBoundedProjection(
404	projection: OrchestrationV2ThreadProjection,
405	turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
406	): OrchestrationV2ThreadProjection["messages"] {
407	const retainedMessageIds = new Set<string>();
408	for (const item of turnItems) {
409	if ("messageId" in item) {
410	retainedMessageIds.add(String(item.messageId));
411	}
412	}
413
414	const latestRun = projection.runs.reduce<(typeof projection.runs)[number] | null>(
415	(latest, run) => (latest === null || run.ordinal > latest.ordinal ? run : latest),
416	null,
417	);
418	const retainedRunIds = new Set<string>();
419	for (const run of projection.runs) {
420	if (
421	run.id === latestRun?.id ||
422	run.status === "preparing" ||
423	run.status === "starting" ||
424	run.status === "running" ||
425	run.status === "waiting" ||
426	run.status === "queued"
427	) {
428	retainedRunIds.add(String(run.id));
429	retainedMessageIds.add(String(run.userMessageId));
430	}
431	}
432
433	return projection.messages.filter(
434	(message) =>
435	retainedMessageIds.has(String(message.id)) ||
436	(message.runId !== null && retainedRunIds.has(String(message.runId))) ||
437	message.delegatedCompletion !== undefined,
438	);
439	}
440
441	/**
442	* Max local turn ordinal over the authoritative full projection turnItems.
443	* Used as a cheap partial-timeline watermark so old missing turn-item updates
444	* cannot append newest when the bounded window has no local rows.
445	*/
446	export function computeLatestLocalTurnOrdinal(
447	turnItems: ReadonlyArray<{ readonly ordinal: number }>,
448	): number | null {
449	let latest: number | null = null;
450	for (const item of turnItems) {
451	if (latest === null || item.ordinal > latest) {
452	latest = item.ordinal;
453	}
454	}
455	return latest;
456	}
457
458	/**
459	* Encoded timeline contribution for a bounded snapshot: visible rows plus the
460	* duplicated local turnItems and any retained interrupt-request dependencies.
461	*/
462	export function boundedTimelineEncodedBytes(input: {
463	readonly visibleTurnItems: ReadonlyArray<OrchestrationV2ProjectedTurnItem>;
464	readonly turnItems: ReadonlyArray<OrchestrationV2TurnItem>;
465	}): number {
466	let total = 0;
467	for (const row of input.visibleTurnItems) {
468	total += projectedRowEncodedBytes(row);
469	}
470	for (const item of input.turnItems) {
471	total += bytesOfJson(item);
472	}
473	return total;
474	}
475
476	/**
477	* Builds a partial thread projection with full control-plane arrays and a
478	* bounded recent timeline window plus matching local turnItems.
479	*/
480	export function buildBoundedThreadProjection(input: {
481	readonly projection: OrchestrationV2ThreadProjection;
482	readonly snapshotSequence: number;
483	readonly policy?: ThreadHistoryPagePolicy | undefined;
484	}): BoundedProjectionResult {
485	const policy = input.policy ?? THREAD_HISTORY_PAGE_POLICY;
486	const threadId = input.projection.thread.id;
487	const controlProjection = {
488	...input.projection,
489	plans: input.projection.plans.filter((plan) => plan.status === "active"),
490	contextHandoffs: input.projection.contextHandoffs.filter(
491	(handoff) => handoff.status === "pending" || handoff.status === "ready",
492	),
493	};
494	const latestLocalTurnOrdinal = computeLatestLocalTurnOrdinal(input.projection.turnItems);
495
496	// Reserve bytes for small interrupt-request dependencies that may sit outside
497	// the recent window but are required for visibility of results inside it.
498	const dependencyReserve = (() => {
499	// Upper bound: all request items in the full projection. Window selection
500	// uses this reserve so the final contribution stays under the cap.
501	let reserve = 0;
502	for (const item of controlProjection.turnItems) {
503	if (item.type === "run_interrupt_request") {
504	reserve += bytesOfJson(item);
505	}
506	}
507	return reserve;
508	})();
509	const controlBytes = bytesOfJson({
510	...controlProjection,
511	messages: [],
512	turnItems: [],
513	visibleTurnItems: [],
514	});
515	const windowBudget = Math.max(
516	0,
517	policy.maxEncodedBytes - dependencyReserve - controlBytes - 1_024,
518	);
519	const windowPolicy: ThreadHistoryPagePolicy = {
520	maxUserTurns: policy.maxUserTurns,
521	maxItems: policy.maxItems,
522	// Zero still admits the first row via the at-least-one rule.
523	maxEncodedBytes: windowBudget,
524	};
525
526	const window = selectRecentTimelineWindow({
527	items: controlProjection.visibleTurnItems,
528	snapshotSequence: input.snapshotSequence,
529	policy: windowPolicy,
530	rowEncodedBytes: (row) => projectedRowBoundedSnapshotEncodedBytes(row, threadId),
531	});
532	const visibleTurnItems = renumberPositions(window.items);
533	const windowTurnItems = localTurnItemsForVisibleWindow(controlProjection, visibleTurnItems);
534	const dependencyTurnItems = retainedInterruptRequestTurnItems(
535	controlProjection,
536	visibleTurnItems,
537	);
538	const turnItemById = new Map<string, OrchestrationV2TurnItem>();
539	for (const item of windowTurnItems) {
540	turnItemById.set(String(item.id), item);
541	}
542	for (const item of dependencyTurnItems) {
543	turnItemById.set(String(item.id), item);
544	}
545	const turnItems = [...turnItemById.values()];
546
547	const visiblePlanIds = new Set(
548	turnItems.flatMap((item) =>
549	item.type === "proposed_plan" || item.type === "todo_list" ? [String(item.planId)] : [],
550	),
551	);
552	const visibleHandoffIds = new Set(
553	turnItems.flatMap((item) => (item.type === "handoff" ? [String(item.contextHandoffId)] : [])),
554	);
555	const plans = input.projection.plans
556	.filter((plan) => plan.status === "active" || visiblePlanIds.has(String(plan.id)))
557	.map((plan) =>
558	plan.status === "active"
559	? plan
560	: plan.kind === "proposed_plan"
561	? { ...plan, markdown: "", detailInTurnItem: true as const }
562	: { ...plan, explanation: undefined, detailInTurnItem: true as const },
563	);
564	const contextHandoffs = input.projection.contextHandoffs
565	.filter(
566	(handoff) =>
567	handoff.status === "pending" ||
568	handoff.status === "ready" ||
569	visibleHandoffIds.has(String(handoff.id)),
570	)
571	.map((handoff) =>
572	handoff.status === "pending" || handoff.status === "ready"
573	? handoff
574	: { ...handoff, summaryText: "", detailInTurnItem: true as const },
575	);
576
577	const projection = {
578	...controlProjection,
579	plans,
580	contextHandoffs,
581	messages: messagesForBoundedProjection(controlProjection, turnItems),
582	turnItems,
583	visibleTurnItems,
584	};
585	return {
586	projection,
587	historyCursor: window.nextCursor,
588	hasMoreHistory: window.hasMoreHistory,
589	// Always from the full projection so inherited-only windows still carry a
590	// watermark for partial live reducers.
591	latestLocalTurnOrdinal,
592	// Adding timeline rows and historical summaries cannot shrink control state.
593	// Avoid serializing the whole timeline when control state already exceeds the cap.
594	payloadBudgetExceeded:
595	controlBytes > policy.maxEncodedBytes || bytesOfJson(projection) > policy.maxEncodedBytes,
596	};
597	}