import * as Cause from "effect/Cause"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL, AuthOrchestrationOperateScope, AuthOrchestrationReadScope, AuthReviewWriteScope, AuthRelayWriteScope, AuthTerminalOperateScope, AuthAccessReadScope, AuthAccessStreamError, type AuthAccessStreamEvent, type AuthEnvironmentScope, AuthSessionId, CommandId, type DiscoveredLocalServerList, EventId, type OrchestrationCommand, type GitActionProgressEvent, type GitManagerServiceError, OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, type ProjectEntriesFailure, type ProjectFileFailure, type ProjectFileOperation, ProjectListEntriesError, ProjectReadFileError, ProjectSearchEntriesError, ProjectWriteFileError, RelayClientInstallFailedError, type RelayClientInstallProgressEvent, OrchestrationReplayEventsError, type FilesystemBrowseFailure, FilesystemBrowseError, AssetWorkspaceContextNotFoundError, AssetWorkspaceContextResolutionError, EnvironmentAuthorizationError, ThreadId, type TerminalAttachStreamEvent, type TerminalError, type TerminalEvent, type TerminalMetadataStreamEvent, WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; import { clamp } from "effect/Number"; import { HttpRouter, HttpServerRequest, HttpServerRespondable } from "effect/unstable/http"; import { RpcSerialization, RpcServer } from "effect/unstable/rpc"; import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; import * as ServerConfig from "./config.ts"; import * as Keybindings from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { observeRpcEffect as instrumentRpcEffect, observeRpcStream as instrumentRpcStream, observeRpcStreamEffect as instrumentRpcStreamEffect, } from "./observability/RpcInstrumentation.ts"; import * as ProviderRegistry from "./provider/Services/ProviderRegistry.ts"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as PreviewManager from "./preview/Manager.ts"; import { issueAssetUrl } from "./assets/AssetAccess.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as GitWorkflowService from "./git/GitWorkflowService.ts"; import * as ReviewService from "./review/ReviewService.ts"; import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts"; import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; import * as SourceControlDiscovery from "./sourceControl/SourceControlDiscovery.ts"; import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts"; import * as AzureDevOpsCli from "./sourceControl/AzureDevOpsCli.ts"; import * as BitbucketApi from "./sourceControl/BitbucketApi.ts"; import * as GitHubCli from "./sourceControl/GitHubCli.ts"; import * as GitLabCli from "./sourceControl/GitLabCli.ts"; import * as SourceControlProviderRegistry from "./sourceControl/SourceControlProviderRegistry.ts"; import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import * as VcsDriverRegistry from "./vcs/VcsDriverRegistry.ts"; import * as VcsProjectConfig from "./vcs/VcsProjectConfig.ts"; import * as VcsProcess from "./vcs/VcsProcess.ts"; import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); function unexpectedCompatibilityError(error: never): never { throw new Error(`Unhandled compatibility error: ${String(error)}`); } /** Preserve the setup runner's broader pre-refactor message normalization. */ function legacySetupFailureDescription(cause: unknown): string { if ( typeof cause === "object" && cause !== null && "message" in cause && typeof cause.message === "string" ) { return cause.message; } return String(cause); } function projectEntriesFailureContext(error: WorkspaceEntries.WorkspaceEntriesError): { readonly failure: ProjectEntriesFailure; readonly normalizedCwd?: string; readonly timeout?: string; readonly detail?: string; } { switch (error._tag) { case "WorkspaceRootNotExistsError": return { failure: "workspace_root_not_found", normalizedCwd: error.normalizedWorkspaceRoot, }; case "WorkspaceRootCreateFailedError": return { failure: "workspace_root_create_failed", normalizedCwd: error.normalizedWorkspaceRoot, }; case "WorkspaceRootStatFailedError": return { failure: "workspace_root_stat_failed", normalizedCwd: error.normalizedWorkspaceRoot, detail: error.phase, }; case "WorkspaceRootNotDirectoryError": return { failure: "workspace_root_not_directory", normalizedCwd: error.normalizedWorkspaceRoot, }; case "WorkspaceSearchIndexCreateFailed": return { failure: "search_index_create_failed", normalizedCwd: error.cwd, detail: error.reason, }; case "WorkspaceSearchIndexScanTimedOut": return { failure: "search_index_scan_timed_out", normalizedCwd: error.cwd, timeout: error.timeout, }; case "WorkspaceSearchIndexSearchFailed": return { failure: "search_index_search_failed", normalizedCwd: error.cwd, detail: error.reason, }; default: return unexpectedCompatibilityError(error); } } function filesystemBrowseFailureContext(error: WorkspaceEntries.WorkspaceEntriesBrowseError): { readonly failure: FilesystemBrowseFailure; readonly parentPath?: string; readonly platform?: string; } { switch (error._tag) { case "WorkspaceEntriesWindowsPathUnsupportedError": return { failure: "windows_path_unsupported", platform: error.platform }; case "WorkspaceEntriesCurrentProjectRequiredError": return { failure: "current_project_required" }; case "WorkspaceEntriesReadDirectoryError": return { failure: "read_directory_failed", parentPath: error.parentPath }; default: return unexpectedCompatibilityError(error); } } function projectFileFailureContext( error: | WorkspaceFileSystem.WorkspaceFileSystemError | WorkspacePaths.WorkspacePathOutsideRootError, ): { readonly failure: ProjectFileFailure; readonly resolvedPath?: string; readonly resolvedWorkspaceRoot?: string; readonly operation?: ProjectFileOperation; readonly operationPath?: string; } { switch (error._tag) { case "WorkspacePathOutsideRootError": return { failure: "workspace_path_outside_root" }; case "WorkspaceFileSystemOperationError": return { failure: "operation_failed", resolvedPath: error.resolvedPath, operation: error.operation, operationPath: error.operationPath, }; case "WorkspaceFilePathEscapeError": return { failure: "resolved_path_outside_root", resolvedPath: error.resolvedPath, resolvedWorkspaceRoot: error.resolvedWorkspaceRoot, }; case "WorkspacePathNotFileError": return { failure: "path_not_file", resolvedPath: error.resolvedPath }; case "WorkspaceBinaryFileError": return { failure: "binary_file", resolvedPath: error.resolvedPath }; default: return unexpectedCompatibilityError(error); } } function projectSetupScriptCompatibilityDetail( error: ProjectSetupScriptRunner.ProjectSetupScriptRunnerError, ): string { switch (error._tag) { case "ProjectSetupScriptOperationError": return legacySetupFailureDescription(error.cause); case "ProjectSetupScriptProjectNotFoundError": return "Project was not found for setup script execution."; default: return unexpectedCompatibilityError(error); } } function isThreadDetailEvent(event: OrchestrationEvent): event is Extract< OrchestrationEvent, { type: | "thread.message-sent" | "thread.proposed-plan-upserted" | "thread.activity-appended" | "thread.turn-diff-completed" | "thread.reverted" | "thread.session-set"; } > { return ( event.type === "thread.message-sent" || event.type === "thread.proposed-plan-upserted" || event.type === "thread.activity-appended" || event.type === "thread.turn-diff-completed" || event.type === "thread.reverted" || event.type === "thread.session-set" ); } const PROVIDER_STATUS_DEBOUNCE_MS = 200; const RPC_REQUIRED_SCOPE = new Map([ [ORCHESTRATION_WS_METHODS.dispatchCommand, AuthOrchestrationOperateScope], [ORCHESTRATION_WS_METHODS.getTurnDiff, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getFullThreadDiff, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.replayEvents, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.subscribeShell, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, AuthOrchestrationReadScope], [ORCHESTRATION_WS_METHODS.subscribeThread, AuthOrchestrationReadScope], [WS_METHODS.serverGetConfig, AuthOrchestrationReadScope], [WS_METHODS.serverRefreshProviders, AuthOrchestrationOperateScope], [WS_METHODS.serverUpdateProvider, AuthOrchestrationOperateScope], [WS_METHODS.serverUpsertKeybinding, AuthOrchestrationOperateScope], [WS_METHODS.serverRemoveKeybinding, AuthOrchestrationOperateScope], [WS_METHODS.serverGetSettings, AuthOrchestrationReadScope], [WS_METHODS.serverUpdateSettings, AuthOrchestrationOperateScope], [WS_METHODS.serverDiscoverSourceControl, AuthOrchestrationReadScope], [WS_METHODS.serverGetTraceDiagnostics, AuthOrchestrationReadScope], [WS_METHODS.serverGetProcessDiagnostics, AuthOrchestrationReadScope], [WS_METHODS.serverGetProcessResourceHistory, AuthOrchestrationReadScope], [WS_METHODS.serverSignalProcess, AuthOrchestrationOperateScope], [WS_METHODS.cloudGetRelayClientStatus, AuthRelayWriteScope], [WS_METHODS.cloudInstallRelayClient, AuthRelayWriteScope], [WS_METHODS.sourceControlLookupRepository, AuthOrchestrationReadScope], [WS_METHODS.sourceControlCloneRepository, AuthOrchestrationOperateScope], [WS_METHODS.sourceControlPublishRepository, AuthOrchestrationOperateScope], [WS_METHODS.projectsListEntries, AuthOrchestrationReadScope], [WS_METHODS.projectsReadFile, AuthOrchestrationReadScope], [WS_METHODS.projectsSearchEntries, AuthOrchestrationReadScope], [WS_METHODS.projectsWriteFile, AuthOrchestrationOperateScope], [WS_METHODS.shellOpenInEditor, AuthOrchestrationOperateScope], [WS_METHODS.filesystemBrowse, AuthOrchestrationReadScope], [WS_METHODS.assetsCreateUrl, AuthOrchestrationReadScope], [WS_METHODS.subscribeVcsStatus, AuthOrchestrationReadScope], [WS_METHODS.vcsRefreshStatus, AuthOrchestrationReadScope], [WS_METHODS.vcsPull, AuthOrchestrationOperateScope], [WS_METHODS.gitRunStackedAction, AuthOrchestrationOperateScope], [WS_METHODS.gitResolvePullRequest, AuthOrchestrationOperateScope], [WS_METHODS.gitPreparePullRequestThread, AuthOrchestrationOperateScope], [WS_METHODS.vcsListRefs, AuthOrchestrationReadScope], [WS_METHODS.vcsCreateWorktree, AuthOrchestrationOperateScope], [WS_METHODS.vcsRemoveWorktree, AuthOrchestrationOperateScope], [WS_METHODS.vcsCreateRef, AuthOrchestrationOperateScope], [WS_METHODS.vcsSwitchRef, AuthOrchestrationOperateScope], [WS_METHODS.vcsInit, AuthOrchestrationOperateScope], [WS_METHODS.reviewGetDiffPreview, AuthReviewWriteScope], [WS_METHODS.terminalOpen, AuthTerminalOperateScope], [WS_METHODS.terminalAttach, AuthTerminalOperateScope], [WS_METHODS.terminalWrite, AuthTerminalOperateScope], [WS_METHODS.terminalResize, AuthTerminalOperateScope], [WS_METHODS.terminalClear, AuthTerminalOperateScope], [WS_METHODS.terminalRestart, AuthTerminalOperateScope], [WS_METHODS.terminalClose, AuthTerminalOperateScope], [WS_METHODS.subscribeTerminalEvents, AuthTerminalOperateScope], [WS_METHODS.subscribeTerminalMetadata, AuthTerminalOperateScope], [WS_METHODS.previewOpen, AuthOrchestrationOperateScope], [WS_METHODS.previewNavigate, AuthOrchestrationOperateScope], [WS_METHODS.previewRefresh, AuthOrchestrationOperateScope], [WS_METHODS.previewClose, AuthOrchestrationOperateScope], [WS_METHODS.previewList, AuthOrchestrationReadScope], [WS_METHODS.previewReportStatus, AuthOrchestrationOperateScope], [WS_METHODS.previewAutomationConnect, AuthOrchestrationOperateScope], [WS_METHODS.previewAutomationRespond, AuthOrchestrationOperateScope], [WS_METHODS.previewAutomationReportOwner, AuthOrchestrationOperateScope], [WS_METHODS.previewAutomationClearOwner, AuthOrchestrationOperateScope], [WS_METHODS.subscribePreviewEvents, AuthOrchestrationReadScope], [WS_METHODS.subscribeDiscoveredLocalServers, AuthOrchestrationReadScope], [WS_METHODS.subscribeServerConfig, AuthOrchestrationReadScope], [WS_METHODS.subscribeServerLifecycle, AuthOrchestrationReadScope], [WS_METHODS.subscribeAuthAccess, AuthAccessReadScope], ]); function toAuthAccessStreamEvent( change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, revision: number, currentSessionId: AuthSessionId, ): AuthAccessStreamEvent { switch (change.type) { case "pairingLinkUpserted": return { version: 1, revision, type: "pairingLinkUpserted", payload: change.pairingLink, }; case "pairingLinkRemoved": return { version: 1, revision, type: "pairingLinkRemoved", payload: { id: change.id }, }; case "clientUpserted": return { version: 1, revision, type: "clientUpserted", payload: { ...change.clientSession, current: change.clientSession.sessionId === currentSessionId, }, }; case "clientRemoved": return { version: 1, revision, type: "clientRemoved", payload: { sessionId: change.sessionId }, }; } } const makeWsRpcLayer = (currentSession: EnvironmentAuth.AuthenticatedSession) => WsRpcGroup.toLayer( Effect.gen(function* () { const currentSessionId = currentSession.sessionId; const crypto = yield* Crypto.Crypto; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngine.OrchestrationEngineService; const checkpointDiffQuery = yield* CheckpointDiffQuery.CheckpointDiffQuery; const keybindings = yield* Keybindings.Keybindings; const externalLauncher = yield* ExternalLauncher.ExternalLauncher; const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; const review = yield* ReviewService.ReviewService; const vcsProvisioning = yield* VcsProvisioningService.VcsProvisioningService; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; const terminalManager = yield* TerminalManager.TerminalManager; const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker; const previewManager = yield* PreviewManager.PreviewManager; const portDiscovery = yield* PortScanner.PortDiscovery; const providerRegistry = yield* ProviderRegistry.ProviderRegistry; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const config = yield* ServerConfig.ServerConfig; const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents; const serverSettings = yield* ServerSettings.ServerSettingsService; const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup; const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem; const projectSetupScriptRunner = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner; const repositoryIdentityResolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; const serverEnvironment = yield* ServerEnvironment.ServerEnvironment; const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const sourceControlDiscovery = yield* SourceControlDiscovery.SourceControlDiscovery; const automaticGitFetchInterval = serverSettings.getSettings.pipe( Effect.map((settings) => settings.automaticGitFetchInterval), Effect.catch((cause) => Effect.logWarning("Failed to read automatic Git fetch interval setting", { detail: cause.message, }).pipe(Effect.as(DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL)), ), ); const sourceControlRepositories = yield* SourceControlRepositoryService.SourceControlRepositoryService; const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; const sessions = yield* SessionStore.SessionStore; const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics; const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor; const relayClient = yield* RelayClient.RelayClient; const authorizationError = (requiredScope: AuthEnvironmentScope) => new EnvironmentAuthorizationError({ message: `The authenticated token is missing required scope: ${requiredScope}.`, requiredScope, }); const authorizeEffect = ( requiredScope: AuthEnvironmentScope, effect: Effect.Effect, ): Effect.Effect => currentSession.scopes.includes(requiredScope) ? effect : Effect.fail(authorizationError(requiredScope)); const authorizeStream = ( requiredScope: AuthEnvironmentScope, stream: Stream.Stream, ): Stream.Stream => currentSession.scopes.includes(requiredScope) ? stream : Stream.fail(authorizationError(requiredScope)); const requiredScopeForMethod = (method: string): AuthEnvironmentScope => { const requiredScope = RPC_REQUIRED_SCOPE.get(method); if (requiredScope === undefined) { throw new Error(`RPC method ${method} has no declared authorization scope.`); } return requiredScope; }; const observeRpcEffect = ( method: string, effect: Effect.Effect, traceAttributes?: Readonly>, ) => instrumentRpcEffect( method, authorizeEffect(requiredScopeForMethod(method), effect), traceAttributes, ); const observeRpcStream = ( method: string, stream: Stream.Stream, traceAttributes?: Readonly>, ) => instrumentRpcStream( method, authorizeStream(requiredScopeForMethod(method), stream), traceAttributes, ); const observeRpcStreamEffect = ( method: string, effect: Effect.Effect< Stream.Stream, EffectError, EffectContext >, traceAttributes?: Readonly>, ) => instrumentRpcStreamEffect( method, authorizeEffect(requiredScopeForMethod(method), effect), traceAttributes, ); const toDispatchCommandError = (cause: unknown, fallbackMessage: string) => isOrchestrationDispatchCommandError(cause) ? cause : new OrchestrationDispatchCommandError({ message: cause instanceof Error ? cause.message : fallbackMessage, cause, }); const randomUUID = crypto.randomUUIDv4.pipe( Effect.mapError((cause) => toDispatchCommandError(cause, "Failed to generate orchestration command identifier."), ), ); const serverEventId = randomUUID.pipe(Effect.map(EventId.make)); const serverCommandId = (tag: string) => randomUUID.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`))); const loadAuthAccessSnapshot = () => Effect.all({ pairingLinks: serverAuth.listPairingLinks(), clientSessions: serverAuth.listClientSessions(currentSessionId), }).pipe( Effect.mapError( (error) => new AuthAccessStreamError({ message: error.message, }), ), ); const appendSetupScriptActivity = (input: { readonly threadId: ThreadId; readonly kind: "setup-script.requested" | "setup-script.started" | "setup-script.failed"; readonly summary: string; readonly createdAt: string; readonly payload: Record; readonly tone: "info" | "error"; }) => Effect.all({ commandId: serverCommandId("setup-script-activity"), activityId: serverEventId, }).pipe( Effect.flatMap(({ commandId, activityId }) => orchestrationEngine.dispatch({ type: "thread.activity.append", commandId, threadId: input.threadId, activity: { id: activityId, tone: input.tone, kind: input.kind, summary: input.summary, payload: input.payload, turnId: null, createdAt: input.createdAt, }, createdAt: input.createdAt, }), ), ); const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause) => { const error = Cause.squash(cause); return isOrchestrationDispatchCommandError(error) ? error : new OrchestrationDispatchCommandError({ message: error instanceof Error ? error.message : "Failed to bootstrap thread turn start.", cause, }); }; const enrichProjectEvent = ( event: OrchestrationEvent, ): Effect.Effect => { switch (event.type) { case "project.created": return repositoryIdentityResolver.resolve(event.payload.workspaceRoot).pipe( Effect.map((repositoryIdentity) => ({ ...event, payload: { ...event.payload, repositoryIdentity, }, })), ); case "project.meta-updated": return Effect.gen(function* () { const workspaceRoot = event.payload.workspaceRoot ?? Option.match( yield* projectionSnapshotQuery.getProjectShellById(event.payload.projectId), { onNone: () => null, onSome: (project) => project.workspaceRoot, }, ) ?? null; if (workspaceRoot === null) { return event; } const repositoryIdentity = yield* repositoryIdentityResolver.resolve(workspaceRoot); return { ...event, payload: { ...event.payload, repositoryIdentity, }, } satisfies OrchestrationEvent; }).pipe(Effect.orElseSucceed(() => event)); default: return Effect.succeed(event); } }; const enrichOrchestrationEvents = (events: ReadonlyArray) => Effect.forEach(events, enrichProjectEvent, { concurrency: 4 }); const toShellStreamEvent = ( event: OrchestrationEvent, ): Effect.Effect, never, never> => { switch (event.type) { case "project.created": case "project.meta-updated": return projectionSnapshotQuery.getProjectShellById(event.payload.projectId).pipe( Effect.map((project) => Option.map(project, (nextProject) => ({ kind: "project-upserted" as const, sequence: event.sequence, project: nextProject, })), ), Effect.orElseSucceed(() => Option.none()), ); case "project.deleted": return Effect.succeed( Option.some({ kind: "project-removed" as const, sequence: event.sequence, projectId: event.payload.projectId, }), ); case "thread.deleted": case "thread.archived": return Effect.succeed( Option.some({ kind: "thread-removed" as const, sequence: event.sequence, threadId: event.payload.threadId, }), ); case "thread.unarchived": return projectionSnapshotQuery.getThreadShellById(event.payload.threadId).pipe( Effect.map((thread) => Option.map(thread, (nextThread) => ({ kind: "thread-upserted" as const, sequence: event.sequence, thread: nextThread, })), ), Effect.orElseSucceed(() => Option.none()), ); default: if (event.aggregateKind !== "thread") { return Effect.succeed(Option.none()); } return projectionSnapshotQuery .getThreadShellById(ThreadId.make(event.aggregateId)) .pipe( Effect.map((thread) => Option.map(thread, (nextThread) => ({ kind: "thread-upserted" as const, sequence: event.sequence, thread: nextThread, })), ), Effect.orElseSucceed(() => Option.none()), ); } }; const dispatchBootstrapTurnStart = ( command: Extract, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => Effect.gen(function* () { const bootstrap = command.bootstrap; const { bootstrap: _bootstrap, ...finalTurnStartCommand } = command; let createdThread = false; let targetProjectId = bootstrap?.createThread?.projectId; let targetProjectCwd = bootstrap?.prepareWorktree?.projectCwd; let targetWorktreePath = bootstrap?.createThread?.worktreePath ?? null; const cleanupCreatedThread = () => createdThread ? serverCommandId("bootstrap-thread-delete").pipe( Effect.flatMap((commandId) => orchestrationEngine.dispatch({ type: "thread.delete", commandId, threadId: command.threadId, }), ), Effect.ignoreCause({ log: true }), ) : Effect.void; const recordSetupScriptLaunchFailure = (input: { readonly error: ProjectSetupScriptRunner.ProjectSetupScriptRunnerError; readonly requestedAt: string; readonly worktreePath: string; }) => { const detail = projectSetupScriptCompatibilityDetail(input.error); return appendSetupScriptActivity({ threadId: command.threadId, kind: "setup-script.failed", summary: "Setup script failed to start", createdAt: input.requestedAt, payload: { detail, worktreePath: input.worktreePath, }, tone: "error", }).pipe( Effect.ignoreCause({ log: false }), Effect.flatMap(() => Effect.logWarning("bootstrap turn start failed to launch setup script", { threadId: command.threadId, worktreePath: input.worktreePath, detail, }), ), ); }; const recordSetupScriptStarted = (input: { readonly requestedAt: string; readonly worktreePath: string; readonly scriptId: string; readonly scriptName: string; readonly terminalId: string; }) => Effect.gen(function* () { const startedAt = yield* nowIso; const payload = { scriptId: input.scriptId, scriptName: input.scriptName, terminalId: input.terminalId, worktreePath: input.worktreePath, }; yield* Effect.all([ appendSetupScriptActivity({ threadId: command.threadId, kind: "setup-script.requested", summary: "Starting setup script", createdAt: input.requestedAt, payload, tone: "info", }), appendSetupScriptActivity({ threadId: command.threadId, kind: "setup-script.started", summary: "Setup script started", createdAt: startedAt, payload, tone: "info", }), ]).pipe( Effect.asVoid, Effect.catch((error) => Effect.logWarning( "bootstrap turn start launched setup script but failed to record setup activity", { threadId: command.threadId, worktreePath: input.worktreePath, scriptId: input.scriptId, terminalId: input.terminalId, detail: error.message, }, ), ), ); }); const runSetupProgram = () => Effect.gen(function* () { if (!bootstrap?.runSetupScript || !targetWorktreePath) { return; } const worktreePath = targetWorktreePath; const requestedAt = yield* nowIso; yield* projectSetupScriptRunner .runForThread({ threadId: command.threadId, ...(targetProjectId ? { projectId: targetProjectId } : {}), ...(targetProjectCwd ? { projectCwd: targetProjectCwd } : {}), worktreePath, }) .pipe( Effect.matchEffect({ onFailure: (error) => recordSetupScriptLaunchFailure({ error, requestedAt, worktreePath, }), onSuccess: (setupResult) => { if (setupResult.status !== "started") { return Effect.void; } return recordSetupScriptStarted({ requestedAt, worktreePath, scriptId: setupResult.scriptId, scriptName: setupResult.scriptName, terminalId: setupResult.terminalId, }); }, }), ); }); const bootstrapProgram = Effect.gen(function* () { if (bootstrap?.createThread) { yield* orchestrationEngine.dispatch({ type: "thread.create", commandId: yield* serverCommandId("bootstrap-thread-create"), threadId: command.threadId, projectId: bootstrap.createThread.projectId, title: bootstrap.createThread.title, modelSelection: bootstrap.createThread.modelSelection, runtimeMode: bootstrap.createThread.runtimeMode, interactionMode: bootstrap.createThread.interactionMode, branch: bootstrap.createThread.branch, worktreePath: bootstrap.createThread.worktreePath, createdAt: bootstrap.createThread.createdAt, }); createdThread = true; } if (bootstrap?.prepareWorktree) { let worktreeBaseRef = bootstrap.prepareWorktree.baseBranch; if (bootstrap.prepareWorktree.startFromOrigin) { yield* gitWorkflow.fetchRemote({ cwd: bootstrap.prepareWorktree.projectCwd, remoteName: "origin", }); const resolvedRemoteBase = yield* gitWorkflow.resolveRemoteTrackingCommit({ cwd: bootstrap.prepareWorktree.projectCwd, refName: bootstrap.prepareWorktree.baseBranch, fallbackRemoteName: "origin", }); worktreeBaseRef = resolvedRemoteBase.commitSha; } const worktree = yield* gitWorkflow.createWorktree({ cwd: bootstrap.prepareWorktree.projectCwd, refName: worktreeBaseRef, newRefName: bootstrap.prepareWorktree.branch, baseRefName: bootstrap.prepareWorktree.baseBranch, path: null, }); targetWorktreePath = worktree.worktree.path; yield* orchestrationEngine.dispatch({ type: "thread.meta.update", commandId: yield* serverCommandId("bootstrap-thread-meta-update"), threadId: command.threadId, branch: worktree.worktree.refName, worktreePath: targetWorktreePath, }); yield* refreshGitStatus(targetWorktreePath); } yield* runSetupProgram(); return yield* orchestrationEngine.dispatch(finalTurnStartCommand); }); return yield* bootstrapProgram.pipe( Effect.catchCause((cause) => { const dispatchError = toBootstrapDispatchCommandCauseError(cause); if (Cause.hasInterruptsOnly(cause)) { return Effect.fail(dispatchError); } return cleanupCreatedThread().pipe(Effect.flatMap(() => Effect.fail(dispatchError))); }), ); }); const dispatchNormalizedCommand = ( normalizedCommand: OrchestrationCommand, ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => { const dispatchEffect = normalizedCommand.type === "thread.turn.start" && normalizedCommand.bootstrap ? dispatchBootstrapTurnStart(normalizedCommand) : orchestrationEngine .dispatch(normalizedCommand) .pipe( Effect.mapError((cause) => toDispatchCommandError(cause, "Failed to dispatch orchestration command"), ), ); return startup .enqueueCommand(dispatchEffect) .pipe( Effect.mapError((cause) => toDispatchCommandError(cause, "Failed to dispatch orchestration command"), ), ); }; const loadServerConfig = Effect.gen(function* () { const keybindingsConfig = yield* keybindings.loadConfigState; const providers = yield* providerRegistry.getProviders; const settings = ServerSettings.redactServerSettingsForClient( yield* serverSettings.getSettings, ); const environment = yield* serverEnvironment.getDescriptor; const auth = yield* serverAuth.getDescriptor(); return { environment, auth, cwd: config.cwd, keybindingsConfigPath: config.keybindingsConfigPath, keybindings: keybindingsConfig.keybindings, issues: keybindingsConfig.issues, providers, availableEditors: yield* externalLauncher.resolveAvailableEditors(), observability: { logsDirectoryPath: config.logsDir, localTracingEnabled: true, ...(config.otlpTracesUrl !== undefined ? { otlpTracesUrl: config.otlpTracesUrl } : {}), otlpTracesEnabled: config.otlpTracesUrl !== undefined, ...(config.otlpMetricsUrl !== undefined ? { otlpMetricsUrl: config.otlpMetricsUrl } : {}), otlpMetricsEnabled: config.otlpMetricsUrl !== undefined, }, settings, }; }); const refreshGitStatus = (cwd: string) => vcsStatusBroadcaster .refreshStatus(cwd) .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid); return WsRpcGroup.of({ [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) => observeRpcEffect( ORCHESTRATION_WS_METHODS.dispatchCommand, Effect.gen(function* () { const normalizedCommand = yield* normalizeDispatchCommand(command); const shouldStopSessionAfterArchive = normalizedCommand.type === "thread.archive" ? yield* projectionSnapshotQuery .getThreadShellById(normalizedCommand.threadId) .pipe( Effect.map( Option.match({ onNone: () => false, onSome: (thread) => thread.session !== null && thread.session.status !== "stopped", }), ), Effect.orElseSucceed(() => false), ) : false; const result = yield* dispatchNormalizedCommand(normalizedCommand); if (normalizedCommand.type === "thread.archive") { if (shouldStopSessionAfterArchive) { yield* Effect.gen(function* () { const stopCommand = yield* normalizeDispatchCommand({ type: "thread.session.stop", commandId: CommandId.make( `session-stop-for-archive:${normalizedCommand.commandId}`, ), threadId: normalizedCommand.threadId, createdAt: yield* nowIso, }); yield* dispatchNormalizedCommand(stopCommand); }).pipe( Effect.catchCause((cause) => Effect.logWarning("failed to stop provider session during archive", { threadId: normalizedCommand.threadId, cause, }), ), ); } yield* terminalManager.close({ threadId: normalizedCommand.threadId }).pipe( Effect.catch((error) => Effect.logWarning("failed to close thread terminals after archive", { threadId: normalizedCommand.threadId, error: error.message, }), ), ); } return result; }).pipe( Effect.mapError((cause) => isOrchestrationDispatchCommandError(cause) ? cause : new OrchestrationDispatchCommandError({ message: "Failed to dispatch orchestration command", cause, }), ), ), { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_WS_METHODS.getTurnDiff]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.getTurnDiff, checkpointDiffQuery.getTurnDiff(input).pipe( Effect.mapError( (cause) => new OrchestrationGetTurnDiffError({ message: "Failed to load turn diff", cause, }), ), ), { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_WS_METHODS.getFullThreadDiff]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.getFullThreadDiff, checkpointDiffQuery.getFullThreadDiff(input).pipe( Effect.mapError( (cause) => new OrchestrationGetFullThreadDiffError({ message: "Failed to load full thread diff", cause, }), ), ), { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_WS_METHODS.replayEvents]: (input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.replayEvents, Stream.runCollect( orchestrationEngine.readEvents( clamp(input.fromSequenceExclusive, { maximum: Number.MAX_SAFE_INTEGER, minimum: 0, }), ), ).pipe( Effect.map((events) => Array.from(events)), Effect.flatMap(enrichOrchestrationEvents), Effect.mapError( (cause) => new OrchestrationReplayEventsError({ message: "Failed to replay orchestration events", cause, }), ), ), { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_WS_METHODS.subscribeShell]: (_input) => observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, Effect.gen(function* () { const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( Effect.tapError((cause) => Effect.logError("orchestration shell snapshot load failed", { cause }), ), Effect.mapError( (cause) => new OrchestrationGetSnapshotError({ message: "Failed to load orchestration shell snapshot", cause, }), ), ); const liveStream = orchestrationEngine.streamDomainEvents.pipe( Stream.mapEffect(toShellStreamEvent), Stream.flatMap((event) => Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, ), ); return Stream.concat( Stream.make({ kind: "snapshot" as const, snapshot, }), liveStream, ); }), { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot]: (_input) => observeRpcEffect( ORCHESTRATION_WS_METHODS.getArchivedShellSnapshot, projectionSnapshotQuery.getArchivedShellSnapshot().pipe( Effect.tapError((cause) => Effect.logError("orchestration archived shell snapshot load failed", { cause }), ), Effect.mapError( (cause) => new OrchestrationGetSnapshotError({ message: "Failed to load archived orchestration shell snapshot", cause, }), ), ), { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_WS_METHODS.subscribeThread]: (input) => observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeThread, Effect.gen(function* () { const [threadDetail, snapshotSequence] = yield* Effect.all([ projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe( Effect.mapError( (cause) => new OrchestrationGetSnapshotError({ message: `Failed to load thread ${input.threadId}`, cause, }), ), ), projectionSnapshotQuery.getSnapshotSequence().pipe( Effect.map(({ snapshotSequence }) => snapshotSequence), Effect.mapError( (cause) => new OrchestrationGetSnapshotError({ message: "Failed to load orchestration snapshot sequence", cause, }), ), ), ]); if (Option.isNone(threadDetail)) { return yield* new OrchestrationGetSnapshotError({ message: `Thread ${input.threadId} was not found`, cause: input.threadId, }); } const liveStream = orchestrationEngine.streamDomainEvents.pipe( Stream.filter( (event) => event.aggregateKind === "thread" && event.aggregateId === input.threadId && isThreadDetailEvent(event), ), Stream.map((event) => ({ kind: "event" as const, event, })), ); return Stream.concat( Stream.make({ kind: "snapshot" as const, snapshot: { snapshotSequence, thread: threadDetail.value, }, }), liveStream, ); }), { "rpc.aggregate": "orchestration" }, ), [WS_METHODS.serverGetConfig]: (_input) => observeRpcEffect(WS_METHODS.serverGetConfig, loadServerConfig, { "rpc.aggregate": "server", }), [WS_METHODS.serverRefreshProviders]: (input) => observeRpcEffect( WS_METHODS.serverRefreshProviders, (input.instanceId !== undefined ? providerRegistry.refreshInstance(input.instanceId) : providerRegistry.refresh() ).pipe(Effect.map((providers) => ({ providers }))), { "rpc.aggregate": "server" }, ), [WS_METHODS.serverUpdateProvider]: (input) => observeRpcEffect( WS_METHODS.serverUpdateProvider, providerMaintenanceRunner.updateProvider(input), { "rpc.aggregate": "server", }, ), [WS_METHODS.serverUpsertKeybinding]: (rule) => observeRpcEffect( WS_METHODS.serverUpsertKeybinding, Effect.gen(function* () { const keybindingsConfig = yield* keybindings.upsertKeybindingRule(rule); return { keybindings: keybindingsConfig, issues: [] }; }), { "rpc.aggregate": "server" }, ), [WS_METHODS.serverRemoveKeybinding]: (rule) => observeRpcEffect( WS_METHODS.serverRemoveKeybinding, Effect.gen(function* () { const keybindingsConfig = yield* keybindings.removeKeybindingRule(rule); return { keybindings: keybindingsConfig, issues: [] }; }), { "rpc.aggregate": "server" }, ), [WS_METHODS.serverGetSettings]: (_input) => observeRpcEffect( WS_METHODS.serverGetSettings, serverSettings.getSettings.pipe( Effect.map(ServerSettings.redactServerSettingsForClient), ), { "rpc.aggregate": "server", }, ), [WS_METHODS.serverUpdateSettings]: ({ patch }) => observeRpcEffect( WS_METHODS.serverUpdateSettings, serverSettings .updateSettings(patch) .pipe(Effect.map(ServerSettings.redactServerSettingsForClient)), { "rpc.aggregate": "server", }, ), [WS_METHODS.serverDiscoverSourceControl]: (_input) => observeRpcEffect( WS_METHODS.serverDiscoverSourceControl, sourceControlDiscovery.discover, { "rpc.aggregate": "server", }, ), [WS_METHODS.serverGetTraceDiagnostics]: (_input) => observeRpcEffect( WS_METHODS.serverGetTraceDiagnostics, TraceDiagnostics.readTraceDiagnostics({ traceFilePath: config.serverTracePath, maxFiles: config.traceMaxFiles, }), { "rpc.aggregate": "server", }, ), [WS_METHODS.serverGetProcessDiagnostics]: (_input) => observeRpcEffect(WS_METHODS.serverGetProcessDiagnostics, processDiagnostics.read, { "rpc.aggregate": "server", }), [WS_METHODS.serverGetProcessResourceHistory]: (input) => observeRpcEffect( WS_METHODS.serverGetProcessResourceHistory, processResourceMonitor.readHistory(input), { "rpc.aggregate": "server", }, ), [WS_METHODS.serverSignalProcess]: (input) => observeRpcEffect(WS_METHODS.serverSignalProcess, processDiagnostics.signal(input), { "rpc.aggregate": "server", }), [WS_METHODS.cloudGetRelayClientStatus]: (_input) => observeRpcEffect(WS_METHODS.cloudGetRelayClientStatus, relayClient.resolve, { "rpc.aggregate": "cloud", }), [WS_METHODS.cloudInstallRelayClient]: (_input) => observeRpcStream( WS_METHODS.cloudInstallRelayClient, Stream.callback( (queue) => relayClient .installWithProgress((event) => Queue.offer(queue, event).pipe(Effect.asVoid)) .pipe( Effect.flatMap((status) => Queue.offer(queue, { type: "complete", status, }), ), Effect.catchTag("RelayClientInstallError", (error) => Queue.fail( queue, new RelayClientInstallFailedError({ reason: error.reason, message: error.message, }), ), ), Effect.andThen(Queue.end(queue)), Effect.forkScoped, ), ), { "rpc.aggregate": "cloud" }, ), [WS_METHODS.sourceControlLookupRepository]: (input) => observeRpcEffect( WS_METHODS.sourceControlLookupRepository, sourceControlRepositories.lookupRepository(input), { "rpc.aggregate": "source-control", }, ), [WS_METHODS.sourceControlCloneRepository]: (input) => observeRpcEffect( WS_METHODS.sourceControlCloneRepository, sourceControlRepositories.cloneRepository(input), { "rpc.aggregate": "source-control", }, ), [WS_METHODS.sourceControlPublishRepository]: (input) => observeRpcEffect( WS_METHODS.sourceControlPublishRepository, sourceControlRepositories .publishRepository(input) .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "source-control", }, ), [WS_METHODS.projectsSearchEntries]: (input) => observeRpcEffect( WS_METHODS.projectsSearchEntries, workspaceEntries.search(input).pipe( Effect.mapError( (cause) => new ProjectSearchEntriesError({ cwd: input.cwd, queryLength: input.query.length, limit: input.limit, ...projectEntriesFailureContext(cause), cause, }), ), ), { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsListEntries]: (input) => observeRpcEffect( WS_METHODS.projectsListEntries, workspaceEntries.list(input).pipe( Effect.mapError( (cause) => new ProjectListEntriesError({ ...input, ...projectEntriesFailureContext(cause), cause, }), ), ), { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsReadFile]: (input) => observeRpcEffect( WS_METHODS.projectsReadFile, workspaceFileSystem.readFile(input).pipe( Effect.mapError( (cause) => new ProjectReadFileError({ ...input, ...projectFileFailureContext(cause), cause, }), ), ), { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsWriteFile]: (input) => observeRpcEffect( WS_METHODS.projectsWriteFile, workspaceFileSystem.writeFile(input).pipe( Effect.mapError( (cause) => new ProjectWriteFileError({ cwd: input.cwd, relativePath: input.relativePath, ...projectFileFailureContext(cause), cause, }), ), ), { "rpc.aggregate": "workspace" }, ), [WS_METHODS.shellOpenInEditor]: (input) => observeRpcEffect(WS_METHODS.shellOpenInEditor, externalLauncher.launchEditor(input), { "rpc.aggregate": "workspace", }), [WS_METHODS.filesystemBrowse]: (input) => observeRpcEffect( WS_METHODS.filesystemBrowse, workspaceEntries.browse(input).pipe( Effect.mapError( (cause) => new FilesystemBrowseError({ ...input, ...filesystemBrowseFailureContext(cause), cause, }), ), ), { "rpc.aggregate": "workspace" }, ), [WS_METHODS.assetsCreateUrl]: (input) => observeRpcEffect( WS_METHODS.assetsCreateUrl, Effect.gen(function* () { if (input.resource._tag !== "workspace-file") { return yield* issueAssetUrl({ resource: input.resource }); } const thread = yield* projectionSnapshotQuery .getThreadShellById(input.resource.threadId) .pipe( Effect.mapError( (cause) => new AssetWorkspaceContextResolutionError({ resource: input.resource, cause, }), ), ); if (Option.isNone(thread)) { return yield* new AssetWorkspaceContextNotFoundError({ resource: input.resource, }); } const project = yield* projectionSnapshotQuery .getProjectShellById(thread.value.projectId) .pipe( Effect.mapError( (cause) => new AssetWorkspaceContextResolutionError({ resource: input.resource, cause, }), ), ); if (Option.isNone(project)) { return yield* new AssetWorkspaceContextNotFoundError({ resource: input.resource, }); } return yield* issueAssetUrl({ resource: input.resource, workspaceRoot: thread.value.worktreePath ?? project.value.workspaceRoot, }); }), { "rpc.aggregate": "workspace" }, ), [WS_METHODS.subscribeVcsStatus]: (input) => observeRpcStream( WS_METHODS.subscribeVcsStatus, vcsStatusBroadcaster.streamStatus(input, { automaticRemoteRefreshInterval: automaticGitFetchInterval, }), { "rpc.aggregate": "vcs", }, ), [WS_METHODS.vcsRefreshStatus]: (input) => observeRpcEffect( WS_METHODS.vcsRefreshStatus, vcsStatusBroadcaster.refreshStatus(input.cwd), { "rpc.aggregate": "vcs", }, ), [WS_METHODS.vcsPull]: (input) => observeRpcEffect( WS_METHODS.vcsPull, gitWorkflow.pullCurrentBranch(input.cwd).pipe( Effect.matchCauseEffect({ onFailure: (cause) => Effect.failCause(cause), onSuccess: (result) => refreshGitStatus(input.cwd).pipe(Effect.ignore({ log: true }), Effect.as(result)), }), ), { "rpc.aggregate": "git" }, ), [WS_METHODS.gitRunStackedAction]: (input) => observeRpcStream( WS_METHODS.gitRunStackedAction, Stream.callback((queue) => gitWorkflow .runStackedAction(input, { actionId: input.actionId, progressReporter: { publish: (event) => Queue.offer(queue, event).pipe(Effect.asVoid), }, }) .pipe( Effect.matchCauseEffect({ onFailure: (cause) => Queue.failCause(queue, cause), onSuccess: () => refreshGitStatus(input.cwd).pipe( Effect.andThen(Queue.end(queue).pipe(Effect.asVoid)), ), }), ), ), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.gitResolvePullRequest]: (input) => observeRpcEffect( WS_METHODS.gitResolvePullRequest, gitWorkflow.resolvePullRequest(input), { "rpc.aggregate": "git", }, ), [WS_METHODS.gitPreparePullRequestThread]: (input) => observeRpcEffect( WS_METHODS.gitPreparePullRequestThread, gitWorkflow .preparePullRequestThread(input) .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "git" }, ), [WS_METHODS.vcsListRefs]: (input) => observeRpcEffect(WS_METHODS.vcsListRefs, gitWorkflow.listRefs(input), { "rpc.aggregate": "vcs", }), [WS_METHODS.vcsCreateWorktree]: (input) => observeRpcEffect( WS_METHODS.vcsCreateWorktree, gitWorkflow.createWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.vcsRemoveWorktree]: (input) => observeRpcEffect( WS_METHODS.vcsRemoveWorktree, gitWorkflow.removeWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.vcsCreateRef]: (input) => observeRpcEffect( WS_METHODS.vcsCreateRef, gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.vcsSwitchRef]: (input) => observeRpcEffect( WS_METHODS.vcsSwitchRef, gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.vcsInit]: (input) => observeRpcEffect( WS_METHODS.vcsInit, vcsProvisioning .initRepository(input) .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), { "rpc.aggregate": "vcs" }, ), [WS_METHODS.reviewGetDiffPreview]: (input) => observeRpcEffect(WS_METHODS.reviewGetDiffPreview, review.getDiffPreview(input), { "rpc.aggregate": "review", }), [WS_METHODS.terminalOpen]: (input) => observeRpcEffect(WS_METHODS.terminalOpen, terminalManager.open(input), { "rpc.aggregate": "terminal", }), [WS_METHODS.terminalAttach]: (input) => observeRpcStream( WS_METHODS.terminalAttach, Stream.callback((queue) => Effect.acquireRelease( terminalManager.attachStream(input, (event) => Queue.offer(queue, event)), (unsubscribe) => Effect.sync(unsubscribe), ), ), { "rpc.aggregate": "terminal" }, ), [WS_METHODS.terminalWrite]: (input) => observeRpcEffect(WS_METHODS.terminalWrite, terminalManager.write(input), { "rpc.aggregate": "terminal", }), [WS_METHODS.terminalResize]: (input) => observeRpcEffect(WS_METHODS.terminalResize, terminalManager.resize(input), { "rpc.aggregate": "terminal", }), [WS_METHODS.terminalClear]: (input) => observeRpcEffect(WS_METHODS.terminalClear, terminalManager.clear(input), { "rpc.aggregate": "terminal", }), [WS_METHODS.terminalRestart]: (input) => observeRpcEffect(WS_METHODS.terminalRestart, terminalManager.restart(input), { "rpc.aggregate": "terminal", }), [WS_METHODS.terminalClose]: (input) => observeRpcEffect(WS_METHODS.terminalClose, terminalManager.close(input), { "rpc.aggregate": "terminal", }), [WS_METHODS.subscribeTerminalEvents]: (_input) => observeRpcStream( WS_METHODS.subscribeTerminalEvents, Stream.callback((queue) => Effect.acquireRelease( terminalManager.subscribe((event) => Queue.offer(queue, event)), (unsubscribe) => Effect.sync(unsubscribe), ), ), { "rpc.aggregate": "terminal" }, ), [WS_METHODS.subscribeTerminalMetadata]: (_input) => observeRpcStream( WS_METHODS.subscribeTerminalMetadata, Stream.callback((queue) => Effect.acquireRelease( terminalManager.subscribeMetadata((event) => Queue.offer(queue, event)), (unsubscribe) => Effect.sync(unsubscribe), ), ), { "rpc.aggregate": "terminal" }, ), [WS_METHODS.previewOpen]: (input) => observeRpcEffect(WS_METHODS.previewOpen, previewManager.open(input), { "rpc.aggregate": "preview", }), [WS_METHODS.previewNavigate]: (input) => observeRpcEffect(WS_METHODS.previewNavigate, previewManager.navigate(input), { "rpc.aggregate": "preview", }), [WS_METHODS.previewRefresh]: (input) => observeRpcEffect(WS_METHODS.previewRefresh, previewManager.refresh(input), { "rpc.aggregate": "preview", }), [WS_METHODS.previewClose]: (input) => observeRpcEffect(WS_METHODS.previewClose, previewManager.close(input), { "rpc.aggregate": "preview", }), [WS_METHODS.previewList]: (input) => observeRpcEffect(WS_METHODS.previewList, previewManager.list(input), { "rpc.aggregate": "preview", }), [WS_METHODS.previewReportStatus]: (input) => observeRpcEffect(WS_METHODS.previewReportStatus, previewManager.reportStatus(input), { "rpc.aggregate": "preview", }), [WS_METHODS.previewAutomationConnect]: (input) => observeRpcStreamEffect( WS_METHODS.previewAutomationConnect, previewAutomationBroker.connect(input), { "rpc.aggregate": "preview-automation" }, ), [WS_METHODS.previewAutomationRespond]: (input) => observeRpcEffect( WS_METHODS.previewAutomationRespond, previewAutomationBroker.respond(input), { "rpc.aggregate": "preview-automation" }, ), [WS_METHODS.previewAutomationReportOwner]: (input) => observeRpcEffect( WS_METHODS.previewAutomationReportOwner, previewAutomationBroker.reportOwner(input), { "rpc.aggregate": "preview-automation" }, ), [WS_METHODS.previewAutomationClearOwner]: (input) => observeRpcEffect( WS_METHODS.previewAutomationClearOwner, previewAutomationBroker.clearOwner(input), { "rpc.aggregate": "preview-automation" }, ), [WS_METHODS.subscribePreviewEvents]: (_input) => observeRpcStream(WS_METHODS.subscribePreviewEvents, previewManager.events, { "rpc.aggregate": "preview", }), [WS_METHODS.subscribeDiscoveredLocalServers]: (_input) => observeRpcStream( WS_METHODS.subscribeDiscoveredLocalServers, Stream.callback((queue) => Effect.gen(function* () { yield* portDiscovery.retain; const initial = yield* portDiscovery.scan(); const initialScannedAt = DateTime.formatIso(yield* DateTime.now); yield* Queue.offer(queue, { servers: initial, scannedAt: initialScannedAt, }); yield* portDiscovery.subscribe((servers) => Effect.gen(function* () { const scannedAt = DateTime.formatIso(yield* DateTime.now); yield* Queue.offer(queue, { servers, scannedAt }); }), ); }), ), { "rpc.aggregate": "preview" }, ), [WS_METHODS.subscribeServerConfig]: (_input) => observeRpcStreamEffect( WS_METHODS.subscribeServerConfig, Effect.gen(function* () { const keybindingsUpdates = keybindings.streamChanges.pipe( Stream.map((event) => ({ version: 1 as const, type: "keybindingsUpdated" as const, payload: { keybindings: event.keybindings, issues: event.issues, }, })), ); const providerStatuses = providerRegistry.streamChanges.pipe( Stream.map((providers) => ({ version: 1 as const, type: "providerStatuses" as const, payload: { providers }, })), Stream.debounce(Duration.millis(PROVIDER_STATUS_DEBOUNCE_MS)), ); const settingsUpdates = serverSettings.streamChanges.pipe( Stream.map((settings) => ServerSettings.redactServerSettingsForClient(settings)), Stream.map((settings) => ({ version: 1 as const, type: "settingsUpdated" as const, payload: { settings }, })), ); yield* providerRegistry .refresh() .pipe(Effect.ignoreCause({ log: true }), Effect.forkScoped); const liveUpdates = Stream.merge( keybindingsUpdates, Stream.merge(providerStatuses, settingsUpdates), ); return Stream.concat( Stream.make({ version: 1 as const, type: "snapshot" as const, config: yield* loadServerConfig, }), liveUpdates, ); }), { "rpc.aggregate": "server" }, ), [WS_METHODS.subscribeServerLifecycle]: (_input) => observeRpcStreamEffect( WS_METHODS.subscribeServerLifecycle, Effect.gen(function* () { const snapshot = yield* lifecycleEvents.snapshot; const snapshotEvents = Array.from(snapshot.events).toSorted( (left, right) => left.sequence - right.sequence, ); const liveEvents = lifecycleEvents.stream.pipe( Stream.filter((event) => event.sequence > snapshot.sequence), ); return Stream.concat(Stream.fromIterable(snapshotEvents), liveEvents); }), { "rpc.aggregate": "server" }, ), [WS_METHODS.subscribeAuthAccess]: (_input) => observeRpcStreamEffect( WS_METHODS.subscribeAuthAccess, Effect.gen(function* () { const initialSnapshot = yield* loadAuthAccessSnapshot(); const revisionRef = yield* Ref.make(1); const accessChanges: Stream.Stream< PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange > = Stream.merge(bootstrapCredentials.streamChanges, sessions.streamChanges); const liveEvents: Stream.Stream = accessChanges.pipe( Stream.mapEffect((change) => Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( Effect.map((revision) => toAuthAccessStreamEvent(change, revision, currentSessionId), ), ), ), ); return Stream.concat( Stream.make({ version: 1 as const, revision: 1, type: "snapshot" as const, payload: initialSnapshot, }), liveEvents, ); }), { "rpc.aggregate": "auth" }, ), }); }), ); export const websocketRpcRouteLayer = Layer.unwrap( Effect.succeed( HttpRouter.add( "GET", "/ws", Effect.gen(function* () { const request = yield* HttpServerRequest.HttpServerRequest; const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const sessions = yield* SessionStore.SessionStore; const session = yield* serverAuth.authenticateWebSocketUpgrade(request).pipe( Effect.catchIf(EnvironmentAuth.isServerAuthCredentialError, (error) => failEnvironmentAuthInvalid(EnvironmentAuth.serverAuthCredentialReason(error)), ), Effect.catchIf(EnvironmentAuth.isServerAuthInternalError, (error) => failEnvironmentInternal("internal_error", error), ), ); const rpcWebSocketHttpEffect = yield* RpcServer.toHttpEffectWebsocket(WsRpcGroup, { disableTracing: true, }).pipe( Effect.provide( makeWsRpcLayer(session).pipe( Layer.provideMerge(RpcSerialization.layerJson), Layer.provide(PreviewAutomationBroker.layer), Layer.provide(ProviderMaintenanceRunner.layer), Layer.provide( SourceControlDiscovery.layer.pipe( Layer.provide( SourceControlProviderRegistry.layer.pipe( Layer.provide( Layer.mergeAll( AzureDevOpsCli.layer, BitbucketApi.layer, GitHubCli.layer, GitLabCli.layer, ), ), Layer.provideMerge(GitVcsDriver.layer), Layer.provide( VcsDriverRegistry.layer.pipe(Layer.provide(VcsProjectConfig.layer)), ), ), ), Layer.provide(VcsProcess.layer), ), ), ), ), ); return yield* Effect.acquireUseRelease( sessions.markConnected(session.sessionId), () => rpcWebSocketHttpEffect, () => sessions.markDisconnected(session.sessionId), ); }).pipe( Effect.catchTags({ EnvironmentAuthInvalidError: HttpServerRespondable.toResponse, EnvironmentInternalError: HttpServerRespondable.toResponse, }), ), ), ), );