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 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 { 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.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 DEFAULT_THREAD_TITLE = "New thread"; 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 canReplaceThreadTitle(currentTitle: string, titleSeed?: string): boolean { const trimmedCurrentTitle = currentTitle.trim(); if (trimmedCurrentTitle === DEFAULT_THREAD_TITLE) { return true; } const trimmedTitleSeed = titleSeed?.trim(); return trimmedTitleSeed !== undefined && trimmedTitleSeed.length > 0 ? trimmedCurrentTitle === trimmedTitleSeed : false;} 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 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)); }); 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) ?.requiresNewThreadForModelChange === true || providers.find((snapshot) => snapshot.instanceId === requestedModelSelection.instanceId) ?.requiresNewThreadForModelChange === true; if (!requiresNewThread) { return; } return yield* new ProviderAdapterRequestError({ provider: providerErrorLabelFromInstanceHint({ instanceId: String(requestedModelSelection.instanceId), modelSelectionInstanceId: String(input.currentModelSelection.instanceId), }), method: "thread.turn.start", detail: `Thread '${input.threadId}' cannot switch models after the conversation has started. Start a new thread to use '${requestedModelSelection.model}'.`, }); }); const ensureSessionForThread = Effect.fn("ensureSessionForThread")(function* ( threadId: ThreadId, createdAt: string, options?: { readonly modelSelection?: ModelSelection; readonly pendingTurnStart?: boolean; }, ) { const thread = yield* resolveThread(threadId); if (!thread) { return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`)); } const desiredRuntimeMode = thread.runtimeMode; const requestedModelSelection = options?.modelSelection; const resolveActiveSession = (threadId: ThreadId) => providerService .listSessions() .pipe(Effect.map((sessions) => sessions.find((session) => session.threadId === threadId))); const activeSession = yield* resolveActiveSession(threadId); const activeThreadSession = thread.session !== null && thread.session.status !== "stopped" && activeSession ? thread.session : null; if ( activeThreadSession !== null && activeSession !== undefined && (activeThreadSession.providerInstanceId === undefined || activeSession.providerInstanceId === undefined) ) { return yield* new ProviderAdapterRequestError({ provider: providerErrorLabel(activeThreadSession.providerName ?? undefined), method: "thread.turn.start", detail: `Thread '${threadId}' has an active provider session without a provider instance id.`, }); } const currentInstanceId = activeThreadSession !== null && activeSession !== undefined && activeSession.providerInstanceId !== undefined ? activeSession.providerInstanceId : thread.modelSelection.instanceId; const desiredModelSelection = requestedModelSelection ?? thread.modelSelection; const desiredInstanceId = desiredModelSelection.instanceId; const currentInfo = yield* providerService.getInstanceInfo(currentInstanceId).pipe( Effect.mapError( () => new ProviderAdapterRequestError({ provider: providerErrorLabelFromInstanceHint({ instanceId: String(currentInstanceId), modelSelectionInstanceId: String(thread.modelSelection.instanceId), sessionProvider: thread.session?.providerName ?? undefined, }), method: "thread.turn.start", detail: `Thread '${threadId}' references unknown provider instance '${currentInstanceId}'. The instance is not configured in this build.`, }), ), ); const desiredInfo = yield* providerService.getInstanceInfo(desiredInstanceId).pipe( Effect.mapError( () => new ProviderAdapterRequestError({ provider: providerErrorLabelFromInstanceHint({ instanceId: String(desiredModelSelection.instanceId), }), method: "thread.turn.start", detail: `Requested provider instance '${desiredInstanceId}' is not configured in this build.`, }), ), ); const desiredDriverKind = desiredInfo.driverKind; if (!isProviderDriverKind(desiredDriverKind)) { return yield* new ProviderAdapterRequestError({ provider: providerErrorLabel(String(desiredDriverKind)), method: "thread.turn.start", detail: `Requested provider instance '${desiredInstanceId}' uses unknown provider driver '${desiredDriverKind}'. The driver is not installed in this build.`, }); } const preferredProvider: ProviderDriverKind = desiredDriverKind; if (options?.pendingTurnStart === true && thread.session?.status !== "running") { yield* setThreadSession({ threadId, session: { threadId, status: "starting", providerName: activeSession?.provider ?? preferredProvider, providerInstanceId: activeSession?.providerInstanceId ?? desiredInstanceId, runtimeMode: desiredRuntimeMode, activeTurnId: null, lastError: null, updatedAt: createdAt, }, createdAt, }); } if (thread.session !== null) { yield* rejectStartedThreadModelChangeIfRequired({ threadId, currentModelSelection: activeSession?.model !== undefined ? { ...thread.modelSelection, instanceId: currentInstanceId, model: activeSession.model, } : thread.modelSelection, requestedModelSelection, }); } if ( thread.session !== null && requestedModelSelection !== undefined && requestedModelSelection.instanceId !== currentInstanceId ) { if (currentInfo.driverKind !== desiredInfo.driverKind) { return yield* new ProviderAdapterRequestError({ provider: preferredProvider, method: "thread.turn.start", detail: `Thread '${threadId}' is bound to driver '${currentInfo.driverKind}' and cannot switch to '${desiredInfo.driverKind}'.`, }); } if ( currentInfo.continuationIdentity.continuationKey !== desiredInfo.continuationIdentity.continuationKey ) { return yield* new ProviderAdapterRequestError({ provider: preferredProvider, method: "thread.turn.start", detail: `Thread '${threadId}' cannot switch from instance '${currentInstanceId}' to '${desiredInstanceId}' because their provider resume state is incompatible.`, }); } } const project = yield* resolveProject(thread.projectId); const effectiveCwd = resolveThreadWorkspaceCwd({ thread, projects: project ? [project] : [], }); const startProviderSession = (input?: { readonly resumeCursor?: unknown; readonly provider?: ProviderDriverKind; }) => providerService.startSession(threadId, {Lines 1–500 of 1112. Load more
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 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 { 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.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 DEFAULT_THREAD_TITLE = "New thread"; 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 canReplaceThreadTitle(currentTitle: string, titleSeed?: string): boolean { const trimmedCurrentTitle = currentTitle.trim(); if (trimmedCurrentTitle === DEFAULT_THREAD_TITLE) { return true; } const trimmedTitleSeed = titleSeed?.trim(); return trimmedTitleSeed !== undefined && trimmedTitleSeed.length > 0 ? trimmedCurrentTitle === trimmedTitleSeed : false; } 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 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)); }); 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) ?.requiresNewThreadForModelChange === true || providers.find((snapshot) => snapshot.instanceId === requestedModelSelection.instanceId) ?.requiresNewThreadForModelChange === true; if (!requiresNewThread) { return; } return yield* new ProviderAdapterRequestError({ provider: providerErrorLabelFromInstanceHint({ instanceId: String(requestedModelSelection.instanceId), modelSelectionInstanceId: String(input.currentModelSelection.instanceId), }), method: "thread.turn.start", detail: `Thread '${input.threadId}' cannot switch models after the conversation has started. Start a new thread to use '${requestedModelSelection.model}'.`, }); }); const ensureSessionForThread = Effect.fn("ensureSessionForThread")(function* ( threadId: ThreadId, createdAt: string, options?: { readonly modelSelection?: ModelSelection; readonly pendingTurnStart?: boolean; }, ) { const thread = yield* resolveThread(threadId); if (!thread) { return yield* Effect.die(new Error(`Thread '${threadId}' was not found in read model.`)); } const desiredRuntimeMode = thread.runtimeMode; const requestedModelSelection = options?.modelSelection; const resolveActiveSession = (threadId: ThreadId) => providerService .listSessions() .pipe(Effect.map((sessions) => sessions.find((session) => session.threadId === threadId))); const activeSession = yield* resolveActiveSession(threadId); const activeThreadSession = thread.session !== null && thread.session.status !== "stopped" && activeSession ? thread.session : null; if ( activeThreadSession !== null && activeSession !== undefined && (activeThreadSession.providerInstanceId === undefined || activeSession.providerInstanceId === undefined) ) { return yield* new ProviderAdapterRequestError({ provider: providerErrorLabel(activeThreadSession.providerName ?? undefined), method: "thread.turn.start", detail: `Thread '${threadId}' has an active provider session without a provider instance id.`, }); } const currentInstanceId = activeThreadSession !== null && activeSession !== undefined && activeSession.providerInstanceId !== undefined ? activeSession.providerInstanceId : thread.modelSelection.instanceId; const desiredModelSelection = requestedModelSelection ?? thread.modelSelection; const desiredInstanceId = desiredModelSelection.instanceId; const currentInfo = yield* providerService.getInstanceInfo(currentInstanceId).pipe( Effect.mapError( () => new ProviderAdapterRequestError({ provider: providerErrorLabelFromInstanceHint({ instanceId: String(currentInstanceId), modelSelectionInstanceId: String(thread.modelSelection.instanceId), sessionProvider: thread.session?.providerName ?? undefined, }), method: "thread.turn.start", detail: `Thread '${threadId}' references unknown provider instance '${currentInstanceId}'. The instance is not configured in this build.`, }), ), ); const desiredInfo = yield* providerService.getInstanceInfo(desiredInstanceId).pipe( Effect.mapError( () => new ProviderAdapterRequestError({ provider: providerErrorLabelFromInstanceHint({ instanceId: String(desiredModelSelection.instanceId), }), method: "thread.turn.start", detail: `Requested provider instance '${desiredInstanceId}' is not configured in this build.`, }), ), ); const desiredDriverKind = desiredInfo.driverKind; if (!isProviderDriverKind(desiredDriverKind)) { return yield* new ProviderAdapterRequestError({ provider: providerErrorLabel(String(desiredDriverKind)), method: "thread.turn.start", detail: `Requested provider instance '${desiredInstanceId}' uses unknown provider driver '${desiredDriverKind}'. The driver is not installed in this build.`, }); } const preferredProvider: ProviderDriverKind = desiredDriverKind; if (options?.pendingTurnStart === true && thread.session?.status !== "running") { yield* setThreadSession({ threadId, session: { threadId, status: "starting", providerName: activeSession?.provider ?? preferredProvider, providerInstanceId: activeSession?.providerInstanceId ?? desiredInstanceId, runtimeMode: desiredRuntimeMode, activeTurnId: null, lastError: null, updatedAt: createdAt, }, createdAt, }); } if (thread.session !== null) { yield* rejectStartedThreadModelChangeIfRequired({ threadId, currentModelSelection: activeSession?.model !== undefined ? { ...thread.modelSelection, instanceId: currentInstanceId, model: activeSession.model, } : thread.modelSelection, requestedModelSelection, }); } if ( thread.session !== null && requestedModelSelection !== undefined && requestedModelSelection.instanceId !== currentInstanceId ) { if (currentInfo.driverKind !== desiredInfo.driverKind) { return yield* new ProviderAdapterRequestError({ provider: preferredProvider, method: "thread.turn.start", detail: `Thread '${threadId}' is bound to driver '${currentInfo.driverKind}' and cannot switch to '${desiredInfo.driverKind}'.`, }); } if ( currentInfo.continuationIdentity.continuationKey !== desiredInfo.continuationIdentity.continuationKey ) { return yield* new ProviderAdapterRequestError({ provider: preferredProvider, method: "thread.turn.start", detail: `Thread '${threadId}' cannot switch from instance '${currentInstanceId}' to '${desiredInstanceId}' because their provider resume state is incompatible.`, }); } } const project = yield* resolveProject(thread.projectId); const effectiveCwd = resolveThreadWorkspaceCwd({ thread, projects: project ? [project] : [], }); const startProviderSession = (input?: { readonly resumeCursor?: unknown; readonly provider?: ProviderDriverKind; }) => providerService.startSession(threadId, {