import { Cause, Effect, Layer, Queue, Ref, Schema, Stream } from "effect";import {  type AuthAccessStreamEvent,  AuthSessionId,  CommandId,  EventId,  type OrchestrationCommand,  type GitActionProgressEvent,  type GitManagerServiceError,  OrchestrationDispatchCommandError,  type OrchestrationEvent,  OrchestrationGetFullThreadDiffError,  OrchestrationGetSnapshotError,  OrchestrationGetTurnDiffError,  ORCHESTRATION_WS_METHODS,  ProjectSearchEntriesError,  ProjectWriteFileError,  OrchestrationReplayEventsError,  ThreadId,  type TerminalEvent,  WS_METHODS,  WsRpcGroup,} from "@t3tools/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";import { ServerConfig } from "./config";import { GitCore } from "./git/Services/GitCore";import { GitManager } from "./git/Services/GitManager";import { GitStatusBroadcaster } from "./git/Services/GitStatusBroadcaster";import { Keybindings } from "./keybindings";import { Open, resolveAvailableEditors } from "./open";import { normalizeDispatchCommand } from "./orchestration/Normalizer";import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine";import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery";import {  observeRpcEffect,  observeRpcStream,  observeRpcStreamEffect,} from "./observability/RpcInstrumentation";import { ProviderRegistry } from "./provider/Services/ProviderRegistry";import { ServerLifecycleEvents } from "./serverLifecycleEvents";import { ServerRuntimeStartup } from "./serverRuntimeStartup";import { ServerSettingsService } from "./serverSettings";import { TerminalManager } from "./terminal/Services/Manager";import { WorkspaceEntries } from "./workspace/Services/WorkspaceEntries";import { WorkspaceFileSystem } from "./workspace/Services/WorkspaceFileSystem";import { WorkspacePathOutsideRootError } from "./workspace/Services/WorkspacePaths";import { ProjectSetupScriptRunner } from "./project/Services/ProjectSetupScriptRunner";import { RepositoryIdentityResolver } from "./project/Services/RepositoryIdentityResolver";import { ServerEnvironment } from "./environment/Services/ServerEnvironment";import { ServerAuth } from "./auth/Services/ServerAuth";import {  BootstrapCredentialService,  type BootstrapCredentialChange,} from "./auth/Services/BootstrapCredentialService";import {  SessionCredentialService,  type SessionCredentialChange,} from "./auth/Services/SessionCredentialService";import { respondToAuthError } from "./auth/http"; 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 = (currentSessionId: AuthSessionId) =>  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 gitManager = yield* GitManager;      const git = yield* GitCore;      const gitStatusBroadcaster = yield* GitStatusBroadcaster;      const terminalManager = yield* TerminalManager;      const providerRegistry = yield* ProviderRegistry;      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 bootstrapCredentials = yield* BootstrapCredentialService;      const sessions = yield* SessionCredentialService;      const serverCommandId = (tag: string) =>        CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);       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.makeUnsafe(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) =>        Schema.is(OrchestrationDispatchCommandError)(cause)          ? cause          : new OrchestrationDispatchCommandError({              message: cause instanceof Error ? cause.message : fallbackMessage,              cause,            });       const toBootstrapDispatchCommandCauseError = (cause: Cause.Cause<unknown>) => {        const error = Cause.squash(cause);        return Schema.is(OrchestrationDispatchCommandError)(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 ??                (yield* orchestrationEngine.getReadModel()).projects.find(                  (project) => project.id === event.payload.projectId,                )?.workspaceRoot ??                null;              if (workspaceRoot === null) {                return event;              }               const repositoryIdentity = yield* repositoryIdentityResolver.resolve(workspaceRoot);              return {                ...event,                payload: {                  ...event.payload,                  repositoryIdentity,                },              } satisfies OrchestrationEvent;            });          default:            return Effect.succeed(event);        }      };       const enrichOrchestrationEvents = (events: ReadonlyArray<OrchestrationEvent>) =>        Effect.forEach(events, enrichProjectEvent, { concurrency: 4 });       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;          }) => {            const payload = {              scriptId: input.scriptId,              scriptName: input.scriptName,              terminalId: input.terminalId,              worktreePath: input.worktreePath,            };            return 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: new Date().toISOString(),                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 instanceof Error                        ? error.message                        : "Unknown setup activity dispatch failure.",                  },                ),              ),            );          };           const runSetupProgram = () =>            bootstrap?.runSetupScript && targetWorktreePath              ? (() => {                  const worktreePath = targetWorktreePath;                  const requestedAt = new Date().toISOString();                  return 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,                          });                        },                      }),                    );                })()              : Effect.void;           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* git.createWorktree({                cwd: bootstrap.prepareWorktree.projectCwd,                branch: bootstrap.prepareWorktree.baseBranch,                newBranch: 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.branch,                worktreePath: targetWorktreePath,              });            }             yield* runSetupProgram();             return yield* orchestrationEngine.dispatch(finalTurnStartCommand);          });           return yield* bootstrapProgram.pipe(            Effect.catchCause((cause) => {              const dispatchError = toBootstrapDispatchCommandCauseError(cause);              if (Cause.hasInterruptsOnly(cause)) {                return Effect.fail(dispatchError);              }              return cleanupCreatedThread().pipe(Effect.flatMap(() => Effect.fail(dispatchError)));            }),          );        });       const dispatchNormalizedCommand = (        normalizedCommand: OrchestrationCommand,      ): Effect.Effect<{ readonly sequence: number }, OrchestrationDispatchCommandError> => {        const dispatchEffect =          normalizedCommand.type === "thread.turn.start" && normalizedCommand.bootstrap            ? dispatchBootstrapTurnStart(normalizedCommand)            : orchestrationEngine                .dispatch(normalizedCommand)                .pipe(                  Effect.mapError((cause) =>                    toDispatchCommandError(cause, "Failed to dispatch orchestration command"),                  ),                );         return startup          .enqueueCommand(dispatchEffect)          .pipe(            Effect.mapError((cause) =>              toDispatchCommandError(cause, "Failed to dispatch orchestration command"),            ),          );      };       const loadServerConfig = Effect.gen(function* () {        const keybindingsConfig = yield* keybindings.loadConfigState;        const providers = yield* providerRegistry.getProviders;        const settings = 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) =>        gitStatusBroadcaster          .refreshStatus(cwd)          .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid);       return WsRpcGroup.of({        [ORCHESTRATION_WS_METHODS.getSnapshot]: (_input) =>          observeRpcEffect(            ORCHESTRATION_WS_METHODS.getSnapshot,            projectionSnapshotQuery.getSnapshot().pipe(              Effect.mapError(                (cause) =>                  new OrchestrationGetSnapshotError({                    message: "Failed to load orchestration snapshot",                    cause,                  }),              ),            ),            { "rpc.aggregate": "orchestration" },          ),        [ORCHESTRATION_WS_METHODS.dispatchCommand]: (command) =>          observeRpcEffect(            ORCHESTRATION_WS_METHODS.dispatchCommand,            Effect.gen(function* () {              const normalizedCommand = yield* normalizeDispatchCommand(command);              const result = yield* dispatchNormalizedCommand(normalizedCommand);              if (normalizedCommand.type === "thread.archive") {                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,                    }),                  ),

Lines 1–500 of 944. Load more