import {  CommandId,  type ChatAttachment,  type MessageId,  type ModelSelection,  type OrchestrationV2Actor,  type OrchestrationV2CreationSource,  type OrchestrationV2ThreadProjection,  type ProviderInteractionMode,  ProjectId,  type RunId,  type RuntimeMode,  ThreadId,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Context from "effect/Context";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Ref from "effect/Ref";import * as Schema from "effect/Schema";import * as Scope from "effect/Scope"; import * as GitWorkflow from "../git/GitWorkflowService.ts";import * as ProjectService from "../project/ProjectService.ts";import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts";import * as TextGeneration from "../textGeneration/TextGeneration.ts";import * as CommandReceiptStore from "./CommandReceiptStore.ts";import * as IdAllocator from "./IdAllocator.ts";import { makeProviderFailure } from "./ProviderFailure.ts";import * as ThreadManagement from "./ThreadManagementService.ts"; export type ThreadLaunchWorkspaceStrategy =  | { readonly type: "root"; readonly branch?: string | undefined }  | {      readonly type: "existing_worktree";      readonly worktreePath: string;      readonly branch?: string | undefined;    }  | {      readonly type: "worktree";      readonly baseRef: string;      readonly branch?: string | undefined;      readonly startFromOrigin?: boolean | undefined;    }; export interface ThreadLaunchInitialMessage {  readonly messageId?: MessageId;  readonly text: string;  readonly attachments: ReadonlyArray<ChatAttachment>;} export interface ThreadLaunchInput {  readonly commandId: CommandId;  readonly threadId?: ThreadId;  readonly reuseExistingThread?: boolean;  readonly projectId: ProjectId;  readonly title: string;  readonly modelSelection: ModelSelection;  readonly runtimeMode: RuntimeMode;  readonly interactionMode: ProviderInteractionMode;  readonly workspaceStrategy: ThreadLaunchWorkspaceStrategy;  readonly initialMessage?: ThreadLaunchInitialMessage;  readonly createdBy: OrchestrationV2Actor;  readonly creationSource: OrchestrationV2CreationSource;} export interface ThreadLaunchResult {  readonly threadId: ThreadId;  readonly projection: OrchestrationV2ThreadProjection;  readonly resumed: boolean;} export class ThreadLaunchError extends Schema.TaggedErrorClass<ThreadLaunchError>()(  "ThreadLaunchError",  {    operation: Schema.Literals([      "resolve-project",      "read-receipt",      "generate-metadata",      "provision-worktree",      "run-setup-script",      "create-thread",      "update-thread",      "dispatch-message",      "release-run",      "fail-run",    ]),    commandId: CommandId,    projectId: ProjectId,    threadId: Schema.optional(ThreadId),    cause: Schema.Defect(),  },) {  override get message(): string {    return `Thread launch ${this.commandId} failed during ${this.operation}.`;  }} export class ThreadLaunchService extends Context.Service<  ThreadLaunchService,  {    readonly launch: (      input: ThreadLaunchInput,    ) => Effect.Effect<ThreadLaunchResult, ThreadLaunchError>;  }>()("t3/orchestration-v2/ThreadLaunchService") {} const isThreadLaunchError = Schema.is(ThreadLaunchError); function fallbackBranchName(threadId: ThreadId): string {  const suffix = String(threadId)    .split(":")    .at(-1)    ?.replace(/[^a-zA-Z0-9-]+/gu, "-")    .replace(/^-+|-+$/gu, "")    .slice(0, 16);  return `thread-${suffix || "new"}`;} function failureDetail(error: unknown): string {  if (isThreadLaunchError(error)) {    const cause = error.cause;    const detail = cause instanceof Error ? cause.message : String(cause);    return `Workspace preparation failed during ${error.operation.replaceAll("-", " ")}: ${detail}`;  }  return `Workspace preparation failed: ${error instanceof Error ? error.message : String(error)}`;} export const make = Effect.gen(function* () {  const projects = yield* ProjectService.ProjectService;  const git = yield* GitWorkflow.GitWorkflowService;  const setupScripts = yield* ProjectSetupScriptRunner.ProjectSetupScriptRunner;  const textGeneration = yield* TextGeneration.TextGeneration;  const receipts = yield* CommandReceiptStore.CommandReceiptStoreV2;  const ids = yield* IdAllocator.IdAllocatorV2;  const threads = yield* ThreadManagement.ThreadManagementService;  const preparationScope = yield* Scope.make("sequential");  const scheduledLaunches = yield* Ref.make<ReadonlySet<CommandId>>(new Set());  yield* Effect.addFinalizer(() => Scope.close(preparationScope, Exit.void));   const mapError =    (input: ThreadLaunchInput, operation: ThreadLaunchError["operation"], threadId?: ThreadId) =>    (cause: unknown) =>      new ThreadLaunchError({        operation,        commandId: input.commandId,        projectId: input.projectId,        ...(threadId === undefined ? {} : { threadId }),        cause,      });   const readReceipt = (input: ThreadLaunchInput, commandId: CommandId) =>    receipts      .getByCommandId(commandId)      .pipe(Effect.mapError(mapError(input, "read-receipt", input.threadId)));   const validateReusableThread = Effect.fn("ThreadLaunchService.validateReusableThread")(function* (    input: ThreadLaunchInput,    threadId: ThreadId,  ) {    const projection = yield* threads      .getThreadProjection(threadId)      .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));    if (      projection.thread.projectId !== input.projectId ||      projection.thread.archivedAt !== null ||      projection.thread.deletedAt !== null ||      projection.messages.length > 0 ||      projection.runs.length > 0    ) {      return yield* mapError(        input,        "update-thread",        threadId,      )("Only an empty active thread in the target project can change workspace during launch.");    }  });   const prepareInBackground = Effect.fn("ThreadLaunchService.prepareInBackground")(function* (    input: ThreadLaunchInput,    threadId: ThreadId,    runId: RunId | null,  ) {    const project = yield* projects.getById(input.projectId).pipe(      Effect.mapError(mapError(input, "resolve-project", threadId)),      Effect.flatMap(        Option.match({          onNone: () =>            Effect.fail(mapError(input, "resolve-project", threadId)("Project no longer exists.")),          onSome: Effect.succeed,        }),      ),    );     if (input.title === "New thread" && input.initialMessage !== undefined) {      yield* textGeneration        .generateThreadTitle({          cwd: project.workspaceRoot,          message: input.initialMessage.text,          attachments: input.initialMessage.attachments,          modelSelection: input.modelSelection,        })        .pipe(          Effect.flatMap((result) =>            threads.dispatch({              type: "thread.metadata.update",              commandId: CommandId.make(`${input.commandId}:title`),              threadId,              title: result.title,            }),          ),          Effect.catchCause((cause) =>            Effect.logWarning("Thread title generation failed", {              commandId: input.commandId,              threadId,              cause,            }),          ),          Effect.forkIn(preparationScope),        );    }     let branch =      input.workspaceStrategy.type === "worktree" &&      input.workspaceStrategy.branch === undefined &&      input.initialMessage !== undefined        ? yield* textGeneration            .generateBranchName({              cwd: project.workspaceRoot,              message: input.initialMessage.text,              attachments: input.initialMessage.attachments,              modelSelection: input.modelSelection,            })            .pipe(              Effect.map((result) => result.branch),              Effect.mapError(mapError(input, "generate-metadata", threadId)),            )        : (input.workspaceStrategy.branch ??          (input.workspaceStrategy.type === "worktree" ? fallbackBranchName(threadId) : null));    let worktreePath =      input.workspaceStrategy.type === "existing_worktree"        ? input.workspaceStrategy.worktreePath        : null;    if (input.workspaceStrategy.type === "worktree") {      if (runId !== null) {        yield* threads          .dispatch({            type: "prepared-run.progress",            commandId: CommandId.make(`${input.commandId}:progress:worktree`),            threadId,            runId,            phase: "worktree",          })          .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));      }      let startRef = input.workspaceStrategy.baseRef;      if (input.workspaceStrategy.startFromOrigin === true) {        yield* git          .fetchRemote({ cwd: project.workspaceRoot, remoteName: "origin" })          .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));        startRef = yield* git          .resolveRemoteTrackingCommit({            cwd: project.workspaceRoot,            refName: input.workspaceStrategy.baseRef,            fallbackRemoteName: "origin",          })          .pipe(            Effect.map((resolved) => resolved.commitSha),            Effect.mapError(mapError(input, "provision-worktree", threadId)),          );      }      const worktree = yield* git        .createWorktree({          cwd: project.workspaceRoot,          refName: startRef,          newRefName: branch!,          baseRefName: input.workspaceStrategy.baseRef,          path: null,        })        .pipe(Effect.mapError(mapError(input, "provision-worktree", threadId)));      worktreePath = worktree.worktree.path;      branch = worktree.worktree.refName;    }     yield* threads      .dispatch({        type: "thread.metadata.update",        commandId: CommandId.make(`${input.commandId}:workspace`),        threadId,        branch,        worktreePath,      })      .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));     const cwd = worktreePath ?? project.workspaceRoot;    if (runId !== null) {      yield* threads        .dispatch({          type: "prepared-run.progress",          commandId: CommandId.make(`${input.commandId}:progress:setup`),          threadId,          runId,          phase: "setup",        })        .pipe(Effect.mapError(mapError(input, "update-thread", threadId)));    }    yield* setupScripts      .runForThread({        threadId,        projectId: input.projectId,        projectCwd: project.workspaceRoot,        worktreePath: cwd,        project: {          workspaceRoot: project.workspaceRoot,          scripts: project.scripts,        },      })      .pipe(Effect.mapError(mapError(input, "run-setup-script", threadId)));     if (runId !== null) {      yield* threads        .dispatch({          type: "prepared-run.release",          commandId: CommandId.make(`${input.commandId}:release`),          threadId,          runId,        })        .pipe(Effect.mapError(mapError(input, "release-run", threadId)));    }  });   const failPreparedRun = (    input: ThreadLaunchInput,    threadId: ThreadId,    runId: RunId | null,    cause: unknown,  ) =>    runId === null      ? Effect.logWarning("Thread workspace preparation failed", {          commandId: input.commandId,          threadId,          cause,        })      : threads          .dispatch({            type: "prepared-run.fail",            commandId: CommandId.make(`${input.commandId}:fail`),            threadId,            runId,            failure: makeProviderFailure({              cause,              message: failureDetail(cause),              class: "validation_error",              retryable: false,            }),          })          .pipe(            Effect.mapError(mapError(input, "fail-run", threadId)),            Effect.catchCause((persistCause) =>              Effect.logWarning("Failed to persist thread workspace preparation failure", {                commandId: input.commandId,                threadId,                cause,                persistCause,              }),            ),          );   const reservePreparation = (commandId: CommandId) =>    Ref.modify(scheduledLaunches, (scheduled) => {      if (scheduled.has(commandId)) return [false, scheduled] as const;      const next = new Set(scheduled);      next.add(commandId);      return [true, next] as const;    });   const releasePreparation = (commandId: CommandId) =>    Ref.update(scheduledLaunches, (scheduled) => {      const next = new Set(scheduled);      next.delete(commandId);      return next;    });   const schedulePreparation = Effect.fn("ThreadLaunchService.schedulePreparation")(function* (    input: ThreadLaunchInput,    threadId: ThreadId,    runId: RunId | null,  ) {    yield* prepareInBackground(input, threadId, runId).pipe(      Effect.catchCause((cause) =>        Cause.hasInterruptsOnly(cause)          ? Effect.void          : failPreparedRun(input, threadId, runId, Cause.squash(cause)),      ),      Effect.ensuring(releasePreparation(input.commandId)),      Effect.forkIn(preparationScope),    );  });   const launch: ThreadLaunchService["Service"]["launch"] = Effect.fn("ThreadLaunchService.launch")(    function* (input) {      const project = yield* projects.getById(input.projectId).pipe(        Effect.mapError(mapError(input, "resolve-project")),        Effect.flatMap(          Option.match({            onNone: () => Effect.fail(mapError(input, "resolve-project")("Project not found.")),            onSome: Effect.succeed,          }),        ),      );      if (input.reuseExistingThread === true && input.threadId === undefined) {        return yield* mapError(          input,          "update-thread",        )("Reusing an existing thread requires a thread id.");      }       const launchReceipt = yield* readReceipt(input, input.commandId);      return yield* Effect.gen(function* () {        const candidateThreadId =          input.threadId ??          (yield* ids.allocate            .thread({ projectId: input.projectId })            .pipe(Effect.mapError(mapError(input, "create-thread"))));         if (input.reuseExistingThread === true && Option.isNone(launchReceipt)) {          yield* validateReusableThread(input, candidateThreadId);        }         const initialBranch = input.workspaceStrategy.branch ?? null;        const initialWorktreePath =          input.workspaceStrategy.type === "existing_worktree"            ? input.workspaceStrategy.worktreePath            : null;        const claimDispatch =          input.reuseExistingThread === true            ? threads.dispatch({                type: "thread.metadata.update",                commandId: input.commandId,                threadId: candidateThreadId,              })            : threads.dispatch({                type: "thread.create",                commandId: input.commandId,                threadId: candidateThreadId,                projectId: input.projectId,                title: input.title,                modelSelection: input.modelSelection,                runtimeMode: input.runtimeMode,                interactionMode: input.interactionMode,                branch: initialBranch,                worktreePath: initialWorktreePath,                createdBy: input.createdBy,                creationSource: input.creationSource,              });        const claimed = yield* claimDispatch.pipe(          Effect.mapError(            mapError(              input,              input.reuseExistingThread === true ? "update-thread" : "create-thread",              candidateThreadId,            ),          ),        );        const threadId =          claimed.storedEvents.find((stored) => stored.event.type.startsWith("thread."))?.event            .threadId ?? candidateThreadId;        if (project.id !== input.projectId) {          return yield* mapError(input, "resolve-project", threadId)("Project identity changed.");        }         let runId: RunId | null = null;        let messageWasAlreadyAccepted = false;        if (input.initialMessage !== undefined) {          const messageCommandId = CommandId.make(`${input.commandId}:initial-message`);          const messageReceipt = yield* readReceipt(input, messageCommandId);          messageWasAlreadyAccepted = Option.isSome(messageReceipt);          const messageId =            input.initialMessage.messageId ??            (yield* ids.allocate              .message({ threadId, ordinal: 1 })              .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId))));          const dispatched = yield* threads            .dispatch({              type: "message.dispatch",              commandId: messageCommandId,              threadId,              messageId,              text: input.initialMessage.text,              attachments: input.initialMessage.attachments,              modelSelection: input.modelSelection,              dispatchMode: { type: "defer_start" },              createdBy: input.createdBy,              creationSource: input.creationSource,            })            .pipe(Effect.mapError(mapError(input, "dispatch-message", threadId)));          const runCreated = dispatched.storedEvents.find(            (stored) => stored.event.type === "run.created",

Lines 1–500 of 553. Load more