Open History to see the last commit
1import {
2 ORCHESTRATION_V2_WS_METHODS,
3 type EnvironmentId,
4 type ExecutionEnvironmentDescriptor,
5 type OrchestrationV2ShellSnapshot,
6 type OrchestrationV2ShellStreamItem,
7 type OrchestrationV2ThreadShell,
8 type ServerConfig,
9 type ThreadId,
10 type ThreadPullRequestLink,
11} from "@t3tools/contracts";
12import * as Cause from "effect/Cause";
13import * as Clock from "effect/Clock";
14import * as Effect from "effect/Effect";
15import * as Exit from "effect/Exit";
16import * as Fiber from "effect/Fiber";
17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Ref from "effect/Ref";
20import * as Semaphore from "effect/Semaphore";
21import * as Stream from "effect/Stream";
22import * as SubscriptionRef from "effect/SubscriptionRef";
23import * as SynchronizedRef from "effect/SynchronizedRef";
24import { AsyncResult, Atom } from "effect/reactivity";
25
26import * as EnvironmentRegistry from "../connection/registry.ts";
27import { connectionProjectionPhase, type PreparedConnection } from "../connection/model.ts";
28import * as EnvironmentSupervisor from "../connection/supervisor.ts";
29import * as ConnectionWakeups from "../connection/wakeups.ts";
30import { safeErrorLogAttributes } from "../errors/safeLog.ts";
31import * as Persistence from "../platform/persistence.ts";
32import { runCachePersistence } from "./cachePersistence.ts";
33import { subscribeDynamicWithSession } from "../rpc/client.ts";
34import type { RpcSession } from "../rpc/session.ts";
35import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
36import type { DeferredShellSnapshot } from "./shellPullRequests.ts";
37import {
38 applyShellStreamEvents,
39 type ShellStreamEvent,
40 mergeShellSnapshotProjects,
41 reuseUnchangedThreadShells,
42 sameThreadPullRequests,
43} from "./shellReducer.ts";
44import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";
45import { followStreamInEnvironment } from "./runtime.ts";
46
47export type EnvironmentShellStatus = "empty" | "cached" | "synchronizing" | "live";
48
49export interface EnvironmentShellState {
50 readonly snapshot: Option.Option<OrchestrationV2ShellSnapshot>;
51 readonly status: EnvironmentShellStatus;
52 readonly error: Option.Option<string>;
53 /**
54 * Changes with each authoritative snapshot (not enrichment refreshes); 0 until the first one.
55 * Thread families read for an older generation may miss changes from before the latest snapshot,
56 * so they are read again.
57 */
58 readonly generation: number;
59 /** Sequences of the thread removals applied since the latest authoritative snapshot. */
60 readonly removedAt: ReadonlyMap<ThreadId, number>;
61}
62
63const EMPTY_REMOVALS: ReadonlyMap<ThreadId, number> = new Map();
64// Shared by every shell state, so a rebuilt one never reuses a generation another already used.
65let lastShellGeneration = 0;
66
67export const EMPTY_SHELL_STATE: EnvironmentShellState = {
68 snapshot: Option.none(),
69 status: "empty",
70 error: Option.none(),
71 generation: 0,
72 removedAt: EMPTY_REMOVALS,
73};
74
75function shellStatusForSnapshot(
76 snapshot: Option.Option<OrchestrationV2ShellSnapshot>,
77): EnvironmentShellStatus {
78 return Option.isSome(snapshot) ? "cached" : "empty";
79}
80
81function withoutDeferredPullRequests(value: DeferredShellSnapshot): OrchestrationV2ShellSnapshot {
82 if (value.loadPullRequests === undefined) return value;
83 const { loadPullRequests: _deferred, ...snapshot } = value;
84 return snapshot;
85}
86
87const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data.";
88
89/**
90 * Replay does not refresh the older-settled cut, rows that left the newest window, or
91 * projections the server rebuilt, so drift grows with the age of the last server snapshot.
92 */
93const CACHE_RESUME_MAX_AGE_MILLIS = 12 * 60 * 60 * 1000;
94
95// How long a ready socket waits for the early HTTP answer before it resumes held rows itself.
96// The supervisor's quick probe allows a live host the same time. A proxy that drops `/api`
97// without an answer would otherwise hold each connect for the full HTTP timeout.
98const EARLY_LOAD_GRACE = "3 seconds";
99
100type ShellOriginIdentity = Omit<Persistence.ShellCacheOrigin, "snapshotAt">;
101
102interface ShellSubscription {
103 readonly session: RpcSession;
104 /** What a snapshot from this subscription descends from; null when the server did not say. */
105 readonly identity: ShellOriginIdentity | null;
106}
107
108function shellIdentity(
109 environment: Pick<ExecutionEnvironmentDescriptor, "serverVersion" | "databaseMigration">,
110 flags: { readonly shellThreadFamilies?: boolean; readonly shellOlderSettled?: boolean },
111): ShellOriginIdentity {
112 const omitIdleSubagents = flags.shellThreadFamilies === true;
113 return {
114 serverVersion: environment.serverVersion,
115 // Migrations rewrite projections without events, so a migrated server reloads.
116 databaseMigration: environment.databaseMigration ?? null,
117 omitIdleSubagents,
118 omitOlderSettled: flags.shellOlderSettled === true && omitIdleSubagents,
119 };
120}
121
122function sameConnection(left: PreparedConnection, right: PreparedConnection): boolean {
123 const keys = Object.keys({ ...left, ...right }) as Array<keyof PreparedConnection>;
124 return keys.every((key) => key === "httpAuthorization" || left[key] === right[key]);
125}
126
127/** Null for a server whose descriptor does not name its shell flags; only its config does. */
128function descriptorShellIdentity(
129 descriptor: ExecutionEnvironmentDescriptor | undefined,
130): ShellOriginIdentity | null {
131 if (descriptor?.capabilities.shellThreadFamilies === undefined) return null;
132 return shellIdentity(descriptor, descriptor.capabilities);
133}
134
135function sameShellIdentity(left: ShellOriginIdentity, right: ShellOriginIdentity): boolean {
136 return (
137 left.serverVersion === right.serverVersion &&
138 left.databaseMigration === right.databaseMigration &&
139 left.omitIdleSubagents === right.omitIdleSubagents &&
140 left.omitOlderSettled === right.omitOlderSettled
141 );
142}
143
144function canResumeFrom(
145 origin: Persistence.ShellCacheOrigin | undefined,
146 identity: ShellOriginIdentity | null,
147 now: number,
148): boolean {
149 if (origin === undefined || identity === null) return false;
150 const age = now - origin.snapshotAt;
151 return sameShellIdentity(origin, identity) && age >= 0 && age < CACHE_RESUME_MAX_AGE_MILLIS;
152}
153
154type HttpShellOutcome =
155 | { readonly kind: "snapshot"; readonly projectIdentities: string | undefined }
156 /** The server can replay from the rows the client holds. */
157 | { readonly kind: "resumable" }
158 | { readonly kind: "failed" };
159
160interface EarlyShellLoad {
161 readonly prepared: PreparedConnection;
162 readonly identity: ShellOriginIdentity;
163 readonly fiber: Fiber.Fiber<HttpShellOutcome>;
164 // One subscription uses a load; a later one on the same connection decides again.
165 readonly claimed: boolean;
166}
167
168interface LiveShell {
169 readonly snapshot: OrchestrationV2ShellSnapshot;
170 // Read together with the rows, so a save never pairs them with another snapshot's origin.
171 readonly origin: Persistence.ShellCacheOrigin | undefined;
172}
173
174export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {
175 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
176 const cache = yield* Persistence.EnvironmentCacheStore;
177 const snapshotLoader = yield* ShellSnapshotLoader.ShellSnapshotLoader;
178 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
179 const environmentId = supervisor.target.environmentId;
180 const cached = yield* cache.loadShell(environmentId).pipe(
181 Effect.catch((error) =>
182 Effect.logWarning("Could not load cached environment shell.").pipe(
183 Effect.annotateLogs({
184 environmentId,
185 ...safeErrorLogAttributes(error),
186 }),
187 Effect.as(Option.none<Persistence.CachedShellSnapshot>()),
188 ),
189 ),
190 );
191 const cachedOrigin = Option.isSome(cached) ? cached.value.origin : undefined;
192 const cachedSnapshot = Option.map(cached, ({ origin: _origin, ...rows }) =>
193 withoutDeferredPullRequests(rows),
194 );
195 const state = yield* SubscriptionRef.make<EnvironmentShellState>({
196 ...EMPTY_SHELL_STATE,
197 snapshot: cachedSnapshot,
198 status: shellStatusForSnapshot(cachedSnapshot),
199 });
200 const awaitingCompletion = yield* Ref.make(false);
201 const activeSubscription = yield* Ref.make<ShellSubscription | null>(null);
202 // The server snapshot the in-memory rows descend from, directly or through the cache.
203 // Only a server snapshot renews it; `snapshot-required` drops it.
204 const origin = yield* Ref.make(cachedOrigin);
205 const reloadRequests = yield* Queue.unbounded<void>();
206 const latestLive = yield* Ref.make<Option.Option<LiveShell>>(Option.none());
207 const persistence = yield* Queue.sliding<OrchestrationV2ShellSnapshot>(1);
208
209 const persistenceLock = yield* Semaphore.make(1);
210 let lastPersisted: LiveShell | undefined;
211 // Rows still waiting for their deferred pull request links, which must not be saved:
212 // the cache would keep them without links until the next server refresh.
213 let pendingLinkRows: ReadonlySet<OrchestrationV2ThreadShell> = new Set();
214 const persistLatest = Effect.fn("EnvironmentShellState.persistLatest")(function* (
215 flush: boolean,
216 ) {
217 while (true) {
218 const latest = yield* Ref.get(latestLive);
219 if (Option.isNone(latest) || latest.value === lastPersisted) return;
220 const { snapshot, origin } = latest.value;
221 if (snapshot.threads.some((thread) => pendingLinkRows.has(thread))) return;
222 if (
223 lastPersisted !== undefined &&
224 origin === lastPersisted.origin &&
225 snapshot.snapshotSequence >= lastPersisted.snapshot.snapshotSequence &&
226 snapshot.schemaVersion === lastPersisted.snapshot.schemaVersion &&
227 snapshot.projects === lastPersisted.snapshot.projects &&
228 snapshot.threads === lastPersisted.snapshot.threads &&
229 snapshot.archivedThreads === lastPersisted.snapshot.archivedThreads
230 ) {
231 // Keep the live cursor without rewriting the rows. Remember it here too so a
232 // later authoritative reset to a lower sequence still reaches the cache.
233 lastPersisted = latest.value;
234 return;
235 }
236 const saved = yield* cache.saveShell(environmentId, snapshot, origin).pipe(
237 Effect.as(true),
238 Effect.catch((error) =>
239 Effect.logWarning("Could not persist environment shell cache.").pipe(
240 Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error) }),
241 Effect.as(false),
242 ),
243 ),
244 );
245 if (!saved) return;
246 lastPersisted = latest.value;
247 if (!flush) return;
248 }
249 }, persistenceLock.withPermit);
250 const persist = () => persistLatest(false);
251 const flushLiveShellSnapshot = persistLatest(true);
252
253 // Register before scoped worker fibers so reverse finalizer order interrupts
254 // those fibers first and this flush sees a stable latestLiveSnapshot.
255 yield* Effect.addFinalizer(() => flushLiveShellSnapshot);
256
257 yield* runCachePersistence(persistence, persist).pipe(Effect.forkScoped);
258
259 const setDisconnected = Ref.set(awaitingCompletion, false).pipe(
260 Effect.andThen(
261 SubscriptionRef.update(state, (current) => ({
262 ...current,
263 status: shellStatusForSnapshot(current.snapshot),
264 })),
265 ),
266 Effect.andThen(flushLiveShellSnapshot.pipe(Effect.forkScoped)),
267 Effect.asVoid,
268 );
269 const setSynchronizing = SubscriptionRef.update(state, (current) => ({
270 ...current,
271 status: "synchronizing" as const,
272 error: Option.none(),
273 }));
274 const setReady = SubscriptionRef.update(state, (current) =>
275 current.status === "live"
276 ? current
277 : {
278 ...current,
279 status: "synchronizing" as const,
280 error: Option.none(),
281 },
282 );
283 const setStreamError = (error: unknown) =>
284 Ref.set(awaitingCompletion, false).pipe(
285 Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")),
286 Effect.annotateLogs({
287 environmentId,
288 ...safeErrorLogAttributes(error),
289 }),
290 Effect.andThen(
291 SubscriptionRef.update(state, (current) => ({
292 ...current,
293 status: shellStatusForSnapshot(current.snapshot),
294 error: Option.some(SHELL_SYNCHRONIZATION_ERROR_MESSAGE),
295 })),
296 ),
297 );
298
299 // Server items and deferred link fills both read the snapshot and write it back; one at a
300 // time, so neither can overwrite what the other wrote in between.
301 const snapshotWriteLock = yield* Semaphore.make(1);
302
303 const applyLocked = Effect.fnUntraced(function* (
304 items: ReadonlyArray<OrchestrationV2ShellStreamItem>,
305 /** What an authoritative snapshot among the items descends from. */
306 identity: ShellOriginIdentity | null,
307 /** Set for an HTTP snapshot: the rows that still wait for their pull request links. */
308 pendingLinkIds?: ReadonlySet<ThreadId>,
309 ) {
310 const initial = yield* SubscriptionRef.get(state);
311 // No socket has confirmed an HTTP snapshot, so it never makes the shell live.
312 const fromHttp = pendingLinkIds !== undefined;
313 const disconnected = initial.status === "empty" || initial.status === "cached";
314 let waiting = yield* Ref.get(awaitingCompletion);
315 let next = initial;
316 let receivedAuthoritativeSnapshot = false;
317 let snapshotRequired = false;
318 let deltas: Array<ShellStreamEvent> = [];
319 const flushDeltas = () => {
320 if (deltas.length > 0 && Option.isSome(next.snapshot)) {
321 let sequence = next.snapshot.value.snapshotSequence;
322 let removedAt: Map<ThreadId, number> | undefined;
323 for (const delta of deltas) {
324 if (delta.sequence <= sequence) continue;
325 switch (delta.kind) {
326 case "thread.removed":
327 removedAt ??= new Map(next.removedAt);
328 removedAt.set(delta.threadId, delta.sequence);
329 break;
330 case "thread.updated":
331 case "project.updated":
332 case "project.removed":
333 break;
334 default:
335 continue;
336 }
337 sequence = delta.sequence;
338 }
339 next = {
340 ...next,
341 removedAt: removedAt ?? next.removedAt,
342 snapshot: Option.some(applyShellStreamEvents(next.snapshot.value, deltas)),
343 status: waiting ? "synchronizing" : "live",
344 error: Option.none(),
345 };
346 }
347 deltas = [];
348 };
349 for (const item of items) {
350 if (item.kind === "snapshot-required") {
351 snapshotRequired = true;
352 continue;
353 }
354 if (item.kind !== "snapshot" && item.kind !== "synchronized") {
355 deltas.push(item);
356 continue;
357 }
358 // Snapshots can reset the cursor; completion markers change the sync status.
359 flushDeltas();
360 if (item.kind === "synchronized") {
361 waiting = false;
362 if (Option.isSome(next.snapshot)) {
363 next = {
364 ...next,
365 status: "live",
366 error: Option.none(),
367 // A cache resume that caught up is as current as a snapshot for generation readers.
368 generation: next.generation === 0 ? ++lastShellGeneration : next.generation,
369 };
370 }
371 continue;
372 }
373 const authoritative = item.resolvedRepositoryIdentityRoots === undefined;
374 const nextSnapshot = mergeShellSnapshotProjects(
375 Option.getOrNull(next.snapshot),
376 item.snapshot,
377 item.resolvedRepositoryIdentityRoots === undefined
378 ? undefined
379 : { resolvedRepositoryIdentityRoots: item.resolvedRepositoryIdentityRoots },
380 );
381 if (authoritative) receivedAuthoritativeSnapshot = true;
382 next = {
383 snapshot: Option.some(nextSnapshot),
384 status:
385 fromHttp && disconnected ? "cached" : waiting || fromHttp ? "synchronizing" : "live",
386 error: Option.none(),
387 generation: authoritative ? ++lastShellGeneration : next.generation,
388 removedAt: authoritative ? EMPTY_REMOVALS : next.removedAt,
389 };
390 }
391 flushDeltas();
392 yield* Ref.set(awaitingCompletion, waiting);
393 let nextOrigin = yield* Ref.get(origin);
394 if (snapshotRequired) {
395 // The rows stay on screen but reload before the next resume; the cache keeps the old
396 // origin until that reload saves, which costs at most one more refusal.
397 nextOrigin = undefined;
398 yield* Ref.set(origin, nextOrigin);
399 yield* Queue.offer(reloadRequests, undefined);
400 } else if (receivedAuthoritativeSnapshot) {
401 nextOrigin =
402 identity === null ? undefined : { ...identity, snapshotAt: yield* Clock.currentTimeMillis };
403 yield* Ref.set(origin, nextOrigin);
404 }
405 if (next === initial) return;
406 if (pendingLinkIds !== undefined && Option.isSome(next.snapshot)) {
407 pendingLinkRows = new Set(
408 next.snapshot.value.threads.filter((thread) => pendingLinkIds.has(thread.id)),
409 );
410 }
411 if (Option.isSome(next.snapshot)) {
412 yield* Ref.set(
413 latestLive,
414 Option.some({ snapshot: next.snapshot.value, origin: nextOrigin }),
415 );
416 }
417 yield* SubscriptionRef.set(state, next);
418 if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) {
419 yield* Queue.offer(persistence, next.snapshot.value);
420 }
421 });
422
423 const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* (
424 entries: ReadonlyArray<readonly [RpcSession, OrchestrationV2ShellStreamItem]>,
425 ) {
426 // Read under the lock: items buffered across a session change must not land on the
427 // rows the next session loaded.
428 const subscription = yield* Ref.get(activeSubscription);
429 yield* applyLocked(
430 entries.flatMap(([session, item]) => (session === subscription?.session ? [item] : [])),
431 subscription?.identity ?? null,
432 );
433 }, snapshotWriteLock.withPermit);
434
435 const stateScope = yield* Effect.scope;
436 // Each full snapshot supersedes the previous one's fill; links from an older snapshot must
437 // not land on rows a newer snapshot reused.
438
439 let activeFill: Fiber.Fiber<void> | undefined;
440 const cancelActiveFill = Effect.suspend(() => {
441 const fill = activeFill;
442 activeFill = undefined;
443 return fill === undefined ? Effect.void : Fiber.interrupt(fill);
444 });
445 const fillDeferredPullRequests = Effect.fnUntraced(function* (
446 source: DeferredShellSnapshot,
447 applied: OrchestrationV2ShellSnapshot,
448 ) {
449 yield* cancelActiveFill;
450 if (source.loadPullRequests === undefined) return;
451 const arrivedRows = new Set(applied.threads);
452 const arrivedIds = new Set(source.threads.map((thread) => thread.id));
453 pendingLinkRows = new Set(applied.threads.filter((thread) => arrivedIds.has(thread.id)));
454 activeFill = yield* source.loadPullRequests.pipe(
455 Effect.flatMap((linksByThreadId) =>
456 writeFilledLinks(linksByThreadId, arrivedRows, arrivedIds),
457 ),
458 Effect.forkIn(stateScope),
459 );
460 });
461 const writeFilledLinks = Effect.fnUntraced(
462 function* (
463 linksByThreadId: ReadonlyMap<ThreadId, ReadonlyArray<ThreadPullRequestLink>>,
464 arrivedRows: ReadonlySet<OrchestrationV2ThreadShell>,
465 arrivedIds: ReadonlySet<ThreadId>,
466 ) {
467 const filled = yield* SubscriptionRef.modify(
468 state,
469 (current): [Option.Option<OrchestrationV2ShellSnapshot>, EnvironmentShellState] => {
470 if (Option.isNone(current.snapshot)) return [Option.none(), current];
471 const snapshot = current.snapshot.value;
472 let changed = false;
473 const threads = snapshot.threads.map((thread) => {
474 if (!arrivedRows.has(thread) || !arrivedIds.has(thread.id)) return thread;
475 const links = linksByThreadId.get(thread.id);
476 if (sameThreadPullRequests(thread.pullRequests, links)) return thread;
477 changed = true;
478 if (links === undefined) {
479 const { pullRequests: _previous, ...row } = thread;
480 return row;
481 }
482 return { ...thread, pullRequests: links };
483 });
484 if (!changed) return [Option.none(), current];
485 const next = { ...snapshot, threads };
486 return [Option.some(next), { ...current, snapshot: Option.some(next) }];
487 },
488 );
489 pendingLinkRows = new Set();
490 const live = yield* Ref.get(latestLive);
491 if (Option.isNone(live)) return;
492 const latest = Option.isSome(filled) ? filled.value : live.value.snapshot;
493 yield* Ref.set(latestLive, Option.some({ snapshot: latest, origin: live.value.origin }));
494 yield* Queue.offer(persistence, latest);
495 },
496 (effect) => snapshotWriteLock.withPermit(effect),
497 );
498 if (Option.isSome(cached) && Option.isSome(cachedSnapshot)) {
499 yield* fillDeferredPullRequests(cached.value, cachedSnapshot.value);
500 }
501
502 const applyHttpSnapshot = Effect.fnUntraced(function* (
503 loaded: ShellSnapshotLoader.ShellHttpSnapshot,
504 identity: ShellOriginIdentity | null,
505 ) {
506 const { projectIdentities: _identities, ...deferred } = loaded;
507 yield* cancelActiveFill;
508 const previous = yield* SubscriptionRef.get(state);
509 const snapshot = reuseUnchangedThreadShells(
510 Option.getOrNull(previous.snapshot),
511 withoutDeferredPullRequests(deferred),
512 );
513 const pendingLinkIds = new Set(
514 deferred.loadPullRequests === undefined ? [] : deferred.threads.map((thread) => thread.id),
515 );
516 yield* snapshotWriteLock.withPermit(
517 applyLocked([{ kind: "snapshot", snapshot }], identity, pendingLinkIds),
518 );
519 const current = yield* SubscriptionRef.get(state);
520 if (Option.isSome(current.snapshot)) {
521 yield* fillDeferredPullRequests(deferred, current.snapshot.value);
522 }
523 }, Effect.uninterruptible);
524
525 /**
526 * One request to the HTTP shell route. Held rows that may resume are named, so the server
527 * answers that they still can, in a few bytes, or sends the snapshot that replaces them.
528 */
529 const loadOverHttp = Effect.fn("EnvironmentShellState.loadOverHttp")(function* (
530 prepared: PreparedConnection,
531 identity: ShellOriginIdentity | null,
532 ) {
533 const held = yield* SubscriptionRef.get(state);
534 const resumeFrom =
535 Option.isSome(held.snapshot) &&
536 canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis)
537 ? held.snapshot.value.snapshotSequence
538 : undefined;
539 // HTTP and the socket must agree, or omitted children would return through the socket.
540 const answer = yield* snapshotLoader.load(prepared, {
541 omitIdleSubagents: identity?.omitIdleSubagents === true,
542 omitOlderSettled: identity?.omitOlderSettled === true,
543 ...(resumeFrom === undefined ? {} : { resumeFrom }),
544 });
545 if (Option.isNone(answer)) return { kind: "failed" } satisfies HttpShellOutcome;
546 if ("resumable" in answer.value) return { kind: "resumable" } satisfies HttpShellOutcome;
547 yield* applyHttpSnapshot(answer.value, identity);
548 return {
549 kind: "snapshot",
550 projectIdentities: answer.value.projectIdentities,
551 } satisfies HttpShellOutcome;
552 });
553
554 // A server whose descriptor names its shell flags is asked for the shell as soon as the
555 // connection is prepared, while the socket opens and the config downloads. The subscription
556 // claims that load, and reloads when the live config disagrees with the descriptor.
557 const earlyLoad = yield* SynchronizedRef.make<EarlyShellLoad | null>(null);
558 const useEarlyLoad = (prepared: PreparedConnection | null, claim: boolean) =>
559 SynchronizedRef.modifyEffect(
560 earlyLoad,
561 Effect.fnUntraced(function* (current) {
562 // A rotated credential replaces the prepared connection of a session that stays up.
563 if (current !== null && prepared !== null && sameConnection(current.prepared, prepared)) {
564 const next = { ...current, prepared, claimed: current.claimed || claim };
565 // A claimed load that still runs is handed out again. A second request could
566 // answer first, and this one would then replace its rows under a passed cursor.
567 const done = current.fiber.pollUnsafe() !== undefined;
568 return [current.claimed && done ? null : current, next] as const;
569 }
570 if (current !== null) yield* Fiber.interrupt(current.fiber);
571 const identity = descriptorShellIdentity(prepared?.descriptor);
572 if (prepared === null || identity === null) return [null, null] as const;
573 // Items a closed session still has in flight must not land on the rows this loads.
574 yield* Ref.set(activeSubscription, null);
575 const fiber = yield* loadOverHttp(prepared, identity).pipe(Effect.forkIn(stateScope));
576 const next: EarlyShellLoad = { prepared, identity, fiber, claimed: claim };
577 return [next, next] as const;
578 }),
579 ).pipe(
580 // Covers the store too. A cancel that lands during the wait above would otherwise take
581 // effect before the store, and the load just started would run on with no record.
582 Effect.uninterruptible,
583 );
584 /**
585 * The subscription waits for the early load, so its answer is applied before the socket
586 * moves the rows on. Held rows that can resume wait a limited time only: after that the
587 * load is ended, so a late answer applies nothing, and the socket resumes them.
588 */
589 const awaitEarlyLoad = Effect.fnUntraced(function* (
590 early: EarlyShellLoad,
591 identity: ShellOriginIdentity | null,
592 ) {
593 const resumable =
594 Option.isSome((yield* SubscriptionRef.get(state)).snapshot) &&
595 canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis);
596 if (!resumable) return yield* Fiber.await(early.fiber);
597 const answered = yield* Fiber.await(early.fiber).pipe(Effect.timeoutOption(EARLY_LOAD_GRACE));
598 if (Option.isSome(answered)) return answered.value;
599 // Returns after an answer that is being applied has finished, so the cursor below is read
600 // from rows that no longer change under it.
601 yield* Fiber.interrupt(early.fiber);
602 return yield* Fiber.await(early.fiber);
603 });
604
605 yield* SubscriptionRef.changes(supervisor.prepared).pipe(
606 // Reads the connection of this moment: a change handled late must not start a load for a
607 // connection that ended, which would also drop the items of the session that replaced it.
608 Stream.runForEach(() =>
609 SubscriptionRef.get(supervisor.prepared).pipe(
610 Effect.flatMap((prepared) => useEarlyLoad(Option.getOrNull(prepared), false)),
611 ),
612 ),
613 Effect.forkScoped,
614 );
615
616 const resubscriptions = Stream.merge(
617 Stream.fromQueue(reloadRequests),
618 Option.match(wakeups, {
619 onNone: () => Stream.never,
620 onSome: (service) =>
621 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
622 }),
623 );
624
625 yield* setSynchronizing;
626 yield* Effect.forkScoped(
627 subscribeDynamicWithSession(
628 ORCHESTRATION_V2_WS_METHODS.subscribeShell,
629 Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) {
630 const identity = yield* session.initialConfig.pipe(
631 Effect.map((config) => shellIdentity(config.environment, config)),
632 Effect.orElseSucceed(() => null),
633 );
634 const { omitIdleSubagents, omitOlderSettled } = identity ?? {
635 omitIdleSubagents: false,
636 omitOlderSettled: false,
637 };
638 const early = yield* useEarlyLoad(
639 Option.getOrNull(yield* SubscriptionRef.get(supervisor.prepared)),
640 true,
641 );
642 const earlyExit = early === null ? null : yield* awaitEarlyLoad(early, identity);
643 // Rows the descriptor's flags or build loaded are not kept when the live config differs:
644 // their origin then cannot resume, so the load below replaces them.
645 let loaded =
646 early !== null &&
647 earlyExit !== null &&
648 Exit.isSuccess(earlyExit) &&
649 identity !== null &&
650 sameShellIdentity(early.identity, identity)
651 ? earlyExit.value
652 : null;
653 yield* Ref.set(activeSubscription, { session, identity });
654 yield* Ref.set(awaitingCompletion, true);
655 yield* setSynchronizing;
656
657 // Rows from a recent server snapshot with the same omissions and server build resume
658 // from their cursor, in memory or from the cache. Anything else reloads the HTTP
659 // snapshot so a valid cursor cannot preserve stale or incomplete data. A `resumable`
660 // answer is checked again too: the rows can pass their age limit while the socket opens.
661 const resumingHeldRows =
662 loaded?.kind !== "snapshot" &&
663 Option.isSome((yield* SubscriptionRef.get(state)).snapshot) &&
664 canResumeFrom(yield* Ref.get(origin), identity, yield* Clock.currentTimeMillis);
665 if (!resumingHeldRows && loaded === null) {
666 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
667 Effect.flatMap(
668 Option.match({
669 onSome: Effect.succeed,
670 onNone: () =>
671 SubscriptionRef.changes(supervisor.prepared).pipe(
672 Stream.filter(Option.isSome),
673 Stream.map((value) => value.value),
674 Stream.runHead,
675 Effect.map(Option.getOrThrow),
676 ),
677 }),
678 ),
679 );
680 loaded = yield* loadOverHttp(prepared, identity);
681 }
682
683 const current = yield* SubscriptionRef.get(state);
684 const omission = {
685 ...(omitIdleSubagents ? { omitIdleSubagents: true as const } : {}),
686 ...(omitOlderSettled ? { omitOlderSettled: true as const } : {}),
687 };
688 if (Option.isNone(current.snapshot) || !(resumingHeldRows || loaded?.kind === "snapshot")) {
689 return { ...omission, requestCompletionMarker: true as const };
690 }
691 return {
692 afterSequence: current.snapshot.value.snapshotSequence,
693 // Only rows the client already held may be refused; a cursor from the snapshot just
694 // loaded falls back to the socket snapshot, so a refusal cannot loop.
695 ...(resumingHeldRows ? { httpSnapshotFallback: true as const } : {}),
696 // No stream item has touched the snapshot just loaded, so its identities still hold.
697 ...(loaded?.kind === "snapshot" && loaded.projectIdentities !== undefined
698 ? { knownProjectIdentities: loaded.projectIdentities }
699 : {}),
700 ...omission,
701 requestCompletionMarker: true as const,
702 };
703 }),
704 {
705 onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
706 retryExpectedFailureAfter: "250 millis",
707 resubscribe: resubscriptions,
708 },
709 ).pipe(Stream.runForEachArray((entries) => applyItems(entries))),
710 );
711 yield* SubscriptionRef.changes(supervisor.state).pipe(
712 Stream.runForEach((connectionState) => {
713 switch (connectionProjectionPhase(connectionState)) {
714 case "synchronizing":
715 return setSynchronizing;
716 case "disconnected":
717 return setDisconnected;
718 case "ready":
719 return setReady;
720 }
721 }),
722 Effect.forkScoped,
723 );
724
725 return state;
726});
727
728function shellStateChanges(environmentId: EnvironmentId) {
729 return followStreamInEnvironment(
730 environmentId,
731 Stream.unwrap(makeEnvironmentShellState().pipe(Effect.map(SubscriptionRef.changes))),
732 );
733}
734
735export interface EnvironmentShellSummary {
736 readonly hasSnapshot: boolean;
737 readonly hasSynchronizingShell: boolean;
738 readonly firstError: string | null;
739}
740
741const EMPTY_ENVIRONMENT_SHELL_SUMMARY: EnvironmentShellSummary = Object.freeze({
742 hasSnapshot: false,
743 hasSynchronizingShell: false,
744 firstError: null,
745});
746
747const EMPTY_SERVER_CONFIGS: ReadonlyMap<EnvironmentId, ServerConfig> = new Map();
748
749function shellSummariesEqual(
750 left: EnvironmentShellSummary,
751 right: EnvironmentShellSummary,
752): boolean {
753 return (
754 left.hasSnapshot === right.hasSnapshot &&
755 left.hasSynchronizingShell === right.hasSynchronizingShell &&
756 left.firstError === right.firstError
757 );
758}
759
760function mapsEqual<K, V>(left: ReadonlyMap<K, V>, right: ReadonlyMap<K, V>): boolean {
761 if (left.size !== right.size) {
762 return false;
763 }
764 for (const [key, value] of left) {
765 if (right.get(key) !== value) {
766 return false;
767 }
768 }
769 return true;
770}
771
772export function createEnvironmentShellSummaryAtom(input: {
773 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
774 readonly shellStateValueAtom: (environmentId: EnvironmentId) => Atom.Atom<EnvironmentShellState>;
775}) {
776 let previousSummary = EMPTY_ENVIRONMENT_SHELL_SUMMARY;
777 return Atom.make((get) => {
778 let hasSnapshot = false;
779 let hasSynchronizingShell = false;
780 let firstError: string | null = null;
781
782 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
783 const state = get(input.shellStateValueAtom(environmentId));
784 hasSynchronizingShell ||= state.status === "synchronizing";
785 if (firstError === null) {
786 firstError = Option.getOrNull(state.error);
787 }
788 if (Option.isNone(state.snapshot)) {
789 continue;
790 }
791 hasSnapshot = true;
792 }
793
794 const next: EnvironmentShellSummary = {
795 hasSnapshot,
796 hasSynchronizingShell,
797 firstError,
798 };
799 if (shellSummariesEqual(previousSummary, next)) {
800 return previousSummary;
801 }
802 previousSummary = next;
803 return previousSummary;
804 }).pipe(Atom.withLabel("environment-shell-summary"));
805}
806
807export function createEnvironmentServerConfigsAtom(input: {
808 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
809 readonly serverConfigValueAtom: (environmentId: EnvironmentId) => Atom.Atom<ServerConfig | null>;
810}) {
811 let previousServerConfigs = EMPTY_SERVER_CONFIGS;
812 return Atom.make((get) => {
813 const next = new Map<EnvironmentId, ServerConfig>();
814 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
815 const config = get(input.serverConfigValueAtom(environmentId));
816 if (config !== null) {
817 next.set(environmentId, config);
818 }
819 }
820 if (mapsEqual(previousServerConfigs, next)) {
821 return previousServerConfigs;
822 }
823 previousServerConfigs = next;
824 return previousServerConfigs;
825 }).pipe(Atom.withLabel("environment-server-configs"));
826}
827
828/**
829 * Every shell push replaces the state, so views that gate on load progress read these
830 * primitives instead and stay put while threads stream.
831 */
832function createEnvironmentShellProgressAtoms(
833 stateValueAtom: (environmentId: EnvironmentId) => Atom.Atom<EnvironmentShellState>,
834) {
835 return {
836 statusAtom: Atom.family((environmentId: EnvironmentId) =>
837 Atom.make((get) => get(stateValueAtom(environmentId)).status).pipe(
838 Atom.withLabel(`environment-shell-status:${environmentId}`),
839 ),
840 ),
841 hasSnapshotAtom: Atom.family((environmentId: EnvironmentId) =>
842 Atom.make((get) => Option.isSome(get(stateValueAtom(environmentId)).snapshot)).pipe(
843 Atom.withLabel(`environment-shell-has-snapshot:${environmentId}`),
844 ),
845 ),
846 };
847}
848
849export function createEnvironmentShellAtoms<R, E>(
850 runtime: Atom.AtomRuntime<
851 | EnvironmentRegistry.EnvironmentRegistry
852 | Persistence.EnvironmentCacheStore
853 | ShellSnapshotLoader.ShellSnapshotLoader
854 | R,
855 E
856 >,
857) {
858 const stateAtom = Atom.family((environmentId: EnvironmentId) =>
859 runtime.atom(shellStateChanges(environmentId), {
860 initialValue: EMPTY_SHELL_STATE,
861 }),
862 );
863
864 const stateValueAtom = Atom.family((environmentId: EnvironmentId) =>
865 Atom.make((get) =>
866 AsyncResult.getOrElse(get(stateAtom(environmentId)), () => EMPTY_SHELL_STATE),
867 ).pipe(Atom.withLabel(`environment-shell-state-value:${environmentId}`)),
868 );
869
870 const { statusAtom, hasSnapshotAtom } = createEnvironmentShellProgressAtoms(stateValueAtom);
871
872 return {
873 stateAtom,
874 stateValueAtom,
875 statusAtom,
876 hasSnapshotAtom,
877 };
878}
879
880export * from "./models.ts";
881export * from "./shellCommands.ts";
882export * from "./shellReducer.ts";
883export * from "./shellPullRequests.ts";
884export * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
885export * from "./snapshots.ts";
886export * from "./steadyShells.ts";