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, requireActiveProjectWorkspaceRootAbsent, requireProject, requireProjectAbsent, requireThread, requireThreadArchived, requireThreadAbsent, requireThreadNotArchived,} from "./commandInvariants.ts";import { projectEvent } from "./projector.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); // Session adoption takes seconds; a user message still unadopted after this// window is a failed/stale start, not pending work. Mirrors the client's// QUEUED_TURN_START_GRACE_MS in client-runtime threadSettled.ts.const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000; /** * 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. That bound is safe here: an OPEN approval/user-input request// blocks its turn, so the thread cannot accumulate hundreds of later// activities while one is outstanding — a request that has scrolled out of// the window is one whose turn kept running, i.e. it was resolved or went// stale. (The projection pipeline's pendingApprovalCount reads the same// capped stream and stays consistent with this view.)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;} /** * A queued turn start — a user message no turn has picked up yet — is work * in flight even though session is still null (turn.start emits * message-sent + turn-start-requested; the session arrives later). Detection * mirrors the client's hasQueuedTurnStart: the newest user message is * strictly newer than every latestTurn timestamp (adoption stamps the new * turn's requestedAt with the message time, clearing this), and only within * the adoption grace window — historical threads whose last user message * postdates their turn timestamps (older-server data, mid-turn messages) * must not be blocked forever. A failed session start (status "error") * clears the block immediately. * * The age check is bounded on BOTH sides: message timestamps are * client-supplied, so a client clock ahead of the server yields a negative * age. Without the lower bound that negative age satisfies `<= grace` for * as long as the skew lasts, extending the block far past the intended two * minutes. */function threadHasQueuedTurnStart( thread: { readonly messages: ReadonlyArray<{ readonly role: string; readonly createdAt: string }>; readonly latestTurn: { readonly requestedAt: string; readonly startedAt: string | null; readonly completedAt: string | null; } | null; readonly session: { readonly status: string } | null; }, occurredAt: string,): boolean { const latestUserMessageAtMs = thread.messages.reduce( (latest, message) => message.role === "user" ? Math.max(latest, Date.parse(message.createdAt)) : latest, Number.NEGATIVE_INFINITY, ); const latestTurnAtMs = thread.latestTurn === null ? Number.NEGATIVE_INFINITY : Math.max( ...[ thread.latestTurn.requestedAt, thread.latestTurn.startedAt, thread.latestTurn.completedAt, ].map((candidate) => candidate == null ? Number.NEGATIVE_INFINITY : Date.parse(candidate), ), ); const queuedAgeMs = Date.parse(occurredAt) - latestUserMessageAtMs; return ( thread.session?.status !== "error" && Number.isFinite(latestUserMessageAtMs) && latestUserMessageAtMs > latestTurnAtMs && Math.abs(queuedAgeMs) <= QUEUED_TURN_START_GRACE_MS );} 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, }); 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, defaultModelSelection: command.defaultModelSelection ?? null, faviconPath: 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.faviconPath !== undefined ? { faviconPath: command.faviconPath } : {}), ...(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.settle": { const thread = yield* requireThreadNotArchived({ readModel, command, threadId: command.threadId, }); // Server-side twin of the client's canSettle session check: a stale // or raced client must not settle a thread whose session is coming // alive or working. if (thread.session?.status === "starting" || thread.session?.status === "running") { return yield* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: `thread ${command.threadId} has an active session and cannot be settled`, }), ); } // 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* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: `thread ${command.threadId} has a pending approval or user-input request and cannot be settled`, }), ); } const occurredAt = yield* nowIso; // Settling inside the adoption window would hide just-requested work. if (threadHasQueuedTurnStart(thread, occurredAt)) { return yield* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: `thread ${command.threadId} has a queued turn start and cannot be settled`, }), ); } // 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,Lines 1–500 of 1402. Load more
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, requireActiveProjectWorkspaceRootAbsent, requireProject, requireProjectAbsent, requireThread, requireThreadArchived, requireThreadAbsent, requireThreadNotArchived, } from "./commandInvariants.ts"; import { projectEvent } from "./projector.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); // Session adoption takes seconds; a user message still unadopted after this // window is a failed/stale start, not pending work. Mirrors the client's // QUEUED_TURN_START_GRACE_MS in client-runtime threadSettled.ts. const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000; /** * 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. That bound is safe here: an OPEN approval/user-input request // blocks its turn, so the thread cannot accumulate hundreds of later // activities while one is outstanding — a request that has scrolled out of // the window is one whose turn kept running, i.e. it was resolved or went // stale. (The projection pipeline's pendingApprovalCount reads the same // capped stream and stays consistent with this view.) 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; } /** * A queued turn start — a user message no turn has picked up yet — is work * in flight even though session is still null (turn.start emits * message-sent + turn-start-requested; the session arrives later). Detection * mirrors the client's hasQueuedTurnStart: the newest user message is * strictly newer than every latestTurn timestamp (adoption stamps the new * turn's requestedAt with the message time, clearing this), and only within * the adoption grace window — historical threads whose last user message * postdates their turn timestamps (older-server data, mid-turn messages) * must not be blocked forever. A failed session start (status "error") * clears the block immediately. * * The age check is bounded on BOTH sides: message timestamps are * client-supplied, so a client clock ahead of the server yields a negative * age. Without the lower bound that negative age satisfies `<= grace` for * as long as the skew lasts, extending the block far past the intended two * minutes. */ function threadHasQueuedTurnStart( thread: { readonly messages: ReadonlyArray<{ readonly role: string; readonly createdAt: string }>; readonly latestTurn: { readonly requestedAt: string; readonly startedAt: string | null; readonly completedAt: string | null; } | null; readonly session: { readonly status: string } | null; }, occurredAt: string, ): boolean { const latestUserMessageAtMs = thread.messages.reduce( (latest, message) => message.role === "user" ? Math.max(latest, Date.parse(message.createdAt)) : latest, Number.NEGATIVE_INFINITY, ); const latestTurnAtMs = thread.latestTurn === null ? Number.NEGATIVE_INFINITY : Math.max( ...[ thread.latestTurn.requestedAt, thread.latestTurn.startedAt, thread.latestTurn.completedAt, ].map((candidate) => candidate == null ? Number.NEGATIVE_INFINITY : Date.parse(candidate), ), ); const queuedAgeMs = Date.parse(occurredAt) - latestUserMessageAtMs; return ( thread.session?.status !== "error" && Number.isFinite(latestUserMessageAtMs) && latestUserMessageAtMs > latestTurnAtMs && Math.abs(queuedAgeMs) <= QUEUED_TURN_START_GRACE_MS ); } 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, }); 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, defaultModelSelection: command.defaultModelSelection ?? null, faviconPath: 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.faviconPath !== undefined ? { faviconPath: command.faviconPath } : {}), ...(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.settle": { const thread = yield* requireThreadNotArchived({ readModel, command, threadId: command.threadId, }); // Server-side twin of the client's canSettle session check: a stale // or raced client must not settle a thread whose session is coming // alive or working. if (thread.session?.status === "starting" || thread.session?.status === "running") { return yield* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: `thread ${command.threadId} has an active session and cannot be settled`, }), ); } // 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* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: `thread ${command.threadId} has a pending approval or user-input request and cannot be settled`, }), ); } const occurredAt = yield* nowIso; // Settling inside the adoption window would hide just-requested work. if (threadHasQueuedTurnStart(thread, occurredAt)) { return yield* Effect.fail( new OrchestrationCommandInvariantError({ commandType: command.type, detail: `thread ${command.threadId} has a queued turn start and cannot be settled`, }), ); } // 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,