import {  CommandId,  EventId,  MessageId,  type ProjectId,  ThreadId,  TurnId,  type OrchestrationEvent,  type ProviderRuntimeEvent,} from "@t3tools/contracts";import { Cause, Effect, Layer, Option, Stream } from "effect";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { parseTurnDiffFilesFromUnifiedDiff } from "../../checkpointing/Diffs.ts";import {  checkpointRefForThreadTurn,  resolveThreadWorkspaceCwd,} from "../../checkpointing/Utils.ts";import { clearWorkspaceIndexCache } from "../../workspaceEntries.ts";import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts";import { ProviderService } from "../../provider/Services/ProviderService.ts";import { CheckpointReactor, type CheckpointReactorShape } from "../Services/CheckpointReactor.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { RuntimeReceiptBus } from "../Services/RuntimeReceiptBus.ts";import { CheckpointStoreError } from "../../checkpointing/Errors.ts";import { OrchestrationDispatchError } from "../Errors.ts";import { isGitRepository } from "../../git/Utils.ts"; type ReactorInput =  | {      readonly source: "runtime";      readonly event: ProviderRuntimeEvent;    }  | {      readonly source: "domain";      readonly event: OrchestrationEvent;    }; function toTurnId(value: string | undefined): TurnId | null {  return value === undefined ? null : TurnId.makeUnsafe(String(value));} function sameId(left: string | null | undefined, right: string | null | undefined): boolean {  if (left === null || left === undefined || right === null || right === undefined) {    return false;  }  return left === right;} function checkpointStatusFromRuntime(status: string | undefined): "ready" | "missing" | "error" {  switch (status) {    case "failed":      return "error";    case "cancelled":    case "interrupted":      return "missing";    case "completed":    default:      return "ready";  }} const serverCommandId = (tag: string): CommandId =>  CommandId.makeUnsafe(`server:${tag}:${crypto.randomUUID()}`); const make = Effect.gen(function* () {  const orchestrationEngine = yield* OrchestrationEngineService;  const providerService = yield* ProviderService;  const checkpointStore = yield* CheckpointStore;  const receiptBus = yield* RuntimeReceiptBus;   const appendRevertFailureActivity = (input: {    readonly threadId: ThreadId;    readonly turnCount: number;    readonly detail: string;    readonly createdAt: string;  }) =>    orchestrationEngine.dispatch({      type: "thread.activity.append",      commandId: serverCommandId("checkpoint-revert-failure"),      threadId: input.threadId,      activity: {        id: EventId.makeUnsafe(crypto.randomUUID()),        tone: "error",        kind: "checkpoint.revert.failed",        summary: "Checkpoint revert failed",        payload: {          turnCount: input.turnCount,          detail: input.detail,        },        turnId: null,        createdAt: input.createdAt,      },      createdAt: input.createdAt,    });   const appendCaptureFailureActivity = (input: {    readonly threadId: ThreadId;    readonly turnId: TurnId | null;    readonly detail: string;    readonly createdAt: string;  }) =>    orchestrationEngine.dispatch({      type: "thread.activity.append",      commandId: serverCommandId("checkpoint-capture-failure"),      threadId: input.threadId,      activity: {        id: EventId.makeUnsafe(crypto.randomUUID()),        tone: "error",        kind: "checkpoint.capture.failed",        summary: "Checkpoint capture failed",        payload: {          detail: input.detail,        },        turnId: input.turnId,        createdAt: input.createdAt,      },      createdAt: input.createdAt,    });   const resolveSessionRuntimeForThread = Effect.fnUntraced(function* (    threadId: ThreadId,  ): Effect.fn.Return<Option.Option<{ readonly threadId: ThreadId; readonly cwd: string }>> {    const readModel = yield* orchestrationEngine.getReadModel();    const thread = readModel.threads.find((entry) => entry.id === threadId);     const sessions = yield* providerService.listSessions();     const findSessionWithCwd = (      session: (typeof sessions)[number] | undefined,    ): Option.Option<{ readonly threadId: ThreadId; readonly cwd: string }> => {      if (!session?.cwd) {        return Option.none();      }      return Option.some({ threadId: session.threadId, cwd: session.cwd });    };     if (thread) {      const projectedSession = sessions.find((session) => session.threadId === thread.id);      const fromProjected = findSessionWithCwd(projectedSession);      if (Option.isSome(fromProjected)) {        return fromProjected;      }    }     return Option.none();  });   const isGitWorkspace = (cwd: string) => isGitRepository(cwd);   // Resolves the workspace CWD for checkpoint operations, preferring the  // active provider session CWD and falling back to the thread/project config.  // Returns undefined when no CWD can be determined or the workspace is not  // a git repository.  const resolveCheckpointCwd = Effect.fnUntraced(function* (input: {    readonly threadId: ThreadId;    readonly thread: { readonly projectId: ProjectId; readonly worktreePath: string | null };    readonly projects: ReadonlyArray<{ readonly id: ProjectId; readonly workspaceRoot: string }>;    readonly preferSessionRuntime: boolean;  }): Effect.fn.Return<string | undefined> {    const fromSession = yield* resolveSessionRuntimeForThread(input.threadId);    const fromThread = resolveThreadWorkspaceCwd({      thread: input.thread,      projects: input.projects,    });     const cwd = input.preferSessionRuntime      ? (Option.match(fromSession, {          onNone: () => undefined,          onSome: (runtime) => runtime.cwd,        }) ?? fromThread)      : (fromThread ??        Option.match(fromSession, {          onNone: () => undefined,          onSome: (runtime) => runtime.cwd,        }));     if (!cwd) {      return undefined;    }    if (!isGitWorkspace(cwd)) {      return undefined;    }    return cwd;  });   // Shared tail for both capture paths: creates the git checkpoint ref, diffs  // it against the previous turn, then dispatches the domain events to update  // the orchestration read model.  const captureAndDispatchCheckpoint = Effect.fnUntraced(function* (input: {    readonly threadId: ThreadId;    readonly turnId: TurnId;    readonly thread: {      readonly messages: ReadonlyArray<{        readonly id: MessageId;        readonly role: string;        readonly turnId: TurnId | null;      }>;    };    readonly cwd: string;    readonly turnCount: number;    readonly status: "ready" | "missing" | "error";    readonly assistantMessageId: MessageId | undefined;    readonly createdAt: string;  }) {    const fromTurnCount = Math.max(0, input.turnCount - 1);    const fromCheckpointRef = checkpointRefForThreadTurn(input.threadId, fromTurnCount);    const targetCheckpointRef = checkpointRefForThreadTurn(input.threadId, input.turnCount);     const fromCheckpointExists = yield* checkpointStore.hasCheckpointRef({      cwd: input.cwd,      checkpointRef: fromCheckpointRef,    });    if (!fromCheckpointExists) {      yield* Effect.logWarning("checkpoint capture missing pre-turn baseline", {        threadId: input.threadId,        turnId: input.turnId,        fromTurnCount,      });    }     yield* checkpointStore.captureCheckpoint({      cwd: input.cwd,      checkpointRef: targetCheckpointRef,    });     // Invalidate the workspace entry cache so the @-mention file picker    // reflects files created or deleted during this turn.    clearWorkspaceIndexCache(input.cwd);     const files = yield* checkpointStore      .diffCheckpoints({        cwd: input.cwd,        fromCheckpointRef,        toCheckpointRef: targetCheckpointRef,        fallbackFromToHead: false,      })      .pipe(        Effect.map((diff) =>          parseTurnDiffFilesFromUnifiedDiff(diff).map((file) => ({            path: file.path,            kind: "modified" as const,            additions: file.additions,            deletions: file.deletions,          })),        ),        Effect.tapError((error) =>          appendCaptureFailureActivity({            threadId: input.threadId,            turnId: input.turnId,            detail: `Checkpoint captured, but turn diff summary is unavailable: ${error.message}`,            createdAt: input.createdAt,          }),        ),        Effect.catch((error) =>          Effect.logWarning("failed to derive checkpoint file summary", {            threadId: input.threadId,            turnId: input.turnId,            turnCount: input.turnCount,            detail: error.message,          }).pipe(Effect.as([])),        ),      );     const assistantMessageId =      input.assistantMessageId ??      input.thread.messages        .toReversed()        .find((entry) => entry.role === "assistant" && entry.turnId === input.turnId)?.id ??      MessageId.makeUnsafe(`assistant:${input.turnId}`);     yield* orchestrationEngine.dispatch({      type: "thread.turn.diff.complete",      commandId: serverCommandId("checkpoint-turn-diff-complete"),      threadId: input.threadId,      turnId: input.turnId,      completedAt: input.createdAt,      checkpointRef: targetCheckpointRef,      status: input.status,      files,      assistantMessageId,      checkpointTurnCount: input.turnCount,      createdAt: input.createdAt,    });    yield* receiptBus.publish({      type: "checkpoint.diff.finalized",      threadId: input.threadId,      turnId: input.turnId,      checkpointTurnCount: input.turnCount,      checkpointRef: targetCheckpointRef,      status: input.status,      createdAt: input.createdAt,    });    yield* receiptBus.publish({      type: "turn.processing.quiesced",      threadId: input.threadId,      turnId: input.turnId,      checkpointTurnCount: input.turnCount,      createdAt: input.createdAt,    });     yield* orchestrationEngine.dispatch({      type: "thread.activity.append",      commandId: serverCommandId("checkpoint-captured-activity"),      threadId: input.threadId,      activity: {        id: EventId.makeUnsafe(crypto.randomUUID()),        tone: "info",        kind: "checkpoint.captured",        summary: "Checkpoint captured",        payload: {          turnCount: input.turnCount,          status: input.status,        },        turnId: input.turnId,        createdAt: input.createdAt,      },      createdAt: input.createdAt,    });  });   // Captures a real git checkpoint when a turn completes via a runtime event.  const captureCheckpointFromTurnCompletion = Effect.fnUntraced(function* (    event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>,  ) {    const turnId = toTurnId(event.turnId);    if (!turnId) {      return;    }     const readModel = yield* orchestrationEngine.getReadModel();    const thread = readModel.threads.find((entry) => entry.id === event.threadId);    if (!thread) {      return;    }     // When a primary turn is active, only that turn may produce completion checkpoints.    if (thread.session?.activeTurnId && !sameId(thread.session.activeTurnId, turnId)) {      return;    }     // Only skip if a real (non-placeholder) checkpoint already exists for this turn.    // ProviderRuntimeIngestion may insert placeholder entries with status "missing"    // before this reactor runs; those must not prevent real git capture.    if (      thread.checkpoints.some(        (checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",      )    ) {      return;    }     const checkpointCwd = yield* resolveCheckpointCwd({      threadId: thread.id,      thread,      projects: readModel.projects,      preferSessionRuntime: true,    });    if (!checkpointCwd) {      return;    }     // If a placeholder checkpoint exists for this turn, reuse its turn count    // instead of incrementing past it.    const existingPlaceholder = thread.checkpoints.find(      (checkpoint) => checkpoint.turnId === turnId && checkpoint.status === "missing",    );    const currentTurnCount = thread.checkpoints.reduce(      (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),      0,    );    const nextTurnCount = existingPlaceholder      ? existingPlaceholder.checkpointTurnCount      : currentTurnCount + 1;     yield* captureAndDispatchCheckpoint({      threadId: thread.id,      turnId,      thread,      cwd: checkpointCwd,      turnCount: nextTurnCount,      status: checkpointStatusFromRuntime(event.payload.state),      assistantMessageId: undefined,      createdAt: event.createdAt,    });  });   // Captures a real git checkpoint when a placeholder checkpoint (status "missing")  // is detected via a domain event. This replaces the placeholder with a real  // git-ref-based checkpoint.  //  // ProviderRuntimeIngestion creates placeholder checkpoints on turn.diff.updated  // events from the Codex runtime. This handler fires when the corresponding  // domain event arrives, allowing the reactor to capture the actual filesystem  // state into a git ref and dispatch a replacement checkpoint.  const captureCheckpointFromPlaceholder = Effect.fnUntraced(function* (    event: Extract<OrchestrationEvent, { type: "thread.turn-diff-completed" }>,  ) {    const { threadId, turnId, checkpointTurnCount, status } = event.payload;     // Only replace placeholders; skip events from our own real captures.    if (status !== "missing") {      return;    }     const readModel = yield* orchestrationEngine.getReadModel();    const thread = readModel.threads.find((entry) => entry.id === threadId);    if (!thread) {      yield* Effect.logWarning("checkpoint capture from placeholder skipped: thread not found", {        threadId,      });      return;    }     // If a real checkpoint already exists for this turn, skip.    if (      thread.checkpoints.some(        (checkpoint) => checkpoint.turnId === turnId && checkpoint.status !== "missing",      )    ) {      yield* Effect.logDebug(        "checkpoint capture from placeholder skipped: real checkpoint already exists",        { threadId, turnId },      );      return;    }     const checkpointCwd = yield* resolveCheckpointCwd({      threadId,      thread,      projects: readModel.projects,      preferSessionRuntime: true,    });    if (!checkpointCwd) {      return;    }     yield* captureAndDispatchCheckpoint({      threadId,      turnId,      thread,      cwd: checkpointCwd,      turnCount: checkpointTurnCount,      status: "ready",      assistantMessageId: event.payload.assistantMessageId ?? undefined,      createdAt: event.payload.completedAt,    });  });   const ensurePreTurnBaselineFromTurnStart = Effect.fnUntraced(function* (    event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>,  ) {    const turnId = toTurnId(event.turnId);    if (!turnId) {      return;    }     const readModel = yield* orchestrationEngine.getReadModel();    const thread = readModel.threads.find((entry) => entry.id === event.threadId);    if (!thread) {      return;    }     const checkpointCwd = yield* resolveCheckpointCwd({      threadId: thread.id,      thread,      projects: readModel.projects,      preferSessionRuntime: false,    });    if (!checkpointCwd) {      return;    }     const currentTurnCount = thread.checkpoints.reduce(      (maxTurnCount, checkpoint) => Math.max(maxTurnCount, checkpoint.checkpointTurnCount),      0,    );    const baselineCheckpointRef = checkpointRefForThreadTurn(thread.id, currentTurnCount);    const baselineExists = yield* checkpointStore.hasCheckpointRef({      cwd: checkpointCwd,      checkpointRef: baselineCheckpointRef,    });    if (baselineExists) {      return;    }     yield* checkpointStore.captureCheckpoint({      cwd: checkpointCwd,      checkpointRef: baselineCheckpointRef,    });    yield* receiptBus.publish({      type: "checkpoint.baseline.captured",      threadId: thread.id,      checkpointTurnCount: currentTurnCount,      checkpointRef: baselineCheckpointRef,      createdAt: event.createdAt,    });  });   const ensurePreTurnBaselineFromDomainTurnStart = Effect.fnUntraced(function* (

Lines 1–500 of 800. Load more