1import type {
2 ApplicationStoredEvent,
3 OrchestrationProjectShell,
4 OrchestrationV2ArchivedShellStreamItem,
5 OrchestrationV2ShellSnapshot,
6 OrchestrationV2ThreadShell,
7 OrchestrationV2ThreadShellSnapshot,
8 OrchestrationV2ShellStreamItem,
9 OrchestrationV2StoredEvent,
10} from "@t3tools/contracts";
11import {
12 OrchestrationProjectShell as ProjectShellSchema,
13 OrchestrationV2ThreadShell as ThreadShellSchema,
14} from "@t3tools/contracts";
15import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
16import * as Clock from "effect/Clock";
17import * as Effect from "effect/Effect";
18import * as Schema from "effect/Schema";
19import type * as SqlClient from "effect/sql/SqlClient";
20import * as Stream from "effect/Stream";
21
22/** Build the regular navigation shell without duplicating the archive dataset. */
23export function buildActiveShellSnapshot(input: {
24 readonly projects: ReadonlyArray<OrchestrationProjectShell>;
25 readonly threads: OrchestrationV2ThreadShellSnapshot;
26 readonly snapshotSequence: number;
27}): OrchestrationV2ShellSnapshot {
28 return {
29 schemaVersion: input.threads.schemaVersion,
30 snapshotSequence: input.snapshotSequence,
31 projects: input.projects,
32 threads: input.threads.threads,
33 archivedThreads: [],
34 };
35}
36
37/**
38 * Loads the thread shell, projects and sequence for the shell snapshots.
39 * One transaction covers the reads so they agree; the threads are decoded
40 * after it commits, because the server shares one SQLite connection and
41 * decoding a large shell takes longer than reading it.
42 */
43export const loadShellSnapshotParts = <E1, E2, E3, E4>(input: {
44 readonly sql: SqlClient.SqlClient;
45 readonly readThreads: Effect.Effect<Effect.Effect<OrchestrationV2ThreadShellSnapshot, E2>, E1>;
46 readonly listProjects: Effect.Effect<ReadonlyArray<OrchestrationProjectShell>, E3>;
47 readonly latestSequence: Effect.Effect<number, E4>;
48}) =>
49 Effect.gen(function* () {
50 // Read on the reader worker, off the event loop and the writer.
51 const read = yield* input.sql
52 .withTransaction(
53 Effect.all({
54 decodeThreads: input.readThreads,
55 projects: input.listProjects,
56 snapshotSequence: input.latestSequence,
57 }),
58 )
59 .pipe(NodeSqliteClient.readOnly);
60 return {
61 projects: read.projects,
62 threads: yield* read.decodeThreads,
63 snapshotSequence: read.snapshotSequence,
64 };
65 });
66
67export type ShellApplicationEvent =
68 | Pick<
69 Extract<ApplicationStoredEvent, { readonly aggregateKind: "project" }>,
70 "aggregateKind" | "aggregateId" | "type" | "sequence"
71 >
72 | {
73 readonly sequence: number;
74 readonly event: Pick<OrchestrationV2StoredEvent["event"], "threadId">;
75 };
76
77/** Shell updates refetch an aggregate; drop transcript bodies before retaining an event. */
78export function toShellApplicationEvent(stored: ApplicationStoredEvent): ShellApplicationEvent {
79 return "aggregateKind" in stored
80 ? {
81 aggregateKind: stored.aggregateKind,
82 aggregateId: stored.aggregateId,
83 type: stored.type,
84 sequence: stored.sequence,
85 }
86 : { sequence: stored.sequence, event: { threadId: stored.event.threadId } };
87}
88
89/** Keep only the newest shell-relevant event per project/thread aggregate. */
90export function coalesceShellApplicationEvents<A extends ShellApplicationEvent>(
91 events: ReadonlyArray<A>,
92): ReadonlyArray<A> {
93 const latestByAggregate = new Map<string, A>();
94 for (const stored of events) {
95 const key =
96 "aggregateKind" in stored
97 ? `project:${stored.aggregateId}`
98 : `thread:${stored.event.threadId}`;
99 latestByAggregate.set(key, stored);
100 }
101 return Array.from(latestByAggregate.values()).sort(
102 (left, right) => left.sequence - right.sequence,
103 );
104}
105
106/**
107 * Emit the initial shell prefix strictly first, then merge the post-prefix
108 * tail with enrichment refreshes. Prevents a newer marked enrichment from
109 * landing before the unmarked authoritative initial snapshot.
110 */
111export function composeShellStreamWithEnrichment<A, E, R, A2, E2, R2, A3, E3, R3>(input: {
112 readonly initial: Stream.Stream<A, E, R>;
113 readonly tail: Stream.Stream<A2, E2, R2>;
114 readonly enrichment: Stream.Stream<A3, E3, R3>;
115}): Stream.Stream<A | A2 | A3, E | E2 | E3, R | R2 | R3> {
116 return Stream.concat(input.initial, Stream.merge(input.tail, input.enrichment));
117}
118
119const sameProjects = Schema.toEquivalence(Schema.Array(ProjectShellSchema));
120
121/** Drop repeated metadata refreshes within one subscription, preserving resolution progress. */
122export function dedupeShellEnrichment<E, R>(
123 stream: Stream.Stream<OrchestrationV2ShellStreamItem, E, R>,
124): Stream.Stream<OrchestrationV2ShellStreamItem, E, R> {
125 return Stream.suspend(() => {
126 let previous:
127 | {
128 readonly projects: ReadonlyArray<OrchestrationProjectShell>;
129 readonly roots: ReadonlySet<string>;
130 }
131 | undefined;
132 return stream.pipe(
133 Stream.filter((item) => {
134 if (item.kind !== "snapshot") {
135 // A project delta may change which projects/identities a later refresh repairs.
136 if (item.kind === "project.updated" || item.kind === "project.removed")
137 previous = undefined;
138 return true;
139 }
140 if (
141 item.resolvedRepositoryIdentityRoots === undefined ||
142 item.snapshot.threads.length > 0 ||
143 item.snapshot.archivedThreads.length > 0
144 ) {
145 previous = undefined;
146 return true;
147 }
148 const roots = new Set(item.resolvedRepositoryIdentityRoots);
149 const prior = previous;
150 if (
151 prior !== undefined &&
152 roots.size === prior.roots.size &&
153 [...roots].every((root) => prior.roots.has(root)) &&
154 sameProjects(prior.projects, item.snapshot.projects)
155 )
156 return false;
157 previous = { projects: item.snapshot.projects, roots };
158 return true;
159 }),
160 );
161 });
162}
163
164/** Build a shell snapshot stream item for a batched enrichment completion. */
165export function shellStreamItemFromEnrichmentRefresh(input: {
166 readonly snapshot: OrchestrationV2ShellSnapshot;
167 readonly changes: ReadonlyArray<{ readonly workspaceRoot: string }>;
168}): Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {
169 const resolvedRepositoryIdentityRoots = [
170 ...new Set(input.changes.map((change) => change.workspaceRoot)),
171 ];
172 return {
173 kind: "snapshot",
174 snapshot: {
175 ...input.snapshot,
176 projects: input.snapshot.projects.filter((project) =>
177 resolvedRepositoryIdentityRoots.includes(project.workspaceRoot),
178 ),
179 threads: [],
180 archivedThreads: [],
181 },
182 resolvedRepositoryIdentityRoots,
183 };
184}
185
186/**
187 * Initial subscribe frames: always emit the unmarked authoritative snapshot,
188 * then a same-sequence enrichment frame only when some roots already resolved.
189 */
190export function shellStreamItemsFromInitialSnapshot(input: {
191 readonly snapshot: OrchestrationV2ShellSnapshot;
192 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
193}): ReadonlyArray<Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }>> {
194 const authoritative = {
195 kind: "snapshot" as const,
196 snapshot: input.snapshot,
197 };
198 if (input.resolvedRepositoryIdentityRoots.length === 0) {
199 return [authoritative];
200 }
201 return [
202 authoritative,
203 {
204 kind: "snapshot" as const,
205 snapshot: {
206 ...input.snapshot,
207 projects: input.snapshot.projects.filter((project) =>
208 input.resolvedRepositoryIdentityRoots.includes(project.workspaceRoot),
209 ),
210 threads: [],
211 archivedThreads: [],
212 },
213 resolvedRepositoryIdentityRoots: [...new Set(input.resolvedRepositoryIdentityRoots)],
214 },
215 ];
216}
217
218/** Resume after HTTP/cache hydration with metadata only, never another full shell body. */
219export function shellStreamItemsFromResumeSnapshot(input: {
220 readonly snapshot: OrchestrationV2ShellSnapshot;
221 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
222}): ReadonlyArray<Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }>> {
223 return shellStreamItemsFromInitialSnapshot(input).filter(
224 (item) => item.resolvedRepositoryIdentityRoots !== undefined,
225 );
226}
227
228const sameThreadShell = Schema.toEquivalence(ThreadShellSchema);
229
230/** How long an unchanged thread shell may go unsent on a live subscription. */
231const UNCHANGED_THREAD_SHELL_RESEND_MS = 5_000;
232
233/**
234 * Drop live `thread.updated` deltas whose shell matches the last one this
235 * subscription sent, apart from `updatedAt`. Every thread event bumps
236 * `updatedAt`, so tool output and streaming text alone used to send the full,
237 * otherwise unchanged shell once per batch.
238 *
239 * A skipped delta leaves the client cursor behind, and a resume replays the
240 * events after that cursor. So an unchanged shell is still sent once its last
241 * send is `UNCHANGED_THREAD_SHELL_RESEND_MS` old. That bounds both the replay
242 * and how stale the client's `updatedAt` can get.
243 */
244export function skipUnchangedThreadShells<A extends OrchestrationV2ShellStreamItem, E, R>(
245 stream: Stream.Stream<A, E, R>,
246): Stream.Stream<A, E, R> {
247 return Stream.suspend(() => {
248 const lastSent = new Map<
249 OrchestrationV2ThreadShell["id"],
250 { readonly thread: OrchestrationV2ThreadShell; readonly sentAt: number }
251 >();
252 return stream.pipe(
253 Stream.filterEffect((item) =>
254 Effect.map(Clock.currentTimeMillis, (now) => {
255 if (item.kind === "thread.removed") {
256 lastSent.delete(item.threadId);
257 return true;
258 }
259 if (item.kind !== "thread.updated") return true;
260 const previous = lastSent.get(item.thread.id);
261 if (
262 previous !== undefined &&
263 now - previous.sentAt < UNCHANGED_THREAD_SHELL_RESEND_MS &&
264 sameThreadShell(
265 { ...item.thread, updatedAt: previous.thread.updatedAt },
266 previous.thread,
267 )
268 ) {
269 return false;
270 }
271 lastSent.set(item.thread.id, { thread: item.thread, sentAt: now });
272 return true;
273 }),
274 ),
275 );
276 });
277}
278
279/** Keep only the newest stored event per thread within a coalescing window. */
280export function coalesceStoredThreadEvents(
281 events: ReadonlyArray<OrchestrationV2StoredEvent>,
282): ReadonlyArray<OrchestrationV2StoredEvent> {
283 const latestByThreadId = new Map<string, OrchestrationV2StoredEvent>();
284 for (const stored of events) {
285 latestByThreadId.set(stored.event.threadId, stored);
286 }
287 return Array.from(latestByThreadId.values()).sort(
288 (left, right) => left.sequence - right.sequence,
289 );
290}
291
292/**
293 * Converts a committed event and the affected thread's current shell into one
294 * delta. `shell` is null when the thread is deleted or unknown.
295 */
296export function shellStreamItemFromThreadShell(input: {
297 readonly stored: Extract<ShellApplicationEvent, { readonly event: unknown }>;
298 readonly shell: OrchestrationV2ThreadShell | null;
299}): Exclude<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {
300 if (input.shell !== null) {
301 if (input.shell.archivedAt !== null) {
302 return {
303 kind: "thread.removed",
304 sequence: input.stored.sequence,
305 location: "active",
306 threadId: input.shell.id,
307 };
308 }
309 return {
310 kind: "thread.updated",
311 sequence: input.stored.sequence,
312 location: "active",
313 thread: input.shell,
314 };
315 }
316
317 return {
318 kind: "thread.removed",
319 sequence: input.stored.sequence,
320 location: "active",
321 threadId: input.stored.event.threadId,
322 };
323}
324
325/** Converts a committed event into an archive-only delta when it changes archive membership. */
326export function archivedShellStreamItemFromThreadShell(input: {
327 readonly stored: OrchestrationV2StoredEvent;
328 readonly shell: OrchestrationV2ThreadShell | null;
329}): Exclude<OrchestrationV2ArchivedShellStreamItem, { readonly kind: "snapshot" }> | null {
330 if (input.shell !== null && input.shell.archivedAt !== null) {
331 return {
332 kind: "thread.updated",
333 sequence: input.stored.sequence,
334 thread: input.shell,
335 };
336 }
337 if (
338 input.stored.event.type === "thread.unarchived" ||
339 (input.stored.event.type === "thread.deleted" && input.stored.event.payload.archivedAt !== null)
340 ) {
341 return {
342 kind: "thread.removed",
343 sequence: input.stored.sequence,
344 threadId: input.stored.event.threadId,
345 };
346 }
347 return null;
348}