1import { applyThreadPullRequestLinkUpdate } from "../threadPullRequests.ts";
2import type {
3 OrchestrationV2DomainEvent,
4 OrchestrationV2Run,
5 OrchestrationV2ThreadProjection,
6 OrchestrationV2TurnItem,
7} from "@t3tools/contracts";
8import {
9 restoreSubagentBodies,
10 restoreTurnItemBodies,
11 type DedupedBodySource,
12} from "@t3tools/shared/orchestrationV2DedupedBodies";
13import {
14 createOrchestrationV2TurnItemVisibility,
15 isOrchestrationV2TurnItemVisible,
16} from "@t3tools/shared/orchestrationV2Timeline";
17
18export type ApplyOrchestrationV2ProjectionEventOptions = {
19 readonly partialTimeline?: boolean;
20 readonly latestLocalTurnOrdinal?: number | null;
21};
22
23// Only this exact pair differs at one index. Both keys are weak so hints
24// neither retain streaming history nor apply to replacement snapshots.
25const arrayReplacements = new WeakMap<
26 ReadonlyArray<unknown>,
27 WeakMap<ReadonlyArray<unknown>, number>
28>();
29
30export function getProjectionArrayReplacementIndex(
31 previous: ReadonlyArray<unknown>,
32 next: ReadonlyArray<unknown>,
33): number | undefined {
34 return arrayReplacements.get(previous)?.get(next);
35}
36
37function replaceProjectionArrayItem<T>(
38 items: ReadonlyArray<T>,
39 index: number,
40 item: T,
41): ReadonlyArray<T> {
42 const next = items.slice();
43 next[index] = item;
44 let replacements = arrayReplacements.get(items);
45 if (replacements === undefined) {
46 replacements = new WeakMap();
47 arrayReplacements.set(items, replacements);
48 }
49 replacements.set(next, index);
50 return next;
51}
52
53function upsertEntity<T extends { readonly id: unknown }>(
54 items: ReadonlyArray<T>,
55 item: T,
56): ReadonlyArray<T> {
57 const index = items.findIndex((candidate) => candidate.id === item.id);
58 if (index === -1) return [...items, item];
59 const next = [...items];
60 next[index] = item;
61 return next;
62}
63
64// A deduped item repeats a record the stream just sent, so search newest first.
65function latestBodies(projection: OrchestrationV2ThreadProjection): DedupedBodySource {
66 return {
67 messageText: (id) => projection.messages.findLast((message) => message.id === id)?.text,
68 checkpointFiles: (id) =>
69 projection.checkpoints.findLast((checkpoint) => checkpoint.id === id)?.files,
70 subagent: (id) => projection.subagents.findLast((subagent) => subagent.id === id),
71 };
72}
73
74function removeVisibleItem(
75 rows: OrchestrationV2ThreadProjection["visibleTurnItems"],
76 sourceItemId: OrchestrationV2TurnItem["id"],
77): OrchestrationV2ThreadProjection["visibleTurnItems"] {
78 const index = rows.findIndex((row) => row.sourceItemId === sourceItemId);
79 if (index === -1) return rows;
80 return renumberVisibleItems([...rows.slice(0, index), ...rows.slice(index + 1)]);
81}
82
83function renumberVisibleItems(
84 rows: OrchestrationV2ThreadProjection["visibleTurnItems"],
85): OrchestrationV2ThreadProjection["visibleTurnItems"] {
86 return rows.map((row, position) => (row.position === position ? row : { ...row, position }));
87}
88
89function shouldShowLocalTurnItem(
90 projection: OrchestrationV2ThreadProjection,
91 item: OrchestrationV2TurnItem,
92): boolean {
93 return isOrchestrationV2TurnItemVisible({
94 item,
95 runs: projection.runs,
96 attempts: projection.attempts,
97 items: projection.turnItems,
98 });
99}
100
101function activeVisibleTurnItems(
102 projection: OrchestrationV2ThreadProjection,
103): OrchestrationV2ThreadProjection["visibleTurnItems"] {
104 const rows = projection.visibleTurnItems;
105 const isVisible = createOrchestrationV2TurnItemVisibility({
106 runs: projection.runs,
107 attempts: projection.attempts,
108 items: projection.turnItems,
109 });
110 let next: Array<(typeof rows)[number]> | null = null;
111 for (let index = 0; index < rows.length; index += 1) {
112 const row = rows[index]!;
113 const keep = row.visibility !== "local" || isVisible(row.item);
114 if (!keep) {
115 next ??= rows.slice(0, index);
116 continue;
117 }
118 next?.push(row);
119 }
120 return next === null ? rows : renumberVisibleItems(next);
121}
122
123function oldestLocalTurnOrdinal(
124 rows: OrchestrationV2ThreadProjection["visibleTurnItems"],
125): number | null {
126 let oldest: number | null = null;
127 for (const row of rows) {
128 if (row.visibility !== "local") continue;
129 if (oldest === null || row.item.ordinal < oldest) {
130 oldest = row.item.ordinal;
131 }
132 }
133 return oldest;
134}
135
136const UNFINISHED_RUN_STATUSES: ReadonlySet<OrchestrationV2Run["status"]> = new Set([
137 "preparing",
138 "queued",
139 "starting",
140 "running",
141 "waiting",
142]);
143
144function shouldDropMissingPartialTurnItem(
145 projection: OrchestrationV2ThreadProjection,
146 item: OrchestrationV2TurnItem,
147 latestLocalTurnOrdinal: number | null | undefined,
148): boolean {
149 if (projection.visibleTurnItems.some((row) => row.sourceItemId === item.id)) {
150 return false;
151 }
152 if (item.runId === null) {
153 return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal);
154 }
155 // A queued run keeps the ordinal it got when queued, so a steer sent later
156 // can raise the watermark past it. Items of a run that has not finished are
157 // new, not unloaded history: the first lands while the run is still queued,
158 // and a bounded window can start after the ones before it.
159 if (
160 projection.runs.some((run) => run.id === item.runId && UNFINISHED_RUN_STATUSES.has(run.status))
161 ) {
162 return false;
163 }
164 if (
165 projection.visibleTurnItems.some(
166 (row) =>
167 row.visibility === "local" &&
168 row.item.runId === item.runId &&
169 row.item.ordinal <= item.ordinal,
170 )
171 ) {
172 return false;
173 }
174 return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal);
175}
176
177function isBelowPartialWindow(
178 projection: OrchestrationV2ThreadProjection,
179 item: OrchestrationV2TurnItem,
180 latestLocalTurnOrdinal: number | null | undefined,
181): boolean {
182 if (
183 latestLocalTurnOrdinal !== null &&
184 latestLocalTurnOrdinal !== undefined &&
185 item.ordinal <= latestLocalTurnOrdinal
186 ) {
187 return true;
188 }
189 const oldest = oldestLocalTurnOrdinal(projection.visibleTurnItems);
190 return oldest !== null && item.ordinal < oldest;
191}
192
193function upsertVisibleTurnItem(
194 projection: OrchestrationV2ThreadProjection,
195 item: OrchestrationV2TurnItem,
196 partialTimeline: boolean,
197 latestLocalTurnOrdinal: number | null | undefined,
198): OrchestrationV2ThreadProjection["visibleTurnItems"] {
199 const rows = projection.visibleTurnItems;
200 // Streamed updates land on the newest rows. Item ids are unique, so the
201 // search can start there.
202 let index = rows.length - 1;
203 while (index >= 0 && rows[index]!.sourceItemId !== item.id) index -= 1;
204 const next = {
205 position: 0,
206 visibility: "local" as const,
207 sourceThreadId: item.threadId,
208 sourceItemId: item.id,
209 item,
210 };
211 const previous = index === -1 ? undefined : rows[index]!;
212 if (
213 previous !== undefined &&
214 previous.visibility === next.visibility &&
215 previous.sourceThreadId === next.sourceThreadId
216 ) {
217 if (previous.item === item) return rows;
218 if (previous.item.ordinal === item.ordinal) {
219 return replaceProjectionArrayItem(rows, index, { ...next, position: previous.position });
220 }
221 }
222 if (
223 index === -1 &&
224 partialTimeline &&
225 shouldDropMissingPartialTurnItem(projection, item, latestLocalTurnOrdinal)
226 ) {
227 return rows;
228 }
229 // The item goes before the first later local row, as if its old row were
230 // removed first. Inserts, reorders, and inherited-to-local changes need
231 // this ordering pass; replacements above preserve every other row.
232 let insertion = rows.length;
233 for (let candidate = 0; candidate < rows.length; candidate += 1) {
234 const row = rows[candidate]!;
235 if (
236 candidate !== index &&
237 row.visibility === "local" &&
238 (row.item.ordinal > item.ordinal ||
239 (row.item.ordinal === item.ordinal &&
240 String(row.item.id).localeCompare(String(item.id)) > 0))
241 ) {
242 insertion = candidate;
243 break;
244 }
245 }
246 const updated: Array<(typeof rows)[number]> = [];
247 for (let source = 0; source < rows.length; source += 1) {
248 if (source === insertion) updated.push(next);
249 if (source !== index) updated.push(rows[source]!);
250 }
251 if (insertion === rows.length) updated.push(next);
252 for (let position = 0; position < updated.length; position += 1) {
253 const row = updated[position]!;
254 if (row.position !== position) updated[position] = { ...row, position };
255 }
256 return updated;
257}
258
259// A user item retained for a queued run stays hidden until the run leaves the
260// queue, and its update can arrive before or after that run update. Items in
261// `turnItems` were loaded or streamed, so they skip the partial-window check;
262// items never received stay absent.
263function restoreQueuedUserItems(
264 projection: OrchestrationV2ThreadProjection,
265 runId: OrchestrationV2Run["id"],
266): OrchestrationV2ThreadProjection["visibleTurnItems"] {
267 let visibleTurnItems = projection.visibleTurnItems;
268 for (const item of projection.turnItems) {
269 if (item.runId !== runId || item.type !== "user_message" || item.inputIntent !== "queued_turn")
270 continue;
271 if (!shouldShowLocalTurnItem(projection, item)) continue;
272 visibleTurnItems = upsertVisibleTurnItem(
273 { ...projection, visibleTurnItems },
274 item,
275 false,
276 null,
277 );
278 }
279 return visibleTurnItems;
280}
281
282export function applyOrchestrationV2ProjectionEvent(
283 projection: OrchestrationV2ThreadProjection | null,
284 event: OrchestrationV2DomainEvent,
285 options?: ApplyOrchestrationV2ProjectionEventOptions,
286): OrchestrationV2ThreadProjection | null {
287 if (projection === null || event.threadId !== projection.thread.id) return projection;
288
289 const partialTimeline = options?.partialTimeline === true;
290 const latestLocalTurnOrdinal = options?.latestLocalTurnOrdinal;
291 const base = { ...projection, updatedAt: event.occurredAt };
292 switch (event.type) {
293 case "thread.pull-request-link-updated":
294 return {
295 ...base,
296 thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload),
297 };
298 case "thread.created":
299 case "thread.archived":
300 case "thread.unarchived":
301 case "thread.deleted":
302 case "thread.settled":
303 case "thread.unsettled":
304 case "thread.snoozed":
305 case "thread.unsnoozed":
306 case "thread.auto-settle-set":
307 case "thread.pinned":
308 case "thread.unpinned":
309 case "thread.pin-reordered":
310 case "thread.active-reordered":
311 case "thread.metadata-updated":
312 case "thread.pull-request-synced":
313 case "thread.runtime-mode-updated":
314 case "thread.interaction-mode-updated":
315 case "thread.model-selection-updated":
316 case "thread.provider-switched":
317 case "thread.guest-share-updated":
318 return { ...base, thread: event.payload };
319 // Visited tracking and tags are not activity: skip the updatedAt bump.
320 case "thread.visited":
321 case "thread.tags-updated":
322 case "thread.marked-unread":
323 return { ...projection, thread: event.payload };
324 case "run.created":
325 case "run.updated": {
326 const next = { ...base, runs: upsertEntity(base.runs, event.payload) };
327 const visible = { ...next, visibleTurnItems: activeVisibleTurnItems(next) };
328 const wasQueued = base.runs.some(
329 (run) => run.id === event.payload.id && run.status === "queued",
330 );
331 return wasQueued && event.payload.status !== "queued"
332 ? { ...visible, visibleTurnItems: restoreQueuedUserItems(visible, event.payload.id) }
333 : visible;
334 }
335 case "run.background-work-cancelled":
336 return {
337 ...base,
338 runs: base.runs.map((run) =>
339 run.id === event.payload.runId
340 ? {
341 ...run,
342 restartCancelledBackgroundWork: event.payload.restartCancelledBackgroundWork,
343 }
344 : run,
345 ),
346 };
347 case "run-attempt.created":
348 case "run-attempt.updated": {
349 const next = { ...base, attempts: upsertEntity(base.attempts, event.payload) };
350 return { ...next, visibleTurnItems: activeVisibleTurnItems(next) };
351 }
352 case "node.updated":
353 return { ...base, nodes: upsertEntity(base.nodes, event.payload) };
354 case "subagent.updated":
355 return {
356 ...base,
357 subagents: upsertEntity(
358 base.subagents,
359 restoreSubagentBodies(
360 event.payload,
361 base.subagents.find((subagent) => subagent.id === event.payload.id),
362 ),
363 ),
364 };
365 case "provider-session.attached":
366 case "provider-session.updated":
367 return {
368 ...base,
369 providerSessions: upsertEntity(base.providerSessions, event.payload),
370 };
371 case "provider-session.detached":
372 return {
373 ...base,
374 providerSessions: base.providerSessions.filter(
375 (session) => session.id !== event.payload.providerSessionId,
376 ),
377 };
378 case "provider-thread.updated":
379 return { ...base, providerThreads: upsertEntity(base.providerThreads, event.payload) };
380 case "provider-turn.updated":
381 return {
382 ...base,
383 providerTurns: upsertEntity(base.providerTurns, {
384 ...event.payload,
385 ...((event.payload.tokenUsage ??
386 base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage) ===
387 undefined
388 ? {}
389 : {
390 tokenUsage:
391 event.payload.tokenUsage ??
392 base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage,
393 }),
394 }),
395 };
396 case "runtime-request.updated":
397 return { ...base, runtimeRequests: upsertEntity(base.runtimeRequests, event.payload) };
398 case "message.updated":
399 return { ...base, messages: upsertEntity(base.messages, event.payload) };
400 case "plan.updated":
401 return { ...base, plans: upsertEntity(base.plans, event.payload) };
402 case "turn-item.updated": {
403 let payload = restoreTurnItemBodies(event.payload, latestBodies(projection));
404 // Streamed updates land on the newest items, so search from the end.
405 let itemIndex = projection.turnItems.length - 1;
406 while (itemIndex >= 0 && projection.turnItems[itemIndex]!.id !== payload.id) {
407 itemIndex -= 1;
408 }
409 const deferredWork =
410 itemIndex === -1 ? undefined : projection.turnItems[itemIndex]?.deferredWork;
411 if (deferredWork !== undefined) payload = { ...payload, deferredWork };
412 if (
413 partialTimeline &&
414 itemIndex === -1 &&
415 shouldDropMissingPartialTurnItem(projection, payload, latestLocalTurnOrdinal)
416 ) {
417 return projection;
418 }
419 const turnItems =
420 itemIndex === -1
421 ? [...projection.turnItems, payload]
422 : replaceProjectionArrayItem(projection.turnItems, itemIndex, payload);
423 const next = { ...base, turnItems };
424 // Only interrupt requests can change another item's visibility. Streaming
425 // text/tool updates must not recheck every row against every run.
426 const previous = itemIndex === -1 ? undefined : projection.turnItems[itemIndex];
427 const visible =
428 payload.type === "run_interrupt_request" || previous?.type === "run_interrupt_request"
429 ? { ...next, visibleTurnItems: activeVisibleTurnItems(next) }
430 : next;
431 return {
432 ...next,
433 visibleTurnItems: shouldShowLocalTurnItem(next, payload)
434 ? upsertVisibleTurnItem(visible, payload, partialTimeline, latestLocalTurnOrdinal)
435 : removeVisibleItem(visible.visibleTurnItems, payload.id),
436 };
437 }
438 case "checkpoint-scope.created":
439 return { ...base, checkpointScopes: upsertEntity(base.checkpointScopes, event.payload) };
440 case "checkpoint.captured":
441 return { ...base, checkpoints: upsertEntity(base.checkpoints, event.payload) };
442 case "checkpoint.rollback-requested":
443 return base;
444 case "context-handoff.updated":
445 return { ...base, contextHandoffs: upsertEntity(base.contextHandoffs, event.payload) };
446 case "context-transfer.created":
447 case "context-transfer.updated":
448 return { ...base, contextTransfers: upsertEntity(base.contextTransfers, event.payload) };
449 }
450}