import {  CommandId,  type CheckpointRef,  EventId,  MessageId,  type ProjectId,  ThreadId,  TurnId,  type OrchestrationEvent,  type ProviderRuntimeEvent,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import type * as PlatformError from "effect/PlatformError";import * as Stream from "effect/Stream";import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { parseTurnDiffFilesFromUnifiedDiff } from "../../checkpointing/Diffs.ts";import {  checkpointRefForThreadTurn,  resolveThreadWorkspaceCwd,} from "../../checkpointing/Utils.ts";import * as CheckpointStore from "../../checkpointing/CheckpointStore.ts";import { ProviderService } from "../../provider/Services/ProviderService.ts";import { CheckpointReactor, type CheckpointReactorShape } from "../Services/CheckpointReactor.ts";import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts";import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts";import { RuntimeReceiptBus } from "../Services/RuntimeReceiptBus.ts";import type { CheckpointStoreError } from "../../checkpointing/Errors.ts";import type { OrchestrationDispatchError } from "../Errors.ts";import { isGitRepository } from "../../git/Utils.ts";import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); 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.make(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 make = Effect.gen(function* () {  const crypto = yield* Crypto.Crypto;  const randomUUID = crypto.randomUUIDv4;  const serverEventId = randomUUID.pipe(Effect.map(EventId.make));  const serverCommandId = (tag: string) =>    randomUUID.pipe(Effect.map((uuid) => CommandId.make(`server:${tag}:${uuid}`)));  const orchestrationEngine = yield* OrchestrationEngineService;  const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;  const providerService = yield* ProviderService;  const checkpointStore = yield* CheckpointStore.CheckpointStore;  const receiptBus = yield* RuntimeReceiptBus;  const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;  const vcsStatusBroadcaster = yield* VcsStatusBroadcaster;   const appendRevertFailureActivity = (input: {    readonly threadId: ThreadId;    readonly turnCount: number;    readonly detail: string;    readonly createdAt: string;  }) =>    Effect.all({      commandId: serverCommandId("checkpoint-revert-failure"),      activityId: serverEventId,    }).pipe(      Effect.flatMap(({ commandId, activityId }) =>        orchestrationEngine.dispatch({          type: "thread.activity.append",          commandId,          threadId: input.threadId,          activity: {            id: activityId,            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;  }) =>    Effect.all({      commandId: serverCommandId("checkpoint-capture-failure"),      activityId: serverEventId,    }).pipe(      Effect.flatMap(({ commandId, activityId }) =>        orchestrationEngine.dispatch({          type: "thread.activity.append",          commandId,          threadId: input.threadId,          activity: {            id: activityId,            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.fn("resolveSessionRuntimeForThread")(function* (    threadId: ThreadId,  ): Effect.fn.Return<Option.Option<{ readonly threadId: ThreadId; readonly cwd: string }>> {    const sessions = yield* providerService.listSessions();    const session = sessions.find((entry) => entry.threadId === threadId);    return session?.cwd      ? Option.some({ threadId: session.threadId, cwd: session.cwd })      : Option.none();  });   const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (threadId: ThreadId) {    return yield* projectionSnapshotQuery      .getThreadDetailById(threadId)      .pipe(Effect.map(Option.getOrUndefined));  });   const resolveThreadProjects = Effect.fn("resolveThreadProjects")(function* (    projectId: ProjectId,  ) {    const project = yield* projectionSnapshotQuery      .getProjectShellById(projectId)      .pipe(Effect.map(Option.getOrUndefined));    return project ? [project] : [];  });   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.fn("resolveCheckpointCwd")(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.fn("captureAndDispatchCheckpoint")(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,    });     // Refresh the workspace entry index so the @-mention file picker    // reflects files created or deleted during this turn.    yield* workspaceEntries.refresh(input.cwd);     const files = yield* checkpointStore      .diffCheckpoints({        cwd: input.cwd,        fromCheckpointRef,        toCheckpointRef: targetCheckpointRef,        fallbackFromToHead: false,        ignoreWhitespace: 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.make(`assistant:${input.turnId}`);     yield* orchestrationEngine.dispatch({      type: "thread.turn.diff.complete",      commandId: yield* 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: yield* serverCommandId("checkpoint-captured-activity"),      threadId: input.threadId,      activity: {        id: EventId.make(yield* 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.fn("captureCheckpointFromTurnCompletion")(    function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>) {      const turnId = toTurnId(event.turnId);      if (!turnId) {        return;      }       const thread = yield* resolveThreadDetail(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 projects = yield* resolveThreadProjects(thread.projectId);      const checkpointCwd = yield* resolveCheckpointCwd({        threadId: thread.id,        thread,        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.fn("captureCheckpointFromPlaceholder")(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 thread = yield* resolveThreadDetail(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 projects = yield* resolveThreadProjects(thread.projectId);    const checkpointCwd = yield* resolveCheckpointCwd({      threadId,      thread,      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.fn("ensurePreTurnBaselineFromTurnStart")(    function* (event: Extract<ProviderRuntimeEvent, { type: "turn.started" }>) {      const turnId = toTurnId(event.turnId);      if (!turnId) {        return;      }       const thread = yield* resolveThreadDetail(event.threadId);      if (!thread) {        return;      }       const projects = yield* resolveThreadProjects(thread.projectId);      const checkpointCwd = yield* resolveCheckpointCwd({        threadId: thread.id,        thread,        projects,        preferSessionRuntime: false,      });      if (!checkpointCwd) {        return;      }

Lines 1–500 of 866. Load more