import {  EventId,  MessageId,  UserInputRequestedPayload,  isImportedAgentSessionMessageId,  type OrchestrationCommand,  type OrchestrationEvent,  type OrchestrationReadModel,  type OrchestrationThread,  type OrchestrationThreadActivity,} from "@t3tools/contracts";import { compareDateTimeStrings } from "@t3tools/shared/dateTime";import * as DateTime from "effect/DateTime";import * as Crypto from "effect/Crypto";import * as Effect from "effect/Effect";import * as Schema from "effect/Schema";import * as Option from "effect/Option";import * as Predicate from "effect/Predicate";import type * as PlatformError from "effect/PlatformError"; import {  OrchestrationCommandInvariantError,  OrchestrationThreadSettleBlockedError,  type OrchestrationCommandRejection,} from "./Errors.ts";import {  listThreadsByProjectId,  requireActiveProjectWorkspaceRootAbsent,  requireProject,  requireProjectAbsent,  requireThread,  requireThreadArchived,  requireThreadAbsent,  requireThreadNotArchived,} from "./commandInvariants.ts";import { projectEvent } from "./projector.ts";import { threadHasQueuedTurnStart } from "./ThreadSettlementPolicy.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso);const decodeUserInputRequestedPayload = Schema.decodeUnknownOption(UserInputRequestedPayload); /** * Blocked-on-you work derived from the thread's retained activities: an * approval or user-input request with no later resolution for the same * requestId. The server-side twin of the shell's hasPendingApprovals / * hasPendingUserInput flags, which the decider read model does not carry. * The clearing rules MUST match ProjectionPipeline's pending accounting — * resolved activities always clear, respond.failed clears only when the * failure detail marks the request stale/unknown — or settle would be * rejected on threads whose shell flags read as clear. */function isStaleRequestFailureDetail(payload: Record<string, unknown> | null): boolean {  const detail = typeof payload?.detail === "string" ? payload.detail.toLowerCase() : null;  if (detail === null) return false;  return (    detail.includes("stale pending approval request") ||    detail.includes("unknown pending approval request") ||    detail.includes("unknown pending permission request") ||    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")  );} // Scans the read model's activities, which the projector caps at the most// recent 500 plus pending async questions. Async questions remain actionable// while the agent works, so they must not expire with the activity window.function hasOpenBlockingRequest(thread: {  readonly activities: ReadonlyArray<{ readonly kind: string; readonly payload: unknown }>;}): boolean {  const openRequestIds = new Set<string>();  for (const activity of thread.activities) {    const payload =      typeof activity.payload === "object" && activity.payload !== null        ? (activity.payload as Record<string, unknown>)        : null;    const requestId = typeof payload?.requestId === "string" ? payload.requestId : null;    if (requestId === null) continue;    if (activity.kind === "approval.requested" || activity.kind === "user-input.requested") {      openRequestIds.add(requestId);    } else if (activity.kind === "approval.resolved" || activity.kind === "user-input.resolved") {      openRequestIds.delete(requestId);    } else if (      (activity.kind === "provider.approval.respond.failed" ||        activity.kind === "provider.user-input.respond.failed") &&      isStaleRequestFailureDetail(payload)    ) {      openRequestIds.delete(requestId);    }  }  return openRequestIds.size > 0;} /** Apply the shared shell-level rule to the detailed command read model. */function hasQueuedTurnStartForThread(  thread: Pick<OrchestrationThread, "messages" | "latestTurn" | "session">,  now: string,): boolean {  let latestUserMessageAt: string | null = null;  let latestUserMessageAtMs = Number.NEGATIVE_INFINITY;  for (const message of thread.messages) {    if (message.role !== "user" || isImportedAgentSessionMessageId(message.id)) continue;    const messageAtMs = Date.parse(message.createdAt);    latestUserMessageAtMs = Math.max(latestUserMessageAtMs, messageAtMs);    if (messageAtMs === latestUserMessageAtMs) {      latestUserMessageAt = message.createdAt;    }  }  return threadHasQueuedTurnStart(    {      latestUserMessageAt: Number.isFinite(latestUserMessageAtMs) ? latestUserMessageAt : null,      latestTurn: thread.latestTurn,      session: thread.session,    },    now,  );} 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>,  OrchestrationCommandRejection | 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,  userInputActivity,}: {  readonly command: OrchestrationCommand;  readonly readModel: OrchestrationReadModel;  readonly userInputActivity?: OrchestrationThreadActivity;}): Effect.fn.Return<  DecideOrchestrationCommandResult,  OrchestrationCommandRejection | PlatformError.PlatformError,  Crypto.Crypto> {  switch (command.type) {    case "project.create": {      yield* requireProjectAbsent({        readModel,        command,        projectId: command.projectId,      });      yield* requireActiveProjectWorkspaceRootAbsent({        readModel,        command,        workspaceRoot: command.workspaceRoot,        exceptProjectId: 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,          // Project creation has no user model choice. Older clients sent an          // automatic seed here, but only a metadata update records an          // explicit project default.          defaultModelSelection: null,          faviconPath: null,          projectIcon: null,          scripts: [],          createdAt: command.createdAt,          updatedAt: command.createdAt,        },      };    }     case "project.meta.update": {      yield* requireProject({        readModel,        command,        projectId: command.projectId,      });      if (command.workspaceRoot !== undefined) {        yield* requireActiveProjectWorkspaceRootAbsent({          readModel,          command,          workspaceRoot: command.workspaceRoot,          exceptProjectId: 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.defaultThreadEnvMode !== undefined            ? { defaultThreadEnvMode: command.defaultThreadEnvMode }            : {}),          ...(command.autoPull !== undefined ? { autoPull: command.autoPull } : {}),          ...(command.faviconPath !== undefined ? { faviconPath: command.faviconPath } : {}),          ...(command.projectIcon !== undefined ? { projectIcon: command.projectIcon } : {}),          ...(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,          ...(command.historyImport === true ? { metadata: { historyImport: true } } : {}),        })),        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.settle":    case "thread.auto-settle": {      const thread = yield* requireThreadNotArchived({        readModel,        command,        threadId: command.threadId,      });      if (command.type === "thread.auto-settle" && thread.settledOverride !== null) {        return yield* Effect.fail(          new OrchestrationCommandInvariantError({            commandType: command.type,            detail: `thread ${command.threadId} changed before automatic settlement`,          }),        );      }      // The server owns settle eligibility. A stale command must not settle      // a thread whose session is coming alive or working.      if (thread.session?.status === "starting" || thread.session?.status === "running") {        return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId });      }      // Pending approval / user-input requests are blocked-on-you work: a      // raced or stale client must not park them behind a settled override      // that would surface only after the request resolves.      if (hasOpenBlockingRequest(thread)) {        return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId });      }      const occurredAt = yield* nowIso;      // Settling inside the adoption window would hide just-requested work.      if (hasQueuedTurnStartForThread(thread, occurredAt)) {        return yield* new OrchestrationThreadSettleBlockedError({ threadId: command.threadId });      }      // Settling an already-settled thread re-emits with the original      // settledAt: the engine rejects zero-event commands, and bulk-settle /      // double-click must stay silent no-ops rather than surface errors.      const alreadySettled = thread.settledOverride === "settled" && thread.settledAt !== null;      const settledEvent = {        ...(yield* withEventBase({          aggregateKind: "thread",          aggregateId: command.threadId,          occurredAt,          commandId: command.commandId,        })),        type: "thread.settled" as const,        payload: {          threadId: command.threadId,          settledAt: alreadySettled            ? thread.settledAt            : command.type === "thread.auto-settle"              ? command.settledAt              : occurredAt,          // A re-emission is a projected no-op: keep the existing updatedAt          // so duplicate settles neither rewind nor churn ordering. A fresh          // settle stamps the command time.          updatedAt: alreadySettled ? thread.updatedAt : occurredAt,        },      };      // Settling is "I'm done with this": clear states that would keep the      // row pinned or snoozed instead of showing the new settled state.      const companionEvents: Array<Omit<OrchestrationEvent, "sequence">> = [];      if (thread.pinnedAt != null) {        companionEvents.push({          ...(yield* withEventBase({            aggregateKind: "thread",            aggregateId: command.threadId,

Lines 1–500 of 1554. Load more