import {  ApprovalRequestId,  type ChatAttachment,  type OrchestrationEvent,} from "@t3tools/contracts";import { Effect, FileSystem, Layer, Option, Path, Stream } from "effect";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]; 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.makeUnsafe(requestId) : 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.catch(() => Effect.succeed([] 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.catch(() => Effect.succeed(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,            scripts: event.payload.scripts,            createdAt: event.payload.createdAt,            updatedAt: event.payload.updatedAt,            deletedAt: null,          });          return;         case "project.meta-updated": {          const existingRow = yield* projectionProjectRepository.getById({            projectId: event.payload.projectId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionProjectRepository.upsert({            ...existingRow.value,            ...(event.payload.title !== undefined ? { title: event.payload.title } : {}),            ...(event.payload.workspaceRoot !== undefined              ? { workspaceRoot: event.payload.workspaceRoot }              : {}),            ...(event.payload.defaultModelSelection !== undefined              ? { defaultModelSelection: event.payload.defaultModelSelection }              : {}),            ...(event.payload.scripts !== undefined ? { scripts: event.payload.scripts } : {}),            updatedAt: event.payload.updatedAt,          });          return;        }         case "project.deleted": {          const existingRow = yield* projectionProjectRepository.getById({            projectId: event.payload.projectId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionProjectRepository.upsert({            ...existingRow.value,            deletedAt: event.payload.deletedAt,            updatedAt: event.payload.deletedAt,          });          return;        }         default:          return;      }    });     const applyThreadsProjection: ProjectorDefinition["apply"] = Effect.fn(      "applyThreadsProjection",    )(function* (event, attachmentSideEffects) {      switch (event.type) {        case "thread.created":          yield* projectionThreadRepository.upsert({            threadId: event.payload.threadId,            projectId: event.payload.projectId,            title: event.payload.title,            modelSelection: event.payload.modelSelection,            runtimeMode: event.payload.runtimeMode,            interactionMode: event.payload.interactionMode,            branch: event.payload.branch,            worktreePath: event.payload.worktreePath,            latestTurnId: null,            createdAt: event.payload.createdAt,            updatedAt: event.payload.updatedAt,            archivedAt: null,            deletedAt: null,          });          return;         case "thread.archived": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            archivedAt: event.payload.archivedAt,            updatedAt: event.payload.updatedAt,          });          return;        }         case "thread.unarchived": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            archivedAt: null,            updatedAt: event.payload.updatedAt,          });          return;        }         case "thread.meta-updated": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            ...(event.payload.title !== undefined ? { title: event.payload.title } : {}),            ...(event.payload.modelSelection !== undefined              ? { modelSelection: event.payload.modelSelection }              : {}),            ...(event.payload.branch !== undefined ? { branch: event.payload.branch } : {}),

Lines 1–500 of 1306. Load more