1import type {
2 ChatAttachment,
3 OrchestrationV2ProjectedTurnItem,
4 OrchestrationV2ProviderCapabilities,
5 OrchestrationV2ThreadProjection,
6} from "@t3tools/contracts";
7import { copySorted } from "@t3tools/shared/Array";
8
9type Projection = OrchestrationV2ThreadProjection;
10type Run = Projection["runs"][number];
11type Message = Projection["messages"][number];
12type ProviderSession = Projection["providerSessions"][number];
13
14const ACTIVE_RUN_STATUSES = new Set<Run["status"]>(["preparing", "starting", "running", "waiting"]);
15const MERGE_BACK_RUN_STATUSES = new Set<Run["status"]>(["waiting", "completed"]);
16const MERGE_BACK_BLOCKING_RUN_STATUSES = new Set<Run["status"]>([
17 "preparing",
18 "starting",
19 "running",
20]);
21
22export interface QueuedThreadRun {
23 readonly run: Run;
24 readonly text: string;
25 readonly attachments: ReadonlyArray<ChatAttachment>;
26 /** Editing replaces this message's content, so its id and context travel with the row. */
27 readonly messageId: Message["id"];
28 readonly context?: Message["context"];
29}
30
31export interface ThreadQueueWorkflowState {
32 readonly activeRun: Run | null;
33 readonly queuedRuns: ReadonlyArray<QueuedThreadRun>;
34 readonly isHeld: boolean;
35 readonly canReorder: boolean;
36 readonly canPromoteToSteer: boolean;
37}
38
39export function resolveActiveThreadRun(projection: Pick<Projection, "runs">): Run | null {
40 return projection.runs.findLast((run) => ACTIVE_RUN_STATUSES.has(run.status)) ?? null;
41}
42
43/**
44 * A successfully finished provider turn remains in `waiting` while its
45 * checkpoint is captured. Keep that newest turn available for merge-back
46 * instead of falling through to an older fully checkpointed run.
47 */
48export function resolveLatestMergeBackRun(projection: Projection): Run | null {
49 const latestProviderFinishedRun = projection.runs.reduce<Run | null>(
50 (latest, run) =>
51 MERGE_BACK_RUN_STATUSES.has(run.status) && (latest === null || run.ordinal > latest.ordinal)
52 ? run
53 : latest,
54 null,
55 );
56 if (latestProviderFinishedRun === null) return null;
57
58 const hasNewerActiveRun = projection.runs.some(
59 (run) =>
60 run.ordinal > latestProviderFinishedRun.ordinal &&
61 MERGE_BACK_BLOCKING_RUN_STATUSES.has(run.status),
62 );
63 return hasNewerActiveRun ? null : latestProviderFinishedRun;
64}
65
66function resolveThreadProviderSession(projection: Projection): ProviderSession | null {
67 const activeRun = resolveActiveThreadRun(projection);
68 const providerThreadId = activeRun?.providerThreadId ?? projection.thread.activeProviderThreadId;
69 const activeProviderThread =
70 providerThreadId === null
71 ? null
72 : (projection.providerThreads.find((thread) => thread.id === providerThreadId) ?? null);
73 const attachedProviderThread =
74 activeProviderThread ??
75 projection.providerThreads.find(
76 (thread) => thread.appThreadId === projection.thread.id && thread.providerSessionId !== null,
77 ) ??
78 null;
79 const sessionId = attachedProviderThread?.providerSessionId ?? null;
80 if (sessionId !== null) {
81 return projection.providerSessions.find((session) => session.id === sessionId) ?? null;
82 }
83 return (
84 projection.providerSessions.findLast(
85 (session) => session.status !== "stopped" && session.status !== "error",
86 ) ?? null
87 );
88}
89
90export function threadSupportsProviderHandoff(projection: Projection | null | undefined): boolean {
91 if (projection == null) return false;
92 const session = resolveThreadProviderSession(projection);
93 if (session !== null) {
94 return session.capabilities.sessions.supportsProviderSwitchingViaHandoff;
95 }
96 if (resolveActiveThreadRun(projection) !== null) return false;
97 if (projection.thread.historyOrigin === "v1_import" || projection.runs.length === 0) return true;
98
99 // Detaching a stopped session removes it from the projection, but its native
100 // provider thread remains available for the next turn's portable handoff.
101 return projection.providerThreads.some(
102 (thread) =>
103 thread.id === projection.thread.activeProviderThreadId &&
104 thread.appThreadId === projection.thread.id &&
105 thread.providerInstanceId === projection.thread.modelSelection.instanceId &&
106 thread.nativeThreadRef !== null,
107 );
108}
109
110/** Automatic completion/notification runs are not messages in the user's queue. */
111export function getUserQueuedThreadRuns(
112 projection: Pick<Projection, "runs" | "messages">,
113): ReadonlyArray<Run> {
114 const automaticCompletionMessageIds = new Set(
115 projection.messages
116 .filter(
117 (message) =>
118 message.delegatedCompletion !== undefined || message.notification !== undefined,
119 )
120 .map((message) => message.id),
121 );
122 return projection.runs.filter(
123 (run) => run.status === "queued" && !automaticCompletionMessageIds.has(run.userMessageId),
124 );
125}
126
127export function deriveThreadQueueWorkflowState(projection: Projection): ThreadQueueWorkflowState {
128 const activeRun = resolveActiveThreadRun(projection);
129 const session = resolveThreadProviderSession(projection);
130 const capabilities = session?.capabilities.turns;
131 const hasSteerableProviderTurn =
132 activeRun?.status === "running" &&
133 activeRun.activeAttemptId !== null &&
134 projection.providerTurns.some(
135 (turn) => turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running",
136 );
137 const queuedRuns = copySorted(
138 getUserQueuedThreadRuns(projection),
139 (left, right) =>
140 (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) ||
141 left.ordinal - right.ordinal,
142 ).map((run) => {
143 const message = projection.messages.find((candidate) => candidate.id === run.userMessageId);
144 return {
145 run,
146 text: message?.text ?? "Queued message",
147 attachments: message?.attachments ?? [],
148 messageId: run.userMessageId,
149 ...(message?.context ? { context: message.context } : {}),
150 };
151 });
152
153 return {
154 activeRun,
155 queuedRuns,
156 isHeld: projection.runs.some((run) => run.status === "queued" && run.queueHeld === true),
157 canReorder: capabilities?.supportsQueuedMessages === true,
158 canPromoteToSteer:
159 hasSteerableProviderTurn &&
160 (capabilities?.supportsActiveSteering === true ||
161 capabilities?.supportsSteeringByInterruptRestart === true),
162 };
163}
164
165export function canForkProjectedAssistantItem(input: {
166 readonly projectedItem: OrchestrationV2ProjectedTurnItem;
167 readonly capabilities?: OrchestrationV2ProviderCapabilities | undefined;
168}): boolean {
169 const item = input.projectedItem.item;
170 if (item.type !== "assistant_message" || item.runId === null || item.status !== "completed") {
171 return false;
172 }
173 if (input.capabilities === undefined) {
174 // Historical and inherited rows may outlive their provider-session record.
175 // Keep the portable server-side fallback available when capability evidence
176 // is absent; a known incapable provider is rejected below.
177 return true;
178 }
179 const capabilities = input.capabilities;
180 const canForkNatively =
181 capabilities.threads.canForkThread &&
182 capabilities.threads.canForkFromTurn &&
183 capabilities.identity.nativeThreadIds === "strong";
184 return canForkNatively || capabilities.context.supportsFullThreadHandoff;
185}
186
187export function canDetachThreadProviderSession(projection: Projection): boolean {
188 const session = resolveThreadProviderSession(projection);
189 return session !== null && session.status !== "stopped" && session.status !== "error";
190}