import {  type ChatAttachment,  CommandId,  DEFAULT_GIT_TEXT_GENERATION_MODEL,  EventId,  type ModelSelection,  type OrchestrationEvent,  type ProviderModelOptions,  ProviderKind,  type ProviderStartOptions,  type OrchestrationSession,  ThreadId,  type ProviderSession,  type RuntimeMode,  type TurnId,} from "@t3tools/contracts";import { Cache, Cause, Duration, Effect, Layer, Option, Schema, Stream } from "effect";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker";import { toProviderModelOptions } from "@t3tools/shared/model"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts";import { GitCore } from "../../git/Services/GitCore.ts";import { ProviderAdapterRequestError, ProviderServiceError } from "../../provider/Errors.ts";import { TextGeneration } from "../../git/Services/TextGeneration.ts";import { ProviderService } from "../../provider/Services/ProviderService.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import {  ProviderCommandReactor,  type ProviderCommandReactorShape,} from "../Services/ProviderCommandReactor.ts"; type ProviderIntentEvent = Extract<  OrchestrationEvent,  {    type:      | "thread.runtime-mode-set"      | "thread.turn-start-requested"      | "thread.turn-interrupt-requested"      | "thread.approval-response-requested"      | "thread.user-input-response-requested"      | "thread.session-stop-requested";  }>; function toNonEmptyProviderInput(value: string | undefined): string | undefined {  const normalized = value?.trim();  return normalized && normalized.length > 0 ? normalized : undefined;} function mapProviderSessionStatusToOrchestrationStatus(  status: "connecting" | "ready" | "running" | "error" | "closed",): OrchestrationSession["status"] {  switch (status) {    case "connecting":      return "starting";    case "running":      return "running";    case "error":      return "error";    case "closed":      return "stopped";    case "ready":    default:      return "ready";  }} const turnStartKeyForEvent = (event: ProviderIntentEvent): string =>  event.commandId !== null ? `command:${event.commandId}` : `event:${event.eventId}`; const serverCommandId = (tag: string): CommandId =>  CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`); const HANDLED_TURN_START_KEY_MAX = 10_000;const HANDLED_TURN_START_KEY_TTL = Duration.minutes(30);const DEFAULT_RUNTIME_MODE: RuntimeMode = "full-access";const WORKTREE_BRANCH_PREFIX = "t3code";const TEMP_WORKTREE_BRANCH_PATTERN = new RegExp(`^${WORKTREE_BRANCH_PREFIX}\\/[0-9a-f]{8}$`); const sameModelOptions = (  left: ProviderModelOptions | undefined,  right: ProviderModelOptions | undefined,): boolean => JSON.stringify(left ?? null) === JSON.stringify(right ?? null); function isUnknownPendingApprovalRequestError(cause: Cause.Cause<ProviderServiceError>): boolean {  const error = Cause.squash(cause);  if (Schema.is(ProviderAdapterRequestError)(error)) {    const detail = error.detail.toLowerCase();    return (      detail.includes("unknown pending approval request") ||      detail.includes("unknown pending permission request")    );  }  const message = Cause.pretty(cause);  return (    message.includes("unknown pending approval request") ||    message.includes("unknown pending permission request")  );} function isUnknownPendingUserInputRequestError(cause: Cause.Cause<ProviderServiceError>): boolean {  const error = Cause.squash(cause);  if (Schema.is(ProviderAdapterRequestError)(error)) {    return error.detail.toLowerCase().includes("unknown pending user-input request");  }  return Cause.pretty(cause).toLowerCase().includes("unknown pending user-input request");} function stalePendingRequestDetail(  requestKind: "approval" | "user-input",  requestId: string,): string {  return `Stale pending ${requestKind} request: ${requestId}. Provider callback state does not survive app restarts or recovered sessions. Restart the turn to continue.`;} function isTemporaryWorktreeBranch(branch: string): boolean {  return TEMP_WORKTREE_BRANCH_PATTERN.test(branch.trim().toLowerCase());} function buildGeneratedWorktreeBranchName(raw: string): string {  const normalized = raw    .trim()    .toLowerCase()    .replace(/^refs\/heads\//, "")    .replace(/['"`]/g, "");   const withoutPrefix = normalized.startsWith(`${WORKTREE_BRANCH_PREFIX}/`)    ? normalized.slice(`${WORKTREE_BRANCH_PREFIX}/`.length)    : normalized;   const branchFragment = withoutPrefix    .replace(/[^a-z0-9/_-]+/g, "-")    .replace(/\/+/g, "/")    .replace(/-+/g, "-")    .replace(/^[./_-]+|[./_-]+$/g, "")    .slice(0, 64)    .replace(/[./_-]+$/g, "");   const safeFragment = branchFragment.length > 0 ? branchFragment : "update";  return `${WORKTREE_BRANCH_PREFIX}/${safeFragment}`;} const make = Effect.gen(function* () {  const orchestrationEngine = yield* OrchestrationEngineService;  const providerService = yield* ProviderService;  const git = yield* GitCore;  const textGeneration = yield* TextGeneration;  const handledTurnStartKeys = yield* Cache.make<string, true>({    capacity: HANDLED_TURN_START_KEY_MAX,    timeToLive: HANDLED_TURN_START_KEY_TTL,    lookup: () => Effect.succeed(true),  });   const hasHandledTurnStartRecently = (key: string) =>    Cache.getOption(handledTurnStartKeys, key).pipe(      Effect.flatMap((cached) =>        Cache.set(handledTurnStartKeys, key, true).pipe(Effect.as(Option.isSome(cached))),      ),    );   const threadProviderOptions = new Map<string, ProviderStartOptions>();  const threadModelOptions = new Map<string, ProviderModelOptions>();   const appendProviderFailureActivity = (input: {    readonly threadId: ThreadId;    readonly kind:      | "provider.turn.start.failed"      | "provider.turn.interrupt.failed"      | "provider.approval.respond.failed"      | "provider.user-input.respond.failed"      | "provider.session.stop.failed";    readonly summary: string;    readonly detail: string;    readonly turnId: TurnId | null;    readonly createdAt: string;    readonly requestId?: string;  }) =>    orchestrationEngine.dispatch({      type: "thread.activity.append",      commandId: serverCommandId("provider-failure-activity"),      threadId: input.threadId,      activity: {        id: EventId.makeUnsafe(crypto.randomUUID()),        tone: "error",        kind: input.kind,        summary: input.summary,        payload: {          detail: input.detail,          ...(input.requestId ? { requestId: input.requestId } : {}),        },        turnId: input.turnId,        createdAt: input.createdAt,      },      createdAt: input.createdAt,    });   const setThreadSession = (input: {    readonly threadId: ThreadId;    readonly session: OrchestrationSession;    readonly createdAt: string;  }) =>    orchestrationEngine.dispatch({      type: "thread.session.set",      commandId: serverCommandId("provider-session-set"),      threadId: input.threadId,      session: input.session,      createdAt: input.createdAt,    });   const resolveThread = Effect.fnUntraced(function* (threadId: ThreadId) {    const readModel = yield* orchestrationEngine.getReadModel();    return readModel.threads.find((entry) => entry.id === threadId);  });   const ensureSessionForThread = Effect.fnUntraced(function* (    threadId: ThreadId,    createdAt: string,    options?: {      readonly modelSelection?: ModelSelection;      readonly modelOptions?: ProviderModelOptions;      readonly providerOptions?: ProviderStartOptions;    },  ) {    const readModel = yield* orchestrationEngine.getReadModel();    const thread = readModel.threads.find((entry) => entry.id === threadId);    if (!thread) {      return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`));    }     const desiredRuntimeMode = thread.runtimeMode;    const currentProvider: ProviderKind | undefined = Schema.is(ProviderKind)(      thread.session?.providerName,    )      ? thread.session.providerName      : undefined;    const requestedModelSelection = options?.modelSelection;    const threadProvider: ProviderKind = currentProvider ?? thread.modelSelection.provider;    if (      requestedModelSelection !== undefined &&      requestedModelSelection.provider !== threadProvider    ) {      return yield* new ProviderAdapterRequestError({        provider: threadProvider,        method: "thread.turn.start",        detail: `Thread '${threadId}' is bound to provider '${threadProvider}' and cannot switch to '${requestedModelSelection.provider}'.`,      });    }    const preferredProvider: ProviderKind = currentProvider ?? threadProvider;    const desiredModelSelection = requestedModelSelection ?? thread.modelSelection;    const desiredModel = desiredModelSelection.model;    const desiredModelOptions =      options?.modelOptions ?? toProviderModelOptions(desiredModelSelection);    const effectiveCwd = resolveThreadWorkspaceCwd({      thread,      projects: readModel.projects,    });     const resolveActiveSession = (threadId: ThreadId) =>      providerService        .listSessions()        .pipe(Effect.map((sessions) => sessions.find((session) => session.threadId === threadId)));     const startProviderSession = (input?: { readonly resumeCursor?: unknown }) =>      providerService.startSession(threadId, {        threadId,        ...(preferredProvider ? { provider: preferredProvider } : {}),        ...(effectiveCwd ? { cwd: effectiveCwd } : {}),        ...(desiredModel ? { model: desiredModel } : {}),        ...(desiredModelOptions !== undefined ? { modelOptions: desiredModelOptions } : {}),        ...(options?.providerOptions !== undefined          ? { providerOptions: options.providerOptions }          : {}),        ...(input?.resumeCursor !== undefined ? { resumeCursor: input.resumeCursor } : {}),        runtimeMode: desiredRuntimeMode,      });     const bindSessionToThread = (session: ProviderSession) =>      setThreadSession({        threadId,        session: {          threadId,          status: mapProviderSessionStatusToOrchestrationStatus(session.status),          providerName: session.provider,          runtimeMode: desiredRuntimeMode,          // Provider turn ids are not orchestration turn ids.          activeTurnId: null,          lastError: session.lastError ?? null,          updatedAt: session.updatedAt,        },        createdAt,      });     const existingSessionThreadId =      thread.session && thread.session.status !== "stopped" ? thread.id : null;    if (existingSessionThreadId) {      const runtimeModeChanged = thread.runtimeMode !== thread.session?.runtimeMode;      const providerChanged =        requestedModelSelection !== undefined &&        requestedModelSelection.provider !== currentProvider;      const activeSession = yield* resolveActiveSession(existingSessionThreadId);      const sessionModelSwitch =        currentProvider === undefined          ? "in-session"          : (yield* providerService.getCapabilities(currentProvider)).sessionModelSwitch;      const modelChanged =        requestedModelSelection !== undefined &&        requestedModelSelection.model !== activeSession?.model;      const shouldRestartForModelChange = modelChanged && sessionModelSwitch === "restart-session";      const previousModelOptions = threadModelOptions.get(threadId);      const shouldRestartForModelOptionsChange =        currentProvider === "claudeAgent" &&        options?.modelOptions !== undefined &&        !sameModelOptions(previousModelOptions, options.modelOptions);       if (        !runtimeModeChanged &&        !providerChanged &&        !shouldRestartForModelChange &&        !shouldRestartForModelOptionsChange      ) {        return existingSessionThreadId;      }       const resumeCursor =        providerChanged || shouldRestartForModelChange          ? undefined          : (activeSession?.resumeCursor ?? undefined);      yield* Effect.logInfo("provider command reactor restarting provider session", {        threadId,        existingSessionThreadId,        currentProvider,        desiredProvider: desiredModelSelection.provider,        currentRuntimeMode: thread.session?.runtimeMode,        desiredRuntimeMode: thread.runtimeMode,        runtimeModeChanged,        providerChanged,        modelChanged,        shouldRestartForModelChange,        shouldRestartForModelOptionsChange,        hasResumeCursor: resumeCursor !== undefined,      });      const restartedSession = yield* startProviderSession(        resumeCursor !== undefined ? { resumeCursor } : undefined,      );      yield* Effect.logInfo("provider command reactor restarted provider session", {        threadId,        previousSessionId: existingSessionThreadId,        restartedSessionThreadId: restartedSession.threadId,        provider: restartedSession.provider,        runtimeMode: restartedSession.runtimeMode,      });      yield* bindSessionToThread(restartedSession);      return restartedSession.threadId;    }     const startedSession = yield* startProviderSession(undefined);    yield* bindSessionToThread(startedSession);    return startedSession.threadId;  });   const sendTurnForThread = Effect.fnUntraced(function* (input: {    readonly threadId: ThreadId;    readonly messageText: string;    readonly attachments?: ReadonlyArray<ChatAttachment>;    readonly modelSelection?: ModelSelection;    readonly modelOptions?: ProviderModelOptions;    readonly providerOptions?: ProviderStartOptions;    readonly interactionMode?: "default" | "plan";    readonly createdAt: string;  }) {    const thread = yield* resolveThread(input.threadId);    if (!thread) {      return;    }    yield* ensureSessionForThread(input.threadId, input.createdAt, {      ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}),      ...(input.modelOptions !== undefined ? { modelOptions: input.modelOptions } : {}),      ...(input.providerOptions !== undefined ? { providerOptions: input.providerOptions } : {}),    });    if (input.providerOptions !== undefined) {      threadProviderOptions.set(input.threadId, input.providerOptions);    }    if (input.modelOptions !== undefined) {      threadModelOptions.set(input.threadId, input.modelOptions);    }    const normalizedInput = toNonEmptyProviderInput(input.messageText);    const normalizedAttachments = input.attachments ?? [];    const activeSession = yield* providerService      .listSessions()      .pipe(        Effect.map((sessions) => sessions.find((session) => session.threadId === input.threadId)),      );    const sessionModelSwitch =      activeSession === undefined        ? "in-session"        : (yield* providerService.getCapabilities(activeSession.provider)).sessionModelSwitch;    const modelForTurn =      sessionModelSwitch === "unsupported" ? activeSession?.model : input.modelSelection?.model;     yield* providerService.sendTurn({      threadId: input.threadId,      ...(normalizedInput ? { input: normalizedInput } : {}),      ...(normalizedAttachments.length > 0 ? { attachments: normalizedAttachments } : {}),      ...(modelForTurn !== undefined ? { model: modelForTurn } : {}),      ...(input.modelOptions !== undefined ? { modelOptions: input.modelOptions } : {}),      ...(input.interactionMode !== undefined ? { interactionMode: input.interactionMode } : {}),    });  });   const maybeGenerateAndRenameWorktreeBranchForFirstTurn = Effect.fnUntraced(function* (input: {    readonly threadId: ThreadId;    readonly branch: string | null;    readonly worktreePath: string | null;    readonly messageId: string;    readonly messageText: string;    readonly attachments?: ReadonlyArray<ChatAttachment>;  }) {    if (!input.branch || !input.worktreePath) {      return;    }    if (!isTemporaryWorktreeBranch(input.branch)) {      return;    }     const thread = yield* resolveThread(input.threadId);    if (!thread) {      return;    }     const userMessages = thread.messages.filter((message) => message.role === "user");    if (userMessages.length !== 1 || userMessages[0]?.id !== input.messageId) {      return;    }     const oldBranch = input.branch;    const cwd = input.worktreePath;    const attachments = input.attachments ?? [];    yield* textGeneration      .generateBranchName({        cwd,        message: input.messageText,        ...(attachments.length > 0 ? { attachments } : {}),        model: DEFAULT_GIT_TEXT_GENERATION_MODEL,      })      .pipe(        Effect.catch((error) =>          Effect.logWarning(            "provider command reactor failed to generate worktree branch name; skipping rename",            { threadId: input.threadId, cwd, oldBranch, reason: error.message },          ),        ),        Effect.flatMap((generated) => {          if (!generated) return Effect.void;           const targetBranch = buildGeneratedWorktreeBranchName(generated.branch);          if (targetBranch === oldBranch) return Effect.void;           return Effect.flatMap(            git.renameBranch({ cwd, oldBranch, newBranch: targetBranch }),            (renamed) =>              orchestrationEngine.dispatch({                type: "thread.meta.update",                commandId: serverCommandId("worktree-branch-rename"),                threadId: input.threadId,                branch: renamed.branch,                worktreePath: cwd,              }),          );        }),        Effect.catchCause((cause) =>          Effect.logWarning(            "provider command reactor failed to generate or rename worktree branch",            { threadId: input.threadId, cwd, oldBranch, cause: Cause.pretty(cause) },          ),        ),      );  });   const processTurnStartRequested = Effect.fnUntraced(function* (    event: Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>,  ) {    const key = turnStartKeyForEvent(event);    if (yield* hasHandledTurnStartRecently(key)) {      return;    }     const thread = yield* resolveThread(event.payload.threadId);    if (!thread) {      return;    }     const message = thread.messages.find((entry) => entry.id === event.payload.messageId);    if (!message || message.role !== "user") {      yield* appendProviderFailureActivity({        threadId: event.payload.threadId,        kind: "provider.turn.start.failed",        summary: "Provider turn start failed",        detail: `User message '${event.payload.messageId}' was not found for turn start request.`,        turnId: null,        createdAt: event.payload.createdAt,

Lines 1–500 of 760. Load more