import {  ORCHESTRATION_V2_WS_METHODS,  type EnvironmentId,  type OrchestrationV2ShellSnapshot,  type OrchestrationV2ShellStreamItem,  type ServerConfig,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { EnvironmentRegistry } from "../connection/registry.ts";import { connectionProjectionPhase } from "../connection/model.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import * as ConnectionWakeups from "../connection/wakeups.ts";import { safeErrorLogAttributes } from "../errors/safeLog.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import { subscribeDynamic } from "../rpc/client.ts";import type { RpcSession } from "../rpc/session.ts";import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";import { applyShellStreamEvent, mergeShellSnapshotProjects } from "./shellReducer.ts";import type { EnvironmentCatalogState } from "./connections.ts";import { followStreamInEnvironment } from "./runtime.ts"; export type EnvironmentShellStatus = "empty" | "cached" | "synchronizing" | "live"; export interface EnvironmentShellState {  readonly snapshot: Option.Option<OrchestrationV2ShellSnapshot>;  readonly status: EnvironmentShellStatus;  readonly error: Option.Option<string>;} const EMPTY_SHELL_STATE: EnvironmentShellState = {  snapshot: Option.none(),  status: "empty",  error: Option.none(),}; function shellStatusForSnapshot(  snapshot: Option.Option<OrchestrationV2ShellSnapshot>,): EnvironmentShellStatus {  return Option.isSome(snapshot) ? "cached" : "empty";} const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {  const supervisor = yield* EnvironmentSupervisor;  const cache = yield* EnvironmentCacheStore;  const snapshotLoader = yield* ShellSnapshotLoader;  const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);  const environmentId = supervisor.target.environmentId;  const cachedSnapshot = yield* cache.loadShell(environmentId).pipe(    Effect.catch((error) =>      Effect.logWarning("Could not load cached environment shell.").pipe(        Effect.annotateLogs({          environmentId,          ...safeErrorLogAttributes(error),        }),        Effect.as(Option.none<OrchestrationV2ShellSnapshot>()),      ),    ),  );  const state = yield* SubscriptionRef.make<EnvironmentShellState>({    snapshot: cachedSnapshot,    status: shellStatusForSnapshot(cachedSnapshot),    error: Option.none(),  });  const awaitingCompletion = yield* Ref.make(false);  const lastAuthoritativeSession = yield* Ref.make<RpcSession | null>(null);  const activeSubscriptionSession = yield* Ref.make<RpcSession | null>(null);  const latestLiveSnapshot = yield* Ref.make<Option.Option<OrchestrationV2ShellSnapshot>>(    Option.none(),  );  const persistence = yield* Queue.sliding<OrchestrationV2ShellSnapshot>(1);   const persist = Effect.fn("EnvironmentShellState.persist")(function* (    snapshot: OrchestrationV2ShellSnapshot,  ) {    let nextSnapshot = snapshot;    while (true) {      yield* cache.saveShell(environmentId, nextSnapshot).pipe(        Effect.catch((error) =>          Effect.logWarning("Could not persist environment shell cache.").pipe(            Effect.annotateLogs({              environmentId,              ...safeErrorLogAttributes(error),            }),          ),        ),      );       const latestSnapshot = yield* Ref.get(latestLiveSnapshot);      // Same sequence can still mean newer content (e.g. repository enrichment).      if (Option.isNone(latestSnapshot) || latestSnapshot.value === nextSnapshot) {        return;      }      nextSnapshot = latestSnapshot.value;    }  });   const flushLiveShellSnapshot = Effect.gen(function* () {    const snapshot = yield* Ref.get(latestLiveSnapshot);    if (Option.isNone(snapshot)) {      return;    }    yield* persist(snapshot.value);  });   // Register before scoped worker fibers so reverse finalizer order interrupts  // those fibers first and this flush sees a stable latestLiveSnapshot.  yield* Effect.addFinalizer(() => flushLiveShellSnapshot);   yield* Stream.fromQueue(persistence).pipe(    Stream.debounce("500 millis"),    Stream.runForEach(persist),    Effect.forkScoped,  );   const setDisconnected = Ref.set(awaitingCompletion, false).pipe(    Effect.andThen(      SubscriptionRef.update(state, (current) => ({        ...current,        status: shellStatusForSnapshot(current.snapshot),      })),    ),    Effect.andThen(flushLiveShellSnapshot.pipe(Effect.forkScoped)),    Effect.asVoid,  );  const setSynchronizing = SubscriptionRef.update(state, (current) => ({    ...current,    status: "synchronizing" as const,    error: Option.none(),  }));  const setReady = SubscriptionRef.update(state, (current) =>    current.status === "live"      ? current      : {          ...current,          status: "synchronizing" as const,          error: Option.none(),        },  );  const setStreamError = (error: unknown) =>    Ref.set(awaitingCompletion, false).pipe(      Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")),      Effect.annotateLogs({        environmentId,        ...safeErrorLogAttributes(error),      }),      Effect.andThen(        SubscriptionRef.update(state, (current) => ({          ...current,          status: shellStatusForSnapshot(current.snapshot),          error: Option.some(SHELL_SYNCHRONIZATION_ERROR_MESSAGE),        })),      ),    );   const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* (    item: OrchestrationV2ShellStreamItem,  ) {    if (item.kind === "synchronized") {      yield* Ref.set(awaitingCompletion, false);      yield* SubscriptionRef.update(state, (current) =>        Option.isSome(current.snapshot)          ? { ...current, status: "live" as const, error: Option.none() }          : current,      );      return;    }     const current = yield* SubscriptionRef.get(state);    const nextSnapshot =      item.kind === "snapshot"        ? mergeShellSnapshotProjects(            Option.getOrNull(current.snapshot),            item.snapshot,            item.resolvedRepositoryIdentityRoots === undefined              ? undefined              : {                  resolvedRepositoryIdentityRoots: item.resolvedRepositoryIdentityRoots,                },          )        : Option.match(current.snapshot, {            onNone: () => null,            onSome: (snapshot) =>              item.sequence > snapshot.snapshotSequence                ? applyShellStreamEvent(snapshot, item)                : snapshot,          });    if (nextSnapshot === null) {      return;    }     yield* Ref.set(latestLiveSnapshot, Option.some(nextSnapshot));    const waiting = yield* Ref.get(awaitingCompletion);    yield* SubscriptionRef.set(state, {      snapshot: Option.some(nextSnapshot),      status: waiting ? "synchronizing" : "live",      error: Option.none(),    });    if (item.kind === "snapshot") {      const session = yield* Ref.get(activeSubscriptionSession);      if (session !== null) {        yield* Ref.set(lastAuthoritativeSession, session);      }    }    yield* Queue.offer(persistence, nextSnapshot);  });   const foregroundResubscriptions = Option.match(wakeups, {    onNone: () => Stream.never,    onSome: (service) =>      service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),  });   yield* setSynchronizing;  yield* Effect.forkScoped(    subscribeDynamic(      ORCHESTRATION_V2_WS_METHODS.subscribeShell,      Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) {        yield* Ref.set(activeSubscriptionSession, session);        const supportsCompletionMarker = yield* session.initialConfig.pipe(          Effect.map((config) => config.shellResumeCompletionMarker === true),          Effect.orElseSucceed(() => false),        );        yield* Ref.set(awaitingCompletion, supportsCompletionMarker);        yield* setSynchronizing;         // Foreground resubscriptions on the same live session can resume from        // the in-memory cursor. A new session reloads the authoritative HTTP        // snapshot so a valid cursor cannot preserve incomplete cached data.        const hasAuthoritativeSnapshot = (yield* Ref.get(lastAuthoritativeSession)) === session;        let canResume = hasAuthoritativeSnapshot;        let current = yield* SubscriptionRef.get(state);        if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) {          const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(            Effect.flatMap(              Option.match({                onSome: Effect.succeed,                onNone: () =>                  SubscriptionRef.changes(supervisor.prepared).pipe(                    Stream.filter(Option.isSome),                    Stream.map((value) => value.value),                    Stream.runHead,                    Effect.map(Option.getOrThrow),                  ),              }),            ),          );          const httpSnapshot = yield* snapshotLoader.load(prepared);          if (Option.isSome(httpSnapshot)) {            yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value });            canResume = true;            current = yield* SubscriptionRef.get(state);          }        }         // If the authoritative refresh failed, omit the cached cursor so the        // socket fallback sends a complete snapshot for this new session.        if (!canResume || Option.isNone(current.snapshot)) {          return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {};        }        if (!supportsCompletionMarker) {          // Without a completion marker there is no synchronized signal for a          // resumed subscription, so report live immediately, like threads.          yield* SubscriptionRef.update(state, (value) => ({            ...value,            status: "live" as const,            error: Option.none(),          }));        }        return {          afterSequence: current.snapshot.value.snapshotSequence,          ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),        };      }),      {        onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),        retryExpectedFailureAfter: "250 millis",        resubscribe: foregroundResubscriptions,      },    ).pipe(Stream.runForEach(applyItem)),  );  yield* SubscriptionRef.changes(supervisor.state).pipe(    Stream.runForEach((connectionState) => {      switch (connectionProjectionPhase(connectionState)) {        case "synchronizing":          return setSynchronizing;        case "disconnected":          return setDisconnected;        case "ready":          return setReady;      }    }),    Effect.forkScoped,  );   return state;}); export function shellStateChanges(environmentId: EnvironmentId) {  return followStreamInEnvironment(    environmentId,    Stream.unwrap(makeEnvironmentShellState().pipe(Effect.map(SubscriptionRef.changes))),  );} export interface EnvironmentShellSummary {  readonly hasSnapshot: boolean;  readonly hasSynchronizingShell: boolean;  readonly hasCachedShell: boolean;  readonly hasLiveShell: boolean;  readonly firstError: string | null;  readonly latestSnapshotUpdatedAt: string | null;} const EMPTY_ENVIRONMENT_SHELL_SUMMARY: EnvironmentShellSummary = Object.freeze({  hasSnapshot: false,  hasSynchronizingShell: false,  hasCachedShell: false,  hasLiveShell: false,  firstError: null,  latestSnapshotUpdatedAt: null,}); const EMPTY_SERVER_CONFIGS: ReadonlyMap<EnvironmentId, ServerConfig> = new Map(); function shellSummariesEqual(  left: EnvironmentShellSummary,  right: EnvironmentShellSummary,): boolean {  return (    left.hasSnapshot === right.hasSnapshot &&    left.hasSynchronizingShell === right.hasSynchronizingShell &&    left.hasCachedShell === right.hasCachedShell &&    left.hasLiveShell === right.hasLiveShell &&    left.firstError === right.firstError &&    left.latestSnapshotUpdatedAt === right.latestSnapshotUpdatedAt  );} function mapsEqual<K, V>(left: ReadonlyMap<K, V>, right: ReadonlyMap<K, V>): boolean {  if (left.size !== right.size) {    return false;  }  for (const [key, value] of left) {    if (right.get(key) !== value) {      return false;    }  }  return true;} export function createEnvironmentShellSummaryAtom(input: {  readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;  readonly shellStateValueAtom: (environmentId: EnvironmentId) => Atom.Atom<EnvironmentShellState>;}) {  let previousSummary = EMPTY_ENVIRONMENT_SHELL_SUMMARY;  return Atom.make((get) => {    let hasSnapshot = false;    let hasSynchronizingShell = false;    let hasCachedShell = false;    let hasLiveShell = false;    let firstError: string | null = null;    let latestSnapshotUpdatedAt: string | null = null;     for (const environmentId of get(input.catalogValueAtom).entries.keys()) {      const state = get(input.shellStateValueAtom(environmentId));      hasSynchronizingShell ||= state.status === "synchronizing";      hasCachedShell ||= state.status === "cached";      hasLiveShell ||= state.status === "live";      if (firstError === null) {        firstError = Option.getOrNull(state.error);      }      if (Option.isNone(state.snapshot)) {        continue;      }      hasSnapshot = true;      const snapshot = state.snapshot.value;      const updatedAt = snapshot.threads.concat(snapshot.archivedThreads).reduce<string | null>(        (latest, thread) => {          const value = DateTime.formatIso(thread.updatedAt);          return latest === null || value > latest ? value : latest;        },        snapshot.projects.reduce<string | null>((latest, project) => {          return latest === null || project.updatedAt > latest ? project.updatedAt : latest;        }, null),      );      if (        updatedAt !== null &&        (latestSnapshotUpdatedAt === null || updatedAt > latestSnapshotUpdatedAt)      ) {        latestSnapshotUpdatedAt = updatedAt;      }    }     const next: EnvironmentShellSummary = {      hasSnapshot,      hasSynchronizingShell,      hasCachedShell,      hasLiveShell,      firstError,      latestSnapshotUpdatedAt,    };    if (shellSummariesEqual(previousSummary, next)) {      return previousSummary;    }    previousSummary = next;    return previousSummary;  }).pipe(Atom.withLabel("environment-shell-summary"));} export function createEnvironmentServerConfigsAtom(input: {  readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;  readonly serverConfigValueAtom: (environmentId: EnvironmentId) => Atom.Atom<ServerConfig | null>;}) {  let previousServerConfigs = EMPTY_SERVER_CONFIGS;  return Atom.make((get) => {    const next = new Map<EnvironmentId, ServerConfig>();    for (const environmentId of get(input.catalogValueAtom).entries.keys()) {      const config = get(input.serverConfigValueAtom(environmentId));      if (config !== null) {        next.set(environmentId, config);      }    }    if (mapsEqual(previousServerConfigs, next)) {      return previousServerConfigs;    }    previousServerConfigs = next;    return previousServerConfigs;  }).pipe(Atom.withLabel("environment-server-configs"));} export function createEnvironmentShellAtoms<R, E>(  runtime: Atom.AtomRuntime<    EnvironmentRegistry | EnvironmentCacheStore | ShellSnapshotLoader | R,    E  >,) {  const stateAtom = Atom.family((environmentId: EnvironmentId) =>    runtime.atom(shellStateChanges(environmentId), {      initialValue: EMPTY_SHELL_STATE,    }),  );   const stateValueAtom = Atom.family((environmentId: EnvironmentId) =>    Atom.make((get) =>      Option.getOrElse(AsyncResult.value(get(stateAtom(environmentId))), () => EMPTY_SHELL_STATE),    ).pipe(Atom.withLabel(`environment-shell-state-value:${environmentId}`)),  );   return {    stateAtom,    stateValueAtom,  };} export * from "./models.ts";export * from "./shellCommands.ts";export * from "./shellReducer.ts";export * from "./shellSnapshotHttp.ts";export * from "./snapshots.ts";