import { Cause, Effect, Layer, Option, Queue, Ref, Schema, Stream } from "effect";import {  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, HttpServerResponse } 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"; const WsRpcLayer = 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 serverCommandId = (tag: string) =>      CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`);     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 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;       return {        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,                  }),                ),              );            }            return result;          }).pipe(            Effect.mapError((cause) =>              Schema.is(OrchestrationDispatchCommandError)(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.mapError(              (cause) =>                new OrchestrationReplayEventsError({                  message: "Failed to replay orchestration events",                  cause,                }),            ),          ),          { "rpc.aggregate": "orchestration" },        ),      [WS_METHODS.subscribeOrchestrationDomainEvents]: (_input) =>        observeRpcStreamEffect(          WS_METHODS.subscribeOrchestrationDomainEvents,          Effect.gen(function* () {            const snapshot = yield* orchestrationEngine.getReadModel();            const fromSequenceExclusive = snapshot.snapshotSequence;            const replayEvents: Array<OrchestrationEvent> = yield* Stream.runCollect(              orchestrationEngine.readEvents(fromSequenceExclusive),            ).pipe(              Effect.map((events) => Array.from(events)),              Effect.catch(() => Effect.succeed([] as Array<OrchestrationEvent>)),            );            const replayStream = Stream.fromIterable(replayEvents);            const source = Stream.merge(replayStream, orchestrationEngine.streamDomainEvents);            type SequenceState = {              readonly nextSequence: number;              readonly pendingBySequence: Map<number, OrchestrationEvent>;            };            const state = yield* Ref.make<SequenceState>({              nextSequence: fromSequenceExclusive + 1,              pendingBySequence: new Map<number, OrchestrationEvent>(),            });             return source.pipe(              Stream.mapEffect((event) =>                Ref.modify(                  state,                  ({                    nextSequence,                    pendingBySequence,                  }): [Array<OrchestrationEvent>, SequenceState] => {                    if (event.sequence < nextSequence || pendingBySequence.has(event.sequence)) {                      return [[], { nextSequence, pendingBySequence }];                    }                     const updatedPending = new Map(pendingBySequence);                    updatedPending.set(event.sequence, event);                     const emit: Array<OrchestrationEvent> = [];                    let expected = nextSequence;                    for (;;) {                      const expectedEvent = updatedPending.get(expected);                      if (!expectedEvent) {                        break;                      }                      emit.push(expectedEvent);                      updatedPending.delete(expected);                      expected += 1;                    }                     return [emit, { nextSequence: expected, pendingBySequence: updatedPending }];                  },                ),

Lines 1–500 of 785. Load more