import {  type DesktopSshEnvironmentTarget,  EnvironmentId,  type OrchestrationShellSnapshot,} from "@t3tools/contracts";import { describe, expect, it } from "@effect/vitest";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Fiber from "effect/Fiber";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Ref from "effect/Ref";import * as Result from "effect/Result";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef"; import { SshEnvironmentGateway } from "../platform/capabilities.ts";import { RemoteDpopAccessToken, RemoteDpopAccessTokenStore } from "../authorization/tokenStore.ts";import {  BearerConnectionCredential,  BearerConnectionProfile,  BearerConnectionRegistration,  type ConnectionRegistration,  ConnectionCredentialStore,  ConnectionProfileStore,  PrimaryConnectionRegistration,  RelayConnectionRegistration,  SshConnectionProfile,  type ConnectionCredential,  type ConnectionProfile,} from "./catalog.ts";import { Connectivity } from "./connectivity.ts";import { ConnectionDriver } from "./driver.ts";import {  ConnectionTransientError,  BearerConnectionTarget,  PrimaryConnectionTarget,  RelayConnectionTarget,  SshConnectionTarget,  type ConnectionTarget,  type PreparedConnection,  type SupervisorConnectionState,} from "./model.ts";import {  ConnectionPersistenceError,  ConnectionRegistrationStore,  ConnectionTargetStore,  EnvironmentCacheStore,  EnvironmentOwnedDataCleanup,} from "../platform/persistence.ts";import { EnvironmentRegistry, environmentRegistryLayer } from "./registry.ts";import type { RpcSession } from "../rpc/session.ts";import { EnvironmentSupervisor } from "./supervisor.ts";import { ConnectionWakeups } from "./wakeups.ts"; const TARGET = new PrimaryConnectionTarget({  environmentId: EnvironmentId.make("environment-1"),  label: "Test environment",  httpBaseUrl: "https://environment.example.test",  wsBaseUrl: "wss://environment.example.test",});const SECOND_TARGET = new PrimaryConnectionTarget({  environmentId: EnvironmentId.make("environment-2"),  label: "Second environment",  httpBaseUrl: "https://environment-2.example.test",  wsBaseUrl: "wss://environment-2.example.test",}); const PREPARED: PreparedConnection = {  environmentId: TARGET.environmentId,  label: TARGET.label,  httpBaseUrl: TARGET.httpBaseUrl,  socketUrl: "wss://environment.example.test/ws",  httpAuthorization: null,  target: TARGET,}; const RELAY_TARGET = new RelayConnectionTarget({  environmentId: EnvironmentId.make("environment-relay"),  label: "Relay environment",});const SECOND_RELAY_TARGET = new RelayConnectionTarget({  environmentId: EnvironmentId.make("environment-relay-2"),  label: "Second relay environment",}); const BEARER_TARGET = new BearerConnectionTarget({  environmentId: EnvironmentId.make("environment-bearer"),  label: "Bearer environment",  connectionId: "bearer-connection",});const BEARER_PROFILE = new BearerConnectionProfile({  connectionId: BEARER_TARGET.connectionId,  environmentId: BEARER_TARGET.environmentId,  label: BEARER_TARGET.label,  httpBaseUrl: "https://bearer.example.test",  wsBaseUrl: "wss://bearer.example.test",});const BEARER_CREDENTIAL = new BearerConnectionCredential({  token: "bearer-token",}); const SSH_TARGET: DesktopSshEnvironmentTarget = {  alias: "test",  hostname: "test.example.test",  username: "developer",  port: 22,};const SSH_CONNECTION = new SshConnectionTarget({  environmentId: EnvironmentId.make("environment-ssh"),  label: "SSH environment",  connectionId: "ssh-connection",});const SSH_PROFILE = new SshConnectionProfile({  connectionId: SSH_CONNECTION.connectionId,  environmentId: SSH_CONNECTION.environmentId,  label: SSH_CONNECTION.label,  target: SSH_TARGET,}); const CACHED_SNAPSHOT: OrchestrationShellSnapshot = {  snapshotSequence: 1,  projects: [],  threads: [],  updatedAt: "2026-06-06T00:00:00.000Z",}; interface SessionControl {  readonly closed: Deferred.Deferred<never, ConnectionTransientError>;} const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (  initialTargets: ReadonlyArray<ConnectionTarget>,  initialProfiles: ReadonlyArray<ConnectionProfile> = [],  initialCredentials: ReadonlyArray<readonly [string, ConnectionCredential]> = [],  options?: {    readonly beforeSessionConnect?: (environmentId: EnvironmentId) => Effect.Effect<void>;    readonly beforeRegistrationRegister?: (      registration: ConnectionRegistration,    ) => Effect.Effect<void, ConnectionPersistenceError>;    readonly beforeRegistrationRemove?: (      target: ConnectionTarget,    ) => Effect.Effect<void, ConnectionPersistenceError>;  },) {  const storedTargets = yield* Ref.make(    new Map(initialTargets.map((target) => [target.environmentId, target])),  );  const shellCache = yield* Ref.make(new Map([[TARGET.environmentId, CACHED_SNAPSHOT]]));  const cacheClears = yield* Ref.make<ReadonlyArray<EnvironmentId>>([]);  const ownedDataClears = yield* Ref.make<ReadonlyArray<EnvironmentId>>([]);  const sessions = yield* Ref.make<ReadonlyArray<SessionControl>>([]);  const releasedSessions = yield* Ref.make(0);  const storedProfiles = yield* Ref.make(    new Map(initialProfiles.map((profile) => [profile.connectionId, profile])),  );  const profileReadCount = yield* Ref.make(0);  const storedCredentials = yield* Ref.make(new Map(initialCredentials));  const storedRemoteTokens = yield* Ref.make(    new Map([      [        SSH_CONNECTION.environmentId,        new RemoteDpopAccessToken({          environmentId: SSH_CONNECTION.environmentId,          label: SSH_CONNECTION.label,          endpoint: {            httpBaseUrl: "https://ssh.example.test",            wsBaseUrl: "wss://ssh.example.test",            providerKind: "cloudflare_tunnel",          },          accessToken: "cached-token",          expiresAtEpochMs: Number.MAX_SAFE_INTEGER,          dpopThumbprint: "thumbprint",        }),      ],    ]),  );  const disconnectedSshTargets = yield* Ref.make<ReadonlyArray<DesktopSshEnvironmentTarget>>([]);   const targetStore = ConnectionTargetStore.of({    list: Ref.get(storedTargets).pipe(Effect.map((targets) => [...targets.values()])),  });  const registrationStore = ConnectionRegistrationStore.of({    register: (registration) =>      Effect.gen(function* () {        yield* options?.beforeRegistrationRegister?.(registration) ?? Effect.void;        yield* Ref.update(storedTargets, (current) => {          const next = new Map(current);          next.set(registration.target.environmentId, registration.target);          return next;        });        switch (registration._tag) {          case "RelayConnectionRegistration":            return;          case "BearerConnectionRegistration":            yield* Ref.update(storedProfiles, (current) => {              const next = new Map(current);              next.set(registration.profile.connectionId, registration.profile);              return next;            });            yield* Ref.update(storedCredentials, (current) => {              const next = new Map(current);              next.set(registration.target.connectionId, registration.credential);              return next;            });            return;          case "SshConnectionRegistration":            yield* Ref.update(storedProfiles, (current) => {              const next = new Map(current);              next.set(registration.profile.connectionId, registration.profile);              return next;            });        }      }),    remove: (target) =>      Effect.gen(function* () {        yield* options?.beforeRegistrationRemove?.(target) ?? Effect.void;        yield* Ref.update(storedTargets, (current) => {          const next = new Map(current);          next.delete(target.environmentId);          return next;        });        if (target._tag === "BearerConnectionTarget" || target._tag === "SshConnectionTarget") {          yield* Ref.update(storedProfiles, (current) => {            const next = new Map(current);            next.delete(target.connectionId);            return next;          });          yield* Ref.update(storedCredentials, (current) => {            const next = new Map(current);            next.delete(target.connectionId);            return next;          });        }        yield* Ref.update(storedRemoteTokens, (current) => {          const next = new Map(current);          next.delete(target.environmentId);          return next;        });      }),  });  const cacheStore = EnvironmentCacheStore.of({    loadShell: (environmentId) =>      Ref.get(shellCache).pipe(        Effect.map((cache) => Option.fromUndefinedOr(cache.get(environmentId))),      ),    saveShell: (environmentId, snapshot) =>      Ref.update(shellCache, (current) => {        const next = new Map(current);        next.set(environmentId, snapshot);        return next;      }),    loadThread: (_environmentId, _threadId) => Effect.succeed(Option.none()),    saveThread: (_environmentId, _thread) => Effect.void,    removeThread: (_environmentId, _threadId) => Effect.void,    clear: (environmentId) =>      Ref.update(shellCache, (current) => {        const next = new Map(current);        next.delete(environmentId);        return next;      }).pipe(        Effect.andThen(          Ref.update(cacheClears, (environmentIds) => [...environmentIds, environmentId]),        ),      ),  });  const ownedDataCleanup = EnvironmentOwnedDataCleanup.of({    clear: (environmentId) =>      Ref.update(ownedDataClears, (environmentIds) => [...environmentIds, environmentId]),  });  const networkStatus = yield* SubscriptionRef.make<"unknown" | "offline" | "online">("online");  const connectivity = Connectivity.of({    status: SubscriptionRef.get(networkStatus),    changes: SubscriptionRef.changes(networkStatus),  });  const profileStore = ConnectionProfileStore.of({    get: (connectionId) =>      Ref.update(profileReadCount, (count) => count + 1).pipe(        Effect.andThen(Ref.get(storedProfiles)),        Effect.map((current) => Option.fromUndefinedOr(current.get(connectionId))),      ),    put: (profile) =>      Ref.update(storedProfiles, (current) => {        const next = new Map(current);        next.set(profile.connectionId, profile);        return next;      }),    remove: (connectionId) =>      Ref.update(storedProfiles, (current) => {        const next = new Map(current);        next.delete(connectionId);        return next;      }),  });  const credentialStore = ConnectionCredentialStore.of({    get: (connectionId) =>      Ref.get(storedCredentials).pipe(        Effect.map((current) => Option.fromUndefinedOr(current.get(connectionId))),      ),    put: (connectionId, credential) =>      Ref.update(storedCredentials, (current) => {        const next = new Map(current);        next.set(connectionId, credential);        return next;      }),    remove: (connectionId) =>      Ref.update(storedCredentials, (current) => {        const next = new Map(current);        next.delete(connectionId);        return next;      }),  });  const tokenStore = RemoteDpopAccessTokenStore.of({    get: (environmentId) =>      Ref.get(storedRemoteTokens).pipe(        Effect.map((current) => Option.fromUndefinedOr(current.get(environmentId))),      ),    put: (token) =>      Ref.update(storedRemoteTokens, (current) => {        const next = new Map(current);        next.set(token.environmentId, token);        return next;      }),    remove: (environmentId) =>      Ref.update(storedRemoteTokens, (current) => {        const next = new Map(current);        next.delete(environmentId);        return next;      }),  });  const sshGateway = SshEnvironmentGateway.of({    provision: () => Effect.die(new Error("SSH provisioning is not used.")),    prepare: () => Effect.die(new Error("SSH preparation is not used.")),    disconnect: (target) => Ref.update(disconnectedSshTargets, (current) => [...current, target]),  });  const driver = ConnectionDriver.of({    connect: (entry, reportProgress) =>      Effect.gen(function* () {        const target = entry.target;        const prepared = {          ...PREPARED,          environmentId: target.environmentId,          label: target.label,          target,        };        yield* reportProgress({ stage: "preparing" });        yield* reportProgress({ stage: "opening", prepared });        yield* options?.beforeSessionConnect?.(target.environmentId) ?? Effect.void;        const closed = yield* Deferred.make<never, ConnectionTransientError>();        yield* Ref.update(sessions, (current) => [...current, { closed }]);        const session = yield* Effect.acquireRelease(          Effect.succeed({            client: {} as RpcSession["client"],            initialConfig: Effect.die(new Error("Config is not used by registry tests.")),            ready: Effect.void,            probe: Effect.void,            closed: Deferred.await(closed),          } satisfies RpcSession),          () => Ref.update(releasedSessions, (count) => count + 1),        );        yield* reportProgress({ stage: "synchronizing", prepared });        yield* session.ready;        return { prepared, session };      }),  });   const cacheLayer = Layer.succeed(EnvironmentCacheStore, cacheStore);  const layer = environmentRegistryLayer.pipe(    Layer.provide(      Layer.mergeAll(        Layer.succeed(ConnectionTargetStore, targetStore),        Layer.succeed(ConnectionRegistrationStore, registrationStore),        Layer.succeed(ConnectionProfileStore, profileStore),        Layer.succeed(ConnectionCredentialStore, credentialStore),        Layer.succeed(RemoteDpopAccessTokenStore, tokenStore),        Layer.succeed(SshEnvironmentGateway, sshGateway),        Layer.succeed(Connectivity, connectivity),        Layer.succeed(ConnectionWakeups, ConnectionWakeups.of({ changes: Stream.never })),        Layer.succeed(ConnectionDriver, driver),        cacheLayer,        Layer.succeed(EnvironmentOwnedDataCleanup, ownedDataCleanup),      ),    ),  );   return {    layer,    storedTargets,    shellCache,    cacheClears,    ownedDataClears,    sessions,    releasedSessions,    storedProfiles,    profileReadCount,    storedCredentials,    storedRemoteTokens,    disconnectedSshTargets,    networkStatus,  };}); function awaitConnectionState(  registry: EnvironmentRegistry["Service"],  environmentId: EnvironmentId,  predicate: (state: SupervisorConnectionState) => boolean,) {  return Effect.gen(function* () {    const current = yield* registry.state(environmentId);    if (predicate(current)) {      return current;    }    return yield* registry      .stateChanges(environmentId)      .pipe(Stream.filter(predicate), Stream.runHead, Effect.map(Option.getOrThrow));  });} describe("EnvironmentRegistry", () => {  it.effect("hydrates connection profiles into catalog entries", () =>    Effect.gen(function* () {      const harness = yield* makeHarness([SSH_CONNECTION], [SSH_PROFILE]);       yield* Effect.gen(function* () {        const registry = yield* EnvironmentRegistry;        const entry = (yield* SubscriptionRef.get(registry.entries)).get(          SSH_CONNECTION.environmentId,        );         expect(entry?.target).toEqual(SSH_CONNECTION);        expect(Option.getOrThrow(entry?.profile ?? Option.none())).toEqual(SSH_PROFILE);      }).pipe(Effect.provide(harness.layer), Effect.scoped);    }),  );   it.effect("publishes network status changes independently of connection state", () =>    Effect.gen(function* () {      const harness = yield* makeHarness([]);       yield* Effect.gen(function* () {        const registry = yield* EnvironmentRegistry;        const offline = yield* Effect.forkChild(          SubscriptionRef.changes(registry.networkStatus).pipe(            Stream.filter((status) => status === "offline"),            Stream.runHead,            Effect.map(Option.getOrThrow),          ),        );         yield* SubscriptionRef.set(harness.networkStatus, "offline");         expect(yield* Fiber.join(offline)).toBe("offline");        expect(yield* SubscriptionRef.get(registry.networkStatus)).toBe("offline");      }).pipe(Effect.provide(harness.layer), Effect.scoped);    }),  );   it.effect("starts persisted environments independently", () =>    Effect.gen(function* () {      const bothLoadsStarted = yield* Deferred.make<void>();      const releaseLoads = yield* Deferred.make<void>();      const loadCount = yield* Ref.make(0);      const harness = yield* makeHarness([TARGET, SECOND_TARGET], [], [], {        beforeSessionConnect: () =>          Ref.updateAndGet(loadCount, (count) => count + 1).pipe(            Effect.tap((count) =>              count === 2 ? Deferred.succeed(bothLoadsStarted, undefined) : Effect.void,            ),            Effect.andThen(Deferred.await(releaseLoads)),          ),      });       yield* Effect.gen(function* () {        const registry = yield* EnvironmentRegistry;        const start = yield* Effect.forkChild(registry.start);         yield* Deferred.await(bothLoadsStarted).pipe(Effect.timeout("1 second"));        yield* Deferred.succeed(releaseLoads, undefined);        yield* Fiber.join(start);         expect(yield* Ref.get(loadCount)).toBe(2);      }).pipe(Effect.provide(harness.layer), Effect.scoped);    }),  );   it.effect("exposes the current RPC generation to late query subscribers", () =>    Effect.gen(function* () {      const harness = yield* makeHarness([TARGET]);      yield* Effect.gen(function* () {        const registry = yield* EnvironmentRegistry;        yield* registry.start;        yield* awaitConnectionState(          registry,          TARGET.environmentId,          (state) => state.phase === "connected",        );         const generation = yield* registry          .runStream(            TARGET.environmentId,

Lines 1–500 of 944. Load more