import * as Cause from "effect/Cause";
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,
  type AuthAccessStreamEvent,
  AuthSessionId,
  CommandId,
  EventId,
  type OrchestrationCommand,
  GitCommandError,
  GitManagerError,
  type GitActionProgressEvent,
  type GitManagerServiceError,
  OrchestrationDispatchCommandError,
  SshError,
  SshHostId,
  SshSessionId,
  SshSetupShellProfileResult,
  SshSpawnError,
  type SshTerminalEvent,
  type OrchestrationEvent,
  type OrchestrationShellStreamEvent,
  OrchestrationGetFullThreadDiffError,
  OrchestrationGetSnapshotError,
  OrchestrationGetTurnDiffError,
  ORCHESTRATION_WS_METHODS,
  ProjectListAgentSkillsError,
  ProjectSearchEntriesError,
  ProjectWriteFileError,
  OrchestrationReplayEventsError,
  FilesystemBrowseError,
  KeybindingsConfigError,
  OpenError,
  ServerProviderUpdateError,
  ServerSettingsError,
  SourceControlRepositoryError,
  ThreadId,
  TerminalCwdError,
  type TerminalEvent,
  VcsUnsupportedOperationError,
  WS_METHODS,
  WsRpcGroup,
} from "@belweave/contracts";
import { clamp } from "effect/Number";
import { HttpRouter, HttpServerRequest } 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 { Open, resolveAvailableEditors } from "./open.ts";
import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts";
import { discoverLocalAgentSkillsFromDisk } from "./provider/localAgentSkills.ts";
import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine.ts";
import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts";
import {
  observeRpcEffect,
  observeRpcStream,
  observeRpcStreamEffect,
} 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 { SshHostProfiles } from "./ssh/Services/SshHostProfiles.ts";
import { SshKnownHosts } from "./ssh/Services/SshKnownHosts.ts";
import { SshAuditLog } from "./ssh/Services/SshAuditLog.ts";
import { SshSessionManager } from "./ssh/Services/SshSessionManager.ts";
import { SshTokenAuthority } from "./ssh/Services/SshTokenAuthority.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 { ProjectSetupScriptRunner } from "./project/Services/ProjectSetupScriptRunner.ts";
import { RepositoryIdentityResolver } from "./project/Services/RepositoryIdentityResolver.ts";
import { ServerEnvironment } from "./environment/Services/ServerEnvironment.ts";
import { ServerAuth, type AuthenticatedSession } from "./auth/Services/ServerAuth.ts";
import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.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 {
  BootstrapCredentialService,
  type BootstrapCredentialChange,
} from "./auth/Services/BootstrapCredentialService.ts";
import {
  SessionCredentialService,
  type SessionCredentialChange,
} from "./auth/Services/SessionCredentialService.ts";
import { respondToAuthError } from "./auth/http.ts";
import { isReviewSession, REVIEW_ACCESS_DENIED_MESSAGE } from "./auth/reviewAccess.ts";
const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError);
const isSshError = Schema.is(SshError);

function mapToSshError(cause: unknown): typeof SshError.Type {
  if (isSshError(cause)) {
    return cause;
  }
  const detail = cause instanceof Error ? cause.message : String(cause);
  return new SshSpawnError({ detail });
}
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;

function toAuthAccessStreamEvent(
  change: BootstrapCredentialChange | 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 = (session: AuthenticatedSession) =>
  WsRpcGroup.toLayer(
    Effect.gen(function* () {
      const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;
      const orchestrationEngine = yield* OrchestrationEngineService;
      const checkpointDiffQuery = yield* CheckpointDiffQuery;
      const keybindings = yield* Keybindings;
      const open = yield* Open;
      const gitWorkflow = yield* GitWorkflowService;
      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* ServerAuth;
      const sourceControlDiscovery = yield* SourceControlDiscoveryLayer.SourceControlDiscovery;
      const sshHostProfiles = yield* SshHostProfiles;
      const sshKnownHosts = yield* SshKnownHosts;
      const sshSessionManager = yield* SshSessionManager;
      const sshAuditLog = yield* SshAuditLog;
      const sshTokenAuthority = yield* SshTokenAuthority;
      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* BootstrapCredentialService;
      const sessions = yield* SessionCredentialService;
      const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics;
      const currentSessionId = session.sessionId;
      const reviewSession = isReviewSession(session, config);
      const serverCommandId = (tag: string) =>
        CommandId.make(`server:${tag}:${crypto.randomUUID()}`);
      const reviewDenied = <E>(error: E): Effect.Effect<void, E> =>
        reviewSession ? Effect.fail(error) : Effect.void;
      const reviewDeniedStream = <A, E>(error: E): Stream.Stream<A, E> =>
        reviewSession ? Stream.fail(error) : Stream.empty;
      const reviewDeniedOrchestration = () =>
        reviewDenied(
          new OrchestrationDispatchCommandError({
            message: REVIEW_ACCESS_DENIED_MESSAGE,
          }),
        );
      const reviewDeniedGit = (operation: string, cwd = config.cwd) =>
        new GitCommandError({
          operation,
          command: operation,
          cwd,
          detail: REVIEW_ACCESS_DENIED_MESSAGE,
        });
      const reviewDeniedGitManager = (operation: string) =>
        new GitManagerError({
          operation,
          detail: REVIEW_ACCESS_DENIED_MESSAGE,
        });
      const reviewDeniedSourceControl = (operation: string, provider: string | undefined) =>
        new SourceControlRepositoryError({
          provider:
            provider === "github" ||
            provider === "gitlab" ||
            provider === "azure-devops" ||
            provider === "bitbucket"
              ? provider
              : "unknown",
          operation,
          detail: REVIEW_ACCESS_DENIED_MESSAGE,
        });
      const reviewDeniedTerminal = () =>
        new TerminalCwdError({
          cwd: config.cwd,
          reason: "statFailed",
          cause: new Error(REVIEW_ACCESS_DENIED_MESSAGE),
        });
      const reviewDeniedSsh = () => new SshSpawnError({ detail: REVIEW_ACCESS_DENIED_MESSAGE });

      const loadAuthAccessSnapshot = () =>
        Effect.all({
          pairingLinks: serverAuth.listPairingLinks().pipe(Effect.orDie),
          clientSessions: serverAuth.listClientSessions(currentSessionId).pipe(Effect.orDie),
        });

      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";
      }) =>
        orchestrationEngine.dispatch({
          type: "thread.activity.append",
          commandId: serverCommandId("setup-script-activity"),
          threadId: input.threadId,
          activity: {
            id: EventId.make(crypto.randomUUID()),
            tone: input.tone,
            kind: input.kind,
            summary: input.summary,
            payload: input.payload,
            turnId: null,
            createdAt: input.createdAt,
          },
          createdAt: input.createdAt,
        });

      const toDispatchCommandError = (cause: unknown, fallbackMessage: string) =>
        isOrchestrationDispatchCommandError(cause)
          ? cause
          : new OrchestrationDispatchCommandError({
              message: cause instanceof Error ? cause.message : fallbackMessage,
              cause,
            });

      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())),
              );
        }
      };

      const dispatchBootstrapTurnStart = (
        command: Extract<OrchestrationCommand, { type: "thread.turn.start" }>,
      ): 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
              ? orchestrationEngine
                  .dispatch({
                    type: "thread.delete",
                    commandId: serverCommandId("bootstrap-thread-delete"),
                    threadId: command.threadId,
                  })
                  .pipe(Effect.ignoreCause({ log: true }))
              : Effect.void;

          const recordSetupScriptLaunchFailure = (input: {
            readonly error: unknown;
            readonly requestedAt: string;
            readonly worktreePath: string;
          }) => {
            const detail =
              input.error instanceof Error ? input.error.message : "Unknown setup failure.";
            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: 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) {
              const worktree = yield* gitWorkflow.createWorktree({
                cwd: bootstrap.prepareWorktree.projectCwd,
                refName: bootstrap.prepareWorktree.baseBranch,
                newRefName: bootstrap.prepareWorktree.branch,
                path: null,
              });
              targetWorktreePath = worktree.worktree.path;
              yield* orchestrationEngine.dispatch({
                type: "thread.meta.update",
                commandId: 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 assertReviewDispatchAllowed = (
        normalizedCommand: OrchestrationCommand,
      ): Effect.Effect<void, OrchestrationDispatchCommandError> => {
        if (!reviewSession) {
          return Effect.void;
        }
        if (
          normalizedCommand.type === "project.create" ||
          normalizedCommand.type === "project.meta.update" ||
          normalizedCommand.type === "project.delete" ||
          normalizedCommand.type === "thread.delete" ||
          normalizedCommand.type === "thread.checkpoint.revert" ||
          (normalizedCommand.type === "thread.turn.start" &&
            (normalizedCommand.bootstrap?.prepareWorktree !== undefined ||
              normalizedCommand.bootstrap?.runSetupScript === true))
        ) {
          return reviewDeniedOrchestration();
        }
        return Effect.void;
      };

      const loadServerConfig = Effect.gen(function* () {
        const keybindingsConfig = yield* keybindings.loadConfigState;
        const providers = yield* providerRegistry.getProviders;
        const settings = 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: 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);
              yield* assertReviewDispatchAllowed(normalizedCommand);
              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.catch(() => Effect.succeed(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,
            reviewDenied(
              new ServerProviderUpdateError({
                provider: input.provider,
                reason: REVIEW_ACCESS_DENIED_MESSAGE,
              }),
            ).pipe(Effect.flatMap(() => providerMaintenanceRunner.updateProvider(input))),
            {
              "rpc.aggregate": "server",
            },
          ),
        [WS_METHODS.serverUpsertKeybinding]: (rule) =>
          observeRpcEffect(
            WS_METHODS.serverUpsertKeybinding,
            Effect.gen(function* () {
              yield* reviewDenied(
                new KeybindingsConfigError({
                  configPath: config.keybindingsConfigPath,
                  detail: REVIEW_ACCESS_DENIED_MESSAGE,
                }),
              );
              const keybindingsConfig = yield* keybindings.upsertKeybindingRule(rule);
              return { keybindings: keybindingsConfig, issues: [] };
            }),
            { "rpc.aggregate": "server" },
          ),
        [WS_METHODS.serverRemoveKeybinding]: (rule) =>
          observeRpcEffect(
            WS_METHODS.serverRemoveKeybinding,
            Effect.gen(function* () {
              yield* reviewDenied(
                new KeybindingsConfigError({
                  configPath: config.keybindingsConfigPath,
                  detail: REVIEW_ACCESS_DENIED_MESSAGE,
                }),
              );
              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(redactServerSettingsForClient)),
            {
              "rpc.aggregate": "server",
            },
          ),
        [WS_METHODS.serverUpdateSettings]: ({ patch }) =>
          observeRpcEffect(
            WS_METHODS.serverUpdateSettings,
            reviewDenied(
              new ServerSettingsError({
                settingsPath: config.settingsPath,
                detail: REVIEW_ACCESS_DENIED_MESSAGE,
              }),
            ).pipe(
              Effect.flatMap(() =>
                serverSettings.updateSettings(patch).pipe(Effect.map(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.serverSignalProcess]: (input) =>
          observeRpcEffect(
            WS_METHODS.serverSignalProcess,
            reviewSession
              ? Effect.succeed({
                  pid: input.pid,
                  signal: input.signal,
                  signaled: false,
                  message: Option.some(REVIEW_ACCESS_DENIED_MESSAGE),
                })
              : processDiagnostics.signal(input),
            {
              "rpc.aggregate": "server",
            },
          ),
        [WS_METHODS.sourceControlLookupRepository]: (input) =>
          observeRpcEffect(
            WS_METHODS.sourceControlLookupRepository,
            sourceControlRepositories.lookupRepository(input),
            {
              "rpc.aggregate": "source-control",
            },
          ),
        [WS_METHODS.sourceControlCloneRepository]: (input) =>
          observeRpcEffect(
            WS_METHODS.sourceControlCloneRepository,
            reviewDenied(reviewDeniedSourceControl("cloneRepository", input.provider)).pipe(
              Effect.flatMap(() => sourceControlRepositories.cloneRepository(input)),
            ),
            {
              "rpc.aggregate": "source-control",
            },
          ),
        [WS_METHODS.sourceControlPublishRepository]: (input) =>
          observeRpcEffect(
            WS_METHODS.sourceControlPublishRepository,
            reviewDenied(reviewDeniedSourceControl("publishRepository", input.provider)).pipe(
              Effect.flatMap(() =>
                sourceControlRepositories
                  .publishRepository(input)
                  .pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
              ),
            ),
            {
              "rpc.aggregate": "source-control",
            },
          ),
        [WS_METHODS.projectsListAgentSkills]: (input) =>
          observeRpcEffect(
            WS_METHODS.projectsListAgentSkills,
            Effect.tryPromise({
              try: async () => ({
                skills: await discoverLocalAgentSkillsFromDisk({ cwd: input.cwd }),
              }),
              catch: (cause) =>
                new ProjectListAgentSkillsError({
                  message: cause instanceof Error ? cause.message : String(cause),
                  ...(cause instanceof Error ? { cause } : {}),
                }),
            }),
            { "rpc.aggregate": "workspace" },
          ),
        [WS_METHODS.projectsSearchEntries]: (input) =>
          observeRpcEffect(
            WS_METHODS.projectsSearchEntries,
            workspaceEntries.search(input).pipe(
              Effect.mapError(
                (cause) =>
                  new ProjectSearchEntriesError({
                    message: `Failed to search workspace entries: ${cause.detail}`,
                    cause,
                  }),
              ),
            ),
            { "rpc.aggregate": "workspace" },
          ),
        [WS_METHODS.projectsWriteFile]: (input) =>
          observeRpcEffect(
            WS_METHODS.projectsWriteFile,
            reviewDenied(
              new ProjectWriteFileError({
                message: REVIEW_ACCESS_DENIED_MESSAGE,
              }),
            ).pipe(
              Effect.flatMap(() =>
                workspaceFileSystem.writeFile(input).pipe(
                  Effect.mapError((cause) => {
                    const message = isWorkspacePathOutsideRootError(cause)
                      ? "Workspace file path must stay within the project root."
                      : "Failed to write workspace file";
                    return new ProjectWriteFileError({
                      message,
                      cause,
                    });
                  }),
                ),
              ),
            ),
            { "rpc.aggregate": "workspace" },
          ),
        [WS_METHODS.shellOpenInEditor]: (input) =>
          observeRpcEffect(
            WS_METHODS.shellOpenInEditor,
            reviewDenied(new OpenError({ message: REVIEW_ACCESS_DENIED_MESSAGE })).pipe(
              Effect.flatMap(() => open.openInEditor(input)),
            ),
            {
              "rpc.aggregate": "workspace",
            },
          ),
        [WS_METHODS.filesystemBrowse]: (input) =>
          observeRpcEffect(
            WS_METHODS.filesystemBrowse,
            reviewDenied(
              new FilesystemBrowseError({
                message: REVIEW_ACCESS_DENIED_MESSAGE,
              }),
            ).pipe(
              Effect.flatMap(() =>
                workspaceEntries.browse(input).pipe(
                  Effect.mapError(
                    (cause) =>
                      new FilesystemBrowseError({
                        message: cause.detail,
                        cause,
                      }),
                  ),
                ),
              ),
            ),
            { "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,
            reviewDenied(reviewDeniedGit("pull", input.cwd)).pipe(
              Effect.flatMap(() =>
                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,
            reviewSession
              ? reviewDeniedStream<GitActionProgressEvent, GitManagerServiceError>(
                  reviewDeniedGitManager("runStackedAction"),
                )
              : Stream.callback<GitActionProgressEvent, GitManagerServiceError>((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,
            reviewDenied(reviewDeniedGitManager("preparePullRequestThread")).pipe(
              Effect.flatMap(() =>
                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,
            reviewDenied(reviewDeniedGit("createWorktree", input.cwd)).pipe(
              Effect.flatMap(() =>
                gitWorkflow.createWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
              ),
            ),
            { "rpc.aggregate": "vcs" },
          ),
        [WS_METHODS.vcsRemoveWorktree]: (input) =>
          observeRpcEffect(
            WS_METHODS.vcsRemoveWorktree,
            reviewDenied(reviewDeniedGit("removeWorktree", input.cwd)).pipe(
              Effect.flatMap(() =>
                gitWorkflow.removeWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
              ),
            ),
            { "rpc.aggregate": "vcs" },
          ),
        [WS_METHODS.vcsCreateRef]: (input) =>
          observeRpcEffect(
            WS_METHODS.vcsCreateRef,
            reviewDenied(reviewDeniedGit("createRef", input.cwd)).pipe(
              Effect.flatMap(() =>
                gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
              ),
            ),
            { "rpc.aggregate": "vcs" },
          ),
        [WS_METHODS.vcsSwitchRef]: (input) =>
          observeRpcEffect(
            WS_METHODS.vcsSwitchRef,
            reviewDenied(reviewDeniedGit("switchRef", input.cwd)).pipe(
              Effect.flatMap(() =>
                gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
              ),
            ),
            { "rpc.aggregate": "vcs" },
          ),
        [WS_METHODS.vcsInit]: (input) =>
          observeRpcEffect(
            WS_METHODS.vcsInit,
            reviewDenied(
              new VcsUnsupportedOperationError({
                operation: "init",
                kind: "git",
                detail: REVIEW_ACCESS_DENIED_MESSAGE,
              }),
            ).pipe(
              Effect.flatMap(() =>
                vcsProvisioning
                  .initRepository(input)
                  .pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
              ),
            ),
            { "rpc.aggregate": "vcs" },
          ),
        [WS_METHODS.terminalOpen]: (input) =>
          observeRpcEffect(
            WS_METHODS.terminalOpen,
            reviewDenied(reviewDeniedTerminal()).pipe(
              Effect.flatMap(() => terminalManager.open(input)),
            ),
            {
              "rpc.aggregate": "terminal",
            },
          ),
        [WS_METHODS.terminalWrite]: (input) =>
          observeRpcEffect(
            WS_METHODS.terminalWrite,
            reviewDenied(reviewDeniedTerminal()).pipe(
              Effect.flatMap(() => terminalManager.write(input)),
            ),
            {
              "rpc.aggregate": "terminal",
            },
          ),
        [WS_METHODS.terminalResize]: (input) =>
          observeRpcEffect(
            WS_METHODS.terminalResize,
            reviewDenied(reviewDeniedTerminal()).pipe(
              Effect.flatMap(() => terminalManager.resize(input)),
            ),
            {
              "rpc.aggregate": "terminal",
            },
          ),
        [WS_METHODS.terminalClear]: (input) =>
          observeRpcEffect(
            WS_METHODS.terminalClear,
            reviewDenied(reviewDeniedTerminal()).pipe(
              Effect.flatMap(() => terminalManager.clear(input)),
            ),
            {
              "rpc.aggregate": "terminal",
            },
          ),
        [WS_METHODS.terminalRestart]: (input) =>
          observeRpcEffect(
            WS_METHODS.terminalRestart,
            reviewDenied(reviewDeniedTerminal()).pipe(
              Effect.flatMap(() => terminalManager.restart(input)),
            ),
            {
              "rpc.aggregate": "terminal",
            },
          ),
        [WS_METHODS.terminalClose]: (input) =>
          observeRpcEffect(
            WS_METHODS.terminalClose,
            reviewDenied(reviewDeniedTerminal()).pipe(
              Effect.flatMap(() => terminalManager.close(input)),
            ),
            {
              "rpc.aggregate": "terminal",
            },
          ),
        [WS_METHODS.subscribeTerminalEvents]: (_input) =>
          observeRpcStream(
            WS_METHODS.subscribeTerminalEvents,
            Stream.callback<TerminalEvent>((queue) =>
              Effect.acquireRelease(
                terminalManager.subscribe((event) => Queue.offer(queue, event)),
                (unsubscribe) => Effect.sync(unsubscribe),
              ),
            ),
            { "rpc.aggregate": "terminal" },
          ),
        [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) => 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.sshListHosts]: (_input) =>
          observeRpcEffect(
            WS_METHODS.sshListHosts,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshHostProfiles.list().pipe(
                  Effect.map((hosts) => ({ hosts })),
                  Effect.mapError((cause) => mapToSshError(cause)),
                ),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshAddHost]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshAddHost,
            Effect.gen(function* () {
              yield* reviewDenied(reviewDeniedSsh());
              const profile = yield* sshHostProfiles.create(input);
              yield* sshAuditLog
                .append({
                  type: "host-profile-created",
                  actorSessionId: currentSessionId,
                  hostId: profile.id,
                  hostname: profile.hostname,
                  port: profile.port,
                  username: profile.username,
                  authMethod: profile.authMethod,
                  sshSessionId: null,
                  message: `Host profile created: ${profile.label}`,
                })
                .pipe(Effect.catch(() => Effect.void));
              return profile;
            }).pipe(Effect.mapError((cause) => mapToSshError(cause))),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshRemoveHost]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshRemoveHost,
            Effect.gen(function* () {
              yield* reviewDenied(reviewDeniedSsh());
              const profile = yield* sshHostProfiles.get(input.hostId);
              yield* sshHostProfiles.remove(input.hostId);
              yield* sshKnownHosts
                .remove({ hostname: profile.hostname, port: profile.port })
                .pipe(Effect.catch(() => Effect.void));
              yield* sshAuditLog
                .append({
                  type: "host-profile-removed",
                  actorSessionId: currentSessionId,
                  hostId: profile.id,
                  hostname: profile.hostname,
                  port: profile.port,
                  username: profile.username,
                  authMethod: profile.authMethod,
                  sshSessionId: null,
                  message: `Host profile removed: ${profile.label}`,
                })
                .pipe(Effect.catch(() => Effect.void));
            }).pipe(Effect.mapError((cause) => mapToSshError(cause))),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshOpenSession]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshOpenSession,
            Effect.gen(function* () {
              yield* reviewDenied(reviewDeniedSsh());
              const snapshot = yield* sshSessionManager.open({
                authSessionId: currentSessionId,
                hostId: input.hostId,
                cols: input.cols,
                rows: input.rows,
              });
              const issued = yield* sshTokenAuthority.issue({
                authSessionId: currentSessionId,
                sshSessionId: snapshot.sessionId,
              });
              return {
                snapshot,
                sessionToken: issued.token,
                sessionTokenExpiresAt: DateTime.formatIso(issued.expiresAt),
              };
            }).pipe(Effect.mapError((cause) => mapToSshError(cause))),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshGetSession]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshGetSession,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshSessionManager
                  .get({ authSessionId: currentSessionId, sshSessionId: input.sessionId })
                  .pipe(Effect.mapError((cause) => mapToSshError(cause))),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshSendInput]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshSendInput,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshSessionManager
                  .sendInput({
                    authSessionId: currentSessionId,
                    sshSessionId: input.sessionId,
                    data: input.data,
                  })
                  .pipe(Effect.mapError((cause) => mapToSshError(cause))),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshResize]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshResize,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshSessionManager
                  .resize({
                    authSessionId: currentSessionId,
                    sshSessionId: input.sessionId,
                    cols: input.cols,
                    rows: input.rows,
                  })
                  .pipe(Effect.mapError((cause) => mapToSshError(cause))),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshConfirmHostKey]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshConfirmHostKey,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshSessionManager
                  .confirmHostKey({
                    authSessionId: currentSessionId,
                    sshSessionId: input.sessionId,
                    fingerprintSha256: input.fingerprintSha256,
                    decision: input.decision,
                    remember: input.remember,
                  })
                  .pipe(Effect.mapError((cause) => mapToSshError(cause))),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshCloseSession]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshCloseSession,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshSessionManager
                  .close({ authSessionId: currentSessionId, sshSessionId: input.sessionId })
                  .pipe(Effect.mapError((cause) => mapToSshError(cause))),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshIssueSessionToken]: (input) =>
          observeRpcEffect(
            WS_METHODS.sshIssueSessionToken,
            Effect.gen(function* () {
              yield* reviewDenied(reviewDeniedSsh());
              // Verify caller still owns the session before minting a fresh token.
              yield* sshSessionManager
                .get({ authSessionId: currentSessionId, sshSessionId: input.sessionId })
                .pipe(Effect.mapError((cause) => mapToSshError(cause)));
              const issued = yield* sshTokenAuthority.issue({
                authSessionId: currentSessionId,
                sshSessionId: input.sessionId,
              });
              return {
                sessionToken: issued.token,
                expiresAt: DateTime.formatIso(issued.expiresAt),
              };
            }),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshListAudit]: (_input) =>
          observeRpcEffect(
            WS_METHODS.sshListAudit,
            reviewDenied(reviewDeniedSsh()).pipe(
              Effect.flatMap(() =>
                sshAuditLog.list().pipe(
                  Effect.map((events) => ({ events })),
                  Effect.mapError((cause) => mapToSshError(cause)),
                ),
              ),
            ),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.sshSetupShellProfile]: (_input: unknown) =>
          observeRpcEffect(
            WS_METHODS.sshSetupShellProfile,
            reviewDenied(reviewDeniedSsh()).pipe(Effect.flatMap(() => Effect.sync(() => {
              if (process.platform !== "darwin") {
                return SshSetupShellProfileResult.make({
                  shellProfile: process.platform === "win32" ? "Windows OpenSSH" : "OpenSSH",
                  alreadyPresent: true,
                });
              }

              const home = process.env.HOME ?? process.cwd();
              const marker = "# >>> trifecta-ssh-keychain >>>";
              const markerEnd = "# <<< trifecta-ssh-keychain <<<";
              const snippet = [
                marker,
                'if [ -n "$SSH_CONNECTION" ] && [ "$TERM" != "dumb" ]; then',
                "  security unlock-keychain ~/Library/Keychains/login.keychain-db 2>/dev/null || true",
                "fi",
                markerEnd,
              ].join("\n");
              const fs = require("node:fs");
              const path = require("node:path");
              for (const rcFile of [".zshrc", ".bashrc", ".bash_profile"]) {
                const rcPath = path.join(home, rcFile);
                try {
                  const content = fs.readFileSync(rcPath, "utf-8");
                  if (content.includes(marker)) {
                    return SshSetupShellProfileResult.make({
                      shellProfile: rcFile as any,
                      alreadyPresent: true,
                    });
                  }
                } catch {
                  continue;
                }
              }
              const shell = process.env.SHELL ?? "/bin/zsh";
              const rcFile = shell.endsWith("zsh") ? ".zshrc" : ".bashrc";
              const rcPath = path.join(home, rcFile);
              let content = "";
              try {
                content = fs.readFileSync(rcPath, "utf-8");
              } catch {
                content = "";
              }
              fs.writeFileSync(rcPath, content.trimEnd() + "\n\n" + snippet + "\n", "utf-8");
              return SshSetupShellProfileResult.make({
                shellProfile: rcFile as any,
                alreadyPresent: false,
              });
            }))),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.subscribeSshTerminal]: (input) =>
          observeRpcStream(
            WS_METHODS.subscribeSshTerminal,
            reviewSession
              ? reviewDeniedStream<SshTerminalEvent, SshError>(reviewDeniedSsh())
              : (sshSessionManager
                  .subscribe({ authSessionId: currentSessionId, sshSessionId: input.sessionId })
                  .pipe(Stream.mapError((cause) => mapToSshError(cause))) as Stream.Stream<
                  SshTerminalEvent,
                  SshError
                >),
            { "rpc.aggregate": "ssh" },
          ),
        [WS_METHODS.subscribeAuthAccess]: (_input) =>
          observeRpcStreamEffect(
            WS_METHODS.subscribeAuthAccess,
            Effect.gen(function* () {
              if (reviewSession) {
                return Stream.empty;
              }
              const initialSnapshot = yield* loadAuthAccessSnapshot();
              const revisionRef = yield* Ref.make(1);
              const accessChanges: Stream.Stream<
                BootstrapCredentialChange | SessionCredentialChange
              > = Stream.merge(bootstrapCredentials.streamChanges, sessions.streamChanges);

              const liveEvents: Stream.Stream<AuthAccessStreamEvent> = 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* ServerAuth;
        const sessions = yield* SessionCredentialService;
        const session = yield* serverAuth.authenticateWebSocketUpgrade(request);
        const rpcWebSocketHttpEffect = yield* RpcServer.toHttpEffectWebsocket(WsRpcGroup, {
          disableTracing: true,
        }).pipe(
          Effect.provide(
            makeWsRpcLayer(session).pipe(
              Layer.provideMerge(RpcSerialization.layerJson),
              Layer.provide(ProviderMaintenanceRunner.layer),
              Layer.provide(
                SourceControlDiscoveryLayer.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.catchTag("AuthError", respondToAuthError)),
    ),
  ),
);
