1import * as DateTime from "effect/DateTime";
2import type { OrchestrationV2Subagent, OrchestrationV2ThreadProjection } from "@t3tools/contracts";
3import { copySorted } from "@t3tools/shared/Array";
4
5import { isActiveSubagentStatus, isTerminalSubagentStatus } from "./subagentRuntime.ts";
6import { resolveActiveThreadRun } from "./threadWorkflows.ts";
7
8type Projection = OrchestrationV2ThreadProjection;
9type Subagent = OrchestrationV2Subagent;
10type RunId = Projection["runs"][number]["id"];
11
12export interface ThreadTurnSubagents {
13 readonly runId: RunId | null;
14 readonly turnActive: boolean;
15 /** No later turn has run since these agents' turn. */
16 readonly latestTurn: boolean;
17 readonly subagents: ReadonlyArray<Subagent>;
18 readonly liveCount: number;
19 readonly settledCount: number;
20}
21
22export interface SubagentPillSegment {
23 readonly label: string;
24 readonly accessibilityLabel: string;
25}
26
27function orderKey(subagent: Subagent): number {
28 return DateTime.toEpochMillis(subagent.startedAt ?? subagent.updatedAt);
29}
30
31export function deriveThreadTurnSubagents(
32 projection: Pick<Projection, "runs" | "subagents">,
33): ThreadTurnSubagents | null {
34 const rolledBackRunIds = new Set(
35 projection.runs.filter((run) => run.status === "rolled_back").map((run) => run.id),
36 );
37 const eligible = projection.subagents.filter(
38 (subagent) => subagent.runId === null || !rolledBackRunIds.has(subagent.runId),
39 );
40 const [first] = eligible;
41 if (first === undefined) return null;
42 const activeRun = resolveActiveThreadRun(projection);
43 const latestUpdated = eligible.reduce(
44 (latest, subagent) =>
45 DateTime.toEpochMillis(subagent.updatedAt) > DateTime.toEpochMillis(latest.updatedAt)
46 ? subagent
47 : latest,
48 first,
49 );
50 // Ordinal, not array position: a reordered queue can leave the projection's runs out of order.
51 // Queued and never-started runs have not replaced anything yet.
52 const latestStartedRun = projection.runs.reduce<Projection["runs"][number] | null>(
53 (latest, run) =>
54 run.startedAt !== null &&
55 !rolledBackRunIds.has(run.id) &&
56 (latest === null || run.ordinal > latest.ordinal)
57 ? run
58 : latest,
59 null,
60 );
61 const latestStartedRunHasAgents =
62 latestStartedRun !== null &&
63 eligible.some((subagent) => subagent.runId === latestStartedRun.id);
64 const runId =
65 activeRun?.id ?? (latestStartedRunHasAgents ? latestStartedRun.id : latestUpdated.runId);
66 const subagents = copySorted(
67 eligible.filter(
68 (subagent) =>
69 subagent.runId === runId ||
70 isActiveSubagentStatus(subagent.status) ||
71 subagent.workflow !== undefined,
72 ),
73 (left, right) => orderKey(left) - orderKey(right) || left.id.localeCompare(right.id),
74 );
75 if (subagents.length === 0) return null;
76
77 let liveCount = 0;
78 let settledCount = 0;
79 for (const subagent of subagents) {
80 if (isActiveSubagentStatus(subagent.status)) liveCount += 1;
81 else if (isTerminalSubagentStatus(subagent.status)) settledCount += 1;
82 }
83 return {
84 runId,
85 turnActive: activeRun !== null,
86 latestTurn: runId === null || latestStartedRun === null || runId === latestStartedRun.id,
87 subagents,
88 liveCount,
89 settledCount,
90 };
91}
92
93export function resolveSubagentPillSegment(
94 turn: ThreadTurnSubagents | null,
95): SubagentPillSegment | null {
96 if (turn === null) return null;
97 // Finished agents stay reachable until a later turn replaces them; workflows always stay reachable.
98 if (
99 !turn.turnActive &&
100 !turn.latestTurn &&
101 turn.liveCount === 0 &&
102 !turn.subagents.some((subagent) => subagent.workflow !== undefined)
103 ) {
104 return null;
105 }
106 const total = turn.subagents.length;
107 if (turn.liveCount > 0) {
108 return {
109 label: `${turn.liveCount}/${total}`,
110 accessibilityLabel: `${turn.liveCount} of ${total} agents working`,
111 };
112 }
113 return {
114 label: `${turn.settledCount} done`,
115 accessibilityLabel: `${turn.settledCount} ${turn.settledCount === 1 ? "agent" : "agents"} done`,
116 };
117}