import {  CommandId,  DEFAULT_MODEL,  DEFAULT_PROVIDER_INTERACTION_MODE,  DEFAULT_SERVER_SETTINGS,  type ModelSelection,  type OrchestrationProjectShell,  ProjectId,  ProviderInstanceId,  ThreadId,  TurnId,} from "@t3tools/contracts";import { resolveProjectAutoPull } from "@t3tools/shared/serverSettings";import * as Cause from "effect/Cause";import * as Console from "effect/Console";import * as Context from "effect/Context";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";import * as Schema from "effect/Schema";import * as Scope from "effect/Scope"; import * as ServerConfig from "./config.ts";import * as Keybindings from "./keybindings.ts";import * as ExternalLauncher from "./process/externalLauncher.ts";import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts";import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts";import * as OrchestrationReactor from "./orchestration/Services/OrchestrationReactor.ts";import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";import * as ServerSettings from "./serverSettings.ts";import * as AnalyticsService from "./telemetry/AnalyticsService.ts";import * as ServerEnvironment from "./environment/ServerEnvironment.ts";import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";import * as ProviderService from "./provider/Services/ProviderService.ts";import * as ProviderSessionDirectory from "./provider/Services/ProviderSessionDirectory.ts";import * as ProviderSessionReaper from "./provider/Services/ProviderSessionReaper.ts";import { forkParked } from "./serverActivation.ts";import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts";import * as GitVcsDriver from "./vcs/GitVcsDriver.ts";import {  formatHeadlessServeOutput,  formatHostForUrl,  isWildcardHost,  issueHeadlessServeAccessInfo,} from "./startupAccess.ts"; export class ServerRuntimeStartupError extends Schema.TaggedError<ServerRuntimeStartupError>()(  "ServerRuntimeStartupError",  {    mode: ServerConfig.RuntimeMode,    host: Schema.NullOr(Schema.String),    port: Schema.Number,    cause: Schema.Defect(),  },) {  override get message(): string {    return "Server runtime startup failed before command readiness.";  }} export class ServerRuntimeStartup extends Context.Service<  ServerRuntimeStartup,  {    readonly awaitCommandReady: Effect.Effect<void, ServerRuntimeStartupError>;    readonly markHttpListening: Effect.Effect<void>;    readonly markRunningProviderSessionsForContinuation: Effect.Effect<      ReadonlyArray<ThreadId>,      ServerUpdateThreadContinuationError    >;    readonly clearProviderSessionContinuationMarkers: (      threadIds: ReadonlyArray<ThreadId>,    ) => Effect.Effect<void, ServerUpdateThreadContinuationError>;    readonly enqueueCommand: <A, E>(      effect: Effect.Effect<A, E>,    ) => Effect.Effect<A, E | ServerRuntimeStartupError>;  }>()("t3/serverRuntimeStartup") {} interface QueuedCommand {  readonly run: Effect.Effect<void, never>;} type CommandReadinessState = "pending" | "ready" | ServerRuntimeStartupError; interface CommandGate {  readonly awaitCommandReady: Effect.Effect<void, ServerRuntimeStartupError>;  readonly signalCommandReady: Effect.Effect<void>;  readonly failCommandReady: (error: ServerRuntimeStartupError) => Effect.Effect<void>;  readonly enqueueCommand: <A, E>(    effect: Effect.Effect<A, E>,  ) => Effect.Effect<A, E | ServerRuntimeStartupError>;} const settleQueuedCommand = <A, E>(deferred: Deferred.Deferred<A, E>, exit: Exit.Exit<A, E>) =>  Exit.isSuccess(exit)    ? Deferred.succeed(deferred, exit.value)    : Deferred.failCause(deferred, exit.cause); export const makeCommandGate = Effect.gen(function* () {  const commandReady = yield* Deferred.make<void, ServerRuntimeStartupError>();  const commandQueue = yield* Queue.unbounded<QueuedCommand>();  const commandReadinessState = yield* Ref.make<CommandReadinessState>("pending");   const commandWorker = Effect.forever(    Queue.take(commandQueue).pipe(Effect.flatMap((command) => command.run)),  );  yield* Effect.forkScoped(commandWorker);   return {    awaitCommandReady: Deferred.await(commandReady),    signalCommandReady: Effect.gen(function* () {      yield* Ref.set(commandReadinessState, "ready");      yield* Deferred.succeed(commandReady, undefined).pipe(Effect.orDie);    }),    failCommandReady: (error) =>      Effect.gen(function* () {        yield* Ref.set(commandReadinessState, error);        yield* Deferred.fail(commandReady, error).pipe(Effect.orDie);      }),    enqueueCommand: <A, E>(effect: Effect.Effect<A, E>) =>      Effect.gen(function* () {        const readinessState = yield* Ref.get(commandReadinessState);        if (readinessState === "ready") {          return yield* effect;        }        if (readinessState !== "pending") {          return yield* readinessState;        }         const result = yield* Deferred.make<A, E | ServerRuntimeStartupError>();        yield* Queue.offer(commandQueue, {          run: Deferred.await(commandReady).pipe(            Effect.flatMap(() => effect),            Effect.exit,            Effect.flatMap((exit) => settleQueuedCommand(result, exit)),          ),        });        return yield* Deferred.await(result);      }),  } satisfies CommandGate;}); const recordStartupHeartbeat = Effect.gen(function* () {  const analytics = yield* AnalyticsService.AnalyticsService;  const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;   const { threadCount, projectCount } = yield* projectionSnapshotQuery.getCounts().pipe(    Effect.catch((cause) =>      Effect.logWarning("failed to gather startup projection counts for telemetry", {        cause,      }).pipe(        Effect.as({          threadCount: 0,          projectCount: 0,        }),      ),    ),  );   yield* analytics.record("server.boot.heartbeat", {    threadCount,    projectCount,  });}); const getAutoBootstrapThreadModelSelection = (): ModelSelection => ({  instanceId: ProviderInstanceId.make("codex"),  model: DEFAULT_MODEL,}); export const resolveWelcomeBase = Effect.gen(function* () {  const serverConfig = yield* ServerConfig.ServerConfig;  const segments = serverConfig.cwd.split(/[/\\]/).filter(Boolean);  const projectName = segments[segments.length - 1] ?? "project";   return {    cwd: serverConfig.cwd,    projectName,  } as const;}); export const resolveAutoBootstrapWelcomeTargets = Effect.gen(function* () {  const crypto = yield* Crypto.Crypto;  const randomUUID = crypto.randomUUIDv4;  const serverConfig = yield* ServerConfig.ServerConfig;  const projectionReadModelQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;  const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;  const path = yield* Path.Path;   let bootstrapProjectId: ProjectId | undefined;  let bootstrapThreadId: ThreadId | undefined;  let bootstrapProjectCreated = false;  let bootstrapThreadCreated = false;   if (serverConfig.autoBootstrapProjectFromCwd) {    const settings = yield* (yield* ServerSettings.ServerSettingsService).getSettings;    const defaultModelSelection =      settings.defaultModelSelection ?? getAutoBootstrapThreadModelSelection();    yield* Effect.gen(function* () {      const existingProject = yield* projectionReadModelQuery.getActiveProjectByWorkspaceRoot(        serverConfig.cwd,      );      let nextProjectId: ProjectId;      let nextThreadModelSelection: ModelSelection;       if (Option.isNone(existingProject)) {        const createdAt = DateTime.formatIso(yield* DateTime.now);        nextProjectId = ProjectId.make(yield* randomUUID);        const bootstrapProjectTitle = path.basename(serverConfig.cwd) || "project";        nextThreadModelSelection = defaultModelSelection;        yield* orchestrationEngine.dispatch({          type: "project.create",          commandId: CommandId.make(yield* randomUUID),          projectId: nextProjectId,          title: bootstrapProjectTitle,          workspaceRoot: serverConfig.cwd,          createdAt,        });        bootstrapProjectId = nextProjectId;        bootstrapProjectCreated = true;      } else {        nextProjectId = existingProject.value.id;        bootstrapProjectId = nextProjectId;        nextThreadModelSelection =          existingProject.value.defaultModelSelection ?? defaultModelSelection;      }       yield* Effect.gen(function* () {        const existingThreadId =          yield* projectionReadModelQuery.getFirstActiveThreadIdByProjectId(nextProjectId);        if (Option.isNone(existingThreadId)) {          const createdAt = DateTime.formatIso(yield* DateTime.now);          const createdThreadId = ThreadId.make(yield* randomUUID);          yield* orchestrationEngine.dispatch({            type: "thread.create",            commandId: CommandId.make(yield* randomUUID),            threadId: createdThreadId,            projectId: nextProjectId,            title: "New thread",            modelSelection: nextThreadModelSelection,            interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,            runtimeMode: "full-access",            branch: null,            worktreePath: null,            createdAt,          });          bootstrapThreadId = createdThreadId;          bootstrapThreadCreated = true;        } else {          bootstrapThreadId = existingThreadId.value;        }      }).pipe(        Effect.catchCause((cause) =>          Cause.hasInterrupts(cause)            ? Effect.failCause(cause)            : Effect.logWarning("startup thread auto-bootstrap failed", {                bootstrapProjectId: nextProjectId,                cause,              }),        ),      );    });  }   return {    ...(bootstrapProjectId ? { bootstrapProjectId } : {}),    ...(bootstrapThreadId ? { bootstrapThreadId } : {}),    ...(bootstrapProjectId ? { bootstrapProjectCreated } : {}),    ...(bootstrapThreadId ? { bootstrapThreadCreated } : {}),  } as const;}); export const completeAutoBootstrapWelcome = <A extends object, E, R>(  bootstrap: Effect.Effect<A, E, R>,) =>  bootstrap.pipe(    Effect.matchCauseEffect({      onFailure: (cause) =>        Cause.hasInterrupts(cause)          ? Effect.failCause(cause)          : Effect.logWarning("startup auto-bootstrap failed", { cause }).pipe(              Effect.as({ bootstrapStatus: "complete" as const }),            ),      onSuccess: (targets) =>        Effect.succeed({          ...targets,          bootstrapStatus: "complete" as const,        }),    }),  ); const resolveStartupBrowserTarget = Effect.gen(function* () {  const serverConfig = yield* ServerConfig.ServerConfig;  const serverAuth = yield* EnvironmentAuth.EnvironmentAuth;  const localUrl = `http://localhost:${serverConfig.port}`;  const bindUrl =    serverConfig.host && !isWildcardHost(serverConfig.host)      ? `http://${formatHostForUrl(serverConfig.host)}:${serverConfig.port}`      : localUrl;  const baseTarget = serverConfig.devUrl?.toString() ?? bindUrl;  return yield* Effect.succeed(serverConfig.mode === "desktop" ? baseTarget : undefined).pipe(    Effect.flatMap((target) =>      target ? Effect.succeed(target) : serverAuth.issueStartupPairingUrl(baseTarget),    ),  );}); const maybeOpenBrowser = (target: string) =>  Effect.gen(function* () {    const serverConfig = yield* ServerConfig.ServerConfig;    if (serverConfig.noBrowser) {      return;    }    const externalLauncher = yield* ExternalLauncher.ExternalLauncher;     yield* externalLauncher.launchBrowser(target).pipe(      Effect.catch(() =>        Effect.logInfo("browser auto-open unavailable", {          hint: `Open ${target} in your browser.`,        }),      ),    );  }); const runStartupPhase = <A, E, R>(phase: string, effect: Effect.Effect<A, E, R>) =>  effect.pipe(    Effect.annotateSpans({ "startup.phase": phase }),    Effect.withSpan(`server.startup.${phase}`),  ); const ORPHANED_PROVIDER_SESSION_ERROR =  "Provider session did not survive a server restart. Send a new message to continue.";const SERVER_UPDATE_CONTINUATION_KEY = "continueAfterServerUpdate";const SERVER_UPDATE_CONTINUATION_PROMPT = "Continue where you left off."; class ProviderSessionContinuationError extends Schema.TaggedError<ProviderSessionContinuationError>()(  "ProviderSessionContinuationError",  {    threadId: ThreadId,  },) {  override get message(): string {    return `Could not continue thread '${this.threadId}': the provider instance is missing.`;  }} export class ServerUpdateThreadContinuationError extends Schema.TaggedError<ServerUpdateThreadContinuationError>()(  "ServerUpdateThreadContinuationError",  {    cause: Schema.Defect(),  },) {  override get message(): string {    return "Could not prepare running threads to continue after the update.";  }} function hasServerUpdateContinuationMarker(  runtimePayload: unknown,): runtimePayload is Record<string, unknown> {  return (    runtimePayload !== null &&    typeof runtimePayload === "object" &&    !Array.isArray(runtimePayload) &&    SERVER_UPDATE_CONTINUATION_KEY in runtimePayload  );} function readRuntimePayload(runtimePayload: unknown): Record<string, unknown> {  return runtimePayload !== null &&    typeof runtimePayload === "object" &&    !Array.isArray(runtimePayload)    ? (runtimePayload as Record<string, unknown>)    : {};} const isServerUpdateThreadContinuationError = Schema.is(ServerUpdateThreadContinuationError); function readServerUpdateContinuationTurnId(runtimePayload: unknown): TurnId | null {  if (!hasServerUpdateContinuationMarker(runtimePayload)) {    return null;  }  const value = runtimePayload[SERVER_UPDATE_CONTINUATION_KEY];  return typeof value === "string" && value.length > 0 ? TurnId.make(value) : null;} const toServerUpdateThreadContinuationError = (cause: unknown) =>  isServerUpdateThreadContinuationError(cause)    ? cause    : new ServerUpdateThreadContinuationError({ cause }); export const markRunningProviderSessionsForContinuation = Effect.gen(function* () {  const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;  const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;  const { threads } = yield* query.getCommandReadModel();  const running = threads.filter(    (thread) =>      thread.archivedAt === null &&      thread.deletedAt === null &&      thread.session?.status === "running" &&      thread.session.activeTurnId !== null,  );   const marked: ThreadId[] = [];  return yield* Effect.gen(function* () {    for (const thread of running) {      const activeTurnId = thread.session?.activeTurnId;      if (activeTurnId === null || activeTurnId === undefined) {        continue;      }      const binding = yield* directory.getBinding(thread.id);      if (Option.isNone(binding)) {        continue;      }      if (binding.value.resumeCursor === null || binding.value.resumeCursor === undefined) {        continue;      }      yield* directory.upsert({        ...binding.value,        runtimePayload: {          ...readRuntimePayload(binding.value.runtimePayload),          [SERVER_UPDATE_CONTINUATION_KEY]: activeTurnId,          continueAfterServerUpdatePrepared: null,        },      });      marked.push(thread.id);    }    return marked;  }).pipe(    Effect.catchCause((cause) =>      clearProviderSessionContinuationMarkers(marked).pipe(Effect.andThen(Effect.failCause(cause))),    ),  );}).pipe(Effect.mapError(toServerUpdateThreadContinuationError)); const clearContinuationMarkers = (  directory: ProviderSessionDirectory.ProviderSessionDirectory["Service"],  threadIds: ReadonlyArray<ThreadId>,) =>  Effect.forEach(    threadIds,    (threadId) =>      directory.getBinding(threadId).pipe(        Effect.flatMap(          Option.match({            onNone: () => Effect.void,            onSome: (binding) =>              directory.upsert({                ...binding,                runtimePayload: {                  ...readRuntimePayload(binding.runtimePayload),                  [SERVER_UPDATE_CONTINUATION_KEY]: null,                  continueAfterServerUpdatePrepared: null,                },              }),          }),        ),      ),    { concurrency: "unbounded", discard: true },  ); const clearProviderSessionContinuationMarkers = (threadIds: ReadonlyArray<ThreadId>) =>  Effect.gen(function* () {    const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;    yield* clearContinuationMarkers(directory, threadIds);  }).pipe(Effect.mapError(toServerUpdateThreadContinuationError)); export const reconcileProviderSessions = Effect.gen(function* () {  const crypto = yield* Crypto.Crypto;  const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory;  const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService;  const providerService = yield* ProviderService.ProviderService;  const query = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery;  const settings = yield* ServerSettings.ServerSettingsService;  const continueAfterRestart = yield* settings.getSettings.pipe(    Effect.map((value) => value.continueThreadsAfterServerUpdate),    Effect.catch((cause) =>      Effect.logWarning("could not read restart continuation preference", { cause }).pipe(        Effect.as(false),      ),    ),  );   const liveThreadIds = new Set(    (yield* providerService.listSessions()).map((session) => session.threadId),  );  const { threads } = yield* query.getCommandReadModel();  // Provider startup can report ready before the continuation is submitted.  // Find those markers in one read rather than querying every idle thread.  const preparedThreadIds = new Set(    (yield* directory.listBindings().pipe(      Effect.catch((cause) =>        Effect.logWarning("failed to read prepared provider continuations", { cause }).pipe(

Lines 1–500 of 1042. Load more