import type {  OrchestrationEvent,  OrchestrationProject,  OrchestrationReadModel,  ThreadId,  ThreadLinkedPullRequest,  ThreadPullRequestKey,  ThreadPullRequestLink,} from "@t3tools/contracts";import {  isImportedAgentSessionMessageId,  OrchestrationCheckpointSummary,  OrchestrationMessage,  OrchestrationSession,  OrchestrationThread,} from "@t3tools/contracts";import {  legacyLinkedPullRequestOf,  legacyThreadPullRequestKey,  threadPullRequestKeysEqual,} from "@t3tools/shared/threadPullRequests";import { compareDateTimeStrings } from "@t3tools/shared/dateTime";import * as Effect from "effect/Effect";import * as Schema from "effect/Schema";import * as Predicate from "effect/Predicate"; import { toProjectorDecodeError, type OrchestrationProjectorDecodeError } from "./Errors.ts";import {  MessageSentPayloadSchema,  ProjectCreatedPayload,  ProjectDeletedPayload,  ProjectMetaUpdatedPayload,  ThreadActivityAppendedPayload,  ThreadArchivedPayload,  ThreadCreatedPayload,  ThreadDeletedPayload,  ThreadInteractionModeSetPayload,  ThreadMetaUpdatedPayload,  ThreadProposedPlanUpsertedPayload,  ThreadRuntimeModeSetPayload,  ThreadSettledPayload,  ThreadPinnedPayload,  ThreadPinReorderedPayload,  ThreadPullRequestLinkedPayload,  ThreadPullRequestSyncedPayload,  ThreadPullRequestUnlinkedPayload,  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; // Async questions can stay open while the agent produces more activity.// Match the database snapshot's pending-question retention.function retainThreadActivities(activities: OrchestrationThread["activities"]) {  const recentStart = activities.length - 500;  if (recentStart <= 0) return activities;  const pending = new Map<string, OrchestrationThread["activities"][number]>();  for (const activity of activities) {    if (!Predicate.isObject(activity.payload)) continue;    const requestId = activity.payload.requestId;    if (typeof requestId !== "string") continue;    if (activity.kind === "user-input.requested" && activity.payload.responseMode === "message") {      pending.set(requestId, activity);    } else if (activity.kind === "user-input.resolved") {      pending.delete(requestId);    }  }  const pendingActivities = new Set(pending.values());  return activities.filter(    (activity, index) => index >= recentStart || pendingActivities.has(activity),  );} function checkpointStatusToLatestTurnState(status: "ready" | "missing" | "error") {  if (status === "error") return "error" as const;  // Match SQL and client projections: a missing git ref is not an interruption.  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));} /** Patch that swaps a thread's links and re-derives the legacy single-PR field from them. */function pullRequestsPatch(  thread: Pick<OrchestrationThread, "projectId">,  pullRequests: ReadonlyArray<ThreadPullRequestLink>,  projects: OrchestrationReadModel["projects"],): Pick<OrchestrationThread, "pullRequests" | "linkedPullRequest"> {  return {    pullRequests,    linkedPullRequest: legacyLinkedPullRequestOf(      pullRequests,      thread.projectId,      projects.find((project) => project.id === thread.projectId)?.repositoryIdentity,    ),  };} function upsertPullRequestLink(  pullRequests: ReadonlyArray<ThreadPullRequestLink>,  link: ThreadPullRequestLink,): ReadonlyArray<ThreadPullRequestLink> {  const index = pullRequests.findIndex((entry) => threadPullRequestKeysEqual(entry, link));  return index === -1    ? [...pullRequests, link]    : pullRequests.map((entry, entryIndex) => (entryIndex === index ? link : entry));} function removePullRequestLink(  pullRequests: ReadonlyArray<ThreadPullRequestLink>,  key: ThreadPullRequestKey,): ReadonlyArray<ThreadPullRequestLink> {  return pullRequests.filter((entry) => !threadPullRequestKeysEqual(entry, key));} /** * Host for a legacy `linkedPullRequest` being replayed into the link array. * Legacy links never carried one; the project's canonical key * (`<host>/<owner>/<name>`) is the best witness, then the link URL. */function legacyPullRequestHost(  project: OrchestrationProject | undefined,  linked: ThreadLinkedPullRequest,): string {  const canonicalHost = project?.repositoryIdentity?.canonicalKey.split("/")[0];  if (canonicalHost) return canonicalHost.toLowerCase();  try {    return new URL(linked.url).hostname.toLowerCase();  } catch {    return "unknown";  }} function legacyLinkToPullRequests(  thread: Pick<OrchestrationThread, "pullRequests">,  project: OrchestrationProject | undefined,  linked: ThreadLinkedPullRequest | null,  linkedAt: string,): ReadonlyArray<ThreadPullRequestLink> {  // The legacy field held one user-chosen link, so null clears exactly the  // manual ones and leaves created/agent/stack links alone.  const withoutManual = thread.pullRequests.filter((entry) => entry.source !== "manual");  if (linked === null) return withoutManual;  return upsertPullRequestLink(withoutManual, {    ...legacyThreadPullRequestKey(linked, legacyPullRequestHost(project, linked)),    url: linked.url,    source: "manual",    linkedAt,    snapshot: null,    stack: null,  });} 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" || isImportedAgentSessionMessageId(message.id)) {      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" &&      !isImportedAgentSessionMessageId(message.id) &&      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) =>          compareDateTimeStrings(left.createdAt, 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" &&      !isImportedAgentSessionMessageId(message.id) &&      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) =>          compareDateTimeStrings(left.createdAt, 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,            autoPull: false,            faviconPath: payload.faviconPath ?? null,            projectIcon: payload.projectIcon ?? 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.autoPull !== undefined ? { autoPull: payload.autoPull } : {}),                  ...(payload.faviconPath !== undefined                    ? { faviconPath: payload.faviconPath }                    : {}),                  ...(payload.projectIcon !== undefined                    ? { projectIcon: payload.projectIcon }                    : {}),                  ...(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,            pullRequests: [],            branchPullRequest: null,            latestTurn: null,            createdAt: payload.createdAt,            updatedAt: payload.updatedAt,            archivedAt: null,            settledOverride: null,            settledAt: null,            unsettledAt: null,            activeOrderKey: 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,            unsettledAt: null,            activeOrderKey: null,            updatedAt: payload.updatedAt,          }),

Lines 1–500 of 1060. Load more