import {  ApprovalRequestId,  isImportedAgentSessionMessageId,  UserInputAttachmentAnswerPayload,  type ChatAttachment,  type OrchestrationEvent,  type OrchestrationSessionStatus,  ThreadId,} from "@t3tools/contracts";import { compareDateTimeStrings } from "@t3tools/shared/dateTime";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 Schema from "effect/Schema";import * as Stream from "effect/Stream";import * as SqlClient from "effect/unstable/sql/SqlClient";import {  legacyThreadPullRequestKey,  threadPullRequestKeysEqual,} from "@t3tools/shared/threadPullRequests"; 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 * as ProjectionThreadPullRequests from "../../persistence/ProjectionThreadPullRequests.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")  );} // A refresh reads each persisted summary source, so skip activities that cannot change the result.function shouldRefreshThreadShellSummary(event: OrchestrationEvent): boolean {  if (event.type !== "thread.activity-appended") {    return true;  }   switch (event.payload.activity.kind) {    case "approval.requested":    case "approval.resolved":    case "provider.approval.respond.failed":    case "user-input.requested":    case "user-input.resolved":    case "provider.user-input.respond.failed":      return true;    default:      return false;  }} 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 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" || isImportedAgentSessionMessageId(message.messageId)) {      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" &&      !isImportedAgentSessionMessageId(message.messageId) &&      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) =>          compareDateTimeStrings(left.createdAt, 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" &&      !isImportedAgentSessionMessageId(message.messageId) &&      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) =>          compareDateTimeStrings(left.createdAt, 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),  );} const decodeQuestionAttachmentAnswer = Schema.decodeUnknownOption(UserInputAttachmentAnswerPayload); 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 ?? []) {      const attachmentThreadSegment = parseThreadSegmentFromAttachmentId(attachment.id);      if (!attachmentThreadSegment || attachmentThreadSegment !== threadSegment) {        continue;      }      const relativePath = attachmentRelativePath(attachment);      if (relativePath) {        relativePaths.add(relativePath);      }    }  }  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,  });   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);    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 projectionThreadPullRequestRepository =      yield* ProjectionThreadPullRequests.ProjectionThreadPullRequestRepository;    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",

Lines 1–500 of 2180. Load more