1import type {
2 OrchestrationV2ThreadProjection,
3 OrchestrationV2ThreadShell,
4 ThreadId,
5} from "@t3tools/contracts";
6import * as DateTime from "effect/DateTime";
7
8export type ThreadRelationshipKind = "parent" | "fork" | "subagent" | "transfer";
9
10export interface ThreadRelationshipNode {
11 readonly threadId: ThreadId;
12 readonly thread: OrchestrationV2ThreadShell | null;
13 readonly missing: boolean;
14}
15
16export interface ThreadRelationshipEdge {
17 readonly sourceThreadId: ThreadId;
18 readonly targetThreadId: ThreadId;
19 readonly kind: ThreadRelationshipKind;
20 readonly status: string | null;
21}
22
23export interface ThreadRelationshipGraph {
24 readonly nodes: ReadonlyMap<ThreadId, ThreadRelationshipNode>;
25 readonly edges: ReadonlyArray<ThreadRelationshipEdge>;
26}
27
28export interface ThreadRelationshipWalkRow {
29 readonly threadId: ThreadId;
30 readonly fromThreadId: ThreadId;
31 readonly depth: number;
32 readonly edge: ThreadRelationshipEdge;
33}
34
35export function resolveMergeBackTargetThreadId(
36 projection: Pick<OrchestrationV2ThreadProjection, "thread"> | null,
37): ThreadId | null {
38 if (projection?.thread.lineage.relationshipToParent !== "fork") return null;
39 return projection.thread.forkedFrom?.type === "run"
40 ? projection.thread.forkedFrom.threadId
41 : projection.thread.lineage.parentThreadId;
42}
43
44function edgeKey(edge: ThreadRelationshipEdge): string {
45 return `${edge.sourceThreadId}\u001f${edge.targetThreadId}\u001f${edge.kind}`;
46}
47
48export function deriveThreadRelationshipGraph(input: {
49 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
50 readonly projection: OrchestrationV2ThreadProjection | null;
51}): ThreadRelationshipGraph {
52 const threadsById = new Map<ThreadId, OrchestrationV2ThreadShell>();
53 for (const thread of input.threads) {
54 // Callers order shells from most to least authoritative. In particular,
55 // live shells precede archived snapshots, which may still contain a stale
56 // copy during archive refresh.
57 if (!threadsById.has(thread.id)) {
58 threadsById.set(thread.id, thread);
59 }
60 }
61 const threads = [...threadsById.values()];
62 const nodes = new Map<ThreadId, ThreadRelationshipNode>(
63 threads.map((thread) => [thread.id, { threadId: thread.id, thread, missing: false }]),
64 );
65 const edgesByKey = new Map<string, ThreadRelationshipEdge>();
66 const ensureNode = (threadId: ThreadId) => {
67 if (!nodes.has(threadId)) {
68 nodes.set(threadId, { threadId, thread: null, missing: true });
69 }
70 };
71 const addEdge = (edge: ThreadRelationshipEdge) => {
72 ensureNode(edge.sourceThreadId);
73 ensureNode(edge.targetThreadId);
74 edgesByKey.set(
75 edgeKey(edge),
76 edge.kind === "subagent" && edge.status === "rolled_back"
77 ? { ...edge, status: "interrupted" }
78 : edge,
79 );
80 };
81
82 for (const thread of threads) {
83 const parentThreadId =
84 thread.forkedFrom?.type === "run"
85 ? thread.forkedFrom.threadId
86 : thread.lineage.parentThreadId;
87 if (parentThreadId === null) continue;
88 addEdge({
89 sourceThreadId: parentThreadId,
90 targetThreadId: thread.id,
91 kind: thread.lineage.relationshipToParent === "subagent" ? "subagent" : "fork",
92 status: thread.activityRunStatus ?? thread.status,
93 });
94 }
95
96 if (input.projection !== null) {
97 const ownerThreadId = input.projection.thread.id;
98 for (const subagent of input.projection.subagents) {
99 if (subagent.childThreadId === null) continue;
100 const child = threadsById.get(subagent.childThreadId);
101 // The row describes the child conversation, including queued follow-ups
102 // and their outcomes. A child without an app run still uses its task.
103 addEdge({
104 sourceThreadId: ownerThreadId,
105 targetThreadId: subagent.childThreadId,
106 kind: "subagent",
107 status:
108 child?.activityRunStatus ?? (child?.latestRunId != null ? child.status : subagent.status),
109 });
110 }
111 for (const transfer of input.projection.contextTransfers) {
112 if (transfer.sourceThreadId === transfer.targetThreadId) continue;
113 addEdge({
114 sourceThreadId: transfer.sourceThreadId,
115 targetThreadId: transfer.targetThreadId,
116 kind: "transfer",
117 status: transfer.status,
118 });
119 }
120 }
121
122 return { nodes, edges: [...edgesByKey.values()] };
123}
124
125export function relatedThreadIds(
126 graph: ThreadRelationshipGraph,
127 threadId: ThreadId,
128): ReadonlyArray<ThreadId> {
129 const ids = new Set<ThreadId>();
130 for (const edge of graph.edges) {
131 if (edge.sourceThreadId === threadId) ids.add(edge.targetThreadId);
132 if (edge.targetThreadId === threadId) ids.add(edge.sourceThreadId);
133 }
134 return [...ids];
135}
136
137export function walkThreadRelationships(
138 graph: ThreadRelationshipGraph,
139 threadId: ThreadId,
140): ReadonlyArray<ThreadRelationshipWalkRow> {
141 const visited = new Set<ThreadId>([threadId]);
142 const pending: Array<{ readonly threadId: ThreadId; readonly depth: number }> = [
143 { threadId, depth: 0 },
144 ];
145 const rows: ThreadRelationshipWalkRow[] = [];
146
147 for (let index = 0; index < pending.length; index += 1) {
148 const current = pending[index];
149 if (current === undefined) continue;
150 for (const edge of graph.edges) {
151 const relatedId =
152 edge.sourceThreadId === current.threadId
153 ? edge.targetThreadId
154 : edge.targetThreadId === current.threadId
155 ? edge.sourceThreadId
156 : null;
157 if (relatedId === null || visited.has(relatedId)) continue;
158 visited.add(relatedId);
159 const depth = current.depth + 1;
160 rows.push({ threadId: relatedId, fromThreadId: current.threadId, depth, edge });
161 pending.push({ threadId: relatedId, depth });
162 }
163 }
164
165 return rows;
166}
167
168export function immediateThreadRelationships(
169 graph: ThreadRelationshipGraph,
170 threadId: ThreadId,
171): ReadonlyArray<ThreadRelationshipWalkRow> {
172 const visited = new Set<ThreadId>();
173 const rows: ThreadRelationshipWalkRow[] = [];
174
175 for (const edge of graph.edges) {
176 const relatedId =
177 edge.sourceThreadId === threadId
178 ? edge.targetThreadId
179 : edge.targetThreadId === threadId
180 ? edge.sourceThreadId
181 : null;
182 if (relatedId === null || visited.has(relatedId)) continue;
183 visited.add(relatedId);
184 rows.push({ threadId: relatedId, fromThreadId: threadId, depth: 1, edge });
185 }
186
187 return rows;
188}
189
190/** True when `edge` reaches `currentThreadId` from its parent or owning agent. */
191export function isParentThreadRelationship(
192 edge: ThreadRelationshipEdge,
193 currentThreadId: ThreadId,
194): boolean {
195 return edge.kind !== "transfer" && edge.targetThreadId === currentThreadId;
196}
197
198/** An incoming parent row shows its own activity, not the child's edge status. */
199export function threadRelationshipRowStatus(
200 graph: ThreadRelationshipGraph,
201 row: Pick<ThreadRelationshipWalkRow, "threadId" | "edge">,
202): string | null {
203 if (row.edge.kind === "transfer" || row.threadId === row.edge.targetThreadId) {
204 return row.edge.status;
205 }
206 const thread = graph.nodes.get(row.threadId)?.thread;
207 return thread?.activityRunStatus ?? thread?.status ?? null;
208}
209
210function threadCreatedAtMillis(node: ThreadRelationshipNode | undefined): number | null {
211 // `createdAt` is typed as a DateTime, but the value reaches here from a
212 // decoded shell that may be missing (a related thread we have no shell for).
213 const createdAt: unknown = node?.thread?.createdAt;
214 if (!DateTime.isDateTime(createdAt)) return null;
215 const millis = DateTime.toEpochMillis(createdAt);
216 return Number.isFinite(millis) ? millis : null;
217}
218
219/**
220 * Orders the web thread-details Lineage rows for display.
221 *
222 * Web-specific by design: the panel pins the parent row first and a distinct
223 * merge-back target second so their actions stay where the user expects, and
224 * only then falls back to newest-created-first. Mobile does not share that
225 * exception, so this is not the canonical relationship order and should not be
226 * reused as one.
227 *
228 * Ordering below the pins is `createdAt` descending, which is immutable, so
229 * rows never move when messages or status changes arrive on a related thread.
230 * Threads whose shell is missing (or whose `createdAt` did not decode) sink to
231 * the bottom. Ties break by thread id ascending so the order is total.
232 */
233export function orderWebThreadLineageRows(input: {
234 readonly graph: ThreadRelationshipGraph;
235 readonly rows: ReadonlyArray<ThreadRelationshipWalkRow>;
236 readonly currentThreadId: ThreadId;
237 readonly mergeTargetThreadId: ThreadId | null;
238}): ReadonlyArray<ThreadRelationshipWalkRow> {
239 const pinRank = (row: ThreadRelationshipWalkRow): number => {
240 if (isParentThreadRelationship(row.edge, input.currentThreadId)) return 0;
241 if (row.threadId === input.mergeTargetThreadId) return 1;
242 return 2;
243 };
244
245 return [...input.rows].sort((left, right) => {
246 const rankDelta = pinRank(left) - pinRank(right);
247 if (rankDelta !== 0) return rankDelta;
248 const leftCreatedAt = threadCreatedAtMillis(input.graph.nodes.get(left.threadId));
249 const rightCreatedAt = threadCreatedAtMillis(input.graph.nodes.get(right.threadId));
250 if (leftCreatedAt !== rightCreatedAt) {
251 if (leftCreatedAt === null) return 1;
252 if (rightCreatedAt === null) return -1;
253 return rightCreatedAt - leftCreatedAt;
254 }
255 return left.threadId < right.threadId ? -1 : left.threadId > right.threadId ? 1 : 0;
256 });
257}