import type { OrchestrationEvent, OrchestrationReadModel, ThreadId } from "@t3tools/contracts";import {  OrchestrationCheckpointSummary,  OrchestrationMessage,  OrchestrationSession,  OrchestrationThread,} from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Schema from "effect/Schema"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts";import {  MessageSentPayloadSchema,  ProjectCreatedPayload,  ProjectDeletedPayload,  ProjectMetaUpdatedPayload,  ThreadActivityAppendedPayload,  ThreadArchivedPayload,  ThreadCreatedPayload,  ThreadDeletedPayload,  ThreadInteractionModeSetPayload,  ThreadMetaUpdatedPayload,  ThreadProposedPlanUpsertedPayload,  ThreadRuntimeModeSetPayload,  ThreadSettledPayload,  ThreadPinnedPayload,  ThreadPinReorderedPayload,  ThreadSnoozedPayload,  ThreadUnpinnedPayload,  ThreadUnarchivedPayload,  ThreadUnsettledPayload,  ThreadUnsnoozedPayload,  ThreadRevertedPayload,  ThreadSessionSetPayload,  ThreadTurnDiffCompletedPayload,} from "./Schemas.ts"; type ThreadPatch = Partial<Omit<OrchestrationThread, "id" | "projectId">>;const MAX_THREAD_MESSAGES = 2_000;const MAX_THREAD_CHECKPOINTS = 500; function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error") {  if (status === "error") return "error" as const;  if (status === "missing") return "interrupted" as const;  return "completed" as const;} /** * Turn state to settle a still-running latest turn with when its session * leaves the "running" status, or null while the session is (re)starting or * running and the turn must stay unsettled. */function settledTurnStateForSessionStatus(  status: OrchestrationSession["status"],): "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;  }} function updateThread(  threads: ReadonlyArray<OrchestrationThread>,  threadId: ThreadId,  patch: ThreadPatch,): OrchestrationThread[] {  return threads.map((thread) => (thread.id === threadId ? { ...thread, ...patch } : thread));} function decodeForEvent<A>(  schema: Schema.Decoder<A, never>,  value: unknown,  eventType: OrchestrationEvent["type"],  field: string,): Effect.Effect<A, OrchestrationProjectorDecodeError> {  return Schema.decodeUnknownEffect(schema)(value).pipe(    Effect.mapError(toProjectorDecodeError(`${eventType}:${field}`)),  );} function retainThreadMessagesAfterRevert(  messages: ReadonlyArray<OrchestrationMessage>,  retainedTurnIds: ReadonlySet<string>,  turnCount: number,): ReadonlyArray<OrchestrationMessage> {  const retainedMessageIds = new Set<string>();  for (const message of messages) {    if (message.role === "system") {      retainedMessageIds.add(message.id);      continue;    }    if (message.turnId !== null && retainedTurnIds.has(message.turnId)) {      retainedMessageIds.add(message.id);    }  }   const retainedUserCount = messages.filter(    (message) => message.role === "user" && retainedMessageIds.has(message.id),  ).length;  const missingUserCount = Math.max(0, turnCount - retainedUserCount);  if (missingUserCount > 0) {    const fallbackUserMessages = messages      .filter(        (message) =>          message.role === "user" &&          !retainedMessageIds.has(message.id) &&          (message.turnId === null || retainedTurnIds.has(message.turnId)),      )      .toSorted(        (left, right) =>          left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id),      )      .slice(0, missingUserCount);    for (const message of fallbackUserMessages) {      retainedMessageIds.add(message.id);    }  }   const retainedAssistantCount = messages.filter(    (message) => message.role === "assistant" && retainedMessageIds.has(message.id),  ).length;  const missingAssistantCount = Math.max(0, turnCount - retainedAssistantCount);  if (missingAssistantCount > 0) {    const fallbackAssistantMessages = messages      .filter(        (message) =>          message.role === "assistant" &&          !retainedMessageIds.has(message.id) &&          (message.turnId === null || retainedTurnIds.has(message.turnId)),      )      .toSorted(        (left, right) =>          left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id),      )      .slice(0, missingAssistantCount);    for (const message of fallbackAssistantMessages) {      retainedMessageIds.add(message.id);    }  }   return messages.filter((message) => retainedMessageIds.has(message.id));} function retainThreadActivitiesAfterRevert(  activities: ReadonlyArray<OrchestrationThread["activities"][number]>,  retainedTurnIds: ReadonlySet<string>,): ReadonlyArray<OrchestrationThread["activities"][number]> {  return activities.filter(    (activity) => activity.turnId === null || retainedTurnIds.has(activity.turnId),  );} function retainThreadProposedPlansAfterRevert(  proposedPlans: ReadonlyArray<OrchestrationThread["proposedPlans"][number]>,  retainedTurnIds: ReadonlySet<string>,): ReadonlyArray<OrchestrationThread["proposedPlans"][number]> {  return proposedPlans.filter(    (proposedPlan) => proposedPlan.turnId === null || retainedTurnIds.has(proposedPlan.turnId),  );} function compareThreadActivities(  left: OrchestrationThread["activities"][number],  right: OrchestrationThread["activities"][number],): number {  if (left.sequence !== undefined && right.sequence !== undefined) {    if (left.sequence !== right.sequence) {      return left.sequence - right.sequence;    }  } else if (left.sequence !== undefined) {    return 1;  } else if (right.sequence !== undefined) {    return -1;  }   return left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id);} export function createEmptyReadModel(nowIso: string): OrchestrationReadModel {  return {    snapshotSequence: 0,    projects: [],    threads: [],    updatedAt: nowIso,  };} export function projectEvent(  model: OrchestrationReadModel,  event: OrchestrationEvent,): Effect.Effect<OrchestrationReadModel, OrchestrationProjectorDecodeError> {  const nextBase: OrchestrationReadModel = {    ...model,    snapshotSequence: event.sequence,    updatedAt: event.occurredAt,  };   switch (event.type) {    case "project.created":      return decodeForEvent(ProjectCreatedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => {          const existing = nextBase.projects.find((entry) => entry.id === payload.projectId);          const nextProject = {            id: payload.projectId,            title: payload.title,            workspaceRoot: payload.workspaceRoot,            defaultModelSelection: payload.defaultModelSelection,            defaultThreadEnvMode: null,            faviconPath: payload.faviconPath ?? null,            scripts: payload.scripts,            createdAt: payload.createdAt,            updatedAt: payload.updatedAt,            deletedAt: null,          };           return {            ...nextBase,            projects: existing              ? nextBase.projects.map((entry) =>                  entry.id === payload.projectId ? nextProject : entry,                )              : [...nextBase.projects, nextProject],          };        }),      );     case "project.meta-updated":      return decodeForEvent(ProjectMetaUpdatedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          projects: nextBase.projects.map((project) =>            project.id === payload.projectId              ? {                  ...project,                  ...(payload.title !== undefined ? { title: payload.title } : {}),                  ...(payload.workspaceRoot !== undefined                    ? { workspaceRoot: payload.workspaceRoot }                    : {}),                  ...(payload.defaultModelSelection !== undefined                    ? { defaultModelSelection: payload.defaultModelSelection }                    : {}),                  ...(payload.defaultThreadEnvMode !== undefined                    ? { defaultThreadEnvMode: payload.defaultThreadEnvMode }                    : {}),                  ...(payload.faviconPath !== undefined                    ? { faviconPath: payload.faviconPath }                    : {}),                  ...(payload.scripts !== undefined ? { scripts: payload.scripts } : {}),                  updatedAt: payload.updatedAt,                }              : project,          ),        })),      );     case "project.deleted":      return decodeForEvent(ProjectDeletedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          projects: nextBase.projects.map((project) =>            project.id === payload.projectId              ? {                  ...project,                  deletedAt: payload.deletedAt,                  updatedAt: payload.deletedAt,                }              : project,          ),        })),      );     case "thread.created":      return Effect.gen(function* () {        const payload = yield* decodeForEvent(          ThreadCreatedPayload,          event.payload,          event.type,          "payload",        );        const thread: OrchestrationThread = yield* decodeForEvent(          OrchestrationThread,          {            id: payload.threadId,            projectId: payload.projectId,            title: payload.title,            modelSelection: payload.modelSelection,            runtimeMode: payload.runtimeMode,            interactionMode: payload.interactionMode,            branch: payload.branch,            worktreePath: payload.worktreePath,            latestTurn: null,            createdAt: payload.createdAt,            updatedAt: payload.updatedAt,            archivedAt: null,            settledOverride: null,            settledAt: null,            snoozedUntil: null,            snoozedAt: null,            deletedAt: null,            messages: [],            activities: [],            checkpoints: [],            session: null,          },          event.type,          "thread",        );        const existing = nextBase.threads.find((entry) => entry.id === thread.id);        return {          ...nextBase,          threads: existing            ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry))            : [...nextBase.threads, thread],        };      });     case "thread.deleted":      return decodeForEvent(ThreadDeletedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            deletedAt: payload.deletedAt,            updatedAt: payload.deletedAt,          }),        })),      );     case "thread.archived":      return decodeForEvent(ThreadArchivedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            archivedAt: payload.archivedAt,            titleRegeneration: null,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.unarchived":      return decodeForEvent(ThreadUnarchivedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            archivedAt: null,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.settled":      return decodeForEvent(ThreadSettledPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            settledOverride: "settled",            settledAt: payload.settledAt,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.unsettled":      return decodeForEvent(ThreadUnsettledPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            settledOverride: payload.reason === "user" ? "active" : null,            settledAt: null,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.snoozed":      return decodeForEvent(ThreadSnoozedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            snoozedUntil: payload.snoozedUntil,            snoozedAt: payload.snoozedAt,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.unsnoozed":      return decodeForEvent(ThreadUnsnoozedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            snoozedUntil: null,            snoozedAt: null,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.pinned":      return decodeForEvent(ThreadPinnedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            pinnedAt: payload.pinnedAt,            ...(payload.pinOrderKey !== undefined ? { pinOrderKey: payload.pinOrderKey } : {}),            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.unpinned":      return decodeForEvent(ThreadUnpinnedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            pinnedAt: null,            // Unpin clears the slot: re-pinning is "pin again", not "restore            // an ancient position".            pinOrderKey: null,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.pin-reordered":      return decodeForEvent(ThreadPinReorderedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            pinOrderKey: payload.orderKey,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.meta-updated":      return decodeForEvent(ThreadMetaUpdatedPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            ...(payload.title !== undefined ? { title: payload.title } : {}),            ...(payload.titleRegeneration !== undefined              ? { titleRegeneration: payload.titleRegeneration }              : {}),            ...(payload.modelSelection !== undefined              ? { modelSelection: payload.modelSelection }              : {}),            ...(payload.branch !== undefined ? { branch: payload.branch } : {}),            ...(payload.worktreePath !== undefined ? { worktreePath: payload.worktreePath } : {}),            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.runtime-mode-set":      return decodeForEvent(ThreadRuntimeModeSetPayload, event.payload, event.type, "payload").pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            runtimeMode: payload.runtimeMode,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.interaction-mode-set":      return decodeForEvent(        ThreadInteractionModeSetPayload,        event.payload,        event.type,        "payload",      ).pipe(        Effect.map((payload) => ({          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            interactionMode: payload.interactionMode,            updatedAt: payload.updatedAt,          }),        })),      );     case "thread.message-sent":      return Effect.gen(function* () {        const payload = yield* decodeForEvent(          MessageSentPayloadSchema,          event.payload,          event.type,          "payload",        );        const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);        if (!thread) {

Lines 1–500 of 806. Load more