import {  ApprovalRequestId,  type ChatAttachment,  type OrchestrationEvent,  type OrchestrationSessionStatus,  ThreadId,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as FileSystem from "effect/FileSystem";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as Stream from "effect/Stream";import * as SqlClient from "effect/unstable/sql/SqlClient"; import { toPersistenceSqlError, type ProjectionRepositoryError } from "../../persistence/Errors.ts";import { OrchestrationEventStore } from "../../persistence/Services/OrchestrationEventStore.ts";import { ProjectionPendingApprovalRepository } from "../../persistence/Services/ProjectionPendingApprovals.ts";import { ProjectionProjectRepository } from "../../persistence/Services/ProjectionProjects.ts";import { ProjectionStateRepository } from "../../persistence/Services/ProjectionState.ts";import { ProjectionThreadActivityRepository } from "../../persistence/Services/ProjectionThreadActivities.ts";import { type ProjectionThreadActivity } from "../../persistence/Services/ProjectionThreadActivities.ts";import {  type ProjectionThreadMessage,  ProjectionThreadMessageRepository,} from "../../persistence/Services/ProjectionThreadMessages.ts";import {  type ProjectionThreadProposedPlan,  ProjectionThreadProposedPlanRepository,} from "../../persistence/Services/ProjectionThreadProposedPlans.ts";import { ProjectionThreadSessionRepository } from "../../persistence/Services/ProjectionThreadSessions.ts";import {  type ProjectionTurn,  ProjectionTurnRepository,} from "../../persistence/Services/ProjectionTurns.ts";import { ProjectionThreadRepository } from "../../persistence/Services/ProjectionThreads.ts";import { ProjectionPendingApprovalRepositoryLive } from "../../persistence/Layers/ProjectionPendingApprovals.ts";import { ProjectionProjectRepositoryLive } from "../../persistence/Layers/ProjectionProjects.ts";import { ProjectionStateRepositoryLive } from "../../persistence/Layers/ProjectionState.ts";import { ProjectionThreadActivityRepositoryLive } from "../../persistence/Layers/ProjectionThreadActivities.ts";import { ProjectionThreadMessageRepositoryLive } from "../../persistence/Layers/ProjectionThreadMessages.ts";import { ProjectionThreadProposedPlanRepositoryLive } from "../../persistence/Layers/ProjectionThreadProposedPlans.ts";import { ProjectionThreadSessionRepositoryLive } from "../../persistence/Layers/ProjectionThreadSessions.ts";import { ProjectionTurnRepositoryLive } from "../../persistence/Layers/ProjectionTurns.ts";import { ProjectionThreadRepositoryLive } from "../../persistence/Layers/ProjectionThreads.ts";import { ServerConfig } from "../../config.ts";import {  OrchestrationProjectionPipeline,  type OrchestrationProjectionPipelineShape,} from "../Services/ProjectionPipeline.ts";import {  attachmentRelativePath,  parseAttachmentIdFromRelativePath,  parseThreadSegmentFromAttachmentId,  toSafeThreadAttachmentSegment,} from "../../attachmentStore.ts"; export const ORCHESTRATION_PROJECTOR_NAMES = {  projects: "projection.projects",  threads: "projection.threads",  threadMessages: "projection.thread-messages",  threadProposedPlans: "projection.thread-proposed-plans",  threadActivities: "projection.thread-activities",  threadSessions: "projection.thread-sessions",  threadTurns: "projection.thread-turns",  checkpoints: "projection.checkpoints",  pendingApprovals: "projection.pending-approvals",} as const; type ProjectorName =  (typeof ORCHESTRATION_PROJECTOR_NAMES)[keyof typeof ORCHESTRATION_PROJECTOR_NAMES]; /** * Turn state to settle still-running turns with when their session leaves the * "running" status, or null while the session is (re)starting or running and * turns must stay unsettled. */function settledTurnStateForSessionStatus(  status: OrchestrationSessionStatus,): "completed" | "interrupted" | "error" | null {  switch (status) {    case "idle":    case "ready":      return "completed";    case "error":      return "error";    case "interrupted":    case "stopped":      return "interrupted";    case "starting":    case "running":      return null;  }} interface ProjectorDefinition {  readonly name: ProjectorName;  readonly apply: (    event: OrchestrationEvent,    attachmentSideEffects: AttachmentSideEffects,  ) => Effect.Effect<void, ProjectionRepositoryError>;} interface AttachmentSideEffects {  readonly deletedThreadIds: Set<string>;  readonly prunedThreadRelativePaths: Map<string, Set<string>>;} const materializeAttachmentsForProjection = Effect.fn("materializeAttachmentsForProjection")(  (input: { readonly attachments: ReadonlyArray<ChatAttachment> }) =>    Effect.succeed(input.attachments.length === 0 ? [] : input.attachments),); function extractActivityRequestId(payload: unknown): ApprovalRequestId | null {  if (typeof payload !== "object" || payload === null) {    return null;  }  const requestId = (payload as Record<string, unknown>).requestId;  return typeof requestId === "string" ? ApprovalRequestId.make(requestId) : null;} function isStalePendingApprovalFailureDetail(detail: string | null): boolean {  if (detail === null) {    return false;  }  return (    detail.includes("stale pending approval request") ||    detail.includes("unknown pending approval request") ||    detail.includes("unknown pending permission request")  );} function derivePendingUserInputCountFromActivities(  activities: ReadonlyArray<ProjectionThreadActivity>,): number {  const openRequestIds = new Set<string>();  const ordered = [...activities].toSorted(    (left, right) =>      left.createdAt.localeCompare(right.createdAt) ||      left.activityId.localeCompare(right.activityId),  );   for (const activity of ordered) {    const requestId = extractActivityRequestId(activity.payload);    if (requestId === null) {      continue;    }    const payload =      typeof activity.payload === "object" && activity.payload !== null        ? (activity.payload as Record<string, unknown>)        : null;    const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;     if (activity.kind === "user-input.requested") {      openRequestIds.add(requestId);      continue;    }     if (activity.kind === "user-input.resolved") {      openRequestIds.delete(requestId);      continue;    }     if (      activity.kind === "provider.user-input.respond.failed" &&      detail !== null &&      (detail.includes("stale pending user-input request") ||        detail.includes("unknown pending user-input request") ||        detail.includes("unknown pending user input request") ||        detail.includes("unknown pending codex user input request"))    ) {      openRequestIds.delete(requestId);    }  }   return openRequestIds.size;} function deriveHasActionableProposedPlan(input: {  readonly latestTurnId: string | null;  readonly proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>;}): boolean {  const sorted = [...input.proposedPlans].toSorted(    (left, right) =>      left.updatedAt.localeCompare(right.updatedAt) || left.planId.localeCompare(right.planId),  );   let latestForTurn: ProjectionThreadProposedPlan | null = null;  if (input.latestTurnId !== null) {    for (let index = sorted.length - 1; index >= 0; index -= 1) {      const plan = sorted[index];      if (plan?.turnId === input.latestTurnId) {        latestForTurn = plan;        break;      }    }  }  if (latestForTurn !== null) {    return latestForTurn.implementedAt === null;  }   const latestPlan = sorted.at(-1) ?? null;  return latestPlan !== null && latestPlan.implementedAt === null;} function retainProjectionMessagesAfterRevert(  messages: ReadonlyArray<ProjectionThreadMessage>,  turns: ReadonlyArray<ProjectionTurn>,  turnCount: number,): ReadonlyArray<ProjectionThreadMessage> {  const retainedMessageIds = new Set<string>();  const retainedTurnIds = new Set<string>();  const keptTurns = turns.filter(    (turn) =>      turn.turnId !== null &&      turn.checkpointTurnCount !== null &&      turn.checkpointTurnCount <= turnCount,  );  for (const turn of keptTurns) {    if (turn.turnId !== null) {      retainedTurnIds.add(turn.turnId);    }    if (turn.pendingMessageId !== null) {      retainedMessageIds.add(turn.pendingMessageId);    }    if (turn.assistantMessageId !== null) {      retainedMessageIds.add(turn.assistantMessageId);    }  }   for (const message of messages) {    if (message.role === "system") {      retainedMessageIds.add(message.messageId);      continue;    }    if (message.turnId !== null && retainedTurnIds.has(message.turnId)) {      retainedMessageIds.add(message.messageId);    }  }   const retainedUserCount = messages.filter(    (message) => message.role === "user" && retainedMessageIds.has(message.messageId),  ).length;  const missingUserCount = Math.max(0, turnCount - retainedUserCount);  if (missingUserCount > 0) {    const fallbackUserMessages = messages      .filter(        (message) =>          message.role === "user" &&          !retainedMessageIds.has(message.messageId) &&          (message.turnId === null || retainedTurnIds.has(message.turnId)),      )      .toSorted(        (left, right) =>          left.createdAt.localeCompare(right.createdAt) ||          left.messageId.localeCompare(right.messageId),      )      .slice(0, missingUserCount);    for (const message of fallbackUserMessages) {      retainedMessageIds.add(message.messageId);    }  }   const retainedAssistantCount = messages.filter(    (message) => message.role === "assistant" && retainedMessageIds.has(message.messageId),  ).length;  const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount);  if (missingAssistantCount > 0) {    const fallbackAssistantMessages = messages      .filter(        (message) =>          message.role === "assistant" &&          !retainedMessageIds.has(message.messageId) &&          (message.turnId === null || retainedTurnIds.has(message.turnId)),      )      .toSorted(        (left, right) =>          left.createdAt.localeCompare(right.createdAt) ||          left.messageId.localeCompare(right.messageId),      )      .slice(0, missingAssistantCount);    for (const message of fallbackAssistantMessages) {      retainedMessageIds.add(message.messageId);    }  }   return messages.filter((message) => retainedMessageIds.has(message.messageId));} function retainProjectionActivitiesAfterRevert(  activities: ReadonlyArray<ProjectionThreadActivity>,  turns: ReadonlyArray<ProjectionTurn>,  turnCount: number,): ReadonlyArray<ProjectionThreadActivity> {  const retainedTurnIds = new Set<string>(    turns      .filter(        (turn) =>          turn.turnId !== null &&          turn.checkpointTurnCount !== null &&          turn.checkpointTurnCount <= turnCount,      )      .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])),  );  return activities.filter(    (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId),  );} function retainProjectionProposedPlansAfterRevert(  proposedPlans: ReadonlyArray<ProjectionThreadProposedPlan>,  turns: ReadonlyArray<ProjectionTurn>,  turnCount: number,): ReadonlyArray<ProjectionThreadProposedPlan> {  const retainedTurnIds = new Set<string>(    turns      .filter(        (turn) =>          turn.turnId !== null &&          turn.checkpointTurnCount !== null &&          turn.checkpointTurnCount <= turnCount,      )      .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])),  );  return proposedPlans.filter(    (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId),  );} function collectThreadAttachmentRelativePaths(  threadId: string,  messages: ReadonlyArray<ProjectionThreadMessage>,): Set<string> {  const threadSegment = toSafeThreadAttachmentSegment(threadId);  if (!threadSegment) {    return new Set();  }  const relativePaths = new Set<string>();  for (const message of messages) {    for (const attachment of message.attachments ?? []) {      if (attachment.type !== "image") {        continue;      }      const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachment.id);      if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) {        continue;      }      relativePaths.add(attachmentRelativePath(attachment));    }  }  return relativePaths;} const runAttachmentSideEffects = Effect.fn("runAttachmentSideEffects")(function* (  sideEffects: AttachmentSideEffects,) {  const serverConfig = yield* Effect.service(ServerConfig);  const fileSystem = yield* Effect.service(FileSystem.FileSystem);  const path = yield* Effect.service(Path.Path);   const attachmentsRootDir = serverConfig.attachmentsDir;  const readAttachmentRootEntries = fileSystem    .readDirectory(attachmentsRootDir, { recursive: false })    .pipe(Effect.orElseSucceed(() => [] as Array<string>));   const removeDeletedThreadAttachmentEntry = Effect.fn("removeDeletedThreadAttachmentEntry")(    function* (threadSegment: string, entry: string) {      const normalizedEntry = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/");      if (normalizedEntry.length === 0 || normalizedEntry.includes("/")) {        return;      }      const attachmentId = parseAttachmentIdFromRelativePath(normalizedEntry);      if (!attachmentId) {        return;      }      const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachmentId);      if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) {        return;      }      yield* fileSystem.remove(path.join(attachmentsRootDir, normalizedEntry), {        force: true,      });    },  );   const deleteThreadAttachments = Effect.fn("deleteThreadAttachments")(function* (    threadId: string,  ) {    const threadSegment = toSafeThreadAttachmentSegment(threadId);    if (!threadSegment) {      yield* Effect.logWarning("skipping attachment cleanup for unsafe thread id", {        threadId,      });      return;    }     const entries = yield* readAttachmentRootEntries;    yield* Effect.forEach(      entries,      (entry) => removeDeletedThreadAttachmentEntry(threadSegment, entry),      {        concurrency: 1,      },    );  });   const pruneThreadAttachmentEntry = Effect.fn("pruneThreadAttachmentEntry")(function* (    threadSegment: string,    keptThreadRelativePaths: Set<string>,    entry: string,  ) {    const relativePath = entry.replace(/^[/\\]+/, "").replace(/\\/g, "/");    if (relativePath.length === 0 || relativePath.includes("/")) {      return;    }    const attachmentId = parseAttachmentIdFromRelativePath(relativePath);    if (!attachmentId) {      return;    }    const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachmentId);    if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) {      return;    }     const absolutePath = path.join(attachmentsRootDir, relativePath);    const fileInfo = yield* fileSystem.stat(absolutePath).pipe(Effect.orElseSucceed(() => null));    if (!fileInfo || fileInfo.type !== "File") {      return;    }     if (!keptThreadRelativePaths.has(relativePath)) {      yield* fileSystem.remove(absolutePath, { force: true });    }  });   const pruneThreadAttachments = Effect.fn("pruneThreadAttachments")(function* (    threadId: string,    keptThreadRelativePaths: Set<string>,  ) {    if (sideEffects.deletedThreadIds.has(threadId)) {      return;    }     const threadSegment = toSafeThreadAttachmentSegment(threadId);    if (!threadSegment) {      yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });      return;    }     const entries = yield* readAttachmentRootEntries;    yield* Effect.forEach(      entries,      (entry) => pruneThreadAttachmentEntry(threadSegment, keptThreadRelativePaths, entry),      { concurrency: 1 },    );  });   yield* Effect.forEach(sideEffects.deletedThreadIds, deleteThreadAttachments, {    concurrency: 1,  });   yield* Effect.forEach(    sideEffects.prunedThreadRelativePaths.entries(),    ([threadId, keptThreadRelativePaths]) =>      pruneThreadAttachments(threadId, keptThreadRelativePaths),    { concurrency: 1 },  );}); const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjectionPipeline")(  function* () {    const sql = yield* SqlClient.SqlClient;    const eventStore = yield* OrchestrationEventStore;    const projectionStateRepository = yield* ProjectionStateRepository;    const projectionProjectRepository = yield* ProjectionProjectRepository;    const projectionThreadRepository = yield* ProjectionThreadRepository;    const projectionThreadMessageRepository = yield* ProjectionThreadMessageRepository;    const projectionThreadProposedPlanRepository = yield* ProjectionThreadProposedPlanRepository;    const projectionThreadActivityRepository = yield* ProjectionThreadActivityRepository;    const projectionThreadSessionRepository = yield* ProjectionThreadSessionRepository;    const projectionTurnRepository = yield* ProjectionTurnRepository;    const projectionPendingApprovalRepository = yield* ProjectionPendingApprovalRepository;     const fileSystem = yield* FileSystem.FileSystem;    const path = yield* Path.Path;    const serverConfig = yield* ServerConfig;     const applyProjectsProjection: ProjectorDefinition["apply"] = Effect.fn(      "applyProjectsProjection",    )(function* (event, _attachmentSideEffects) {      switch (event.type) {        case "project.created":          yield* projectionProjectRepository.upsert({            projectId: event.payload.projectId,            title: event.payload.title,            workspaceRoot: event.payload.workspaceRoot,            defaultModelSelection: event.payload.defaultModelSelection,            defaultThreadEnvMode: null,            faviconPath: event.payload.faviconPath ?? null,            scripts: event.payload.scripts,

Lines 1–500 of 1749. Load more