import {  ApprovalRequestId,  type ChatAttachment,  type OrchestrationEvent,} from "@t3tools/contracts";import * as NodeServices from "@effect/platform-node/NodeServices";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 {  type ProjectionThreadWorkUnit,  ProjectionThreadWorkUnitRepository,} from "../../persistence/Services/ProjectionThreadWorkUnits.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 { ProjectionThreadWorkUnitRepositoryLive } from "../../persistence/Layers/ProjectionThreadWorkUnits.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",  threadWorkUnits: "projection.thread-work-units",  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(  (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 retainProjectionWorkUnitsAfterRevert(  workUnits: ReadonlyArray<ProjectionThreadWorkUnit>,  turns: ReadonlyArray<ProjectionTurn>,  turnCount: number,): ReadonlyArray<ProjectionThreadWorkUnit> {  const retainedTurnIds = new Set<string>(    turns      .filter(        (turn) =>          turn.turnId !== null &&          turn.checkpointTurnCount !== null &&          turn.checkpointTurnCount <= turnCount,      )      .flatMap((turn) => (turn.turnId === null ? [] : [turn.turnId])),  );  const retained = workUnits.filter((workUnit) => retainedTurnIds.has(workUnit.turnId));  const retainedIds = new Set(retained.map((workUnit) => workUnit.workUnitId));  return retained.filter(    (workUnit) => workUnit.parentWorkUnitId === null || retainedIds.has(workUnit.parentWorkUnitId),  );} 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(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 = path.join(serverConfig.stateDir, "attachments");   yield* Effect.forEach(    sideEffects.deletedThreadIds,    (threadId) =>      Effect.gen(function* () {        const threadSegment = toSafeThreadAttachmentSegment(threadId);        if (!threadSegment) {          yield* Effect.logWarning("skipping attachment cleanup for unsafe thread id", {            threadId,          });          return;        }        const entries = yield* fileSystem          .readDirectory(attachmentsRootDir, { recursive: false })          .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));        yield* Effect.forEach(          entries,          (entry) =>            Effect.gen(function* () {              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,              });            }),          { concurrency: 1 },        );      }),    { concurrency: 1 },  );   yield* Effect.forEach(    sideEffects.prunedThreadRelativePaths.entries(),    ([threadId, keptThreadRelativePaths]) => {      if (sideEffects.deletedThreadIds.has(threadId)) {        return Effect.void;      }      return Effect.gen(function* () {        const threadSegment = toSafeThreadAttachmentSegment(threadId);        if (!threadSegment) {          yield* Effect.logWarning("skipping attachment prune for unsafe thread id", { threadId });          return;        }        const entries = yield* fileSystem          .readDirectory(attachmentsRootDir, { recursive: false })          .pipe(Effect.catch(() => Effect.succeed([] as Array<string>)));        yield* Effect.forEach(          entries,          (entry) =>            Effect.gen(function* () {              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 });              }            }),          { concurrency: 1 },        );      });    },    { concurrency: 1 },  );}); const makeOrchestrationProjectionPipeline = Effect.gen(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 projectionThreadWorkUnitRepository = yield* ProjectionThreadWorkUnitRepository;  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"] = (event, _attachmentSideEffects) =>    Effect.gen(function* () {      switch (event.type) {        case "project.created":          yield* projectionProjectRepository.upsert({            projectId: event.payload.projectId,            title: event.payload.title,            workspaceRoot: event.payload.workspaceRoot,            defaultModel: event.payload.defaultModel,            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.defaultModel !== undefined              ? { defaultModel: event.payload.defaultModel }              : {}),            ...(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"] = (event, attachmentSideEffects) =>    Effect.gen(function* () {      switch (event.type) {        case "thread.created":          yield* projectionThreadRepository.upsert({            threadId: event.payload.threadId,            projectId: event.payload.projectId,            title: event.payload.title,            model: event.payload.model,            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,            deletedAt: null,          });          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.model !== undefined ? { model: event.payload.model } : {}),            ...(event.payload.branch !== undefined ? { branch: event.payload.branch } : {}),            ...(event.payload.worktreePath !== undefined              ? { worktreePath: event.payload.worktreePath }              : {}),            updatedAt: event.payload.updatedAt,          });          return;        }         case "thread.runtime-mode-set": {          const existingRow = yield* projectionThreadRepository.getById({            threadId: event.payload.threadId,          });          if (Option.isNone(existingRow)) {            return;          }          yield* projectionThreadRepository.upsert({            ...existingRow.value,            runtimeMode: event.payload.runtimeMode,            updatedAt: event.payload.updatedAt,          });          return;        }         case "thread.interaction-mode-set": {          const existingRow = yield* projectionThreadRepository.getById({

Lines 1–500 of 1351. Load more