import {  type EnvironmentId,  type ServerConfig,  type ServerConfigStreamEvent,  type ServerLifecycleWelcomePayload,  type ServerLifecycleStreamReadyEvent,  type ServerSelfUpdateProgressEvent,  type ServerSelfUpdateResult,  type ServerSelfUpdateCapability,  ServerAutomaticUpdateDeferredError,  WS_METHODS,} from "@t3tools/contracts";import {  type AutomaticServerUpdateThreadActivity,  isAutomaticServerUpdateThreadActive,} from "@t3tools/shared/automaticServerUpdate";import { compareSemverVersions, parseSemver } from "@t3tools/shared/semver";import * as Cause from "effect/Cause";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Result from "effect/Result";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { AsyncResult, Atom } from "effect/unstable/reactivity"; import {  createAtomCommandScheduler,  createEnvironmentRpcCommand,  createEnvironmentRpcQueryAtomFamily,  createEnvironmentRpcSubscriptionAtomFamily,  createRuntimeCommand,  scheduleAtomCommandEffect,} from "./runtime.ts";import { EnvironmentRegistry } from "../connection/registry.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import { safeErrorLogAttributes } from "../errors/safeLog.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import {  isRpcClientError,  request,  runStream,  subscribe,  type EnvironmentRpcInput,} from "../rpc/client.ts";import { followStreamInEnvironment } from "./runtime.ts"; const isServerAutomaticUpdateDeferredError = Schema.is(ServerAutomaticUpdateDeferredError); export type ServerUpdateStage = "downloading" | "installing" | "resuming"; export type AutomaticServerUpdateDecision =  | { readonly status: "ineligible" }  | {      readonly status: "pending";      readonly reason: "synchronizing" | "active-work";      readonly fromVersion: string;      readonly targetVersion: string;    }  | {      readonly status: "ready";      readonly fromVersion: string;      readonly targetVersion: string;    }; export function automaticServerUpdateDecision(input: {  readonly enabled: boolean;  readonly selfUpdate: ServerSelfUpdateCapability | null;  readonly clientVersion: string;  readonly serverVersion: string;  readonly shellStatus: "empty" | "cached" | "synchronizing" | "live";  readonly threads: ReadonlyArray<AutomaticServerUpdateThreadActivity>;}): AutomaticServerUpdateDecision {  if (    !input.enabled ||    input.selfUpdate !== "boot-service" ||    parseSemver(input.clientVersion) === null ||    parseSemver(input.serverVersion) === null ||    compareSemverVersions(input.clientVersion, input.serverVersion) <= 0  ) {    return { status: "ineligible" };  }   const target = {    fromVersion: input.serverVersion,    targetVersion: input.clientVersion,  };  if (input.shellStatus !== "live") {    return { status: "pending", reason: "synchronizing", ...target };  }  if (input.threads.some(isAutomaticServerUpdateThreadActive)) {    return { status: "pending", reason: "active-work", ...target };  }  return { status: "ready", ...target };} export type ServerUpdateState =  | { readonly status: "idle" }  | {      readonly status: "pending";      readonly reason: "synchronizing" | "active-work";      readonly fromVersion: string;      readonly targetVersion: string;    }  | {      readonly status: "running";      readonly stage: ServerUpdateStage;      readonly fromVersion: string;      readonly targetVersion: string;    }  | {      readonly status: "failed";      readonly stage: ServerUpdateStage;      readonly fromVersion: string;      readonly targetVersion: string;      readonly message: string;    }; export interface ServerUpdateTarget {  readonly environmentId: EnvironmentId;  readonly input: EnvironmentRpcInput<typeof WS_METHODS.serverUpdateServer>;} const IDLE_SERVER_UPDATE_STATE: ServerUpdateState = { status: "idle" };const EMPTY_SERVER_UPDATE_STATE_ATOM = Atom.make<ServerUpdateState>(IDLE_SERVER_UPDATE_STATE).pipe(  Atom.withLabel("environment-data:server:update-state:empty"),);const serverUpdateStateAtom = Atom.family((environmentId: EnvironmentId) =>  Atom.make<ServerUpdateState>(IDLE_SERVER_UPDATE_STATE).pipe(    Atom.withLabel(`environment-data:server:update-state:${environmentId}`),  ),); export class ServerUpdateResumeTimeoutError extends Schema.TaggedErrorClass<ServerUpdateResumeTimeoutError>()(  "ServerUpdateResumeTimeoutError",  {    environmentId: Schema.String,    targetVersion: Schema.String,  },) {  override get message(): string {    return `The server did not resume on t3@${this.targetVersion}.`;  }} export class ServerUpdateProgressIncompleteError extends Schema.TaggedErrorClass<ServerUpdateProgressIncompleteError>()(  "ServerUpdateProgressIncompleteError",  {    targetVersion: Schema.String,  },) {  override get message(): string {    return `The t3@${this.targetVersion} update ended before the server accepted the restart.`;  }} export class ServerUpdateTerminalError extends Schema.TaggedErrorClass<ServerUpdateTerminalError>()(  "ServerUpdateTerminalError",  {    targetVersion: Schema.String,    status: Schema.Literals(["committed", "rolled-back", "failed"]),    reason: Schema.optional(Schema.String),  },) {  override get message(): string {    return this.reason ?? `The t3@${this.targetVersion} update ${this.status}.`;  }} // Covers the 120-second trial deadline and a final restart of the previous// version when the trial rolls back.const SERVER_UPDATE_RESUME_TIMEOUT = Duration.minutes(4); export function matchesServerUpdateReadyEvent(  result: ServerSelfUpdateResult,  event: ServerLifecycleStreamReadyEvent,): boolean {  return result.updateId === undefined    ? event.payload.environment.serverVersion === result.targetVersion    : event.payload.updateOutcome?.id === result.updateId;} export function validateServerUpdateReadyEvent(  result: ServerSelfUpdateResult,  event: ServerLifecycleStreamReadyEvent,): Effect.Effect<void, ServerUpdateTerminalError> {  if (result.updateId === undefined) return Effect.void;  const outcome = event.payload.updateOutcome;  if (    outcome?.id === result.updateId &&    outcome.status === "committed" &&    outcome.targetVersion === result.targetVersion &&    event.payload.environment.serverVersion === result.targetVersion  ) {    return Effect.void;  }  return Effect.fail(    new ServerUpdateTerminalError({      targetVersion: result.targetVersion,      status: outcome?.status ?? "failed",      reason:        outcome?.reason ??        "The service launcher resumed without committing the requested server version.",    }),  );} /** * Keeps reconnect attempts ~1s apart for the whole update restart. * * A restart takes the server down for ~15 seconds, but the supervisor's normal * backoff ladder (1/2/4/8/16s) assumes an unexpected failure and lands attempts * at ~3, 5, 9, 17 and 33 seconds — so a 15-second restart is observed as a * 33-second "Resuming". Nudging on every backoff entry (not just the first) * holds the retry cadence flat until the server answers again. The sleep before * each nudge is the pacer: a connection that fails instantly re-enters backoff * immediately and would otherwise spin a tight retry loop. * * Callers fork this as a child of the update command so it is interrupted as * soon as the update settles, whether it succeeds, fails, or times out. */export function nudgeReconnectDuringUpdateRestart(input: {  readonly stateChanges: Stream.Stream<{ readonly phase: string }, unknown>;  readonly retryNow: Effect.Effect<void>;  readonly interval?: Duration.Duration;}): Effect.Effect<void> {  return input.stateChanges.pipe(    Stream.filter((state) => state.phase === "backoff"),    Stream.runForEach(() =>      Effect.sleep(input.interval ?? Duration.seconds(1)).pipe(Effect.andThen(input.retryNow)),    ),    Effect.timeoutOption(SERVER_UPDATE_RESUME_TIMEOUT),    Effect.ignore,  );} export function serverUpdateStateForProgressEvent(  fromVersion: string,  targetVersion: string,  event: ServerSelfUpdateProgressEvent,): Extract<ServerUpdateState, { status: "running" }> {  return {    status: "running",    stage: event.type === "complete" ? "resuming" : event.stage,    fromVersion,    targetVersion,  };} export function serverUpdateStateForAutomaticDecision(  current: ServerUpdateState,  pending: Extract<ServerUpdateState, { status: "pending" }> | null,): ServerUpdateState {  if (current.status === "running") {    return current;  }  if (    current.status === "failed" &&    pending !== null &&    current.fromVersion === pending.fromVersion &&    current.targetVersion === pending.targetVersion  ) {    return current;  }  return pending ?? IDLE_SERVER_UPDATE_STATE;} export function serverUpdateStateForServerVersion(  state: ServerUpdateState,  serverVersion: string | null,): ServerUpdateState {  return state.status === "idle" ||    state.status === "running" ||    serverVersion === null ||    state.fromVersion === serverVersion    ? state    : IDLE_SERVER_UPDATE_STATE;} function serverUpdateFailureMessage(error: unknown): string {  return error instanceof Error ? error.message : "Server update failed.";} function isRpcSocketError(error: unknown): boolean {  if (!isRpcClientError(error)) {    return false;  }  switch (error.reason._tag) {    case "SocketReadError":    case "SocketWriteError":    case "SocketCloseError":      return true;    default:      return false;  }} export function isLegacyUpdateHandoffLoss(cause: Cause.Cause<unknown>): boolean {  if (Cause.hasInterruptsOnly(cause)) {    return true;  }  return (    cause.reasons.length > 0 &&    cause.reasons.every((reason) => Cause.isFailReason(reason) && isRpcSocketError(reason.error))  );} export function resolveServerUpdateProgressResult<E>(  targetVersion: string,  terminal: Option.Option<ServerSelfUpdateResult>,  streamExit: Exit.Exit<void, E>,): Effect.Effect<ServerSelfUpdateResult, E | ServerUpdateProgressIncompleteError> {  if (    Option.isSome(terminal) &&    (Exit.isSuccess(streamExit) || isLegacyUpdateHandoffLoss(streamExit.cause))  ) {    return Effect.succeed(terminal.value);  }  if (Exit.isFailure(streamExit)) {    return Effect.failCause(streamExit.cause);  }  return Effect.fail(new ServerUpdateProgressIncompleteError({ targetVersion }));} export interface ServerConfigProjection {  readonly config: ServerConfig;  readonly latestEvent: ServerConfigStreamEvent;  readonly source: "cache" | "live";} export function applyServerConfigProjection(  current: Option.Option<ServerConfigProjection>,  event: ServerConfigStreamEvent,): Option.Option<ServerConfigProjection> {  switch (event.type) {    case "snapshot":      return Option.some({        config: event.config,        latestEvent: event,        source: "live",      });    case "keybindingsUpdated":      return Option.map(current, (projection) => ({        config: {          ...projection.config,          keybindings: event.payload.keybindings,          issues: event.payload.issues,        },        latestEvent: event,        source: "live",      }));    case "providerStatuses":      return Option.map(current, (projection) => ({        config: {          ...projection.config,          providers: event.payload.providers,        },        latestEvent: event,        source: "live",      }));    case "settingsUpdated":      return Option.map(current, (projection) => ({        config: {          ...projection.config,          settings: event.payload.settings,        },        latestEvent: event,        source: "live",      }));  }} export function projectServerConfig(  current: Option.Option<ServerConfigProjection>,  event: ServerConfigStreamEvent,): readonly [Option.Option<ServerConfigProjection>, ReadonlyArray<ServerConfigProjection>] {  const next = applyServerConfigProjection(current, event);  return [next, Option.toArray(next)];} const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEvent => ({  version: 1,  type: "snapshot",  config,}); /** * Keeps a complete server configuration available during reconnects. Server * config carries the provider/model catalogue used by task creation, so it is * useful—and safe—to retain after a transport session ends. */export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")(  function* () {    const supervisor = yield* EnvironmentSupervisor;    const cache = yield* EnvironmentCacheStore;    const environmentId = supervisor.target.environmentId;    const cachedConfig = yield* cache.loadServerConfig(environmentId).pipe(      Effect.catch((error) =>        Effect.logWarning("Could not load cached server configuration.").pipe(          Effect.annotateLogs({            environmentId,            ...safeErrorLogAttributes(error),          }),          Effect.as(Option.none<ServerConfig>()),        ),      ),    );    const state = yield* SubscriptionRef.make<Option.Option<ServerConfigProjection>>(      Option.map(cachedConfig, (config) => ({        config,        latestEvent: cachedConfigSnapshotEvent(config),        source: "cache" as const,      })),    );    const persistence = yield* Queue.sliding<ServerConfig>(1);    const pendingPersistence = yield* Ref.make<Option.Option<ServerConfig>>(Option.none());     const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* (      config: ServerConfig,    ) {      return yield* cache.saveServerConfig(environmentId, config).pipe(        Effect.as(true),        Effect.catch((error) =>          Effect.logWarning("Could not persist cached server configuration.").pipe(            Effect.annotateLogs({              environmentId,              ...safeErrorLogAttributes(error),            }),            Effect.as(false),          ),        ),      );    });     const persistPending = Effect.fn("EnvironmentServerConfigState.persistPending")(function* (      config: ServerConfig,    ) {      if (!(yield* persist(config))) {        return;      }      yield* Ref.update(pendingPersistence, (pending) =>        Option.isSome(pending) && pending.value === config ? Option.none() : pending,      );    });     yield* Stream.fromQueue(persistence).pipe(      Stream.debounce("500 millis"),      Stream.runForEach(persistPending),      Effect.forkScoped,    );     yield* subscribe(WS_METHODS.subscribeServerConfig, {}).pipe(      Stream.runForEach((event) =>        Effect.gen(function* () {          const next = applyServerConfigProjection(yield* SubscriptionRef.get(state), event);          if (Option.isNone(next)) {            return;          }          yield* Ref.set(pendingPersistence, Option.some(next.value.config));          yield* SubscriptionRef.set(state, next);          yield* Queue.offer(persistence, next.value.config);        }),      ),      Effect.forkScoped,    );     yield* Effect.addFinalizer(() =>      Ref.get(pendingPersistence).pipe(        Effect.flatMap(          Option.match({            onNone: () => Effect.void,            onSome: (config) => persist(config).pipe(Effect.asVoid),          }),        ),      ),    );     return state;  },); export function serverConfigStateChanges(environmentId: EnvironmentId) {  return followStreamInEnvironment(    environmentId,    Stream.unwrap(      makeEnvironmentServerConfigState().pipe(        Effect.map((state) =>          SubscriptionRef.changes(state).pipe(            Stream.filterMap((projection) =>              Option.match(projection, {                onNone: () => Result.failVoid,                onSome: (value) => Result.succeed(value),              }),            ),          ),        ),

Lines 1–500 of 873. Load more