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 RelayClientInstallFailureReason, 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 { catchEnvironmentAuthenticationErrors } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); function relayClientInstallFailureReason( error: RelayClient.RelayClientInstallError, ): RelayClientInstallFailureReason { switch (error._tag) { case "RelayClientDownloadError": case "RelayClientDownloadReadError": return "download_failed"; case "RelayClientChecksumMismatchError": return "invalid_checksum"; case "RelayClientInstallLockedError": return "install_locked"; case "RelayClientOverrideMissingError": return "override_missing"; case "RelayClientUnsupportedPlatformError": return "unsupported_platform"; case "RelayClientChecksumVerificationError": case "RelayClientExecutableValidationError": return "validation_failed"; case "RelayClientDirectoryCreateError": case "RelayClientInstallLockAcquireError": case "RelayClientDownloadWriteError": case "RelayClientArchiveExtractError": case "RelayClientExecutablePermissionError": case "RelayClientStageError": case "RelayClientActivationError": case "RelayClientInstallWriteError": return "write_failed"; } } 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 authorizeEffect = ( requiredScope: AuthEnvironmentScope, effect: Effect.Effect, ): Effect.Effect => currentSession.scopes.includes(requiredScope) ? effect : Effect.fail(new EnvironmentAuthorizationError({ requiredScope })); const authorizeStream = ( requiredScope: AuthEnvironmentScope, stream: Stream.Stream, ): Stream.Stream => currentSession.scopes.includes(requiredScope) ? stream : Stream.fail(new EnvironmentAuthorizationError({ 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().pipe( Effect.mapError( (cause) => new AuthAccessStreamError({ operation: "list-pairing-links", cause, }), ), ), clientSessions: serverAuth.listClientSessions(currentSessionId).pipe( Effect.mapError( (cause) => new AuthAccessStreamError({ operation: "list-client-sessions", currentSessionId, cause, }), ), ), }); 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.mapError( (error) => new RelayClientInstallFailedError({ reason: relayClientInstallFailureReason(error), cause: error, }), ), Effect.flatMap((status) => Queue.offer(queue, { type: "complete", status, }), ), Effect.catchTags({ RelayClientInstallFailedError: (error) => Queue.fail(queue, error), }), 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* catchEnvironmentAuthenticationErrors( serverAuth.authenticateWebSocketUpgrade(request), ); 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, }), ), ), ), );