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 { assistantCitationsToPlainText } from "@t3tools/shared/assistantCitations";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 DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";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 Schedule from "effect/Schedule";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, ProviderAdapterValidationError, ProviderWorkspaceMissingError,} from "../../provider/Errors.ts";import type { ProviderServiceError } from "../../provider/Errors.ts";import { TextGeneration } from "../../textGeneration/TextGeneration.ts";import { ProviderAuthService } from "../../provider/Services/ProviderAuthService.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 isProviderAdapterValidationError = Schema.is(ProviderAdapterValidationError);const isProviderWorkspaceMissingError = Schema.is(ProviderWorkspaceMissingError);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" | "thread.settled"; }>; function toNonEmptyProviderInput(value: string | undefined): string | undefined { const normalized = value?.trim(); return normalized && normalized.length > 0 ? normalized : undefined;} const isCompactCommandMessage = (message: ThreadTitleMessage): boolean => message.role === "user" && (message.attachments?.length ?? 0) === 0 && message.text.trim().toLowerCase() === "/compact";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 = assistantCitationsToPlainText(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)), ), ], };} 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") || detail.includes("unknown pending codex approval request") ); } const message = Cause.pretty(cause).toLowerCase(); return ( message.includes("unknown pending approval request") || message.includes("unknown pending permission request") || message.includes("unknown pending codex approval 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 providerAuthService = yield* ProviderAuthService; 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 compactingThreadIds = new Set<ThreadId>(); type QueuedTurnStart = Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>; // Turn starts received while a thread compacts, replayed in order once its session is restored. const turnsAfterCompaction = new Map<ThreadId, Array<QueuedTurnStart>>(); // Replay command id → the queued turn start it re-requests. `sent` settles once the replay's // provider send finishes, which is what lets the next queued turn follow it in order. const resumedTurnStarts = new Map< CommandId, { readonly event: QueuedTurnStart; readonly queued: Array<QueuedTurnStart>; readonly sent: Deferred.Deferred<void>; } >(); const stoppingThreadIds = new Set<ThreadId>(); 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 cancelTurnsAfterCompaction = Effect.fn("cancelTurnsAfterCompaction")(function* ( threadId: ThreadId, detail: string, ) { const queued = turnsAfterCompaction.get(threadId) ?? []; turnsAfterCompaction.delete(threadId); for (const event of queued) { yield* appendProviderFailureActivity({ threadId, kind: "provider.turn.start.failed", summary: "Queued message was not sent", detail, turnId: null, createdAt: DateTime.formatIso(yield* DateTime.now), requestId: event.payload.messageId, }).pipe(Effect.ignore({ log: true, message: "failed to report canceled queued message" })); } }); const resumeTurnsAfterCompaction = Effect.fn("resumeTurnsAfterCompaction")(function* ( threadId: ThreadId, ) { const queued = turnsAfterCompaction.get(threadId) ?? []; while (queued.length > 0 && turnsAfterCompaction.get(threadId) === queued) { const event = queued[0]!; const turnStart = yield* projectionSnapshotQuery.getTurnStartMessage({ threadId, messageId: event.payload.messageId, }); if (turnsAfterCompaction.get(threadId) !== queued) return; // In flight from here on: a cancellation reports it when the replay runs, not from the queue. queued.shift(); if (Option.isNone(turnStart)) continue; // Reissue the durable request after restoration clears compaction's // pending slot. Reusing the message id preserves a single user bubble. const commandId = yield* serverCommandId("after-compaction"); const sent = yield* Deferred.make<void>(); resumedTurnStarts.set(commandId, { event, queued, sent }); const { messageId, ...request } = event.payload; yield* orchestrationEngine .dispatch({ type: "thread.turn.start", commandId, ...request, message: { messageId, role: "user", text: turnStart.value.message.text, attachments: turnStart.value.message.attachments ?? [], }, }) .pipe( Effect.onError(() => Effect.sync(() => { resumedTurnStarts.delete(commandId); queued.unshift(event); }), ), ); yield* Deferred.await(sent); resumedTurnStarts.delete(commandId); } if (turnsAfterCompaction.get(threadId) === queued) turnsAfterCompaction.delete(threadId); }); const formatFailureDetail = (cause: Cause.Cause<unknown>): string => { const failReason = cause.reasons.find(Cause.isFailReason); if (isProviderAdapterRequestError(failReason?.error)) { return failReason.error.detail; } if (isProviderAdapterValidationError(failReason?.error)) { return failReason.error.issue; } if (isProviderWorkspaceMissingError(failReason?.error)) { return failReason.error.message; } 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, }), ), ); Lines 1–500 of 1972. 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 { assistantCitationsToPlainText } from "@t3tools/shared/assistantCitations"; 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 DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; 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 Schedule from "effect/Schedule"; 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, ProviderAdapterValidationError, ProviderWorkspaceMissingError, } from "../../provider/Errors.ts"; import type { ProviderServiceError } from "../../provider/Errors.ts"; import { TextGeneration } from "../../textGeneration/TextGeneration.ts"; import { ProviderAuthService } from "../../provider/Services/ProviderAuthService.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 isProviderAdapterValidationError = Schema.is(ProviderAdapterValidationError); const isProviderWorkspaceMissingError = Schema.is(ProviderWorkspaceMissingError); 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" | "thread.settled"; } >; function toNonEmptyProviderInput(value: string | undefined): string | undefined { const normalized = value?.trim(); return normalized && normalized.length > 0 ? normalized : undefined; } const isCompactCommandMessage = (message: ThreadTitleMessage): boolean => message.role === "user" && (message.attachments?.length ?? 0) === 0 && message.text.trim().toLowerCase() === "/compact"; 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 = assistantCitationsToPlainText(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)), ), ], }; } 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") || detail.includes("unknown pending codex approval request") ); } const message = Cause.pretty(cause).toLowerCase(); return ( message.includes("unknown pending approval request") || message.includes("unknown pending permission request") || message.includes("unknown pending codex approval 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 providerAuthService = yield* ProviderAuthService; 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 compactingThreadIds = new Set<ThreadId>(); type QueuedTurnStart = Extract<ProviderIntentEvent, { type: "thread.turn-start-requested" }>; // Turn starts received while a thread compacts, replayed in order once its session is restored. const turnsAfterCompaction = new Map<ThreadId, Array<QueuedTurnStart>>(); // Replay command id → the queued turn start it re-requests. `sent` settles once the replay's // provider send finishes, which is what lets the next queued turn follow it in order. const resumedTurnStarts = new Map< CommandId, { readonly event: QueuedTurnStart; readonly queued: Array<QueuedTurnStart>; readonly sent: Deferred.Deferred<void>; } >(); const stoppingThreadIds = new Set<ThreadId>(); 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 cancelTurnsAfterCompaction = Effect.fn("cancelTurnsAfterCompaction")(function* ( threadId: ThreadId, detail: string, ) { const queued = turnsAfterCompaction.get(threadId) ?? []; turnsAfterCompaction.delete(threadId); for (const event of queued) { yield* appendProviderFailureActivity({ threadId, kind: "provider.turn.start.failed", summary: "Queued message was not sent", detail, turnId: null, createdAt: DateTime.formatIso(yield* DateTime.now), requestId: event.payload.messageId, }).pipe(Effect.ignore({ log: true, message: "failed to report canceled queued message" })); } }); const resumeTurnsAfterCompaction = Effect.fn("resumeTurnsAfterCompaction")(function* ( threadId: ThreadId, ) { const queued = turnsAfterCompaction.get(threadId) ?? []; while (queued.length > 0 && turnsAfterCompaction.get(threadId) === queued) { const event = queued[0]!; const turnStart = yield* projectionSnapshotQuery.getTurnStartMessage({ threadId, messageId: event.payload.messageId, }); if (turnsAfterCompaction.get(threadId) !== queued) return; // In flight from here on: a cancellation reports it when the replay runs, not from the queue. queued.shift(); if (Option.isNone(turnStart)) continue; // Reissue the durable request after restoration clears compaction's // pending slot. Reusing the message id preserves a single user bubble. const commandId = yield* serverCommandId("after-compaction"); const sent = yield* Deferred.make<void>(); resumedTurnStarts.set(commandId, { event, queued, sent }); const { messageId, ...request } = event.payload; yield* orchestrationEngine .dispatch({ type: "thread.turn.start", commandId, ...request, message: { messageId, role: "user", text: turnStart.value.message.text, attachments: turnStart.value.message.attachments ?? [], }, }) .pipe( Effect.onError(() => Effect.sync(() => { resumedTurnStarts.delete(commandId); queued.unshift(event); }), ), ); yield* Deferred.await(sent); resumedTurnStarts.delete(commandId); } if (turnsAfterCompaction.get(threadId) === queued) turnsAfterCompaction.delete(threadId); }); const formatFailureDetail = (cause: Cause.Cause<unknown>): string => { const failReason = cause.reasons.find(Cause.isFailReason); if (isProviderAdapterRequestError(failReason?.error)) { return failReason.error.detail; } if (isProviderAdapterValidationError(failReason?.error)) { return failReason.error.issue; } if (isProviderWorkspaceMissingError(failReason?.error)) { return failReason.error.message; } 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, }), ), );