import {  EnvironmentId,  ORCHESTRATION_V2_WS_METHODS,  type OrchestrationV2ShellSnapshot,  type OrchestrationV2ShellStreamItem,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Scope from "effect/Scope";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef"; import {  AVAILABLE_CONNECTION_STATE,  PrimaryConnectionTarget,  type PreparedConnection,} from "../connection/model.ts";import * as EnvironmentSupervisor from "../connection/supervisor.ts";import * as ConnectionWakeups from "../connection/wakeups.ts";import * as Persistence from "../platform/persistence.ts";import * as RpcSession from "../rpc/session.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts";import { v2Project, v2ShellSnapshot } from "./orchestrationV2TestFixtures.ts"; const TARGET = new PrimaryConnectionTarget({  environmentId: EnvironmentId.make("environment-1"),  label: "Test environment",  httpBaseUrl: "https://environment.example.test",  wsBaseUrl: "wss://environment.example.test",}); const PREPARED: PreparedConnection = {  environmentId: TARGET.environmentId,  label: TARGET.label,  httpBaseUrl: TARGET.httpBaseUrl,  socketUrl: TARGET.wsBaseUrl,  httpAuthorization: null,  target: TARGET,}; const LIVE_SHELL_SNAPSHOT: OrchestrationV2ShellSnapshot = {  ...v2ShellSnapshot,  snapshotSequence: 1,}; function session(client: WsRpcProtocolClient): RpcSession.RpcSession {  return {    client,    initialConfig: Effect.succeed({ shellResumeCompletionMarker: true } as never),    subscribeServerConfig: (input) => client.subscribeServerConfig(input),    ready: Effect.void,    probe: Effect.void,    closed: Effect.never,  };} describe("environment shell synchronization", () => {  it.effect("publishes live state before persistence and preserves it when ready", () =>    Effect.gen(function* () {      const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();      const client = {        [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),      } as unknown as WsRpcProtocolClient;      const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);      const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(        Option.some(session(client)),      );      const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({        target: TARGET,        state: supervisorState,        session: activeSession,        prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),        connect: Effect.void,        disconnect: Effect.void,        retryNow: Effect.void,      } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);      const cache = Persistence.EnvironmentCacheStore.of({        loadShell: () => Effect.succeed(Option.none()),        saveShell: () => Effect.void,        loadThread: () => Effect.succeed(Option.none()),        saveThread: () => Effect.void,        removeThread: () => Effect.void,        loadServerConfig: () => Effect.succeed(Option.none()),        saveServerConfig: () => Effect.void,        loadVcsRefs: () => Effect.succeed(Option.none()),        saveVcsRefs: () => Effect.void,        removeVcsRefs: () => Effect.void,        clearVcsRefs: () => Effect.void,        clear: () => Effect.void,      });      // Cold cache with no HTTP snapshot available → falls back to the      // socket-embedded snapshot.      const snapshotLoader = ShellSnapshotLoader.of({        load: () => Effect.succeed(Option.none()),      });      const shellState = yield* makeEnvironmentShellState().pipe(        Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),        Effect.provideService(Persistence.EnvironmentCacheStore, cache),        Effect.provideService(ShellSnapshotLoader, snapshotLoader),      );       yield* SubscriptionRef.set(supervisorState, {        desired: true,        network: "online",        phase: "connecting",        stage: "synchronizing",        attempt: 1,        generation: 0,        lastFailure: null,        retryAt: null,      });      yield* Queue.offer(events, {        kind: "snapshot",        snapshot: LIVE_SHELL_SNAPSHOT,      });      const synchronizing = yield* SubscriptionRef.changes(shellState).pipe(        Stream.filter((state) => state.status === "synchronizing" && Option.isSome(state.snapshot)),        Stream.runHead,      );      expect(Option.getOrThrow(Option.getOrThrow(synchronizing).snapshot)).toEqual(        LIVE_SHELL_SNAPSHOT,      );       yield* Queue.offer(events, { kind: "synchronized" });      yield* SubscriptionRef.changes(shellState).pipe(        Stream.filter((state) => state.status === "live"),        Stream.runHead,      );       yield* SubscriptionRef.set(supervisorState, {        desired: true,        network: "online",        phase: "connected",        stage: null,        attempt: 1,        generation: 1,        lastFailure: null,        retryAt: null,      });      for (let index = 0; index < 10; index += 1) {        yield* Effect.yieldNow;      }       const state = yield* SubscriptionRef.get(shellState);      expect(state.status).toBe("live");      expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT);    }),  );   it.effect("refreshes a warm shell cache from HTTP before resuming", () =>    Effect.gen(function* () {      const cachedSnapshot: OrchestrationV2ShellSnapshot = {        ...v2ShellSnapshot,        snapshotSequence: 5,      };      const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();      const capturedAfterSequence = yield* SubscriptionRef.make<number | undefined>(undefined);      const capturedCompletionMarker = yield* Ref.make(false);      const loaderCalls = yield* SubscriptionRef.make(0);      const httpSnapshot: OrchestrationV2ShellSnapshot = {        ...v2ShellSnapshot,        snapshotSequence: 9,      };      const client = {        [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: {          readonly afterSequence?: number;          readonly requestCompletionMarker?: true;        }) =>          Stream.unwrap(            Effect.all([              Ref.set(capturedCompletionMarker, input.requestCompletionMarker === true),              SubscriptionRef.set(capturedAfterSequence, input.afterSequence),            ]).pipe(Effect.as(Stream.fromQueue(events))),          ),      } as unknown as WsRpcProtocolClient;      const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);      const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(        Option.some(session(client)),      );      const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({        target: TARGET,        state: supervisorState,        session: activeSession,        prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),        connect: Effect.void,        disconnect: Effect.void,        retryNow: Effect.void,      } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);      const cache = Persistence.EnvironmentCacheStore.of({        loadShell: () => Effect.succeed(Option.some(cachedSnapshot)),        saveShell: () => Effect.void,        loadThread: () => Effect.succeed(Option.none()),        saveThread: () => Effect.void,        removeThread: () => Effect.void,        loadServerConfig: () => Effect.succeed(Option.none()),        saveServerConfig: () => Effect.void,        loadVcsRefs: () => Effect.succeed(Option.none()),        saveVcsRefs: () => Effect.void,        removeVcsRefs: () => Effect.void,        clearVcsRefs: () => Effect.void,        clear: () => Effect.void,      });      const snapshotLoader = ShellSnapshotLoader.of({        load: () =>          SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe(            Effect.as(Option.some(httpSnapshot)),          ),      });      const shellState = yield* makeEnvironmentShellState().pipe(        Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),        Effect.provideService(Persistence.EnvironmentCacheStore, cache),        Effect.provideService(ShellSnapshotLoader, snapshotLoader),      );       // Wait until the subscription is established from the warm cache.      yield* SubscriptionRef.changes(capturedAfterSequence).pipe(        Stream.filter((value) => value !== undefined),        Stream.runHead,      );       expect(yield* SubscriptionRef.get(capturedAfterSequence)).toBe(9);      expect(yield* Ref.get(capturedCompletionMarker)).toBe(true);      expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1);      const synchronizing = yield* SubscriptionRef.get(shellState);      expect(synchronizing.status).toBe("synchronizing");      expect(Option.getOrThrow(synchronizing.snapshot)).toEqual(httpSnapshot);       yield* Queue.offer(events, { kind: "synchronized" });      yield* SubscriptionRef.changes(shellState).pipe(        Stream.filter((value) => value.status === "live"),        Stream.runHead,      );    }),  );   it.effect("resubscribes from the in-memory shell cursor when the app becomes active", () =>    Effect.gen(function* () {      const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();      const wakeups = yield* Queue.unbounded<ConnectionWakeups.ConnectionWakeup>();      const loaderCalls = yield* Ref.make(0);      const capturedAfterSequences = yield* Ref.make<ReadonlyArray<number | undefined>>([]);      const client = {        [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input: {          readonly afterSequence?: number;        }) =>          Stream.unwrap(            Ref.update(capturedAfterSequences, (captured) => [              ...captured,              input.afterSequence,            ]).pipe(Effect.as(Stream.fromQueue(events))),          ),      } as unknown as WsRpcProtocolClient;      const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);      const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(        Option.some(session(client)),      );      const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({        target: TARGET,        state: supervisorState,        session: activeSession,        prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),        connect: Effect.void,        disconnect: Effect.void,        retryNow: Effect.void,      } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);      const cache = Persistence.EnvironmentCacheStore.of({        loadShell: () => Effect.succeed(Option.some(LIVE_SHELL_SNAPSHOT)),        saveShell: () => Effect.void,        loadThread: () => Effect.succeed(Option.none()),        saveThread: () => Effect.void,        removeThread: () => Effect.void,        loadServerConfig: () => Effect.succeed(Option.none()),        saveServerConfig: () => Effect.void,        loadVcsRefs: () => Effect.succeed(Option.none()),        saveVcsRefs: () => Effect.void,        removeVcsRefs: () => Effect.void,        clearVcsRefs: () => Effect.void,        clear: () => Effect.void,      });      const snapshotLoader = ShellSnapshotLoader.of({        load: () =>          Ref.updateAndGet(loaderCalls, (count) => count + 1).pipe(            Effect.map((count) =>              Option.some({ ...LIVE_SHELL_SNAPSHOT, snapshotSequence: count * 10 }),            ),          ),      });      const shellState = yield* makeEnvironmentShellState().pipe(        Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),        Effect.provideService(Persistence.EnvironmentCacheStore, cache),        Effect.provideService(ShellSnapshotLoader, snapshotLoader),        Effect.provideService(          ConnectionWakeups.ConnectionWakeups,          ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }),        ),      );       // A new session starts from an authoritative HTTP snapshot.      for (let attempt = 0; attempt < 100; attempt += 1) {        if ((yield* Ref.get(capturedAfterSequences)).length >= 1) break;        yield* Effect.yieldNow;      }      expect(yield* Ref.get(capturedAfterSequences)).toEqual([10]);      yield* Queue.offer(events, { kind: "synchronized" });      yield* SubscriptionRef.changes(shellState).pipe(        Stream.filter((value) => value.status === "live"),        Stream.runHead,      );       // A newer snapshot arrives on the stream and advances the cursor.      yield* Queue.offer(events, {        kind: "snapshot",        snapshot: { ...LIVE_SHELL_SNAPSHOT, snapshotSequence: 40 },      });      yield* SubscriptionRef.changes(shellState).pipe(        Stream.filter(          (value) => Option.isSome(value.snapshot) && value.snapshot.value.snapshotSequence === 40,        ),        Stream.runHead,      );       yield* Queue.offer(wakeups, "application-active");      for (let attempt = 0; attempt < 100; attempt += 1) {        if ((yield* Ref.get(capturedAfterSequences)).length >= 2) break;        yield* Effect.yieldNow;      }      expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40]);      yield* Queue.offer(events, { kind: "synchronized" });       yield* Queue.offer(wakeups, "application-active-probe");      for (let attempt = 0; attempt < 100; attempt += 1) {        if ((yield* Ref.get(capturedAfterSequences)).length >= 3) break;        yield* Effect.yieldNow;      }      expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40]);       yield* Queue.offer(wakeups, "application-active-reconnect");      for (let attempt = 0; attempt < 10; attempt += 1) {        yield* Effect.yieldNow;      }      expect((yield* Ref.get(capturedAfterSequences)).length).toBe(3);      expect(yield* Ref.get(loaderCalls)).toBe(1);       // Replacing the session performs another authoritative refresh.      yield* SubscriptionRef.set(activeSession, Option.some(session(client)));      for (let attempt = 0; attempt < 100; attempt += 1) {        if ((yield* Ref.get(capturedAfterSequences)).length >= 4) break;        yield* Effect.yieldNow;      }      expect(yield* Ref.get(capturedAfterSequences)).toEqual([10, 40, 40, 20]);      expect(yield* Ref.get(loaderCalls)).toBe(2);    }),  );   it.effect("flushes the latest live snapshot without blocking connection state updates", () =>    Effect.gen(function* () {      const events = yield* Queue.unbounded<OrchestrationV2ShellStreamItem>();      const saved = yield* Ref.make<ReadonlyArray<OrchestrationV2ShellSnapshot>>([]);      const saveStarted = yield* Deferred.make<void>();      const releaseSave = yield* Deferred.make<void>();      const client = {        [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events),      } as unknown as WsRpcProtocolClient;      const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE);      const activeSession = yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(        Option.some(session(client)),      );      const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({        target: TARGET,        state: supervisorState,        session: activeSession,        prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),        connect: Effect.void,        disconnect: Effect.void,        retryNow: Effect.void,      } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);      const cache = Persistence.EnvironmentCacheStore.of({        loadShell: () => Effect.succeed(Option.none()),        saveShell: (_environmentId, snapshot) =>          Deferred.succeed(saveStarted, undefined).pipe(            Effect.andThen(Deferred.await(releaseSave)),            Effect.andThen(Ref.update(saved, (values) => [...values, snapshot])),          ),        loadThread: () => Effect.succeed(Option.none()),        saveThread: () => Effect.void,        removeThread: () => Effect.void,        loadServerConfig: () => Effect.succeed(Option.none()),        saveServerConfig: () => Effect.void,        loadVcsRefs: () => Effect.succeed(Option.none()),        saveVcsRefs: () => Effect.void,        removeVcsRefs: () => Effect.void,        clearVcsRefs: () => Effect.void,        clear: () => Effect.void,      });      const snapshotLoader = ShellSnapshotLoader.of({        load: () => Effect.succeed(Option.none()),      });       const shellState = yield* makeEnvironmentShellState().pipe(        Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),        Effect.provideService(Persistence.EnvironmentCacheStore, cache),        Effect.provideService(ShellSnapshotLoader, snapshotLoader),      );       yield* SubscriptionRef.set(supervisorState, {        desired: true,        network: "online",        phase: "connecting",        stage: "synchronizing",        attempt: 1,        generation: 0,        lastFailure: null,        retryAt: null,      });      yield* Queue.offer(events, {        kind: "snapshot",        snapshot: LIVE_SHELL_SNAPSHOT,      });      yield* Queue.offer(events, { kind: "synchronized" });      yield* SubscriptionRef.changes(shellState).pipe(        Stream.filter((state) => state.status === "live"),        Stream.runHead,      );       yield* SubscriptionRef.set(supervisorState, {        desired: true,        network: "online",        phase: "connecting",        stage: "synchronizing",        attempt: 2,        generation: 1,        lastFailure: null,        retryAt: null,      });      yield* SubscriptionRef.set(supervisorState, {        desired: false,        network: "online",        phase: "available",        stage: null,        attempt: 2,        generation: 2,        lastFailure: null,        retryAt: null,      });      yield* Deferred.await(saveStarted);       yield* SubscriptionRef.set(supervisorState, {        desired: true,        network: "online",        phase: "connecting",        stage: "synchronizing",        attempt: 3,        generation: 3,        lastFailure: null,        retryAt: null,      });      for (let index = 0; index < 100; index += 1) {        if ((yield* SubscriptionRef.get(shellState)).status === "synchronizing") break;        yield* Effect.yieldNow;      }       expect((yield* SubscriptionRef.get(shellState)).status).toBe("synchronizing");      expect(yield* Ref.get(saved)).toEqual([]);       yield* Deferred.succeed(releaseSave, undefined);      for (let index = 0; index < 100; index += 1) {        if ((yield* Ref.get(saved)).length > 0) break;        yield* Effect.yieldNow;      }       expect(yield* Ref.get(saved)).toEqual([LIVE_SHELL_SNAPSHOT]);    }),  );   it.effect("re-persists same-sequence enrichment when an older in-flight save completes", () =>    Effect.gen(function* () {      const repositoryIdentity = {        canonicalKey: "github.com/example/repo",        locator: {          source: "git-remote" as const,          remoteName: "origin",          remoteUrl: "https://github.com/example/repo.git",        },      };      const unenrichedSnapshot: OrchestrationV2ShellSnapshot = {        ...LIVE_SHELL_SNAPSHOT,        snapshotSequence: 1,        projects: [{ ...v2Project, repositoryIdentity: null }],      };      const enrichedSnapshot: OrchestrationV2ShellSnapshot = {        ...LIVE_SHELL_SNAPSHOT,        snapshotSequence: 1,        projects: [{ ...v2Project, repositoryIdentity }],

Lines 1–500 of 780. Load more