import type { OrchestrationEvent, OrchestrationReadModel, ThreadId } from "@t3tools/contracts";import {  OrchestrationCheckpointSummary,  OrchestrationMessage,  OrchestrationSession,  OrchestrationThread,} from "@t3tools/contracts";import { Effect, Schema } from "effect"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts";import {  MessageSentPayloadSchema,  ProjectCreatedPayload,  ProjectDeletedPayload,  ProjectMetaUpdatedPayload,  ThreadActivityAppendedPayload,  ThreadCreatedPayload,  ThreadDeletedPayload,  ThreadInteractionModeSetPayload,  ThreadMetaUpdatedPayload,  ThreadProposedPlanUpsertedPayload,  ThreadRuntimeModeSetPayload,  ThreadRevertedPayload,  ThreadSessionSetPayload,  ThreadTurnDiffCompletedPayload,  ThreadWorkUnitUpsertedPayload,} 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;} 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.Schema<A>,  value: unknown,  eventType: OrchestrationEvent["type"],  field: string,): Effect.Effect<A, OrchestrationProjectorDecodeError> {  return Effect.try({    try: () => Schema.decodeUnknownSync(schema as any)(value),    catch: (error) => toProjectorDecodeError(`${eventType}:${field}`)(error as Schema.SchemaError),  });} 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 retainThreadWorkUnitsAfterRevert(  workUnits: ReadonlyArray<OrchestrationThread["workUnits"][number]>,  retainedTurnIds: ReadonlySet<string>,): ReadonlyArray<OrchestrationThread["workUnits"][number]> {  const retained = workUnits.filter((workUnit) => retainedTurnIds.has(workUnit.turnId));  const retainedIds = new Set(retained.map((workUnit) => workUnit.id));  return retained.filter(    (workUnit) => workUnit.parentWorkUnitId === null || retainedIds.has(workUnit.parentWorkUnitId),  );} 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,            defaultModel: payload.defaultModel,            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.defaultModel !== undefined                    ? { defaultModel: payload.defaultModel }                    : {}),                  ...(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,            model: payload.model,            runtimeMode: payload.runtimeMode,            interactionMode: payload.interactionMode,            branch: payload.branch,            worktreePath: payload.worktreePath,            latestTurn: null,            createdAt: payload.createdAt,            updatedAt: payload.updatedAt,            deletedAt: null,            messages: [],            proposedPlans: [],            activities: [],            workUnits: [],            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.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.model !== undefined ? { model: payload.model } : {}),            ...(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) {          return nextBase;        }         const message: OrchestrationMessage = yield* decodeForEvent(          OrchestrationMessage,          {            id: payload.messageId,            role: payload.role,            text: payload.text,            ...(payload.attachments !== undefined ? { attachments: payload.attachments } : {}),            turnId: payload.turnId,            streaming: payload.streaming,            createdAt: payload.createdAt,            updatedAt: payload.updatedAt,          },          event.type,          "message",        );         const existingMessage = thread.messages.find((entry) => entry.id === message.id);        const messages = existingMessage          ? thread.messages.map((entry) =>              entry.id === message.id                ? {                    ...entry,                    text: message.streaming                      ? `${entry.text}${message.text}`                      : message.text.length > 0                        ? message.text                        : entry.text,                    streaming: message.streaming,                    updatedAt: message.updatedAt,                    turnId: message.turnId,                    ...(message.attachments !== undefined                      ? { attachments: message.attachments }                      : {}),                  }                : entry,            )          : [...thread.messages, message];        const cappedMessages = messages.slice(-MAX_THREAD_MESSAGES);         return {          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            messages: cappedMessages,            updatedAt: event.occurredAt,          }),        };      });     case "thread.session-set":      return Effect.gen(function* () {        const payload = yield* decodeForEvent(          ThreadSessionSetPayload,          event.payload,          event.type,          "payload",        );        const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);        if (!thread) {          return nextBase;        }         const session: OrchestrationSession = yield* decodeForEvent(          OrchestrationSession,          payload.session,          event.type,          "session",        );         return {          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            session,            latestTurn:              session.status === "running" && session.activeTurnId !== null                ? {                    turnId: session.activeTurnId,                    state: "running",                    requestedAt:                      thread.latestTurn?.turnId === session.activeTurnId                        ? thread.latestTurn.requestedAt                        : session.updatedAt,                    startedAt:                      thread.latestTurn?.turnId === session.activeTurnId                        ? (thread.latestTurn.startedAt ?? session.updatedAt)                        : session.updatedAt,                    completedAt: null,                    assistantMessageId:                      thread.latestTurn?.turnId === session.activeTurnId                        ? thread.latestTurn.assistantMessageId                        : null,                  }                : thread.latestTurn,            updatedAt: event.occurredAt,          }),        };      });     case "thread.proposed-plan-upserted":      return Effect.gen(function* () {        const payload = yield* decodeForEvent(          ThreadProposedPlanUpsertedPayload,          event.payload,          event.type,          "payload",        );        const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);        if (!thread) {          return nextBase;        }         const proposedPlans = [          ...thread.proposedPlans.filter((entry) => entry.id !== payload.proposedPlan.id),          payload.proposedPlan,        ]          .toSorted(            (left, right) =>              left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id),          )          .slice(-200);         return {          ...nextBase,          threads: updateThread(nextBase.threads, payload.threadId, {            proposedPlans,            updatedAt: event.occurredAt,          }),        };      });     case "thread.turn-diff-completed":      return Effect.gen(function* () {        const payload = yield* decodeForEvent(          ThreadTurnDiffCompletedPayload,          event.payload,          event.type,          "payload",        );        const thread = nextBase.threads.find((entry) => entry.id === payload.threadId);        if (!thread) {          return nextBase;        }

Lines 1–500 of 674. Load more