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, AuthTerminalOperateScope, AuthAccessReadScope, AuthAccessStreamError, type AuthAccessStreamEvent, type AuthEnvironmentScope, AuthSessionId, CommandId, EventId, type OrchestrationCommand, type GitActionProgressEvent, type GitManagerServiceError, OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, ProjectSearchEntriesError, ProjectWriteFileError, OrchestrationReplayEventsError, FilesystemBrowseError, 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 { CheckpointDiffQuery } from "./checkpointing/Services/CheckpointDiffQuery.ts";import { ServerConfig } from "./config.ts";import { Keybindings } from "./keybindings.ts";import * as ExternalLauncher from "./process/externalLauncher.ts";import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts";import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine.ts";import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts";import { observeRpcEffect as instrumentRpcEffect, observeRpcStream as instrumentRpcStream, observeRpcStreamEffect as instrumentRpcStreamEffect,} from "./observability/RpcInstrumentation.ts";import { ProviderRegistry } from "./provider/Services/ProviderRegistry.ts";import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts";import { ServerLifecycleEvents } from "./serverLifecycleEvents.ts";import { ServerRuntimeStartup } from "./serverRuntimeStartup.ts";import { redactServerSettingsForClient, ServerSettingsService } from "./serverSettings.ts";import { TerminalManager } from "./terminal/Services/Manager.ts";import { WorkspaceEntries } from "./workspace/Services/WorkspaceEntries.ts";import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem.ts";import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths.ts";import { VcsStatusBroadcaster } from "./vcs/VcsStatusBroadcaster.ts";import { VcsProvisioningService } from "./vcs/VcsProvisioningService.ts";import { GitWorkflowService } from "./git/GitWorkflowService.ts";import { ReviewService } from "./review/ReviewService.ts";import { ProjectSetupScriptRunner } from "./project/Services/ProjectSetupScriptRunner.ts";import { RepositoryIdentityResolver } from "./project/Services/RepositoryIdentityResolver.ts";import { ServerEnvironment } from "./environment/Services/ServerEnvironment.ts";import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";import type { AuthenticatedSession } 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 SourceControlDiscoveryLayer from "./sourceControl/SourceControlDiscovery.ts";import { 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";const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError);const isWorkspacePathOutsideRootError = Schema.is(WorkspacePathOutsideRootError); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); 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<string, AuthEnvironmentScope>([ [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.sourceControlLookupRepository, AuthOrchestrationReadScope], [WS_METHODS.sourceControlCloneRepository, AuthOrchestrationOperateScope], [WS_METHODS.sourceControlPublishRepository, AuthOrchestrationOperateScope], [WS_METHODS.projectsSearchEntries, AuthOrchestrationReadScope], [WS_METHODS.projectsWriteFile, AuthOrchestrationOperateScope], [WS_METHODS.shellOpenInEditor, AuthOrchestrationOperateScope], [WS_METHODS.filesystemBrowse, 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.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: AuthenticatedSession) => WsRpcGroup.toLayer( Effect.gen(function* () { const currentSessionId = currentSession.sessionId; const crypto = yield* Crypto.Crypto; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngineService; const checkpointDiffQuery = yield* CheckpointDiffQuery; const keybindings = yield* Keybindings; const externalLauncher = yield* ExternalLauncher.ExternalLauncher; const gitWorkflow = yield* GitWorkflowService; const review = yield* ReviewService; const vcsProvisioning = yield* VcsProvisioningService; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; const terminalManager = yield* TerminalManager; const providerRegistry = yield* ProviderRegistry; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const config = yield* ServerConfig; const lifecycleEvents = yield* ServerLifecycleEvents; const serverSettings = yield* ServerSettingsService; const startup = yield* ServerRuntimeStartup; const workspaceEntries = yield* WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem; const projectSetupScriptRunner = yield* ProjectSetupScriptRunner; const repositoryIdentityResolver = yield* RepositoryIdentityResolver; const serverEnvironment = yield* ServerEnvironment; const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const sourceControlDiscovery = yield* SourceControlDiscoveryLayer.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; const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; const sessions = yield* SessionStore.SessionStore; const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics; const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor; const authorizationError = (requiredScope: AuthEnvironmentScope) => new EnvironmentAuthorizationError({ message: `The authenticated token is missing required scope: ${requiredScope}.`, requiredScope, }); const authorizeEffect = <A, E, R>( requiredScope: AuthEnvironmentScope, effect: Effect.Effect<A, E, R>, ): Effect.Effect<A, E | EnvironmentAuthorizationError, R> => currentSession.scopes.includes(requiredScope) ? effect : Effect.fail(authorizationError(requiredScope)); const authorizeStream = <A, E, R>( requiredScope: AuthEnvironmentScope, stream: Stream.Stream<A, E, R>, ): Stream.Stream<A, E | EnvironmentAuthorizationError, R> => 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 = <A, E, R>( method: string, effect: Effect.Effect<A, E, R>, traceAttributes?: Readonly<Record<string, unknown>>, ) => instrumentRpcEffect( method, authorizeEffect(requiredScopeForMethod(method), effect), traceAttributes, ); const observeRpcStream = <A, E, R>( method: string, stream: Stream.Stream<A, E, R>, traceAttributes?: Readonly<Record<string, unknown>>, ) => instrumentRpcStream( method, authorizeStream(requiredScopeForMethod(method), stream), traceAttributes, ); const observeRpcStreamEffect = <A, StreamError, StreamContext, EffectError, EffectContext>( method: string, effect: Effect.Effect< Stream.Stream<A, StreamError, StreamContext>, EffectError, EffectContext >, traceAttributes?: Readonly<Record<string, unknown>>, ) => 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<string, unknown>; 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<unknown>) => { 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<OrchestrationEvent, never, never> => { 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.catch(() => Effect.succeed(event))); default: return Effect.succeed(event); } }; const enrichOrchestrationEvents = (events: ReadonlyArray<OrchestrationEvent>) => Effect.forEach(events, enrichProjectEvent, { concurrency: 4 }); const toShellStreamEvent = ( event: OrchestrationEvent, ): Effect.Effect<Option.Option<OrchestrationShellStreamEvent>, 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.catch(() => Effect.succeed(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.catch(() => Effect.succeed(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.catch(() => Effect.succeed(Option.none())), );Lines 1–500 of 1474. Load more
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, AuthTerminalOperateScope, AuthAccessReadScope, AuthAccessStreamError, type AuthAccessStreamEvent, type AuthEnvironmentScope, AuthSessionId, CommandId, EventId, type OrchestrationCommand, type GitActionProgressEvent, type GitManagerServiceError, OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, ProjectSearchEntriesError, ProjectWriteFileError, OrchestrationReplayEventsError, FilesystemBrowseError, 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 { CheckpointDiffQuery } from "./checkpointing/Services/CheckpointDiffQuery.ts"; import { ServerConfig } from "./config.ts"; import { Keybindings } from "./keybindings.ts"; import * as ExternalLauncher from "./process/externalLauncher.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine.ts"; import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import { observeRpcEffect as instrumentRpcEffect, observeRpcStream as instrumentRpcStream, observeRpcStreamEffect as instrumentRpcStreamEffect, } from "./observability/RpcInstrumentation.ts"; import { ProviderRegistry } from "./provider/Services/ProviderRegistry.ts"; import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; import { ServerLifecycleEvents } from "./serverLifecycleEvents.ts"; import { ServerRuntimeStartup } from "./serverRuntimeStartup.ts"; import { redactServerSettingsForClient, ServerSettingsService } from "./serverSettings.ts"; import { TerminalManager } from "./terminal/Services/Manager.ts"; import { WorkspaceEntries } from "./workspace/Services/WorkspaceEntries.ts"; import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem.ts"; import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths.ts"; import { VcsStatusBroadcaster } from "./vcs/VcsStatusBroadcaster.ts"; import { VcsProvisioningService } from "./vcs/VcsProvisioningService.ts"; import { GitWorkflowService } from "./git/GitWorkflowService.ts"; import { ReviewService } from "./review/ReviewService.ts"; import { ProjectSetupScriptRunner } from "./project/Services/ProjectSetupScriptRunner.ts"; import { RepositoryIdentityResolver } from "./project/Services/RepositoryIdentityResolver.ts"; import { ServerEnvironment } from "./environment/Services/ServerEnvironment.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import type { AuthenticatedSession } 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 SourceControlDiscoveryLayer from "./sourceControl/SourceControlDiscovery.ts"; import { 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"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); const isWorkspacePathOutsideRootError = Schema.is(WorkspacePathOutsideRootError); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); 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<string, AuthEnvironmentScope>([ [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.sourceControlLookupRepository, AuthOrchestrationReadScope], [WS_METHODS.sourceControlCloneRepository, AuthOrchestrationOperateScope], [WS_METHODS.sourceControlPublishRepository, AuthOrchestrationOperateScope], [WS_METHODS.projectsSearchEntries, AuthOrchestrationReadScope], [WS_METHODS.projectsWriteFile, AuthOrchestrationOperateScope], [WS_METHODS.shellOpenInEditor, AuthOrchestrationOperateScope], [WS_METHODS.filesystemBrowse, 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.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: AuthenticatedSession) => WsRpcGroup.toLayer( Effect.gen(function* () { const currentSessionId = currentSession.sessionId; const crypto = yield* Crypto.Crypto; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; const orchestrationEngine = yield* OrchestrationEngineService; const checkpointDiffQuery = yield* CheckpointDiffQuery; const keybindings = yield* Keybindings; const externalLauncher = yield* ExternalLauncher.ExternalLauncher; const gitWorkflow = yield* GitWorkflowService; const review = yield* ReviewService; const vcsProvisioning = yield* VcsProvisioningService; const vcsStatusBroadcaster = yield* VcsStatusBroadcaster; const terminalManager = yield* TerminalManager; const providerRegistry = yield* ProviderRegistry; const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const config = yield* ServerConfig; const lifecycleEvents = yield* ServerLifecycleEvents; const serverSettings = yield* ServerSettingsService; const startup = yield* ServerRuntimeStartup; const workspaceEntries = yield* WorkspaceEntries; const workspaceFileSystem = yield* WorkspaceFileSystem; const projectSetupScriptRunner = yield* ProjectSetupScriptRunner; const repositoryIdentityResolver = yield* RepositoryIdentityResolver; const serverEnvironment = yield* ServerEnvironment; const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const sourceControlDiscovery = yield* SourceControlDiscoveryLayer.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; const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; const sessions = yield* SessionStore.SessionStore; const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics; const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor; const authorizationError = (requiredScope: AuthEnvironmentScope) => new EnvironmentAuthorizationError({ message: `The authenticated token is missing required scope: ${requiredScope}.`, requiredScope, }); const authorizeEffect = <A, E, R>( requiredScope: AuthEnvironmentScope, effect: Effect.Effect<A, E, R>, ): Effect.Effect<A, E | EnvironmentAuthorizationError, R> => currentSession.scopes.includes(requiredScope) ? effect : Effect.fail(authorizationError(requiredScope)); const authorizeStream = <A, E, R>( requiredScope: AuthEnvironmentScope, stream: Stream.Stream<A, E, R>, ): Stream.Stream<A, E | EnvironmentAuthorizationError, R> => 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 = <A, E, R>( method: string, effect: Effect.Effect<A, E, R>, traceAttributes?: Readonly<Record<string, unknown>>, ) => instrumentRpcEffect( method, authorizeEffect(requiredScopeForMethod(method), effect), traceAttributes, ); const observeRpcStream = <A, E, R>( method: string, stream: Stream.Stream<A, E, R>, traceAttributes?: Readonly<Record<string, unknown>>, ) => instrumentRpcStream( method, authorizeStream(requiredScopeForMethod(method), stream), traceAttributes, ); const observeRpcStreamEffect = <A, StreamError, StreamContext, EffectError, EffectContext>( method: string, effect: Effect.Effect< Stream.Stream<A, StreamError, StreamContext>, EffectError, EffectContext >, traceAttributes?: Readonly<Record<string, unknown>>, ) => 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<string, unknown>; 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<unknown>) => { 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<OrchestrationEvent, never, never> => { 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.catch(() => Effect.succeed(event))); default: return Effect.succeed(event); } }; const enrichOrchestrationEvents = (events: ReadonlyArray<OrchestrationEvent>) => Effect.forEach(events, enrichProjectEvent, { concurrency: 4 }); const toShellStreamEvent = ( event: OrchestrationEvent, ): Effect.Effect<Option.Option<OrchestrationShellStreamEvent>, 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.catch(() => Effect.succeed(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.catch(() => Effect.succeed(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.catch(() => Effect.succeed(Option.none())), );