import {  CommandId,  type CheckpointRef,  EventId,  MessageId,  type ProjectId,  ThreadId,  TurnId,  type OrchestrationEvent,  type ProviderRuntimeEvent,  type VcsStatusLocalResult,} 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 { isTemporaryWorktreeBranch } from "@t3tools/shared/git"; import { parseTurnDiffFilesFromNumstat } 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 { forkParked } from "../../serverActivation.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 { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts";import * as WorkspaceEntries from "../../workspace/WorkspaceEntries.ts";import * as PullRequestService from "../../pullRequest/PullRequestService.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 pullRequests = yield* PullRequestService.PullRequestService;  const startedTurns = new Map<ThreadId, TurnId>();  const pending = new Set<ThreadId>();   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, { activityKinds: [] })      .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] : [];  });   // 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, CheckpointStoreError> {    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 (!(yield* checkpointStore.isGitRepository(cwd))) {      return undefined;    }    return cwd;  });   // Capture the completed turn's files, then publish its summary and receipts.  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);     // Git may have been initialized during this turn, leaving no pre-turn    // snapshot. Keep the completion checkpoint for future turns, but do not    // invent a baseline or attempt a diff against a ref that does not exist.    const files = yield* (      fromCheckpointExists        ? checkpointStore.diffCheckpoints({            cwd: input.cwd,            fromCheckpointRef,            toCheckpointRef: targetCheckpointRef,            fallbackFromToHead: false,            ignoreWhitespace: false,            format: "numstat",          })        : Effect.succeed("")    ).pipe(      Effect.map((diff) =>        parseTurnDiffFilesFromNumstat(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,    });  });   // Capture the files left by a completed or interrupted turn.  const captureCheckpointFromTurnCompletion = Effect.fn("captureCheckpointFromTurnCompletion")(    function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" | "turn.aborted" }>) {      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:          event.type === "turn.aborted"            ? "ready"            : checkpointStatusFromRuntime(event.payload.state),        assistantMessageId: existingPlaceholder?.assistantMessageId ?? undefined,        createdAt: event.createdAt,      });    },  );   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;      }       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 refreshLocalGitStatusFromTurnCompletion = Effect.fn(    "refreshLocalGitStatusFromTurnCompletion",  )(function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>) {    const sessionRuntime = yield* resolveSessionRuntimeForThread(event.threadId);    if (Option.isNone(sessionRuntime)) {      return;    }     const local = yield* vcsStatusBroadcaster.refreshLocalStatus(sessionRuntime.value.cwd).pipe(      Effect.catch((error) =>        Effect.logWarning("failed to refresh local git status after turn completion", {          threadId: event.threadId,          turnId: event.turnId ?? null,          cwd: sessionRuntime.value.cwd,          detail: error.message,        }).pipe(Effect.as(null)),      ),    );    if (local !== null) {      yield* followWorktreeBranchDrift({        threadId: event.threadId,        cwd: sessionRuntime.value.cwd,        local,

Lines 1–500 of 967. Load more