import {  EnvironmentId,  ORCHESTRATION_WS_METHODS,  type OrchestrationShellSnapshot,  type OrchestrationShellStreamItem,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef"; import {  AVAILABLE_CONNECTION_STATE,  PrimaryConnectionTarget,  type PreparedConnection,} from "../connection/model.ts";import {  EnvironmentSupervisor,  type EnvironmentSupervisorService,} from "../connection/supervisor.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import type { RpcSession } from "../rpc/session.ts";import type { WsRpcProtocolClient } from "../rpc/protocol.ts";import { makeEnvironmentShellState } from "./shell.ts"; const TARGET = new PrimaryConnectionTarget({  environmentId: EnvironmentId.make("environment-1"),  label: "Test environment",  httpBaseUrl: "https://environment.example.test",  wsBaseUrl: "wss://environment.example.test",}); const LIVE_SHELL_SNAPSHOT: OrchestrationShellSnapshot = {  snapshotSequence: 1,  projects: [],  threads: [],  updatedAt: "2026-06-06T00:00:00.000Z",}; function session(client: WsRpcProtocolClient): RpcSession {  return {    client,    initialConfig: Effect.never,    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<OrchestrationShellStreamItem>();      const client = {        [ORCHESTRATION_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>>(        Option.some(session(client)),      );      const supervisor = EnvironmentSupervisor.of({        target: TARGET,        state: supervisorState,        session: activeSession,        prepared: yield* SubscriptionRef.make(Option.none<PreparedConnection>()),        connect: Effect.void,        disconnect: Effect.void,        retryNow: Effect.void,      } satisfies EnvironmentSupervisorService);      const cache = EnvironmentCacheStore.of({        loadShell: () => Effect.succeed(Option.none()),        saveShell: () => Effect.never,        loadThread: () => Effect.succeed(Option.none()),        saveThread: () => Effect.void,        removeThread: () => Effect.void,        clear: () => Effect.void,      });      const shellState = yield* makeEnvironmentShellState().pipe(        Effect.provideService(EnvironmentSupervisor, supervisor),        Effect.provideService(EnvironmentCacheStore, cache),      );       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* 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);    }),  );});