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