import {  EventId,  type OrchestrationCommand,  type OrchestrationEvent,  type OrchestrationReadModel,} from "@t3tools/contracts";import * as DateTime from "effect/DateTime";import * as Crypto from "effect/Crypto";import * as Effect from "effect/Effect";import type * as PlatformError from "effect/PlatformError"; import { OrchestrationCommandInvariantError } from "./Errors.ts";import {  listThreadsByProjectId,  requireProject,  requireProjectAbsent,  requireThread,  requireThreadArchived,  requireThreadAbsent,  requireThreadNotArchived,} from "./commandInvariants.ts";import { projectEvent } from "./projector.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); function withEventBase(  input: Pick<OrchestrationCommand, "commandId"> & {    readonly aggregateKind: OrchestrationEvent["aggregateKind"];    readonly aggregateId: OrchestrationEvent["aggregateId"];    readonly occurredAt: string;    readonly metadata?: OrchestrationEvent["metadata"];  },): Effect.Effect<  Omit<OrchestrationEvent, "sequence" | "type" | "payload">,  PlatformError.PlatformError,  Crypto.Crypto> {  return Crypto.Crypto.pipe(    Effect.flatMap((crypto) =>      crypto.randomUUIDv4.pipe(        Effect.map((eventId) => ({          eventId: EventId.make(eventId),          aggregateKind: input.aggregateKind,          aggregateId: input.aggregateId,          occurredAt: input.occurredAt,          commandId: input.commandId,          causationEventId: null,          correlationId: input.commandId,          metadata: input.metadata ?? {},        })),      ),    ),  );} type PlannedOrchestrationEvent = Omit<OrchestrationEvent, "sequence">; type DecideOrchestrationCommandResult =  | PlannedOrchestrationEvent  | ReadonlyArray<PlannedOrchestrationEvent>; const decideCommandSequence = Effect.fn("decideCommandSequence")(function* ({  commands,  readModel,}: {  readonly commands: ReadonlyArray<OrchestrationCommand>;  readonly readModel: OrchestrationReadModel;}): Effect.fn.Return<  ReadonlyArray<PlannedOrchestrationEvent>,  OrchestrationCommandInvariantError | PlatformError.PlatformError,  Crypto.Crypto> {  let nextReadModel = readModel;  let nextSequence = readModel.snapshotSequence;  const plannedEvents: PlannedOrchestrationEvent[] = [];   for (const nextCommand of commands) {    const decided = yield* decideOrchestrationCommand({      command: nextCommand,      readModel: nextReadModel,    });    const nextEvents = Array.isArray(decided) ? decided : [decided];    for (const nextEvent of nextEvents) {      plannedEvents.push(nextEvent);      nextSequence += 1;      nextReadModel = yield* projectEvent(nextReadModel, {        ...nextEvent,        sequence: nextSequence,      }).pipe(Effect.orDie);    }  }   return plannedEvents;}); export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand")(function* ({  command,  readModel,}: {  readonly command: OrchestrationCommand;  readonly readModel: OrchestrationReadModel;}): Effect.fn.Return<  DecideOrchestrationCommandResult,  OrchestrationCommandInvariantError | PlatformError.PlatformError,  Crypto.Crypto> {  switch (command.type) {    case "project.create": {      yield* requireProjectAbsent({        readModel,        command,        projectId: command.projectId,      });       return {        ...(yield* withEventBase({          aggregateKind: "project",          aggregateId: command.projectId,          occurredAt: command.createdAt,          commandId: command.commandId,        })),        type: "project.created",        payload: {          projectId: command.projectId,          title: command.title,          workspaceRoot: command.workspaceRoot,          defaultModelSelection: command.defaultModelSelection ?? null,          scripts: [],          createdAt: command.createdAt,          updatedAt: command.createdAt,        },      };    }     case "project.meta.update": {      yield* requireProject({        readModel,        command,        projectId: command.projectId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "project",          aggregateId: command.projectId,          occurredAt,          commandId: command.commandId,        })),        type: "project.meta-updated",        payload: {          projectId: command.projectId,          ...(command.title !== undefined ? { title: command.title } : {}),          ...(command.workspaceRoot !== undefined ? { workspaceRoot: command.workspaceRoot } : {}),          ...(command.defaultModelSelection !== undefined            ? { defaultModelSelection: command.defaultModelSelection }            : {}),          ...(command.scripts !== undefined ? { scripts: command.scripts } : {}),          updatedAt: occurredAt,        },      };    }     case "project.delete": {      yield* requireProject({        readModel,        command,        projectId: command.projectId,      });      const activeThreads = listThreadsByProjectId(readModel, command.projectId).filter(        (thread) => thread.deletedAt === null,      );      if (activeThreads.length > 0 && command.force !== true) {        return yield* new OrchestrationCommandInvariantError({          commandType: command.type,          detail: `Project '${command.projectId}' is not empty and cannot be deleted without force=true.`,        });      }      if (activeThreads.length > 0) {        return yield* decideCommandSequence({          readModel,          commands: [            ...activeThreads.map(              (thread): Extract<OrchestrationCommand, { type: "thread.delete" }> => ({                type: "thread.delete",                commandId: command.commandId,                threadId: thread.id,              }),            ),            {              type: "project.delete",              commandId: command.commandId,              projectId: command.projectId,            },          ],        });      }       const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "project",          aggregateId: command.projectId,          occurredAt,          commandId: command.commandId,        })),        type: "project.deleted" as const,        payload: {          projectId: command.projectId,          deletedAt: occurredAt,        },      };    }     case "thread.create": {      yield* requireProject({        readModel,        command,        projectId: command.projectId,      });      yield* requireThreadAbsent({        readModel,        command,        threadId: command.threadId,      });      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt: command.createdAt,          commandId: command.commandId,        })),        type: "thread.created",        payload: {          threadId: command.threadId,          projectId: command.projectId,          title: command.title,          modelSelection: command.modelSelection,          runtimeMode: command.runtimeMode,          interactionMode: command.interactionMode,          branch: command.branch,          worktreePath: command.worktreePath,          createdAt: command.createdAt,          updatedAt: command.createdAt,        },      };    }     case "thread.delete": {      yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.deleted",        payload: {          threadId: command.threadId,          deletedAt: occurredAt,        },      };    }     case "thread.archive": {      yield* requireThreadNotArchived({        readModel,        command,        threadId: command.threadId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.archived",        payload: {          threadId: command.threadId,          archivedAt: occurredAt,          updatedAt: occurredAt,        },      };    }     case "thread.unarchive": {      yield* requireThreadArchived({        readModel,        command,        threadId: command.threadId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.unarchived",        payload: {          threadId: command.threadId,          updatedAt: occurredAt,        },      };    }     case "thread.meta.update": {      yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.meta-updated",        payload: {          threadId: command.threadId,          ...(command.title !== undefined ? { title: command.title } : {}),          ...(command.modelSelection !== undefined            ? { modelSelection: command.modelSelection }            : {}),          ...(command.branch !== undefined ? { branch: command.branch } : {}),          ...(command.worktreePath !== undefined ? { worktreePath: command.worktreePath } : {}),          updatedAt: occurredAt,        },      };    }     case "thread.runtime-mode.set": {      yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.runtime-mode-set",        payload: {          threadId: command.threadId,          runtimeMode: command.runtimeMode,          updatedAt: occurredAt,        },      };    }     case "thread.interaction-mode.set": {      yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      const occurredAt = yield* nowIso;      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.interaction-mode-set",        payload: {          threadId: command.threadId,          interactionMode: command.interactionMode,          updatedAt: occurredAt,        },      };    }     case "thread.turn.start": {      const targetThread = yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      const sourceProposedPlan = command.sourceProposedPlan;      const sourceThread = sourceProposedPlan        ? yield* requireThread({            readModel,            command,            threadId: sourceProposedPlan.threadId,          })        : null;      const sourcePlan =        sourceProposedPlan && sourceThread          ? sourceThread.proposedPlans.find((entry) => entry.id === sourceProposedPlan.planId)          : null;      if (sourceProposedPlan && !sourcePlan) {        return yield* new OrchestrationCommandInvariantError({          commandType: command.type,          detail: `Proposed plan '${sourceProposedPlan.planId}' does not exist on thread '${sourceProposedPlan.threadId}'.`,        });      }      if (sourceThread && sourceThread.projectId !== targetThread.projectId) {        return yield* new OrchestrationCommandInvariantError({          commandType: command.type,          detail: `Proposed plan '${sourceProposedPlan?.planId}' belongs to thread '${sourceThread.id}' in a different project.`,        });      }      const userMessageEvent: Omit<OrchestrationEvent, "sequence"> = {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt: command.createdAt,          commandId: command.commandId,        })),        type: "thread.message-sent",        payload: {          threadId: command.threadId,          messageId: command.message.messageId,          role: "user",          text: command.message.text,          attachments: command.message.attachments,          turnId: null,          streaming: false,          createdAt: command.createdAt,          updatedAt: command.createdAt,        },      };      const turnStartRequestedEvent: Omit<OrchestrationEvent, "sequence"> = {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt: command.createdAt,          commandId: command.commandId,        })),        causationEventId: userMessageEvent.eventId,        type: "thread.turn-start-requested",        payload: {          threadId: command.threadId,          messageId: command.message.messageId,          ...(command.modelSelection !== undefined            ? { modelSelection: command.modelSelection }            : {}),          ...(command.titleSeed !== undefined ? { titleSeed: command.titleSeed } : {}),          runtimeMode: targetThread.runtimeMode,          interactionMode: targetThread.interactionMode,          ...(sourceProposedPlan !== undefined ? { sourceProposedPlan } : {}),          createdAt: command.createdAt,        },      };      return [userMessageEvent, turnStartRequestedEvent];    }     case "thread.turn.interrupt": {      yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt: command.createdAt,          commandId: command.commandId,        })),        type: "thread.turn-interrupt-requested",        payload: {          threadId: command.threadId,          ...(command.turnId !== undefined ? { turnId: command.turnId } : {}),          createdAt: command.createdAt,        },      };    }     case "thread.approval.respond": {      yield* requireThread({        readModel,        command,        threadId: command.threadId,      });      return {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt: command.createdAt,          commandId: command.commandId,          metadata: {            requestId: command.requestId,          },

Lines 1–500 of 765. Load more