1import type {
2 ApplicationStoredEvent,
3 OrchestrationProjectShell,
4 OrchestrationV2OlderSettledCount,
5 OrchestrationV2OlderSettledThreads,
6 OrchestrationV2ShellSnapshot,
7 OrchestrationV2ThreadShell,
8 OrchestrationV2ThreadShellSnapshot,
9 OrchestrationV2ShellStreamItem,
10 OrchestrationV2StoredEvent,
11 ProjectId,
12 ThreadId,
13} from "@t3tools/contracts";
14import {
15 isListedThreadLineage,
16 isLiveSubagentShell,
17 isOlderSettledThreadShell,
18 OrchestrationProjectShell as ProjectShellSchema,
19 OrchestrationV2ThreadShell as ThreadShellSchema,
20} from "@t3tools/contracts";
21import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
22// @effect-diagnostics-next-line nodeBuiltinImport:off - a synchronous digest of a few project rows.
23import * as NodeCrypto from "node:crypto";
24import type * as Arr from "effect/Array";
25import type * as Cause from "effect/Cause";
26import * as Clock from "effect/Clock";
27import * as DateTime from "effect/DateTime";
28import * as Deferred from "effect/Deferred";
29import * as Effect from "effect/Effect";
30import * as Exit from "effect/Exit";
31import * as Queue from "effect/Queue";
32import * as Schema from "effect/Schema";
33import type * as SqlClient from "effect/sql/SqlClient";
34import * as Stream from "effect/Stream";
35
36import type * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
37import { bufferLatestLiveStream } from "./LiveStreamBudget.ts";
38import type { ActiveShellIndexEntry } from "./ProjectionStore.ts";
39
40type ShellDelta = Extract<OrchestrationV2ShellStreamItem, { readonly sequence: number }>;
41
42// Replaying each intervening event costs a shell refetch; past this gap a
43// single O(active-threads) snapshot is cheaper and bounded.
44const SHELL_RESUME_MAX_GAP = 1_000;
45// Row counts do not bound large tool payloads; sum stored JSON bytes in SQL and snapshot above budget.
46const SHELL_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024;
47
48/**
49 * Whether the events after a shell cursor can be replayed. The socket resume and the HTTP
50 * snapshot's `resumeFrom` ask the same question, so a client the route tells to resume is not
51 * refused by the socket.
52 */
53export const canResumeShellFrom = Effect.fnUntraced(function* (
54 events: Pick<
55 OrchestrationEventStore.OrchestrationEventStore["Service"],
56 "prunedThroughSequence" | "getReplayStats"
57 >,
58 afterSequence: number,
59 headSequence: number,
60) {
61 // A cursor at the head missed nothing, even where pruning has passed the head.
62 if (afterSequence === headSequence) return true;
63 const replayGap = headSequence - afterSequence;
64 if (replayGap < 0 || replayGap > SHELL_RESUME_MAX_GAP) return false;
65 if (afterSequence < (yield* events.prunedThroughSequence)) return false;
66 const stats = yield* events.getReplayStats({ afterSequence, throughSequence: headSequence });
67 if (stats.rawPayloadBytes > SHELL_REPLAY_PAYLOAD_BUDGET_BYTES) {
68 yield* Effect.logDebug("orchestration replay replaced by snapshot", {
69 afterSequence,
70 headSequence,
71 replayGap,
72 eventCount: stats.eventCount,
73 payloadBytes: stats.rawPayloadBytes,
74 payloadBudgetBytes: SHELL_REPLAY_PAYLOAD_BUDGET_BYTES,
75 });
76 return false;
77 }
78 return true;
79});
80
81/**
82 * Answers a client that asks for the shell with the cursor of the rows it holds: `resumable`
83 * when the socket can replay from that cursor, so the client keeps its rows, else the snapshot.
84 */
85export const loadShellUnlessResumable = Effect.fnUntraced(function* <A, E, R>(input: {
86 readonly events: Pick<
87 OrchestrationEventStore.OrchestrationEventStore["Service"],
88 "prunedThroughSequence" | "getReplayStats" | "latestApplicationSequence"
89 >;
90 readonly resumeFrom: number | undefined;
91 readonly loadSnapshot: Effect.Effect<A, E, R>;
92}) {
93 if (
94 input.resumeFrom !== undefined &&
95 (yield* canResumeShellFrom(
96 input.events,
97 input.resumeFrom,
98 yield* input.events.latestApplicationSequence,
99 ))
100 ) {
101 return { resumable: true as const };
102 }
103 return yield* input.loadSnapshot;
104});
105
106/**
107 * Names the repository identities a list of project shells carries. A client sends back the one
108 * its HTTP snapshot had, and a resume that computes the same one skips its marked snapshot.
109 */
110export function projectIdentitiesDigest(
111 projects: ReadonlyArray<Pick<OrchestrationProjectShell, "id" | "repositoryIdentity">>,
112): string {
113 const hash = NodeCrypto.createHash("sha256");
114 for (const project of projects) {
115 hash.update(JSON.stringify([project.id, project.repositoryIdentity ?? null]));
116 }
117 return hash.digest("base64url");
118}
119
120/**
121 * The resolved roots a resume sends. A client that names the current identities holds them
122 * already, but it keeps a cached identity over an absent one, so a root resolved to no identity
123 * still goes out.
124 */
125export function resumeIdentityRoots(input: {
126 readonly projects: ReadonlyArray<
127 Pick<OrchestrationProjectShell, "id" | "workspaceRoot" | "repositoryIdentity">
128 >;
129 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
130 readonly knownProjectIdentities: string | undefined;
131}): ReadonlyArray<string> {
132 if (input.knownProjectIdentities !== projectIdentitiesDigest(input.projects)) {
133 return input.resolvedRepositoryIdentityRoots;
134 }
135 const withoutIdentity = new Set(
136 input.projects.flatMap((project) =>
137 project.repositoryIdentity == null ? [project.workspaceRoot] : [],
138 ),
139 );
140 return input.resolvedRepositoryIdentityRoots.filter((root) => withoutIdentity.has(root));
141}
142
143/** Snapshots, synchronization, and enrichment markers must bypass full-state coalescing. */
144export const bufferShellLiveStream = <E, R>(
145 source: Stream.Stream<ShellDelta, E, R>,
146 limits?: { readonly maxItems?: number; readonly maxSerializedBytes?: number },
147) =>
148 bufferLatestLiveStream(
149 source,
150 (item) => {
151 switch (item.kind) {
152 case "project.updated":
153 return `project:${item.project.id}`;
154 case "project.removed":
155 return `project:${item.projectId}`;
156 case "thread.updated":
157 return `thread:${item.location}:${item.thread.id}`;
158 case "thread.removed":
159 return `thread:${item.location}:${item.threadId}`;
160 }
161 },
162 limits,
163 );
164
165function buildActiveShellSnapshot(input: {
166 readonly projects: ReadonlyArray<OrchestrationProjectShell>;
167 readonly threads: OrchestrationV2ThreadShellSnapshot;
168 readonly snapshotSequence: number;
169}): OrchestrationV2ShellSnapshot {
170 return {
171 schemaVersion: input.threads.schemaVersion,
172 snapshotSequence: input.snapshotSequence,
173 projects: input.projects,
174 threads: input.threads.threads,
175 archivedThreads: [],
176 };
177}
178
179type ShellTrimRow = Pick<
180 OrchestrationV2ThreadShell,
181 "id" | "projectId" | "lineage" | "settledOverride" | "settledAt"
182> & { readonly hasRemoteTasks?: boolean };
183
184/**
185 * What a subscriber's trims read across every active row. Computed from full shells or from the
186 * store's index, so a snapshot read only for the rows it sends trims the same way.
187 */
188interface ShellTrimFacts {
189 readonly subagentParentIds: ReadonlySet<string>;
190 readonly subagentThreadIds: Set<string>;
191 /** Only the store's index knows these; shells alone leave it empty. */
192 readonly remoteTaskParentIds: ReadonlySet<string>;
193 /** Undefined when no thread is settled, so none is older. */
194 readonly olderSettledBeforeMs: number | undefined;
195 readonly olderSettledCounts: ReadonlyArray<OrchestrationV2OlderSettledCount>;
196}
197
198const NEWEST_SETTLED_KEPT = 25;
199
200function shellTrimFacts(rows: ReadonlyArray<ShellTrimRow>): ShellTrimFacts {
201 const subagentParentIds = new Set<string>();
202 const subagentThreadIds = new Set<string>();
203 const remoteTaskParentIds = new Set<string>();
204 const settledTimes: Array<number> = [];
205 for (const row of rows) {
206 if (row.hasRemoteTasks === true) remoteTaskParentIds.add(row.id);
207 if (row.lineage.relationshipToParent === "subagent") {
208 subagentThreadIds.add(row.id);
209 if (row.lineage.parentThreadId) subagentParentIds.add(row.lineage.parentThreadId);
210 } else if (
211 isListedThreadLineage(row.lineage) &&
212 row.settledOverride === "settled" &&
213 row.settledAt !== null
214 ) {
215 settledTimes.push(DateTime.toEpochMillis(row.settledAt));
216 }
217 }
218 settledTimes.sort((left, right) => right - left);
219 // With fewer than 25, the oldest one is the cut, so nothing is older.
220 const olderSettledBeforeMs = settledTimes[NEWEST_SETTLED_KEPT - 1] ?? settledTimes.at(-1);
221 const counts = new Map<ProjectId, number>();
222 if (olderSettledBeforeMs !== undefined) {
223 for (const row of rows) {
224 if (isOlderSettledThreadShell(row, olderSettledBeforeMs)) {
225 counts.set(row.projectId, (counts.get(row.projectId) ?? 0) + 1);
226 }
227 }
228 }
229 return {
230 subagentParentIds,
231 subagentThreadIds,
232 remoteTaskParentIds,
233 olderSettledBeforeMs,
234 olderSettledCounts: Array.from(counts, ([projectId, count]) => ({ projectId, count })),
235 };
236}
237
238function hiddenChildAncestorIds(
239 threads: ReadonlyArray<Pick<OrchestrationV2ThreadShell, "lineage">>,
240) {
241 const ancestorIds = new Set<string>();
242 for (const thread of threads) {
243 if (isListedThreadLineage(thread.lineage)) continue;
244 if (thread.lineage.parentThreadId !== null) ancestorIds.add(thread.lineage.parentThreadId);
245 ancestorIds.add(thread.lineage.rootThreadId);
246 }
247 return ancestorIds;
248}
249
250function omitIdleSubagentShellsWith(
251 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
252 facts: Pick<ShellTrimFacts, "subagentParentIds" | "remoteTaskParentIds">,
253): ReadonlyArray<OrchestrationV2ThreadShell> {
254 return threads.filter(
255 (thread) =>
256 thread.lineage.relationshipToParent !== "subagent" ||
257 isLiveSubagentShell(thread) ||
258 facts.subagentParentIds.has(thread.id) ||
259 facts.remoteTaskParentIds.has(thread.id),
260 );
261}
262
263function omitOlderSettledShellsWith(
264 input: {
265 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
266 readonly snapshotSequence: number;
267 readonly now: DateTime.Utc;
268 },
269 facts: Pick<
270 ShellTrimFacts,
271 "olderSettledBeforeMs" | "olderSettledCounts" | "remoteTaskParentIds"
272 >,
273): {
274 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
275 readonly olderSettled: OrchestrationV2OlderSettledThreads;
276} {
277 const nowMs = DateTime.toEpochMillis(input.now);
278 const beforeMs = facts.olderSettledBeforeMs ?? nowMs;
279 const ancestorIds = hiddenChildAncestorIds(input.threads);
280 const threads = input.threads.filter(
281 (thread) =>
282 !isOlderSettledThreadShell(thread, beforeMs) ||
283 thread.pinnedAt != null ||
284 (thread.snoozedUntil != null && DateTime.toEpochMillis(thread.snoozedUntil) > nowMs) ||
285 isLiveSubagentShell(thread) ||
286 ancestorIds.has(thread.id) ||
287 facts.remoteTaskParentIds.has(thread.id),
288 );
289 return {
290 threads,
291 olderSettled: {
292 before: DateTime.formatIso(DateTime.makeUnsafe(beforeMs)),
293 asOfSequence: input.snapshotSequence,
294 counts: facts.olderSettledCounts,
295 },
296 };
297}
298
299/**
300 * Idle subagent children have no sidebar row, so clients that opt in read them from their thread
301 * family instead. A child stays while it is live or the parent of another child, so the request
302 * and working walks over shell rows never meet a gap in a nested chain. It also stays as the
303 * parent of a task in a linked environment, whose clients nest that task under it.
304 */
305export function omitIdleSubagentShells(
306 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
307): ReadonlyArray<OrchestrationV2ThreadShell> {
308 return omitIdleSubagentShellsWith(threads, shellTrimFacts(threads));
309}
310
311/**
312 * The active rows a shell subscriber gets, and the subagent ids whose transcript events its stream
313 * drops. The ids come from every row, so an omitted child that wakes keeps that filter.
314 */
315export function activeShellRowsForSubscriber(
316 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
317 omitIdleSubagents: boolean,
318): {
319 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
320 readonly subagentThreadIds: Set<string>;
321} {
322 const facts = shellTrimFacts(threads);
323 return {
324 threads: omitIdleSubagents ? omitIdleSubagentShellsWith(threads, facts) : threads,
325 subagentThreadIds: facts.subagentThreadIds,
326 };
327}
328
329/**
330 * Opted-in snapshots keep the 25 newest settled top-level threads (and ties) and count the rest per
331 * project in `olderSettled`; clients page them with `getOlderSettledThreads`. Older rows stay while
332 * pinned, snoozed, live, above a kept subagent or side chat, or the parent of a task in a linked
333 * environment, so walks over shell rows and other environments' tasks still find them.
334 */
335export function omitOlderSettledShells(input: {
336 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
337 readonly snapshotSequence: number;
338 readonly now: DateTime.Utc;
339}): {
340 readonly threads: ReadonlyArray<OrchestrationV2ThreadShell>;
341 readonly olderSettled: OrchestrationV2OlderSettledThreads;
342} {
343 return omitOlderSettledShellsWith(input, shellTrimFacts(input.threads));
344}
345
346interface ShellSubscriberOptions {
347 readonly omitIdleSubagents: boolean;
348 readonly omitOlderSettled: boolean;
349}
350
351function trimShellSnapshot(
352 base: OrchestrationV2ShellSnapshot,
353 facts: ShellTrimFacts,
354 options: ShellSubscriberOptions & { readonly now: DateTime.Utc },
355): OrchestrationV2ShellSnapshot {
356 const threads = options.omitIdleSubagents
357 ? omitIdleSubagentShellsWith(base.threads, facts)
358 : base.threads;
359 if (!options.omitOlderSettled) return { ...base, threads };
360 const older = omitOlderSettledShellsWith(
361 { threads, snapshotSequence: base.snapshotSequence, now: options.now },
362 facts,
363 );
364 return { ...base, threads: older.threads, olderSettled: older.olderSettled };
365}
366
367/**
368 * The threads whose shells a trimmed snapshot can contain: a superset of what the trims keep, so
369 * trimming the shells read for these ids gives the snapshot trimming every shell would.
370 */
371function shellSubscriberThreadIds(
372 index: ReadonlyArray<ActiveShellIndexEntry>,
373 options: ShellSubscriberOptions,
374): ReadonlyArray<ThreadId> {
375 const facts = shellTrimFacts(index);
376 const subagents = index.filter(
377 (entry) =>
378 entry.lineage.relationshipToParent === "subagent" &&
379 (!options.omitIdleSubagents ||
380 entry.mayBeLive ||
381 entry.hasRemoteTasks ||
382 facts.subagentParentIds.has(entry.id)),
383 );
384 const ancestorIds = hiddenChildAncestorIds([
385 ...subagents,
386 ...index.filter((entry) => entry.lineage.sideChat === true),
387 ]);
388 const keptSubagentIds = new Set(subagents.map((entry) => entry.id));
389 const beforeMs = facts.olderSettledBeforeMs;
390 return index.flatMap((entry) =>
391 entry.lineage.relationshipToParent === "subagent"
392 ? keptSubagentIds.has(entry.id)
393 ? [entry.id]
394 : []
395 : !options.omitOlderSettled ||
396 beforeMs === undefined ||
397 !isOlderSettledThreadShell(entry, beforeMs) ||
398 entry.pinnedAt != null ||
399 entry.snoozedUntil != null ||
400 entry.mayBeLive ||
401 entry.hasRemoteTasks ||
402 ancestorIds.has(entry.id)
403 ? [entry.id]
404 : [],
405 );
406}
407
408/**
409 * A subscriber's active shell snapshot, read like `loadShellSnapshotParts`. With an opt-in, the
410 * store's index decides which shells to read, so threads the trims drop are never built.
411 */
412export const loadSubscriberShellSnapshot = Effect.fn("ShellStream.loadSubscriberShellSnapshot")(
413 function* <E1, E2, E3, E4, E5>(input: {
414 readonly sql: SqlClient.SqlClient;
415 readonly listShellIndex: Effect.Effect<ReadonlyArray<ActiveShellIndexEntry>, E1>;
416 readonly readThreads: (
417 threadIds: ReadonlyArray<ThreadId> | undefined,
418 ) => Effect.Effect<Effect.Effect<OrchestrationV2ThreadShellSnapshot, E3>, E2>;
419 readonly listProjects: Effect.Effect<ReadonlyArray<OrchestrationProjectShell>, E4>;
420 readonly latestSequence: Effect.Effect<number, E5>;
421 readonly options: ShellSubscriberOptions;
422 }) {
423 const trims = input.options.omitIdleSubagents || input.options.omitOlderSettled;
424 const read = yield* input.sql
425 .withTransaction(
426 Effect.gen(function* () {
427 const index = trims ? yield* input.listShellIndex : null;
428 return {
429 index,
430 decodeThreads: yield* input.readThreads(
431 index === null ? undefined : shellSubscriberThreadIds(index, input.options),
432 ),
433 projects: yield* input.listProjects,
434 snapshotSequence: yield* input.latestSequence,
435 };
436 }),
437 )
438 .pipe(NodeSqliteClient.readOnly);
439 const base = buildActiveShellSnapshot({
440 projects: read.projects,
441 threads: yield* read.decodeThreads,
442 snapshotSequence: read.snapshotSequence,
443 });
444 const facts = shellTrimFacts(read.index ?? base.threads);
445 return {
446 snapshot: trimShellSnapshot(base, facts, { ...input.options, now: yield* DateTime.now }),
447 subagentThreadIds: facts.subagentThreadIds,
448 };
449 },
450);
451
452/**
453 * One transaction covers the reads so they agree; the threads are decoded after it commits,
454 * because the server shares one SQLite connection and decoding a large shell takes longer than
455 * reading it.
456 */
457export const loadShellSnapshotParts = Effect.fn("ShellStream.loadShellSnapshotParts")(function* <
458 E1,
459 E2,
460 E3,
461 E4,
462>(input: {
463 readonly sql: SqlClient.SqlClient;
464 readonly readThreads: Effect.Effect<Effect.Effect<OrchestrationV2ThreadShellSnapshot, E2>, E1>;
465 readonly listProjects: Effect.Effect<ReadonlyArray<OrchestrationProjectShell>, E3>;
466 readonly latestSequence: Effect.Effect<number, E4>;
467}) {
468 const read = yield* input.sql
469 .withTransaction(
470 Effect.all({
471 decodeThreads: input.readThreads,
472 projects: input.listProjects,
473 snapshotSequence: input.latestSequence,
474 }),
475 )
476 .pipe(NodeSqliteClient.readOnly);
477 return {
478 projects: read.projects,
479 threads: yield* read.decodeThreads,
480 snapshotSequence: read.snapshotSequence,
481 };
482});
483
484/** A page of older settled threads, read and decoded like `loadShellSnapshotParts`. */
485export const loadOlderSettledThreadsPage = Effect.fn("ShellStream.loadOlderSettledThreadsPage")(
486 function* <E1, E2, E3, E4>(input: {
487 readonly sql: SqlClient.SqlClient;
488 readonly listThreads: Effect.Effect<
489 {
490 readonly threadIds: ReadonlyArray<ThreadId>;
491 readonly counts: ReadonlyArray<OrchestrationV2OlderSettledCount>;
492 },
493 E1
494 >;
495 readonly readThreads: (
496 threadIds: ReadonlyArray<ThreadId>,
497 ) => Effect.Effect<Effect.Effect<OrchestrationV2ThreadShellSnapshot, E3>, E2>;
498 readonly latestSequence: Effect.Effect<number, E4>;
499 }) {
500 const read = yield* input.sql
501 .withTransaction(
502 Effect.gen(function* () {
503 const listed = yield* input.listThreads;
504 return {
505 listed,
506 decodeThreads:
507 listed.threadIds.length === 0 ? null : yield* input.readThreads(listed.threadIds),
508 snapshotSequence: yield* input.latestSequence,
509 };
510 }),
511 )
512 .pipe(NodeSqliteClient.readOnly);
513 const threads = read.decodeThreads === null ? [] : (yield* read.decodeThreads).threads;
514 const order = new Map(read.listed.threadIds.map((threadId, index) => [threadId, index]));
515 return {
516 snapshotSequence: read.snapshotSequence,
517 threads: threads.toSorted((left, right) => order.get(left.id)! - order.get(right.id)!),
518 counts: read.listed.counts,
519 };
520 },
521);
522
523export type ShellApplicationEvent =
524 | Pick<
525 Extract<ApplicationStoredEvent, { readonly aggregateKind: "project" }>,
526 "aggregateKind" | "aggregateId" | "type" | "sequence"
527 >
528 | {
529 readonly sequence: number;
530 readonly event: Pick<OrchestrationV2StoredEvent["event"], "threadId">;
531 };
532
533/** Focused subagents receive these updates through their thread subscription. */
534export function shouldPublishShellApplicationEvent(
535 stored: ApplicationStoredEvent,
536 subagentThreadIds: Set<string>,
537): boolean {
538 if ("aggregateKind" in stored) return true;
539 const { event } = stored;
540 if (event.type === "thread.created") {
541 if (event.payload.lineage.relationshipToParent === "subagent") {
542 subagentThreadIds.add(event.threadId);
543 }
544 return true;
545 }
546 if (event.type === "thread.deleted") {
547 // Replay and the live tail overlap; keep known IDs so late events stay filtered.
548 return true;
549 }
550 return !(
551 subagentThreadIds.has(event.threadId) &&
552 (event.type === "node.updated" ||
553 event.type === "turn-item.updated" ||
554 event.type === "provider-turn.updated" ||
555 event.type === "message.updated")
556 );
557}
558
559/** Shell updates refetch an aggregate; drop transcript bodies before retaining an event. */
560export function toShellApplicationEvent(stored: ApplicationStoredEvent): ShellApplicationEvent {
561 return "aggregateKind" in stored
562 ? {
563 aggregateKind: stored.aggregateKind,
564 aggregateId: stored.aggregateId,
565 type: stored.type,
566 sequence: stored.sequence,
567 }
568 : { sequence: stored.sequence, event: { threadId: stored.event.threadId } };
569}
570
571export function coalesceShellApplicationEvents<A extends ShellApplicationEvent>(
572 events: ReadonlyArray<A>,
573): ReadonlyArray<A> {
574 const latestByAggregate = new Map<string, A>();
575 for (const stored of events) {
576 const key =
577 "aggregateKind" in stored
578 ? `project:${stored.aggregateId}`
579 : `thread:${stored.event.threadId}`;
580 latestByAggregate.set(key, stored);
581 }
582 return Array.from(latestByAggregate.values()).sort(
583 (left, right) => left.sequence - right.sequence,
584 );
585}
586
587/** Emit the authoritative initial snapshot before marked enrichment. */
588export function composeShellStreamWithEnrichment<A, E, R, A2, E2, R2, A3, E3, R3>(input: {
589 readonly initial: Stream.Stream<A, E, R>;
590 readonly tail: Stream.Stream<A2, E2, R2>;
591 readonly enrichment: Stream.Stream<A3, E3, R3>;
592}): Stream.Stream<A | A2 | A3, E | E2 | E3, R | R2 | R3> {
593 const merged = Stream.fromPull(
594 Effect.gen(function* () {
595 const output = yield* Queue.bounded<
596 {
597 readonly items: Arr.NonEmptyReadonlyArray<A2 | A3>;
598 readonly acknowledged?: Deferred.Deferred<void>;
599 },
600 E2 | E3 | Cause.Done
601 >(0);
602 yield* Effect.addFinalizer(() => Queue.shutdown(output));
603 let remaining = 2;
604 const drain = (effect: Effect.Effect<void, E2 | E3, R2 | R3>) =>
605 effect.pipe(
606 Effect.exit,
607 Effect.flatMap((exit) =>
608 Exit.isFailure(exit)
609 ? Queue.failCause(output, exit.cause)
610 : --remaining === 0
611 ? Queue.end(output)
612 : Effect.void,
613 ),
614 Effect.forkScoped,
615 );
616 yield* drain(
617 input.tail.pipe(
618 Stream.runForEachArray((items) =>
619 Effect.gen(function* () {
620 const acknowledged = yield* Deferred.make<void>();
621 yield* Queue.offer(output, { items, acknowledged });
622 yield* Deferred.await(acknowledged);
623 }),
624 ),
625 ),
626 );
627 // Keep draining the sliding enrichment subscription after each handoff,
628 // as Stream.merge does, while only the budgeted tail waits for an ACK.
629 yield* drain(
630 input.enrichment.pipe(Stream.runForEachArray((items) => Queue.offer(output, { items }))),
631 );
632 let previous: Deferred.Deferred<void> | undefined;
633 // Stream.merge pulls ahead after handing off each batch. Wait for the next
634 // downstream pull so shell delivery keeps that batch charged until its RPC ACK.
635 return Effect.gen(function* () {
636 if (previous) yield* Deferred.succeed(previous, undefined);
637 const next = yield* Queue.take(output);
638 previous = next.acknowledged;
639 return next.items;
640 });
641 }),
642 ).pipe(Stream.scoped);
643 return Stream.concat(input.initial, merged);
644}
645
646const sameProjects = Schema.toEquivalence(Schema.Array(ProjectShellSchema));
647
648export function dedupeShellEnrichment<E, R>(
649 stream: Stream.Stream<OrchestrationV2ShellStreamItem, E, R>,
650): Stream.Stream<OrchestrationV2ShellStreamItem, E, R> {
651 return Stream.suspend(() => {
652 let previous:
653 | {
654 readonly projects: ReadonlyArray<OrchestrationProjectShell>;
655 readonly roots: ReadonlySet<string>;
656 }
657 | undefined;
658 return stream.pipe(
659 Stream.filter((item) => {
660 if (item.kind !== "snapshot") {
661 if (item.kind === "project.updated" || item.kind === "project.removed")
662 previous = undefined;
663 return true;
664 }
665 if (
666 item.resolvedRepositoryIdentityRoots === undefined ||
667 item.snapshot.threads.length > 0 ||
668 item.snapshot.archivedThreads.length > 0
669 ) {
670 previous = undefined;
671 return true;
672 }
673 const roots = new Set(item.resolvedRepositoryIdentityRoots);
674 const prior = previous;
675 if (
676 prior !== undefined &&
677 roots.size === prior.roots.size &&
678 [...roots].every((root) => prior.roots.has(root)) &&
679 sameProjects(prior.projects, item.snapshot.projects)
680 )
681 return false;
682 previous = { projects: item.snapshot.projects, roots };
683 return true;
684 }),
685 );
686 });
687}
688
689export function shellStreamItemFromEnrichmentRefresh(input: {
690 readonly snapshot: OrchestrationV2ShellSnapshot;
691 readonly changes: ReadonlyArray<{ readonly workspaceRoot: string }>;
692}): Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }> {
693 const resolvedRepositoryIdentityRoots = [
694 ...new Set(input.changes.map((change) => change.workspaceRoot)),
695 ];
696 return {
697 kind: "snapshot",
698 snapshot: {
699 ...input.snapshot,
700 projects: input.snapshot.projects.filter((project) =>
701 resolvedRepositoryIdentityRoots.includes(project.workspaceRoot),
702 ),
703 threads: [],
704 archivedThreads: [],
705 },
706 resolvedRepositoryIdentityRoots,
707 };
708}
709
710export function shellStreamItemsFromInitialSnapshot(input: {
711 readonly snapshot: OrchestrationV2ShellSnapshot;
712 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
713}): ReadonlyArray<Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }>> {
714 const authoritative = {
715 kind: "snapshot" as const,
716 snapshot: input.snapshot,
717 };
718 if (input.resolvedRepositoryIdentityRoots.length === 0) {
719 return [authoritative];
720 }
721 return [
722 authoritative,
723 {
724 kind: "snapshot" as const,
725 snapshot: {
726 ...input.snapshot,
727 projects: input.snapshot.projects.filter((project) =>
728 input.resolvedRepositoryIdentityRoots.includes(project.workspaceRoot),
729 ),
730 threads: [],
731 archivedThreads: [],
732 },
733 resolvedRepositoryIdentityRoots: [...new Set(input.resolvedRepositoryIdentityRoots)],
734 },
735 ];
736}
737
738export function shellStreamItemsFromResumeSnapshot(input: {
739 readonly snapshot: OrchestrationV2ShellSnapshot;
740 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
741}): ReadonlyArray<Extract<OrchestrationV2ShellStreamItem, { readonly kind: "snapshot" }>> {
742 return shellStreamItemsFromInitialSnapshot(input).filter(
743 (item) => item.resolvedRepositoryIdentityRoots !== undefined,
744 );
745}
746
747const sameThreadShell = Schema.toEquivalence(ThreadShellSchema);
748
749const UNCHANGED_THREAD_SHELL_RESEND_MS = 5_000;
750
751/** Resend unchanged shells after UNCHANGED_THREAD_SHELL_RESEND_MS so skipped deltas cannot leave resume cursors behind. */
752export function skipUnchangedThreadShells<A extends OrchestrationV2ShellStreamItem, E, R>(
753 stream: Stream.Stream<A, E, R>,
754): Stream.Stream<A, E, R> {
755 return Stream.suspend(() => {
756 const lastSent = new Map<
757 OrchestrationV2ThreadShell["id"],
758 { readonly thread: OrchestrationV2ThreadShell; readonly sentAt: number }
759 >();
760 return stream.pipe(
761 Stream.filterEffect((item) =>
762 Effect.map(Clock.currentTimeMillis, (now) => {
763 if (item.kind === "thread.removed") {
764 lastSent.delete(item.threadId);
765 return true;
766 }
767 if (item.kind !== "thread.updated") return true;
768 const previous = lastSent.get(item.thread.id);
769 if (
770 previous !== undefined &&
771 now - previous.sentAt < UNCHANGED_THREAD_SHELL_RESEND_MS &&
772 sameThreadShell(
773 { ...item.thread, updatedAt: previous.thread.updatedAt },
774 previous.thread,
775 )
776 ) {
777 return false;
778 }
779 lastSent.set(item.thread.id, { thread: item.thread, sentAt: now });
780 return true;
781 }),
782 ),
783 );
784 });
785}
786
787export function shellStreamItemFromThreadShell(input: {
788 readonly stored: Extract<ShellApplicationEvent, { readonly event: unknown }>;
789 readonly shell: OrchestrationV2ThreadShell | null;
790}): Extract<
791 OrchestrationV2ShellStreamItem,
792 { readonly kind: "thread.updated" | "thread.removed" }
793> {
794 if (input.shell !== null) {
795 if (input.shell.archivedAt !== null) {
796 return {
797 kind: "thread.removed",
798 sequence: input.stored.sequence,
799 location: "active",
800 threadId: input.shell.id,
801 };
802 }
803 return {
804 kind: "thread.updated",
805 sequence: input.stored.sequence,
806 location: "active",
807 thread: input.shell,
808 };
809 }
810
811 return {
812 kind: "thread.removed",
813 sequence: input.stored.sequence,
814 location: "active",
815 threadId: input.stored.event.threadId,
816 };
817}