import type { ChatAttachment, OrchestrationV2ProjectedTurnItem, OrchestrationV2ProviderCapabilities, OrchestrationV2ThreadProjection,} from "@t3tools/contracts";import { copySorted } from "@t3tools/shared/Array"; type Projection = OrchestrationV2ThreadProjection;type Run = Projection["runs"][number];type ProviderSession = Projection["providerSessions"][number]; const ACTIVE_RUN_STATUSES = new Set<Run["status"]>(["preparing", "starting", "running", "waiting"]);const MERGE_BACK_RUN_STATUSES = new Set<Run["status"]>(["waiting", "completed"]);const MERGE_BACK_BLOCKING_RUN_STATUSES = new Set<Run["status"]>([ "preparing", "starting", "running",]); export interface QueuedThreadRun { readonly run: Run; readonly text: string; readonly attachments: ReadonlyArray<ChatAttachment>;} export interface ThreadQueueWorkflowState { readonly activeRun: Run | null; readonly queuedRuns: ReadonlyArray<QueuedThreadRun>; readonly canReorder: boolean; readonly canPromoteToSteer: boolean;} export function resolveActiveThreadRun(projection: Projection): Run | null { return projection.runs.findLast((run) => ACTIVE_RUN_STATUSES.has(run.status)) ?? null;} /** * A successfully finished provider turn remains in `waiting` while its * checkpoint is captured. Keep that newest turn available for merge-back * instead of falling through to an older fully checkpointed run. */export function resolveLatestMergeBackRun(projection: Projection): Run | null { const latestProviderFinishedRun = projection.runs.reduce<Run | null>( (latest, run) => MERGE_BACK_RUN_STATUSES.has(run.status) && (latest === null || run.ordinal > latest.ordinal) ? run : latest, null, ); if (latestProviderFinishedRun === null) return null; const hasNewerActiveRun = projection.runs.some( (run) => run.ordinal > latestProviderFinishedRun.ordinal && MERGE_BACK_BLOCKING_RUN_STATUSES.has(run.status), ); return hasNewerActiveRun ? null : latestProviderFinishedRun;} export function resolveThreadProviderSession(projection: Projection): ProviderSession | null { const activeRun = resolveActiveThreadRun(projection); const providerThreadId = activeRun?.providerThreadId ?? projection.thread.activeProviderThreadId; const activeProviderThread = providerThreadId === null ? null : (projection.providerThreads.find((thread) => thread.id === providerThreadId) ?? null); const attachedProviderThread = activeProviderThread ?? projection.providerThreads.find( (thread) => thread.appThreadId === projection.thread.id && thread.providerSessionId !== null, ) ?? null; const sessionId = attachedProviderThread?.providerSessionId ?? null; if (sessionId !== null) { return projection.providerSessions.find((session) => session.id === sessionId) ?? null; } return ( projection.providerSessions.findLast( (session) => session.status !== "stopped" && session.status !== "error", ) ?? null );} export function deriveThreadQueueWorkflowState(projection: Projection): ThreadQueueWorkflowState { const activeRun = resolveActiveThreadRun(projection); const session = resolveThreadProviderSession(projection); const capabilities = session?.capabilities.turns; const hasSteerableProviderTurn = activeRun?.status === "running" && activeRun.activeAttemptId !== null && projection.providerTurns.some( (turn) => turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running", ); const automaticCompletionMessageIds = new Set( projection.messages .filter((message) => message.delegatedCompletion !== undefined) .map((message) => message.id), ); const queuedRuns = copySorted( projection.runs.filter( (run) => run.status === "queued" && !automaticCompletionMessageIds.has(run.userMessageId), ), (left, right) => (left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) || left.ordinal - right.ordinal, ).map((run) => { const message = projection.messages.find((candidate) => candidate.id === run.userMessageId); return { run, text: message?.text ?? "Queued message", attachments: message?.attachments ?? [], }; }); return { activeRun, queuedRuns, canReorder: capabilities?.supportsQueuedMessages === true, canPromoteToSteer: hasSteerableProviderTurn && (capabilities?.supportsActiveSteering === true || capabilities?.supportsSteeringByInterruptRestart === true), };} export function canForkProjectedAssistantItem(input: { readonly projectedItem: OrchestrationV2ProjectedTurnItem; readonly capabilities?: OrchestrationV2ProviderCapabilities | undefined;}): boolean { const item = input.projectedItem.item; if (item.type !== "assistant_message" || item.runId === null || item.status !== "completed") { return false; } if (input.capabilities === undefined) { // Historical and inherited rows may outlive their provider-session record. // Keep the portable server-side fallback available when capability evidence // is absent; a known incapable provider is rejected below. return true; } const capabilities = input.capabilities; const canForkNatively = capabilities.threads.canForkThread && capabilities.threads.canForkFromTurn && capabilities.identity.nativeThreadIds === "strong"; return canForkNatively || capabilities.context.supportsFullThreadHandoff;} export function canDetachThreadProviderSession(projection: Projection): boolean { const session = resolveThreadProviderSession(projection); return session !== null && session.status !== "stopped" && session.status !== "error";}
import type {
ChatAttachment,
OrchestrationV2ProjectedTurnItem,
OrchestrationV2ProviderCapabilities,
OrchestrationV2ThreadProjection,
} from "@t3tools/contracts";
import { copySorted } from "@t3tools/shared/Array";
type Projection = OrchestrationV2ThreadProjection;
type Run = Projection["runs"][number];
type ProviderSession = Projection["providerSessions"][number];
const ACTIVE_RUN_STATUSES = new Set<Run["status"]>(["preparing", "starting", "running", "waiting"]);
const MERGE_BACK_RUN_STATUSES = new Set<Run["status"]>(["waiting", "completed"]);
const MERGE_BACK_BLOCKING_RUN_STATUSES = new Set<Run["status"]>([
"preparing",
"starting",
"running",
]);
export interface QueuedThreadRun {
readonly run: Run;
readonly text: string;
readonly attachments: ReadonlyArray<ChatAttachment>;
}
export interface ThreadQueueWorkflowState {
readonly activeRun: Run | null;
readonly queuedRuns: ReadonlyArray<QueuedThreadRun>;
readonly canReorder: boolean;
readonly canPromoteToSteer: boolean;
}
export function resolveActiveThreadRun(projection: Projection): Run | null {
return projection.runs.findLast((run) => ACTIVE_RUN_STATUSES.has(run.status)) ?? null;
}
/**
* A successfully finished provider turn remains in `waiting` while its
* checkpoint is captured. Keep that newest turn available for merge-back
* instead of falling through to an older fully checkpointed run.
*/
export function resolveLatestMergeBackRun(projection: Projection): Run | null {
const latestProviderFinishedRun = projection.runs.reduce<Run | null>(
(latest, run) =>
MERGE_BACK_RUN_STATUSES.has(run.status) && (latest === null || run.ordinal > latest.ordinal)
? run
: latest,
null,
);
if (latestProviderFinishedRun === null) return null;
const hasNewerActiveRun = projection.runs.some(
(run) =>
run.ordinal > latestProviderFinishedRun.ordinal &&
MERGE_BACK_BLOCKING_RUN_STATUSES.has(run.status),
);
return hasNewerActiveRun ? null : latestProviderFinishedRun;
}
export function resolveThreadProviderSession(projection: Projection): ProviderSession | null {
const activeRun = resolveActiveThreadRun(projection);
const providerThreadId = activeRun?.providerThreadId ?? projection.thread.activeProviderThreadId;
const activeProviderThread =
providerThreadId === null
? null
: (projection.providerThreads.find((thread) => thread.id === providerThreadId) ?? null);
const attachedProviderThread =
activeProviderThread ??
projection.providerThreads.find(
(thread) => thread.appThreadId === projection.thread.id && thread.providerSessionId !== null,
) ??
null;
const sessionId = attachedProviderThread?.providerSessionId ?? null;
if (sessionId !== null) {
return projection.providerSessions.find((session) => session.id === sessionId) ?? null;
}
return (
projection.providerSessions.findLast(
(session) => session.status !== "stopped" && session.status !== "error",
) ?? null
);
}
export function deriveThreadQueueWorkflowState(projection: Projection): ThreadQueueWorkflowState {
const activeRun = resolveActiveThreadRun(projection);
const session = resolveThreadProviderSession(projection);
const capabilities = session?.capabilities.turns;
const hasSteerableProviderTurn =
activeRun?.status === "running" &&
activeRun.activeAttemptId !== null &&
projection.providerTurns.some(
(turn) => turn.runAttemptId === activeRun.activeAttemptId && turn.status === "running",
);
const automaticCompletionMessageIds = new Set(
projection.messages
.filter((message) => message.delegatedCompletion !== undefined)
.map((message) => message.id),
);
const queuedRuns = copySorted(
projection.runs.filter(
(run) => run.status === "queued" && !automaticCompletionMessageIds.has(run.userMessageId),
),
(left, right) =>
(left.queuePosition ?? left.ordinal) - (right.queuePosition ?? right.ordinal) ||
left.ordinal - right.ordinal,
).map((run) => {
const message = projection.messages.find((candidate) => candidate.id === run.userMessageId);
return {
run,
text: message?.text ?? "Queued message",
attachments: message?.attachments ?? [],
};
});
return {
activeRun,
queuedRuns,
canReorder: capabilities?.supportsQueuedMessages === true,
canPromoteToSteer:
hasSteerableProviderTurn &&
(capabilities?.supportsActiveSteering === true ||
capabilities?.supportsSteeringByInterruptRestart === true),
};
}
export function canForkProjectedAssistantItem(input: {
readonly projectedItem: OrchestrationV2ProjectedTurnItem;
readonly capabilities?: OrchestrationV2ProviderCapabilities | undefined;
}): boolean {
const item = input.projectedItem.item;
if (item.type !== "assistant_message" || item.runId === null || item.status !== "completed") {
return false;
}
if (input.capabilities === undefined) {
// Historical and inherited rows may outlive their provider-session record.
// Keep the portable server-side fallback available when capability evidence
// is absent; a known incapable provider is rejected below.
return true;
}
const capabilities = input.capabilities;
const canForkNatively =
capabilities.threads.canForkThread &&
capabilities.threads.canForkFromTurn &&
capabilities.identity.nativeThreadIds === "strong";
return canForkNatively || capabilities.context.supportsFullThreadHandoff;
}
export function canDetachThreadProviderSession(projection: Projection): boolean {
const session = resolveThreadProviderSession(projection);
return session !== null && session.status !== "stopped" && session.status !== "error";
}