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