1import { applyThreadPullRequestLinkUpdate } from "@t3tools/shared/threadPullRequests";
2import type {
3 OrchestrationV2DomainEvent,
4 OrchestrationV2Run,
5 OrchestrationV2ThreadProjection,
6 OrchestrationV2TurnItem,
7} from "@t3tools/contracts";
8import {
9 createOrchestrationV2TurnItemVisibility,
10 isOrchestrationV2TurnItemVisible,
11} from "@t3tools/shared/orchestrationV2Timeline";
12
13export type ApplyOrchestrationV2ProjectionEventOptions = {
14 readonly partialTimeline?: boolean;
15 readonly latestLocalTurnOrdinal?: number | null;
16};
17
18function upsertEntity<T extends { readonly id: unknown }>(
19 items: ReadonlyArray<T>,
20 item: T,
21): ReadonlyArray<T> {
22 const index = items.findIndex((candidate) => candidate.id === item.id);
23 if (index === -1) return [...items, item];
24 const next = [...items];
25 next[index] = item;
26 return next;
27}
28
29function removeVisibleItem(
30 rows: OrchestrationV2ThreadProjection["visibleTurnItems"],
31 sourceItemId: OrchestrationV2TurnItem["id"],
32): OrchestrationV2ThreadProjection["visibleTurnItems"] {
33 const index = rows.findIndex((row) => row.sourceItemId === sourceItemId);
34 if (index === -1) return rows;
35 return renumberVisibleItems([...rows.slice(0, index), ...rows.slice(index + 1)]);
36}
37
38function renumberVisibleItems(
39 rows: OrchestrationV2ThreadProjection["visibleTurnItems"],
40): OrchestrationV2ThreadProjection["visibleTurnItems"] {
41 return rows.map((row, position) => (row.position === position ? row : { ...row, position }));
42}
43
44function shouldShowLocalTurnItem(
45 projection: OrchestrationV2ThreadProjection,
46 item: OrchestrationV2TurnItem,
47): boolean {
48 return isOrchestrationV2TurnItemVisible({
49 item,
50 runs: projection.runs,
51 attempts: projection.attempts,
52 items: projection.turnItems,
53 });
54}
55
56function activeVisibleTurnItems(
57 projection: OrchestrationV2ThreadProjection,
58): OrchestrationV2ThreadProjection["visibleTurnItems"] {
59 const rows = projection.visibleTurnItems;
60 const isVisible = createOrchestrationV2TurnItemVisibility({
61 runs: projection.runs,
62 attempts: projection.attempts,
63 items: projection.turnItems,
64 });
65 let next: Array<(typeof rows)[number]> | null = null;
66 for (let index = 0; index < rows.length; index += 1) {
67 const row = rows[index]!;
68 const keep = row.visibility !== "local" || isVisible(row.item);
69 if (!keep) {
70 next ??= rows.slice(0, index);
71 continue;
72 }
73 next?.push(row);
74 }
75 return next === null ? rows : renumberVisibleItems(next);
76}
77
78function oldestLocalTurnOrdinal(
79 rows: OrchestrationV2ThreadProjection["visibleTurnItems"],
80): number | null {
81 let oldest: number | null = null;
82 for (const row of rows) {
83 if (row.visibility !== "local") continue;
84 if (oldest === null || row.item.ordinal < oldest) {
85 oldest = row.item.ordinal;
86 }
87 }
88 return oldest;
89}
90
91const UNFINISHED_RUN_STATUSES: ReadonlySet<OrchestrationV2Run["status"]> = new Set([
92 "preparing",
93 "queued",
94 "starting",
95 "running",
96 "waiting",
97]);
98
99function shouldDropMissingPartialTurnItem(
100 projection: OrchestrationV2ThreadProjection,
101 item: OrchestrationV2TurnItem,
102 latestLocalTurnOrdinal: number | null | undefined,
103): boolean {
104 if (projection.visibleTurnItems.some((row) => row.sourceItemId === item.id)) {
105 return false;
106 }
107 if (item.runId === null) {
108 return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal);
109 }
110 // A queued run keeps the ordinal it got when queued, so a steer sent later
111 // can raise the watermark past it. Items of a run that has not finished are
112 // new, not unloaded history: the first lands while the run is still queued,
113 // and a bounded window can start after the ones before it.
114 if (
115 projection.runs.some((run) => run.id === item.runId && UNFINISHED_RUN_STATUSES.has(run.status))
116 ) {
117 return false;
118 }
119 // Later items of a run already in the window can raise the watermark first.
120 if (
121 projection.visibleTurnItems.some(
122 (row) =>
123 row.visibility === "local" &&
124 row.item.runId === item.runId &&
125 row.item.ordinal <= item.ordinal,
126 )
127 ) {
128 return false;
129 }
130 return isBelowPartialWindow(projection, item, latestLocalTurnOrdinal);
131}
132
133function isBelowPartialWindow(
134 projection: OrchestrationV2ThreadProjection,
135 item: OrchestrationV2TurnItem,
136 latestLocalTurnOrdinal: number | null | undefined,
137): boolean {
138 if (
139 latestLocalTurnOrdinal !== null &&
140 latestLocalTurnOrdinal !== undefined &&
141 item.ordinal <= latestLocalTurnOrdinal
142 ) {
143 return true;
144 }
145 const oldest = oldestLocalTurnOrdinal(projection.visibleTurnItems);
146 return oldest !== null && item.ordinal < oldest;
147}
148
149function upsertVisibleTurnItem(
150 projection: OrchestrationV2ThreadProjection,
151 item: OrchestrationV2TurnItem,
152 partialTimeline: boolean,
153 latestLocalTurnOrdinal: number | null | undefined,
154): OrchestrationV2ThreadProjection["visibleTurnItems"] {
155 const rows = projection.visibleTurnItems;
156 const index = rows.findIndex((row) => row.sourceItemId === item.id);
157 const next = {
158 position: 0,
159 visibility: "local" as const,
160 sourceThreadId: item.threadId,
161 sourceItemId: item.id,
162 item,
163 };
164 const previous = index === -1 ? undefined : rows[index]!;
165 if (
166 previous !== undefined &&
167 previous.visibility === next.visibility &&
168 previous.sourceThreadId === next.sourceThreadId &&
169 previous.sourceItemId === next.sourceItemId &&
170 previous.item === item
171 ) {
172 return rows;
173 }
174 if (
175 index === -1 &&
176 partialTimeline &&
177 shouldDropMissingPartialTurnItem(projection, item, latestLocalTurnOrdinal)
178 ) {
179 return rows;
180 }
181 const updated = index === -1 ? [...rows] : [...rows.slice(0, index), ...rows.slice(index + 1)];
182 const insertionIndex = updated.findIndex(
183 (row) =>
184 row.visibility === "local" &&
185 (row.item.ordinal > item.ordinal ||
186 (row.item.ordinal === item.ordinal &&
187 String(row.item.id).localeCompare(String(item.id)) > 0)),
188 );
189 updated.splice(insertionIndex === -1 ? updated.length : insertionIndex, 0, next);
190 return renumberVisibleItems(updated);
191}
192
193/** Applies one committed event to a matching thread projection. */
194export function applyOrchestrationV2ProjectionEvent(
195 projection: OrchestrationV2ThreadProjection | null,
196 event: OrchestrationV2DomainEvent,
197 options?: ApplyOrchestrationV2ProjectionEventOptions,
198): OrchestrationV2ThreadProjection | null {
199 if (projection === null || event.threadId !== projection.thread.id) return projection;
200
201 const partialTimeline = options?.partialTimeline === true;
202 const latestLocalTurnOrdinal = options?.latestLocalTurnOrdinal;
203 const base = { ...projection, updatedAt: event.occurredAt };
204 switch (event.type) {
205 case "thread.pull-request-link-updated":
206 return {
207 ...base,
208 thread: applyThreadPullRequestLinkUpdate(projection.thread, event.payload),
209 };
210 case "thread.created":
211 case "thread.archived":
212 case "thread.unarchived":
213 case "thread.deleted":
214 case "thread.settled":
215 case "thread.unsettled":
216 case "thread.snoozed":
217 case "thread.unsnoozed":
218 case "thread.auto-settle-set":
219 case "thread.pinned":
220 case "thread.unpinned":
221 case "thread.pin-reordered":
222 case "thread.active-reordered":
223 case "thread.metadata-updated":
224 case "thread.pull-request-synced":
225 case "thread.runtime-mode-updated":
226 case "thread.interaction-mode-updated":
227 case "thread.model-selection-updated":
228 case "thread.provider-switched":
229 return { ...base, thread: event.payload };
230 // Visited tracking is read state, not activity: skip the updatedAt bump.
231 case "thread.visited":
232 case "thread.marked-unread":
233 return { ...projection, thread: event.payload };
234 case "run.created":
235 case "run.updated": {
236 const next = { ...base, runs: upsertEntity(base.runs, event.payload) };
237 return { ...next, visibleTurnItems: activeVisibleTurnItems(next) };
238 }
239 case "run.background-work-cancelled":
240 return {
241 ...base,
242 runs: base.runs.map((run) =>
243 run.id === event.payload.runId
244 ? {
245 ...run,
246 restartCancelledBackgroundWork: event.payload.restartCancelledBackgroundWork,
247 }
248 : run,
249 ),
250 };
251 case "run-attempt.created":
252 case "run-attempt.updated": {
253 const next = { ...base, attempts: upsertEntity(base.attempts, event.payload) };
254 return { ...next, visibleTurnItems: activeVisibleTurnItems(next) };
255 }
256 case "node.updated":
257 return { ...base, nodes: upsertEntity(base.nodes, event.payload) };
258 case "subagent.updated":
259 return { ...base, subagents: upsertEntity(base.subagents, event.payload) };
260 case "provider-session.attached":
261 case "provider-session.updated":
262 return {
263 ...base,
264 providerSessions: upsertEntity(base.providerSessions, event.payload),
265 };
266 case "provider-session.detached":
267 return {
268 ...base,
269 providerSessions: base.providerSessions.filter(
270 (session) => session.id !== event.payload.providerSessionId,
271 ),
272 };
273 case "provider-thread.updated":
274 return { ...base, providerThreads: upsertEntity(base.providerThreads, event.payload) };
275 case "provider-turn.updated":
276 return {
277 ...base,
278 providerTurns: upsertEntity(base.providerTurns, {
279 ...event.payload,
280 ...((event.payload.tokenUsage ??
281 base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage) ===
282 undefined
283 ? {}
284 : {
285 tokenUsage:
286 event.payload.tokenUsage ??
287 base.providerTurns.find((turn) => turn.id === event.payload.id)?.tokenUsage,
288 }),
289 }),
290 };
291 case "runtime-request.updated":
292 return { ...base, runtimeRequests: upsertEntity(base.runtimeRequests, event.payload) };
293 case "message.updated":
294 return { ...base, messages: upsertEntity(base.messages, event.payload) };
295 case "plan.updated":
296 return { ...base, plans: upsertEntity(base.plans, event.payload) };
297 case "turn-item.updated": {
298 if (
299 partialTimeline &&
300 !projection.turnItems.some((candidate) => candidate.id === event.payload.id) &&
301 shouldDropMissingPartialTurnItem(projection, event.payload, latestLocalTurnOrdinal)
302 ) {
303 return projection;
304 }
305 const next = { ...base, turnItems: upsertEntity(base.turnItems, event.payload) };
306 // Only interrupt requests can change another item's visibility. Streaming
307 // text/tool updates must not recheck every row against every run.
308 const previous = projection.turnItems.find((item) => item.id === event.payload.id);
309 const visible =
310 event.payload.type === "run_interrupt_request" || previous?.type === "run_interrupt_request"
311 ? { ...next, visibleTurnItems: activeVisibleTurnItems(next) }
312 : next;
313 return {
314 ...next,
315 visibleTurnItems: shouldShowLocalTurnItem(next, event.payload)
316 ? upsertVisibleTurnItem(visible, event.payload, partialTimeline, latestLocalTurnOrdinal)
317 : removeVisibleItem(visible.visibleTurnItems, event.payload.id),
318 };
319 }
320 case "checkpoint-scope.created":
321 return { ...base, checkpointScopes: upsertEntity(base.checkpointScopes, event.payload) };
322 case "checkpoint.captured":
323 return { ...base, checkpoints: upsertEntity(base.checkpoints, event.payload) };
324 case "checkpoint.rollback-requested":
325 return base;
326 case "context-handoff.updated":
327 return { ...base, contextHandoffs: upsertEntity(base.contextHandoffs, event.payload) };
328 case "context-transfer.created":
329 case "context-transfer.updated":
330 return { ...base, contextTransfers: upsertEntity(base.contextTransfers, event.payload) };
331 }
332}