import { CommandId, DEFAULT_MODEL, DEFAULT_PROVIDER_INTERACTION_MODE, DEFAULT_SERVER_SETTINGS, type ModelSelection, type Project, ProjectId, ProviderInstanceId, ThreadId, } from "@t3tools/contracts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; 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 Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; 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 ServerConfig from "./config.ts"; import * as ServiceLauncherClient from "./cloud/serviceLauncherClient.ts"; import { flushCompileCache } from "./compileCache.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as EffectWorker from "./orchestration-v2/EffectWorker.ts"; import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; import * as ThreadLaunch from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; import * as ProjectService from "./project/ProjectService.ts"; import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerSettings from "./serverSettings.ts"; import { forkParked, forkParkedFiber } from "./serverActivation.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import { getServerOwners } from "./telemetry/Identify.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import { formatHeadlessServeOutput, formatHostForUrl, isWildcardHost, issueHeadlessServeAccessInfo, } from "./startupAccess.ts"; export class ServerRuntimeStartupError extends Schema.TaggedError()( "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; readonly markHttpListening: Effect.Effect; readonly enqueueCommand: ( effect: Effect.Effect, ) => Effect.Effect; } >()("t3/serverRuntimeStartup") {} interface QueuedCommand { readonly run: Effect.Effect; } type CommandReadinessState = "pending" | "ready" | ServerRuntimeStartupError; interface CommandGate { readonly awaitCommandReady: Effect.Effect; readonly signalCommandReady: Effect.Effect; readonly failCommandReady: (error: ServerRuntimeStartupError) => Effect.Effect; readonly enqueueCommand: ( effect: Effect.Effect, ) => Effect.Effect; } const settleQueuedCommand = (deferred: Deferred.Deferred, exit: Exit.Exit) => Exit.isSuccess(exit) ? Deferred.succeed(deferred, exit.value) : Deferred.failCause(deferred, exit.cause); export const makeCommandGate = Effect.gen(function* () { const commandReady = yield* Deferred.make(); const commandQueue = yield* Queue.unbounded(); const commandReadinessState = yield* Ref.make("pending"); const commandWorker = Effect.forever( Queue.take(commandQueue).pipe(Effect.flatMap((command) => command.run)), ); yield* Effect.forkScoped(commandWorker); const enqueueCommand = Effect.fn("serverRuntimeStartup.enqueueCommand")(function* ( effect: Effect.Effect, ): Effect.fn.Return { const readinessState = yield* Ref.get(commandReadinessState); if (readinessState === "ready") { return yield* effect; } if (readinessState !== "pending") { return yield* readinessState; } const result = yield* Deferred.make(); 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); }); return { awaitCommandReady: Deferred.await(commandReady), signalCommandReady: Effect.gen(function* () { yield* Ref.set(commandReadinessState, "ready"); yield* Deferred.succeed(commandReady, undefined).pipe(Effect.orDie); }), failCommandReady: Effect.fn("serverRuntimeStartup.failCommandReady")(function* (error) { yield* Ref.set(commandReadinessState, error); yield* Deferred.fail(commandReady, error).pipe(Effect.orDie); }), enqueueCommand, } satisfies CommandGate; }); const recordStartupHeartbeat = Effect.gen(function* () { const analytics = yield* AnalyticsService.AnalyticsService; const projects = yield* ProjectService.ProjectService; const threads = yield* ThreadManagement.ThreadManagementService; const { threadCount, projectCount } = yield* Effect.all({ projects: projects.snapshot, threads: threads.getShellSnapshot(), }).pipe( Effect.map(({ projects: projectSnapshot, threads: shellSnapshot }) => ({ projectCount: projectSnapshot.projects.length, threadCount: shellSnapshot.threads.length + shellSnapshot.archivedThreads.length, })), Effect.catch((cause) => Effect.logWarning("failed to gather V2 startup counts for telemetry", { cause, }).pipe( Effect.as({ threadCount: 0, projectCount: 0, }), ), ), ); yield* analytics.record("server.boot.heartbeat", { threadCount, projectCount, }); yield* analytics.recordOwners(yield* getServerOwners, "server"); }); const getAutoBootstrapThreadModelSelection = (): ModelSelection => ({ instanceId: ProviderInstanceId.make("codex"), model: DEFAULT_MODEL, }); interface AutoBootstrapWelcomeTargets { readonly bootstrapProjectId?: ProjectId; readonly bootstrapThreadId?: ThreadId; } export const autoPullProjects = Effect.fn("autoPullProjects")(function* ( projects: ReadonlyArray>, settings = DEFAULT_SERVER_SETTINGS, ) { const git = yield* GitVcsDriver.GitVcsDriver; const workspaceRoots = [ ...new Set( projects .filter((project) => resolveProjectSettings(settings, project.id).settings.defaultAutoPull) .map((project) => project.workspaceRoot), ), ]; yield* Effect.forEach( workspaceRoots, (cwd) => Effect.gen(function* () { const status = yield* git.statusDetails(cwd); if ( !status.isRepo || !status.isDefaultBranch || !status.hasUpstream || status.hasWorkingTreeChanges || status.aheadCount > 0 ) { yield* Effect.logDebug("Skipped automatic project pull", { cwd, reason: !status.isRepo ? "not-a-repository" : !status.isDefaultBranch ? "not-on-default-branch" : !status.hasUpstream ? "no-upstream" : status.hasWorkingTreeChanges ? "working-tree-changes" : "local-commits", }); return; } if (status.behindCount <= 0) return; const result = yield* git.pullCurrentBranch(cwd); yield* Effect.logDebug("Automatic project pull completed", { cwd, status: result.status, refName: result.refName, }); }).pipe( Effect.catch((cause) => Effect.logWarning("Automatic project pull failed", { cwd, cause, }), ), ), { concurrency: 4, discard: true }, ); }); 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; }); const resolveAutoBootstrapWelcomeTargets = Effect.gen(function* () { const crypto = yield* Crypto.Crypto; const randomUUID = crypto.randomUUIDv4(); const serverConfig = yield* ServerConfig.ServerConfig; const projects = yield* ProjectService.ProjectService; const threads = yield* ThreadManagement.ThreadManagementService; const threadLaunch = yield* ThreadLaunch.ThreadLaunchService; const path = yield* Path.Path; let bootstrapProjectId: ProjectId | undefined; let bootstrapThreadId: ThreadId | undefined; if (serverConfig.autoBootstrapProjectFromCwd) { const threadModelSelection = getAutoBootstrapThreadModelSelection(); const { project } = yield* projects.bootstrap({ commandId: CommandId.make(yield* randomUUID), projectId: ProjectId.make(yield* randomUUID), title: path.basename(serverConfig.cwd) || "project", workspaceRoot: serverConfig.cwd, }); const shell = yield* threads.getShellSnapshot(); const existingThread = shell.threads.find( (thread) => thread.projectId === project.id && thread.lineage.relationshipToParent !== "subagent", ); if (existingThread === undefined) { const serverSettings = yield* ServerSettings.ServerSettingsService; const settings = yield* serverSettings.getSettings; const launched = yield* threadLaunch .launch({ commandId: CommandId.make(yield* randomUUID), projectId: project.id, title: "New thread", modelSelection: resolveProjectSettings(settings, project.id, project).settings.defaultModelSelection ?? threadModelSelection, interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, runtimeMode: resolveProjectSettings(settings, project.id, project).settings .defaultRuntimeMode, workspaceStrategy: { type: "root" }, createdBy: "system", creationSource: "server", }) .pipe( // The bootstrap thread is optional; its launch must not prevent opening the app. Effect.catchTags({ ThreadManagementProviderInstanceDisabledError: (error) => Effect.logInfo("Skipped bootstrap thread because its provider instance is disabled", { projectId: project.id, instanceId: error.instanceId, }).pipe(Effect.as(undefined)), ThreadLaunchError: (error) => Effect.logWarning("Failed to launch bootstrap thread", { projectId: project.id, operation: error.operation, }).pipe(Effect.as(undefined)), }), ); bootstrapProjectId = project.id; bootstrapThreadId = launched?.threadId; } else { bootstrapProjectId = project.id; bootstrapThreadId = existingThread.id; } } return { ...(bootstrapProjectId ? { bootstrapProjectId } : {}), ...(bootstrapThreadId ? { bootstrapThreadId } : {}), } satisfies AutoBootstrapWelcomeTargets; }); 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 serverConfig.mode === "desktop" ? baseTarget : yield* serverAuth.issueStartupPairingUrl(baseTarget); }); /** Prints the one-time admin pairing URL to stdout only; logger records reach trace files and collectors. */ export const announceStartupPairingUrl = (pairingUrl: string) => Effect.logInfo("Authentication required. Open T3 Code using the pairing URL.").pipe( Effect.andThen(Console.log(` pairingUrl: ${pairingUrl}`)), ); // Pairing tokens belong only in announceStartupPairingUrl's stdout output. export const logBrowserAutoOpenUnavailable = Effect.logInfo("browser auto-open unavailable", { hint: "Open the pairing URL above in your browser.", }); export const maybeOpenBrowser = Effect.fn("serverRuntimeStartup.maybeOpenBrowser")(function* ( target: string, ) { const serverConfig = yield* ServerConfig.ServerConfig; if (serverConfig.noBrowser) { return; } const externalLauncher = yield* ExternalLauncher.ExternalLauncher; yield* externalLauncher.launchBrowser(target).pipe( // The target carries the pairing token, and a failed launch records its // command line on the span, so this call stays out of traces. Effect.withTracerEnabled(false), Effect.catch(() => logBrowserAutoOpenUnavailable), ); }); const runStartupPhase = (phase: string, effect: Effect.Effect) => effect.pipe( // Use span attributes; annotations would follow long-lived forked fibers for the server's lifetime. Effect.withSpan(`server.startup.${phase}`, { attributes: { "startup.phase": phase } }), ); interface StartupOptions { readonly activate?: Effect.Effect; readonly awaitAuxiliaryParked?: Effect.Effect; readonly abort?: (error: ServerRuntimeStartupError) => Effect.Effect; } export const startEffectWorkerWithRelay = Effect.fn( "ServerRuntimeStartup.startEffectWorkerWithRelay", )(function* (input: { readonly runWorker: Effect.Effect; readonly startRelay: Effect.Effect; readonly workerFiberRef: Ref.Ref | null>; }) { const workerFiber = yield* forkParkedFiber(input.runWorker); yield* Ref.set(input.workerFiberRef, workerFiber); yield* input.startRelay.pipe( Effect.onExit((exit) => { if (Exit.isSuccess(exit)) { return Effect.void; } return Ref.getAndSet(input.workerFiberRef, null).pipe( Effect.flatMap((ownedWorkerFiber) => ownedWorkerFiber === null ? Effect.void : Fiber.interrupt(ownedWorkerFiber).pipe(Effect.asVoid), ), ); }), ); }); const runOrderedV2StartupPhases = Effect.fn("serverRuntimeStartup.runOrderedV2StartupPhases")( function* < Import, Recovery, Bootstrap, ImportError, RecoveryError, DelegationError, WorkerError, BootstrapError, ImportContext, RecoveryContext, DelegationContext, WorkerContext, BootstrapContext, >(input: { readonly importLegacyShells: Effect.Effect; readonly recover: Effect.Effect; readonly recoverDelegatedTasks: Effect.Effect; readonly startEffectWorker: Effect.Effect; readonly autoBootstrap: Effect.Effect; }) { yield* input.importLegacyShells; const recovery = yield* input.recover; yield* input.recoverDelegatedTasks; yield* input.startEffectWorker; const bootstrap = yield* input.autoBootstrap; return { recovery, bootstrap } as const; }, ); const make = Effect.fn("serverRuntimeStartup.make")(function* (options?: StartupOptions) { const serverConfig = yield* ServerConfig.ServerConfig; const keybindings = yield* Keybindings.Keybindings; const legacyV1ThreadImporter = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; const providerRuntimeRecovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; const orchestrator = yield* Orchestrator.OrchestratorV2; const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const agentAwarenessRelay = yield* AgentAwarenessRelay.AgentAwarenessRelay; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; const serverSettings = yield* ServerSettings.ServerSettingsService; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const crypto = yield* Crypto.Crypto; const launcher = yield* ServiceLauncherClient.ServiceLauncherClient; const commandGate = yield* makeCommandGate; const httpListening = yield* Deferred.make(); const effectWorkerFiber = yield* Ref.make | null>(null); yield* Effect.addFinalizer(() => Effect.gen(function* () { yield* commandGate.failCommandReady( new ServerRuntimeStartupError({ mode: serverConfig.mode, host: serverConfig.host ?? null, port: serverConfig.port, cause: "Server runtime is shutting down.", }), ); const workerFiber = yield* Ref.getAndSet(effectWorkerFiber, null); if (workerFiber !== null) { yield* Fiber.interrupt(workerFiber).pipe(Effect.ignore); } yield* providerRuntimeRecovery.prepareForShutdown.pipe( Effect.ensuring(providerSessions.shutdown), ); const reconciliation = yield* providerRuntimeRecovery.reconcile("shutdown"); yield* Effect.logInfo("V2 orchestration shutdown reconciliation completed", reconciliation); }).pipe( Effect.catchCause((cause) => Effect.logWarning("V2 orchestration shutdown reconciliation failed", { cause: Cause.pretty(cause), }), ), ), ); const startup = Effect.gen(function* () { yield* Effect.logDebug("startup phase: starting keybindings runtime"); yield* runStartupPhase( "keybindings.start", keybindings.start.pipe( Effect.catch((error) => Effect.logWarning("failed to start keybindings runtime", { path: error.configPath, detail: error.detail, cause: error.cause, }), ), ), ); yield* Effect.logDebug("startup phase: starting server settings runtime"); yield* runStartupPhase( "settings.start", serverSettings.start.pipe( Effect.catch((error) => Effect.logWarning("failed to start server settings runtime", { path: error.settingsPath, operation: error.operation, providerInstanceId: error.providerInstanceId, environmentVariable: error.environmentVariable, cause: error.cause, }), ), ), ); const welcomeBase = yield* resolveWelcomeBase; const environment = yield* serverEnvironment.getDescriptor; const legacyMigrationThreadCount = yield* legacyV1ThreadImporter.pendingThreadCount; if (legacyMigrationThreadCount > 0) { yield* lifecycleEvents.publish({ version: 1, type: "legacyThreadMigration", payload: { status: "running", totalThreadCount: legacyMigrationThreadCount, }, }); } const { recovery, bootstrap: bootstrapTargets } = yield* runOrderedV2StartupPhases({ importLegacyShells: runStartupPhase( "orchestration-v2.legacy-v1.import-shells", legacyV1ThreadImporter.reconcileShells.pipe( Effect.tap((summary) => summary.importedThreadCount === 0 ? Effect.void : Effect.logInfo("Imported legacy v1 thread shells", summary), ), ), ), recover: runStartupPhase("orchestration-v2.recovery", providerRuntimeRecovery.recover), recoverDelegatedTasks: runStartupPhase( "orchestration-v2.delegated-tasks.recover", orchestrator.recoverDelegatedTasks, ), startEffectWorker: runStartupPhase( "orchestration-v2.effect-worker.start", startEffectWorkerWithRelay({ runWorker: EffectWorker.runDaemon, startRelay: agentAwarenessRelay.start(), workerFiberRef: effectWorkerFiber, }), ), autoBootstrap: (serverConfig.autoBootstrapProjectFromCwd ? runStartupPhase( "welcome.autobootstrap", resolveAutoBootstrapWelcomeTargets.pipe(Effect.provideService(Crypto.Crypto, crypto)), ) : Effect.succeed({}) ).pipe(Effect.map((targets): AutoBootstrapWelcomeTargets => targets)), }); yield* Effect.logInfo("V2 orchestration recovery completed", recovery); // Runs after activation: the status check fetches every enabled project's // remote, and awaiting it here held command readiness for that long. yield* runStartupPhase( "projects.auto-pull", Effect.gen(function* () { const projects = yield* (yield* ProjectStore.ProjectStoreV2).listShells(); const settings = yield* serverSettings.getSettings; yield* autoPullProjects(projects, settings); }), ).pipe( Effect.catch((cause) => Effect.logWarning("Failed to load projects for automatic pull", { cause }), ), forkParked, ); const importPendingTranscripts = legacyV1ThreadImporter.importPendingTranscripts.pipe( Effect.tap((summary) => summary.importedThreadCount === 0 ? Effect.void : Effect.logInfo("Hydrated legacy v1 thread transcripts", summary), ), ); yield* ( legacyMigrationThreadCount > 0 ? importPendingTranscripts.pipe( Effect.tap(() => lifecycleEvents.publish({ version: 1, type: "legacyThreadMigration", payload: { status: "complete", totalThreadCount: legacyMigrationThreadCount, }, }), ), ) : importPendingTranscripts ).pipe(forkParked); yield* forkParked( Effect.gen(function* () { yield* Effect.logDebug("startup phase: recording startup heartbeat"); yield* recordStartupHeartbeat.pipe( Effect.withSpan("server.startup.heartbeat.record", { attributes: { "startup.phase": "heartbeat.record" }, }), Effect.ignoreCause({ log: true }), ); if (serverConfig.startupPresentation === "headless") { yield* Effect.logDebug("startup phase: headless access info"); const accessInfo = yield* issueHeadlessServeAccessInfo(); yield* runStartupPhase( "headless.output", Console.log(formatHeadlessServeOutput(accessInfo)), ); } else { yield* Effect.logDebug("startup phase: browser open check"); const startupBrowserTarget = yield* resolveStartupBrowserTarget; if (serverConfig.mode !== "desktop") { yield* announceStartupPairingUrl(startupBrowserTarget); } yield* runStartupPhase("browser.open", maybeOpenBrowser(startupBrowserTarget)); } }), ); yield* Effect.logDebug("startup phase: waiting for http listener"); yield* runStartupPhase("http.wait", Deferred.await(httpListening)); yield* runStartupPhase("auxiliary-roots.parked", options?.awaitAuxiliaryParked ?? Effect.void); const updateOutcome = yield* launcher.prepareTrial; yield* Effect.logDebug("startup phase: publishing welcome event", { environmentId: environment.environmentId, cwd: welcomeBase.cwd, projectName: welcomeBase.projectName, bootstrapProjectId: bootstrapTargets.bootstrapProjectId, bootstrapThreadId: bootstrapTargets.bootstrapThreadId, }); yield* runStartupPhase( "welcome.publish", lifecycleEvents.publish({ version: 1, type: "welcome", payload: { environment, ...welcomeBase, ...bootstrapTargets, }, }), ); yield* options?.activate ?? Effect.void; yield* Effect.logDebug("Accepting commands"); yield* commandGate.signalCommandReady; yield* Effect.logDebug("startup phase: publishing ready event"); yield* runStartupPhase( "ready.publish", lifecycleEvents.publish({ version: 1, type: "ready", payload: { at: DateTime.formatIso(yield* DateTime.now), environment, ...(updateOutcome === undefined ? {} : { updateOutcome }), }, }), ); yield* Effect.logDebug("startup phase: complete"); yield* flushCompileCache; }).pipe( Effect.withSpan("server.startup", { kind: "server", root: true, attributes: { "server.mode": serverConfig.mode, "server.port": serverConfig.port, "server.host": serverConfig.host ?? "default", }, }), ); yield* Effect.forkScoped( Effect.exit(startup).pipe( Effect.flatMap((startupExit) => { if (Exit.isSuccess(startupExit)) return Effect.void; const error = new ServerRuntimeStartupError({ mode: serverConfig.mode, host: serverConfig.host ?? null, port: serverConfig.port, cause: startupExit.cause, }); return Effect.logError("server runtime startup failed", { cause: Cause.pretty(startupExit.cause), }).pipe( Effect.andThen(commandGate.failCommandReady(error)), Effect.andThen(options?.abort?.(error) ?? Effect.void), ); }), ), ); return { awaitCommandReady: commandGate.awaitCommandReady, markHttpListening: Deferred.succeed(httpListening, undefined), enqueueCommand: commandGate.enqueueCommand, } satisfies ServerRuntimeStartup["Service"]; }); export const layerWithOptions = (options?: StartupOptions) => Layer.effect(ServerRuntimeStartup, make(options));