import {  type ChatAttachment,  CommandId,  EventId,  type ModelSelection,  type OrchestrationEvent,  ProviderDriverKind,  type ProjectId,  type OrchestrationSession,  ThreadId,  type ProviderSession,  type RuntimeMode,  type TurnId,} from "@t3tools/contracts";import { isTemporaryWorktreeBranch, WORKTREE_BRANCH_PREFIX } from "@t3tools/shared/git";import * as Cache from "effect/Cache";import * as Cause from "effect/Cause";import * as Crypto from "effect/Crypto";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Equal from "effect/Equal";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Schema from "effect/Schema";import * as Stream from "effect/Stream";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts";import { increment, orchestrationEventsProcessedTotal } from "../../observability/Metrics.ts";import { ProviderAdapterRequestError } from "../../provider/Errors.ts";import type { ProviderServiceError } from "../../provider/Errors.ts";import { TextGeneration } from "../../textGeneration/TextGeneration.ts";import { ProviderService } from "../../provider/Services/ProviderService.ts";import { ProviderRegistry } from "../../provider/Services/ProviderRegistry.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import {  ProviderCommandReactor,  type ProviderCommandReactorShape,} from "../Services/ProviderCommandReactor.ts";import { forkParked, ServerActivation } from "../../serverActivation.ts";import { canReplaceThreadTitle, DEFAULT_THREAD_TITLE } from "../threadTitles.ts";import {  resolveSourceControlWriterModelSelection,  ServerSettingsService,} from "../../serverSettings.ts";import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";import { GitWorkflowService } from "../../git/GitWorkflowService.ts";const isProviderAdapterRequestError = Schema.is(ProviderAdapterRequestError);const isProviderDriverKind = Schema.is(ProviderDriverKind); type ProviderIntentEvent = Extract<  OrchestrationEvent,  {    type:      | "thread.meta-updated"      | "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 HANDLED_TURN_START_KEY_MAX = 10_000;const HANDLED_TURN_START_KEY_TTL = Duration.minutes(30);const DEFAULT_RUNTIME_MODE: RuntimeMode = "full-access";const MAX_REGENERATION_ATTACHMENTS = 4;const MAX_THREAD_TITLE_CONTEXT_CHARS = 8_000;const MAX_FIRST_USER_TITLE_CONTEXT_CHARS = 2_000;const THREAD_TITLE_CONTEXT_TRUNCATION_MARKER = "[Earlier content truncated]\n\n";const FIRST_USER_CONTEXT_TRUNCATION_MARKER = "\n[First user message truncated]"; type ThreadTitleMessage = {  readonly role: "user" | "assistant" | "system";  readonly text: string;  readonly attachments?: ReadonlyArray<ChatAttachment> | undefined;}; function formatThreadTitleSection(message: ThreadTitleMessage): string | undefined {  if (message.role === "system") {    return undefined;  }  const text = message.text.trim();  const attachmentSummary = (message.attachments ?? [])    .map((attachment) => attachment.name)    .join(", ");  const contents = [    ...(text.length > 0 ? [text] : []),    ...(attachmentSummary.length > 0 ? [`[Attachments: ${attachmentSummary}]`] : []),  ].join("\n");  return contents.length > 0 ? `${message.role.toUpperCase()}:\n${contents}` : undefined;} function limitFirstUserSection(section: string): string {  if (section.length <= MAX_FIRST_USER_TITLE_CONTEXT_CHARS) {    return section;  }  return `${section.slice(    0,    MAX_FIRST_USER_TITLE_CONTEXT_CHARS - FIRST_USER_CONTEXT_TRUNCATION_MARKER.length,  )}${FIRST_USER_CONTEXT_TRUNCATION_MARKER}`;} function collectRecentThreadTitleContext(  messages: ReadonlyArray<ThreadTitleMessage>,  maxChars: number,): {  readonly context: string;  readonly attachments: ReadonlyArray<ChatAttachment>;  readonly truncated: boolean;} {  let context = "";  let truncated = false;  const retainedAttachments: Array<ChatAttachment> = [];   for (const message of messages.toReversed()) {    const section = formatThreadTitleSection(message);    if (section === undefined) {      continue;    }     const separator = context.length > 0 ? "\n\n" : "";    const available = maxChars - context.length - separator.length;    if (section.length > available) {      if (available > 0) {        context = `${section.slice(-available)}${separator}${context}`;        retainedAttachments.unshift(...(message.attachments ?? []));      }      truncated = true;      break;    }    context = `${section}${separator}${context}`;    retainedAttachments.unshift(...(message.attachments ?? []));  }   return { context, attachments: retainedAttachments, truncated };} function formatThreadTitleContext(messages: ReadonlyArray<ThreadTitleMessage>): {  readonly message: string;  readonly attachments: ReadonlyArray<ChatAttachment>;} {  const recent = collectRecentThreadTitleContext(messages, MAX_THREAD_TITLE_CONTEXT_CHARS);  if (!recent.truncated) {    return {      message: recent.context,      attachments: recent.attachments.slice(-MAX_REGENERATION_ATTACHMENTS),    };  }   const firstUserMessage = messages.find(    (message) => message.role === "user" && formatThreadTitleSection(message),  );  const firstUserSection = firstUserMessage    ? formatThreadTitleSection(firstUserMessage)    : undefined;  if (!firstUserMessage || !firstUserSection) {    return {      message: `${THREAD_TITLE_CONTEXT_TRUNCATION_MARKER}${recent.context}`,      attachments: recent.attachments.slice(-MAX_REGENERATION_ATTACHMENTS),    };  }   const pinnedSection = limitFirstUserSection(firstUserSection);  const recentContextBudget =    MAX_THREAD_TITLE_CONTEXT_CHARS -    pinnedSection.length -    "\n\n".length -    THREAD_TITLE_CONTEXT_TRUNCATION_MARKER.length;  const retainedRecent = collectRecentThreadTitleContext(messages, recentContextBudget);  const pinnedAttachment = firstUserMessage.attachments?.[0];  const recentAttachments = retainedRecent.attachments.filter(    (attachment) => attachment.id !== pinnedAttachment?.id,  );   return {    message: `${pinnedSection}\n\n${THREAD_TITLE_CONTEXT_TRUNCATION_MARKER}${retainedRecent.context}`,    attachments: [      ...(pinnedAttachment ? [pinnedAttachment] : []),      ...recentAttachments.slice(        -(MAX_REGENERATION_ATTACHMENTS - (pinnedAttachment === undefined ? 0 : 1)),      ),    ],  };} export function providerErrorLabel(value: string | undefined): string {  const normalized = value?.trim();  return normalized && normalized.length > 0 ? normalized : "unknown";} export function providerErrorLabelFromInstanceHint(input: {  readonly instanceId?: string | undefined;  readonly modelSelectionInstanceId?: string | undefined;  readonly sessionProvider?: string | undefined;}): string {  return providerErrorLabel(    input.instanceId ?? input.modelSelectionInstanceId ?? input.sessionProvider,  );} function findProviderAdapterRequestError(  cause: Cause.Cause<ProviderServiceError>,): ProviderAdapterRequestError | undefined {  const failReason = cause.reasons.find(Cause.isFailReason);  return isProviderAdapterRequestError(failReason?.error) ? failReason.error : undefined;} function isUnknownPendingApprovalRequestError(cause: Cause.Cause<ProviderServiceError>): boolean {  const error = findProviderAdapterRequestError(cause);  if (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 = findProviderAdapterRequestError(cause);  if (error) {    const detail = error.detail.toLowerCase();    return (      detail.includes("unknown pending user-input request") ||      detail.includes("unknown pending user input request") ||      detail.includes("unknown pending codex user input request")    );  }  const message = Cause.pretty(cause).toLowerCase();  return (    message.includes("unknown pending user-input request") ||    message.includes("unknown pending user input request") ||    message.includes("unknown pending codex 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 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 crypto = yield* Crypto.Crypto;  const orchestrationEngine = yield* OrchestrationEngineService;  const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;  const providerService = yield* ProviderService;  const providerRegistry = yield* ProviderRegistry;  const gitWorkflow = yield* GitWorkflowService;  const fileSystem = yield* FileSystem.FileSystem;  const vcsStatusBroadcaster = yield* VcsStatusBroadcaster;  const textGeneration = yield* TextGeneration;  const serverSettingsService = yield* ServerSettingsService;  const serverCommandId = (tag: string) =>    crypto.randomUUIDv4.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`)));  const serverEventId = () => crypto.randomUUIDv4.pipe(Effect.map(EventId.make));  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 threadModelSelections = new Map<string, ModelSelection>();   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;  }) =>    Effect.all({      commandId: serverCommandId("provider-failure-activity"),      eventId: serverEventId(),    }).pipe(      Effect.flatMap(({ commandId, eventId }) =>        orchestrationEngine.dispatch({          type: "thread.activity.append",          commandId,          threadId: input.threadId,          activity: {            id: eventId,            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 formatFailureDetail = (cause: Cause.Cause<unknown>): string => {    const failReason = cause.reasons.find(Cause.isFailReason);    const providerError = isProviderAdapterRequestError(failReason?.error)      ? failReason.error      : undefined;    if (providerError) {      return providerError.detail;    }    return Cause.pretty(cause);  };   const setThreadSession = (input: {    readonly threadId: ThreadId;    readonly session: OrchestrationSession;    readonly createdAt: string;  }) =>    serverCommandId("provider-session-set").pipe(      Effect.flatMap((commandId) =>        orchestrationEngine.dispatch({          type: "thread.session.set",          commandId,          threadId: input.threadId,          session: input.session,          createdAt: input.createdAt,        }),      ),    );   const setThreadSessionErrorOnTurnStartFailure = Effect.fnUntraced(function* (input: {    readonly threadId: ThreadId;    readonly detail: string;    readonly createdAt: string;  }) {    const thread = yield* resolveThread(input.threadId);    if (!thread) {      return;    }    const session = thread.session;    yield* setThreadSession({      threadId: input.threadId,      session: {        ...(session ?? {          threadId: input.threadId,          providerName: null,          providerInstanceId: thread.modelSelection.instanceId,          runtimeMode: thread.runtimeMode,        }),        status: session?.status === "stopped" ? "stopped" : "error",        activeTurnId: null,        lastError: input.detail,        updatedAt: input.createdAt,      },      createdAt: input.createdAt,    });  });   const resolveProject = Effect.fnUntraced(function* (projectId: ProjectId) {    return yield* projectionSnapshotQuery      .getProjectShellById(projectId)      .pipe(Effect.map(Option.getOrUndefined));  });   /**   * Recreates a thread's worktree from its branch when the directory has   * disappeared. Provider sessions resume into the persisted cwd, so a missing   * worktree makes every later turn fail as a bogus "session not found".   * Best-effort: on failure the turn proceeds and reports the real error.   */  const ensureThreadWorktree = Effect.fnUntraced(function* (thread: {    readonly id: ThreadId;    readonly projectId: ProjectId;    readonly branch: string | null;    readonly worktreePath: string | null;  }) {    const { worktreePath, branch } = thread;    if (!worktreePath || !branch) {      return;    }    const exists = yield* fileSystem.exists(worktreePath).pipe(Effect.orElseSucceed(() => true));    if (exists) {      return;    }    const project = yield* resolveProject(thread.projectId);    if (!project) {      return;    }    const cwd = project.workspaceRoot;    yield* Effect.logWarning("provider command reactor recreating missing worktree", {      threadId: thread.id,      worktreePath,      branch,    });    // A directory deleted without `git worktree remove` leaves an admin entry    // that makes `git worktree add` refuse the path; prune clears it.    yield* gitWorkflow.pruneWorktrees({ cwd }).pipe(      Effect.andThen(gitWorkflow.createWorktree({ cwd, refName: branch, path: worktreePath })),      Effect.catchCause((cause) =>        Cause.hasInterruptsOnly(cause)          ? Effect.failCause(cause)          : Effect.logWarning("provider command reactor failed to recreate worktree", {              threadId: thread.id,              worktreePath,              cause: Cause.pretty(cause),            }),      ),    );  });   const resolveThread = Effect.fnUntraced(function* (threadId: ThreadId) {    return yield* projectionSnapshotQuery      .getThreadDetailById(threadId)      .pipe(Effect.map(Option.getOrUndefined));  });   const rejectStartedThreadModelChangeIfRequired = Effect.fnUntraced(function* (input: {    readonly threadId: ThreadId;    readonly currentModelSelection: ModelSelection;    readonly requestedModelSelection: ModelSelection | undefined;  }) {    const requestedModelSelection = input.requestedModelSelection;    if (      requestedModelSelection === undefined ||      (input.currentModelSelection.instanceId === requestedModelSelection.instanceId &&        input.currentModelSelection.model === requestedModelSelection.model)    ) {      return;    }    const providers = yield* providerRegistry.getProviders;    const requiresNewThread =      providers.find((snapshot) => snapshot.instanceId === input.currentModelSelection.instanceId)

Lines 1–500 of 1564. Load more