import type {
  OrchestrationV2AppThread,
  OrchestrationV2ConversationMessage,
  OrchestrationV2DomainEvent,
  OrchestrationV2ProjectedTurnItem,
  OrchestrationV2ProviderTurn,
  OrchestrationV2Run,
  OrchestrationV2Subagent,
  OrchestrationV2ThreadShellSnapshot,
  OrchestrationV2ShellThreadStatus,
  OrchestrationV2ThreadShell,
  OrchestrationV2ThreadProjection,
  OrchestrationV2TurnItem,
  ProviderSessionId,
  ProviderThreadId,
  ProviderTurnId,
  RunAttemptId,
  RuntimeRequestId,
  MessageId,
} from "@t3tools/contracts";
import {
  OrchestrationV2AppThreadJson as OrchestrationV2AppThreadJsonSchema,
  OrchestrationV2CheckpointJson as OrchestrationV2CheckpointJsonSchema,
  OrchestrationV2CheckpointScopeJson as OrchestrationV2CheckpointScopeJsonSchema,
  OrchestrationV2ContextHandoffJson as OrchestrationV2ContextHandoffJsonSchema,
  OrchestrationV2ContextTransferJson as OrchestrationV2ContextTransferJsonSchema,
  OrchestrationV2ConversationMessageJson as OrchestrationV2ConversationMessageJsonSchema,
  OrchestrationV2ExecutionNodeJson as OrchestrationV2ExecutionNodeJsonSchema,
  OrchestrationV2PlanArtifact as OrchestrationV2PlanArtifactSchema,
  OrchestrationV2ProviderSessionJson as OrchestrationV2ProviderSessionJsonSchema,
  OrchestrationV2ProviderThreadJson as OrchestrationV2ProviderThreadJsonSchema,
  OrchestrationV2ProviderTurnJson as OrchestrationV2ProviderTurnJsonSchema,
  OrchestrationV2RunAttemptJson as OrchestrationV2RunAttemptJsonSchema,
  OrchestrationV2RunJson as OrchestrationV2RunJsonSchema,
  OrchestrationV2RuntimeRequestJson as OrchestrationV2RuntimeRequestJsonSchema,
  OrchestrationV2SubagentJson as OrchestrationV2SubagentJsonSchema,
  OrchestrationV2TurnItemJson as OrchestrationV2TurnItemJsonSchema,
  RunId,
  ThreadId,
  TurnItemId,
} from "@t3tools/contracts";
import {
  isOrchestrationV2SupersededInterrupt,
  isOrchestrationV2TurnItemVisible,
} from "@t3tools/shared/orchestrationV2Timeline";
import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
import * as Context from "effect/Context";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Ref from "effect/Ref";
import * as Schema from "effect/Schema";
import * as SqlClient from "effect/unstable/sql/SqlClient";

export class ProjectionStoreApplyEventError extends Schema.TaggedErrorClass<ProjectionStoreApplyEventError>()(
  "ProjectionStoreApplyEventError",
  {
    eventType: Schema.String,
    cause: Schema.optional(Schema.Defect()),
  },
) {
  override get message(): string {
    return `Failed to apply orchestration projection event ${this.eventType}.`;
  }
}

export class ProjectionStoreSetupError extends Schema.TaggedErrorClass<ProjectionStoreSetupError>()(
  "ProjectionStoreSetupError",
  {
    cause: Schema.optional(Schema.Defect()),
  },
) {
  override get message(): string {
    return "Failed to initialize orchestration projection store.";
  }
}

export class ProjectionStoreThreadNotFoundError extends Schema.TaggedErrorClass<ProjectionStoreThreadNotFoundError>()(
  "ProjectionStoreThreadNotFoundError",
  {
    threadId: ThreadId,
  },
) {
  override get message(): string {
    return `No orchestration projection exists for thread ${this.threadId}.`;
  }
}

export class ProjectionStoreReadError extends Schema.TaggedErrorClass<ProjectionStoreReadError>()(
  "ProjectionStoreReadError",
  {
    threadId: ThreadId,
    cause: Schema.optional(Schema.Defect()),
  },
) {
  override get message(): string {
    return `Failed to read orchestration projection for thread ${this.threadId}.`;
  }
}

export const ProjectionStoreV2Error = Schema.Union([
  ProjectionStoreSetupError,
  ProjectionStoreApplyEventError,
  ProjectionStoreThreadNotFoundError,
  ProjectionStoreReadError,
]);
export type ProjectionStoreV2Error = typeof ProjectionStoreV2Error.Type;

export type ProjectionRecoveryKind =
  | "queued-runs"
  | "runtime"
  | "subagent-results"
  | "delegated-completions";

/** Thread activity needed by settlement, without transcript or fork history. */
export type ProjectionSettlementCandidate = Pick<
  OrchestrationV2ThreadShell,
  | "id"
  | "projectId"
  | "branch"
  | "worktreePath"
  | "linkedPullRequest"
  | "createdAt"
  | "updatedAt"
  | "archivedAt"
  | "settledOverride"
  | "pinnedAt"
  | "snoozedUntil"
  | "snoozedAt"
  | "latestRunId"
  | "latestRunRequestedAt"
  | "latestRunStartedAt"
  | "latestRunCompletedAt"
  | "latestUserMessageAt"
  | "status"
  | "activityRunStatus"
  | "pendingRuntimeRequest"
  | "pendingBackgroundTasks"
>;

const ProjectionCheckpointContext = Schema.Struct({
  runs: Schema.Array(
    OrchestrationV2RunJsonSchema.mapFields(({ id, ordinal, status }) => ({ id, ordinal, status })),
  ),
  checkpointScopes: Schema.Array(
    OrchestrationV2CheckpointScopeJsonSchema.mapFields(({ id, runId, kind, cwd }) => ({
      id,
      runId,
      kind,
      cwd,
    })),
  ),
  checkpoints: Schema.Array(
    OrchestrationV2CheckpointJsonSchema.mapFields(
      ({ scopeId, runId, appRunOrdinal, status, ref }) => ({
        scopeId,
        runId,
        appRunOrdinal,
        status,
        ref,
      }),
    ),
  ),
});
export type ProjectionCheckpointContext = typeof ProjectionCheckpointContext.Type;
const decodeCheckpointContext = Schema.decodeUnknownEffect(ProjectionCheckpointContext);

/** Exact durable targets used by interrupt, restart, and steering effects. */
export interface ProjectionProviderControlContext {
  readonly providerThread: OrchestrationV2ThreadProjection["providerThreads"][number] | undefined;
  readonly providerTurn: OrchestrationV2ProviderTurn | undefined;
  readonly attempt: OrchestrationV2ThreadProjection["attempts"][number] | undefined;
  readonly message: OrchestrationV2ConversationMessage | undefined;
  readonly run: OrchestrationV2Run | undefined;
}

export interface ProjectionProviderControlTarget {
  readonly providerThreadId: ProviderThreadId;
  readonly providerTurnId: ProviderTurnId;
  readonly attemptId?: RunAttemptId;
  readonly messageId?: MessageId;
}

export type ProjectionPendingUserInputs = Pick<
  OrchestrationV2ThreadProjection,
  "runtimeRequests" | "nodes" | "turnItems"
>;

export interface ProjectionStoreV2Shape {
  readonly apply: (
    event: OrchestrationV2DomainEvent,
  ) => Effect.Effect<void, ProjectionStoreV2Error>;
  readonly getShellSnapshot: (options?: {
    readonly location?: "active" | "archive";
  }) => Effect.Effect<OrchestrationV2ThreadShellSnapshot, ProjectionStoreV2Error>;
  readonly getThreadShell: (
    threadId: ThreadId,
  ) => Effect.Effect<OrchestrationV2ThreadShell | null, ProjectionStoreV2Error>;
  readonly getThread: (
    threadId: ThreadId,
  ) => Effect.Effect<OrchestrationV2AppThread, ProjectionStoreV2Error>;
  readonly getSettlementCandidates: () => Effect.Effect<
    ReadonlyArray<ProjectionSettlementCandidate>,
    ProjectionStoreV2Error
  >;
  readonly getThreadProjection: (
    threadId: ThreadId,
  ) => Effect.Effect<OrchestrationV2ThreadProjection, ProjectionStoreV2Error>;
  readonly getPendingNativeUserInputs: (
    threadId: ThreadId,
    providerTurnId: ProviderTurnId,
  ) => Effect.Effect<ProjectionPendingUserInputs, ProjectionStoreV2Error>;
  readonly getRuntimeRequest: (
    threadId: ThreadId,
    requestId: RuntimeRequestId,
  ) => Effect.Effect<
    OrchestrationV2ThreadProjection["runtimeRequests"][number] | undefined,
    ProjectionStoreV2Error
  >;
  readonly getProviderControlContext: (
    threadId: ThreadId,
    target: ProjectionProviderControlTarget,
  ) => Effect.Effect<ProjectionProviderControlContext, ProjectionStoreV2Error>;
  readonly getCheckpointContext: (
    threadId: ThreadId,
  ) => Effect.Effect<ProjectionCheckpointContext, ProjectionStoreV2Error>;
  readonly getRecoveryThreadIds: (
    kind: ProjectionRecoveryKind,
  ) => Effect.Effect<ReadonlyArray<ThreadId>, ProjectionStoreV2Error>;
  readonly getUnreadableThreadIds: () => Effect.Effect<
    ReadonlyArray<ThreadId>,
    ProjectionStoreV2Error
  >;
  readonly getThreadSnapshot: (threadId: ThreadId) => Effect.Effect<
    {
      readonly schemaVersion: number;
      readonly snapshotSequence: number;
      readonly projection: OrchestrationV2ThreadProjection;
    },
    ProjectionStoreV2Error
  >;
  readonly getThreadSnapshotWindow: (
    threadId: ThreadId,
    options: {
      readonly rowLimit: number;
      readonly anchorItemId?: TurnItemId | undefined;
      readonly anchorThreadId?: ThreadId | undefined;
      readonly requiredRunId?: RunId | undefined;
    },
  ) => Effect.Effect<
    {
      readonly schemaVersion: number;
      readonly snapshotSequence: number;
      readonly projection: OrchestrationV2ThreadProjection;
    },
    ProjectionStoreV2Error
  >;
}

export class ProjectionStoreV2 extends Context.Service<ProjectionStoreV2, ProjectionStoreV2Shape>()(
  "t3/orchestration-v2/ProjectionStore/ProjectionStoreV2",
) {}

export const ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION = 2;

function needsRecovery(
  projection: OrchestrationV2ThreadProjection,
  kind: ProjectionRecoveryKind,
): boolean {
  if (projection.thread.deletedAt !== null) return false;
  switch (kind) {
    case "queued-runs":
      return (
        projection.thread.archivedAt === null &&
        projection.runs.some((run) => run.status === "queued") &&
        !projection.runs.some((run) =>
          ["preparing", "starting", "running", "waiting"].includes(run.status),
        )
      );
    case "delegated-completions":
      return projection.runs.some((run) => run.delegatedCompletion?.delivery != null);
    case "subagent-results": {
      const parentThreadId = projection.thread.lineage.parentThreadId;
      return (
        projection.thread.lineage.relationshipToParent === "subagent" &&
        parentThreadId !== null &&
        projection.thread.forkedFrom?.type === "node" &&
        ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
          projection.runs.at(-1)?.status ?? "idle",
        ) &&
        !projection.contextTransfers.some(
          (transfer) =>
            transfer.type === "subagent_result" &&
            transfer.sourceThreadId === projection.thread.id &&
            transfer.targetThreadId === parentThreadId,
        )
      );
    }
    case "runtime":
      return (
        projection.runs.some((run) =>
          ["queued", "preparing", "starting", "running", "waiting"].includes(run.status),
        ) ||
        projection.runtimeRequests.some((request) => request.status === "pending") ||
        projection.providerSessions.some(
          (session) => session.status !== "stopped" && session.status !== "error",
        ) ||
        projection.providerThreads.some(
          (thread) =>
            thread.status === "active" || (thread.pendingBackgroundTasks?.length ?? 0) > 0,
        ) ||
        projection.turnItems.some(
          (item) =>
            ["command_execution", "dynamic_tool", "subagent"].includes(item.type) &&
            ["pending", "running", "waiting"].includes(item.status),
        )
      );
  }
}

function upsertById<T extends { readonly id: string }>(items: ReadonlyArray<T>, next: T): Array<T> {
  const index = items.findIndex((item) => item.id === next.id);
  if (index === -1) {
    return [...items, next];
  }

  const updated = [...items];
  updated[index] = next;
  return updated;
}

export function upsertProviderTurn(
  turns: ReadonlyArray<OrchestrationV2ProviderTurn>,
  next: OrchestrationV2ProviderTurn,
): Array<OrchestrationV2ProviderTurn> {
  const current = turns.find((turn) => turn.id === next.id);
  return upsertById(turns, {
    ...next,
    ...((next.tokenUsage ?? current?.tokenUsage) === undefined
      ? {}
      : { tokenUsage: next.tokenUsage ?? current?.tokenUsage }),
  });
}

function preserveDelegatedCompletion(
  current: OrchestrationV2Run | undefined,
  next: OrchestrationV2Run,
): OrchestrationV2Run {
  if (next.delegatedCompletion !== undefined || current?.delegatedCompletion === undefined) {
    return next;
  }
  return { ...next, delegatedCompletion: current.delegatedCompletion };
}

function preserveCompletionDelivery(
  current: OrchestrationV2Subagent | undefined,
  next: OrchestrationV2Subagent,
): OrchestrationV2Subagent {
  if (next.completionDelivery !== undefined || current?.completionDelivery === undefined) {
    return next;
  }
  return { ...next, completionDelivery: current.completionDelivery };
}

export function emptyProjection(
  event: Extract<OrchestrationV2DomainEvent, { readonly type: "thread.created" }>,
): OrchestrationV2ThreadProjection {
  return {
    thread: event.payload,
    runs: [],
    attempts: [],
    nodes: [],
    subagents: [],
    providerSessions: [],
    providerThreads: [],
    providerTurns: [],
    runtimeRequests: [],
    messages: [],
    plans: [],
    turnItems: [],
    checkpointScopes: [],
    checkpoints: [],
    contextHandoffs: [],
    contextTransfers: [],
    visibleTurnItems: [],
    updatedAt: event.occurredAt,
  };
}

export function applyToProjection(
  projection: OrchestrationV2ThreadProjection,
  event: OrchestrationV2DomainEvent,
): OrchestrationV2ThreadProjection {
  const base = {
    ...projection,
    thread: {
      ...projection.thread,
      updatedAt: event.occurredAt,
    },
    updatedAt: event.occurredAt,
  };

  switch (event.type) {
    case "thread.created":
    case "thread.archived":
    case "thread.unarchived":
    case "thread.deleted":
    case "thread.settled":
    case "thread.unsettled":
    case "thread.snoozed":
    case "thread.unsnoozed":
    case "thread.pinned":
    case "thread.unpinned":
    case "thread.pin-reordered":
    case "thread.metadata-updated":
    case "thread.runtime-mode-updated":
    case "thread.interaction-mode-updated":
    case "thread.model-selection-updated":
    case "thread.provider-switched":
      return {
        ...base,
        thread: event.payload,
      };
    // Visited tracking is read state, not activity: skip the updatedAt bump so
    // viewing a thread does not surface it as recently active.
    case "thread.visited":
    case "thread.marked-unread":
      return {
        ...projection,
        thread: event.payload,
      };
    case "run.created":
    case "run.updated":
      return withLocalVisibleTurnItems({
        ...base,
        runs: upsertById(
          base.runs,
          preserveDelegatedCompletion(
            base.runs.find((run) => run.id === event.payload.id),
            event.payload,
          ),
        ),
      });
    case "run-attempt.created":
    case "run-attempt.updated":
      return withLocalVisibleTurnItems({
        ...base,
        attempts: upsertById(base.attempts, event.payload),
      });
    case "node.updated":
      return {
        ...base,
        nodes: upsertById(base.nodes, event.payload),
      };
    case "subagent.updated":
      return {
        ...base,
        subagents: upsertById(
          base.subagents,
          preserveCompletionDelivery(
            base.subagents.find((task) => task.id === event.payload.id),
            event.payload,
          ),
        ),
      };
    case "provider-session.attached":
    case "provider-session.updated":
      return {
        ...base,
        providerSessions: upsertById(base.providerSessions, event.payload),
      };
    case "provider-session.detached":
      return {
        ...base,
        providerSessions: base.providerSessions.filter(
          (session) => session.id !== event.payload.providerSessionId,
        ),
      };
    case "provider-thread.updated":
      return {
        ...base,
        thread:
          event.payload.appThreadId === base.thread.id
            ? {
                ...base.thread,
                activeProviderThreadId: event.payload.id,
              }
            : base.thread,
        providerThreads: upsertById(base.providerThreads, event.payload),
      };
    case "provider-turn.updated":
      return {
        ...base,
        providerTurns: upsertProviderTurn(base.providerTurns, event.payload),
      };
    case "runtime-request.updated":
      return {
        ...base,
        runtimeRequests: upsertById(base.runtimeRequests, event.payload),
      };
    case "message.updated":
      return {
        ...base,
        messages: upsertById(base.messages, event.payload),
      };
    case "turn-item.updated":
      return withLocalVisibleTurnItems({
        ...base,
        turnItems: upsertById(base.turnItems, event.payload),
      });
    case "plan.updated":
      return {
        ...base,
        plans: upsertById(base.plans, event.payload),
      };
    case "checkpoint-scope.created":
      return {
        ...base,
        checkpointScopes: upsertById(base.checkpointScopes, event.payload),
      };
    case "checkpoint.captured":
      return {
        ...base,
        checkpoints: upsertById(base.checkpoints, event.payload),
      };
    case "checkpoint.rollback-requested":
      return base;
    case "context-handoff.updated":
      return {
        ...base,
        contextHandoffs: upsertById(base.contextHandoffs, event.payload),
      };
    case "context-transfer.created":
    case "context-transfer.updated":
      return {
        ...base,
        contextTransfers: upsertById(base.contextTransfers, event.payload),
      };
  }
}

/**
 * Replay state for entities whose persisted projection is shared across thread bindings.
 *
 * Provider sessions are process-scoped: one session row can be bound to several app
 * threads. Updating that row changes what every bound thread reads, even though the
 * application event itself belongs to one thread stream. Keeping the binding index here
 * makes in-memory replay match the normalized SQL projection without scanning every
 * thread for every session event.
 */
export interface ProjectionReplayState {
  readonly projections: Map<ThreadId, OrchestrationV2ThreadProjection>;
  readonly providerSessionThreadIds: Map<ProviderSessionId, ReadonlySet<ThreadId>>;
}

export function makeProjectionReplayState(): ProjectionReplayState {
  return {
    projections: new Map(),
    providerSessionThreadIds: new Map(),
  };
}

export function applyToProjectionReplayState(
  state: ProjectionReplayState,
  event: OrchestrationV2DomainEvent,
): boolean {
  if (event.type === "thread.created" && !state.projections.has(event.threadId)) {
    state.projections.set(event.threadId, emptyProjection(event));
    return true;
  }

  const current = state.projections.get(event.threadId);
  if (current === undefined) {
    return false;
  }

  let next = applyToProjection(current, event);
  if (event.type === "provider-session.updated") {
    const boundThreadIds = state.providerSessionThreadIds.get(event.payload.id);
    if (boundThreadIds?.has(event.threadId) !== true) {
      // The SQL projection updates the global session row but does not implicitly
      // create a thread binding for an update event.
      next = { ...next, providerSessions: current.providerSessions };
    }
  }
  state.projections.set(event.threadId, next);

  switch (event.type) {
    case "provider-session.attached": {
      const boundThreadIds = new Set(state.providerSessionThreadIds.get(event.payload.id) ?? []);
      boundThreadIds.add(event.threadId);
      state.providerSessionThreadIds.set(event.payload.id, boundThreadIds);
      for (const threadId of boundThreadIds) {
        if (threadId === event.threadId) continue;
        const projection = state.projections.get(threadId);
        if (projection === undefined) continue;
        state.projections.set(threadId, {
          ...projection,
          providerSessions: upsertById(projection.providerSessions, event.payload),
        });
      }
      break;
    }
    case "provider-session.updated": {
      const boundThreadIds = state.providerSessionThreadIds.get(event.payload.id) ?? [];
      for (const threadId of boundThreadIds) {
        if (threadId === event.threadId) continue;
        const projection = state.projections.get(threadId);
        if (projection === undefined) continue;
        state.projections.set(threadId, {
          ...projection,
          providerSessions: upsertById(projection.providerSessions, event.payload),
        });
      }
      break;
    }
    case "provider-session.detached": {
      const boundThreadIds = new Set(
        state.providerSessionThreadIds.get(event.payload.providerSessionId) ?? [],
      );
      boundThreadIds.delete(event.threadId);
      if (boundThreadIds.size === 0) {
        state.providerSessionThreadIds.delete(event.payload.providerSessionId);
      } else {
        state.providerSessionThreadIds.set(event.payload.providerSessionId, boundThreadIds);
      }
      break;
    }
  }

  return true;
}

type PayloadRow = {
  readonly payload_json: string;
};

type ShellThreadRow = {
  readonly thread_id: string;
  readonly payload_json: string;
  readonly forked_from_run_source_thread_id: string | null;
  readonly latest_run_id: string | null;
  readonly latest_run_status: string | null;
  readonly latest_run_requested_at: string | null;
  readonly latest_run_started_at: string | null;
  readonly latest_run_completed_at: string | null;
  readonly active_run_id: string | null;
  readonly activity_run_status: string | null;
  readonly last_error: string | null;
  readonly pending_request_payload_json: string | null;
  readonly latest_user_message_at: string | null;
  readonly has_actionable_proposed_plan: number;
  readonly item_count: number;
  readonly runless_item_count: number;
};

type ShellRunRow = {
  readonly thread_id: string;
  readonly run_id: string;
  readonly ordinal: number;
};

type SettlementThreadRow = Pick<
  ShellThreadRow,
  | "thread_id"
  | "payload_json"
  | "latest_run_id"
  | "latest_run_status"
  | "latest_run_requested_at"
  | "latest_run_started_at"
  | "latest_run_completed_at"
  | "latest_user_message_at"
>;

type ShellRunItemCountRow = {
  readonly thread_id: string;
  readonly run_id: string;
  readonly item_count: number;
};

const encodeThreadPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema),
);
const encodeRunPayload = Schema.encodeEffect(Schema.fromJsonString(OrchestrationV2RunJsonSchema));
const encodeRunAttemptPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2RunAttemptJsonSchema),
);
const encodeNodePayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ExecutionNodeJsonSchema),
);
const encodeSubagentPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2SubagentJsonSchema),
);
const encodeProviderSessionPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema),
);
const encodeProviderThreadPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ProviderThreadJsonSchema),
);
const encodeProviderTurnPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ProviderTurnJsonSchema),
);
const encodeRuntimeRequestPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2RuntimeRequestJsonSchema),
);
const encodeMessagePayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ConversationMessageJsonSchema),
);
const encodePlanPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2PlanArtifactSchema),
);
const encodeTurnItemPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2TurnItemJsonSchema),
);
const encodeCheckpointScopePayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2CheckpointScopeJsonSchema),
);
const encodeCheckpointPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2CheckpointJsonSchema),
);
const encodeContextHandoffPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ContextHandoffJsonSchema),
);
const encodeContextTransferPayload = Schema.encodeEffect(
  Schema.fromJsonString(OrchestrationV2ContextTransferJsonSchema),
);

const decodeThreadPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2AppThreadJsonSchema),
);
const decodeRunPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2RunJsonSchema),
);
const decodeRunAttemptPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2RunAttemptJsonSchema),
);
const decodeNodePayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ExecutionNodeJsonSchema),
);
const decodeSubagentPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2SubagentJsonSchema),
);
const decodeProviderSessionPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ProviderSessionJsonSchema),
);
const decodeProviderThreadPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ProviderThreadJsonSchema),
);
const decodeProviderTurnPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ProviderTurnJsonSchema),
);
const decodeRuntimeRequestPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2RuntimeRequestJsonSchema),
);
const decodeMessagePayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ConversationMessageJsonSchema),
);
const decodePlanArtifact = Schema.decodeUnknownEffect(OrchestrationV2PlanArtifactSchema);
const decodePlanPayload = (json: string) => decodePlanArtifact(parseEncodedPayload(json));
const decodeTurnItemPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2TurnItemJsonSchema),
);
const decodeCheckpointScopePayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2CheckpointScopeJsonSchema),
);
const decodeCheckpointPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2CheckpointJsonSchema),
);
const decodeContextHandoffPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ContextHandoffJsonSchema),
);
const decodeContextTransferPayload = Schema.decodeUnknownEffect(
  Schema.fromJsonString(OrchestrationV2ContextTransferJsonSchema),
);

const isProjectionStoreThreadNotFoundError = Schema.is(ProjectionStoreThreadNotFoundError);
const isProjectionStoreReadError = Schema.is(ProjectionStoreReadError);

function parseEncodedPayload(json: string): Record<string, unknown> {
  return JSON.parse(json) as Record<string, unknown>;
}

function stringField(payload: Record<string, unknown>, field: string): string {
  const value = payload[field];
  return typeof value === "string" ? value : String(value);
}

function nullableStringField(payload: Record<string, unknown>, field: string): string | null {
  const value = payload[field];
  if (value === null || value === undefined) {
    return null;
  }
  return typeof value === "string" ? value : String(value);
}

function booleanInt(value: boolean): 0 | 1 {
  return value ? 1 : 0;
}

const decodeRows =
  <A, E>(decode: (json: string) => Effect.Effect<A, E>, threadId: ThreadId) =>
  (rows: ReadonlyArray<PayloadRow>): Effect.Effect<Array<A>, ProjectionStoreReadError> =>
    Effect.forEach(rows, (row) => decode(row.payload_json)).pipe(
      Effect.mapError(
        (cause) =>
          new ProjectionStoreReadError({
            threadId,
            cause,
          }),
      ),
    );

function messageIdForTurnItem(item: OrchestrationV2TurnItem): string | null {
  switch (item.type) {
    case "user_message":
    case "assistant_message":
      return item.messageId;
    default:
      return null;
  }
}

function sortMessagesByTurnItemOrder(
  messages: ReadonlyArray<OrchestrationV2ConversationMessage>,
  turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
): Array<OrchestrationV2ConversationMessage> {
  const messageOrdinals = new Map<string, number>();
  for (const turnItem of turnItems) {
    const messageId = messageIdForTurnItem(turnItem);
    if (messageId === null) {
      continue;
    }
    const existing = messageOrdinals.get(messageId);
    if (existing === undefined || turnItem.ordinal < existing) {
      messageOrdinals.set(messageId, turnItem.ordinal);
    }
  }

  return messages.toSorted((left, right) => {
    const leftOrdinal = messageOrdinals.get(left.id) ?? Number.MAX_SAFE_INTEGER;
    const rightOrdinal = messageOrdinals.get(right.id) ?? Number.MAX_SAFE_INTEGER;
    if (leftOrdinal !== rightOrdinal) {
      return leftOrdinal - rightOrdinal;
    }

    const leftCreatedAt = DateTime.toEpochMillis(left.createdAt);
    const rightCreatedAt = DateTime.toEpochMillis(right.createdAt);
    if (leftCreatedAt !== rightCreatedAt) {
      return leftCreatedAt - rightCreatedAt;
    }

    return left.id.localeCompare(right.id);
  });
}

function activeLocalTurnItems(
  projection: OrchestrationV2ThreadProjection,
): Array<OrchestrationV2ProjectedTurnItem> {
  return projection.turnItems
    .filter((item) =>
      isOrchestrationV2TurnItemVisible({
        item,
        runs: projection.runs,
        attempts: projection.attempts,
        items: projection.turnItems,
      }),
    )
    .map((item, position) => ({
      position,
      visibility: "local" as const,
      sourceThreadId: item.threadId,
      sourceItemId: item.id,
      item,
    }));
}

function localVisibleTurnItems(
  projection: OrchestrationV2ThreadProjection,
): Array<OrchestrationV2ProjectedTurnItem> {
  return activeLocalTurnItems(projection);
}

function inheritedVisibleTurnItemsFromLocalItems(
  items: ReadonlyArray<OrchestrationV2TurnItem>,
): Array<Omit<OrchestrationV2ProjectedTurnItem, "position">> {
  return items.map((item) => ({
    visibility: "inherited" as const,
    sourceThreadId: item.threadId,
    sourceItemId: item.id,
    item,
  }));
}

function withLocalVisibleTurnItems(
  projection: OrchestrationV2ThreadProjection,
): OrchestrationV2ThreadProjection {
  return {
    ...projection,
    visibleTurnItems: localVisibleTurnItems(projection),
  };
}

function renumberVisibleTurnItems(
  rows: ReadonlyArray<Omit<OrchestrationV2ProjectedTurnItem, "position">>,
): Array<OrchestrationV2ProjectedTurnItem> {
  return rows.map((row, position) => ({ ...row, position }));
}

function makeForkMarkerTurnItem(input: {
  readonly targetProjection: OrchestrationV2ThreadProjection;
  readonly sourceThreadId: ThreadId;
  readonly sourceRunId: NonNullable<OrchestrationV2TurnItem["runId"]>;
}): OrchestrationV2TurnItem {
  const createdAt = input.targetProjection.thread.createdAt;
  return {
    id: TurnItemId.make(`turn-item:fork:${input.targetProjection.thread.id}`),
    threadId: input.targetProjection.thread.id,
    runId: null,
    nodeId: null,
    providerTurnId: null,
    nativeItemRef: null,
    parentItemId: null,
    ordinal: 0,
    status: "completed",
    title: "Forked from conversation",
    startedAt: null,
    completedAt: createdAt,
    updatedAt: createdAt,
    type: "fork",
    source: { type: "run", threadId: input.sourceThreadId, runId: input.sourceRunId },
    targetThreadId: input.targetProjection.thread.id,
  };
}

export function isTurnItemAtOrBeforeRun(input: {
  readonly historyOrigin: OrchestrationV2ThreadProjection["thread"]["historyOrigin"];
  readonly itemRunId: OrchestrationV2TurnItem["runId"];
  readonly runOrdinalById: ReadonlyMap<NonNullable<OrchestrationV2TurnItem["runId"]>, number>;
  readonly sourceRunOrdinal: number;
}): boolean {
  if (input.itemRunId === null) {
    return input.historyOrigin === "v1_import";
  }
  const ordinal = input.runOrdinalById.get(input.itemRunId);
  return ordinal !== undefined && ordinal <= input.sourceRunOrdinal;
}

function visibleTurnItemsThroughRun(input: {
  readonly sourceProjection: OrchestrationV2ThreadProjection;
  readonly sourceRunId: NonNullable<OrchestrationV2TurnItem["runId"]>;
}): Array<Omit<OrchestrationV2ProjectedTurnItem, "position">> {
  const sourceRun = input.sourceProjection.runs.find((run) => run.id === input.sourceRunId);
  if (sourceRun === undefined) {
    return [];
  }

  const runOrdinalById = new Map(input.sourceProjection.runs.map((run) => [run.id, run.ordinal]));
  const inheritedPrefix = input.sourceProjection.visibleTurnItems
    .filter(
      (row) => row.item.threadId !== input.sourceProjection.thread.id || row.item.type === "fork",
    )
    .map((row) => ({
      visibility: "inherited" as const,
      sourceThreadId: row.sourceThreadId,
      sourceItemId: row.sourceItemId,
      item: row.item,
    }));
  const localPrefix = inheritedVisibleTurnItemsFromLocalItems(
    input.sourceProjection.turnItems.filter((item) => {
      if (
        isOrchestrationV2SupersededInterrupt({
          item,
          attempts: input.sourceProjection.attempts,
          items: input.sourceProjection.turnItems,
        })
      ) {
        return false;
      }
      return isTurnItemAtOrBeforeRun({
        historyOrigin: input.sourceProjection.thread.historyOrigin,
        itemRunId: item.runId,
        runOrdinalById,
        sourceRunOrdinal: sourceRun.ordinal,
      });
    }),
  );

  return [...inheritedPrefix, ...localPrefix];
}

function buildVisibleTurnItems(input: {
  readonly projection: OrchestrationV2ThreadProjection;
  readonly sourceProjection: OrchestrationV2ThreadProjection | null;
}): Array<OrchestrationV2ProjectedTurnItem> {
  const forkedFrom = input.projection.thread.forkedFrom;
  if (forkedFrom?.type !== "run" || input.sourceProjection === null) {
    return localVisibleTurnItems(input.projection);
  }

  const inherited = visibleTurnItemsThroughRun({
    sourceProjection: input.sourceProjection,
    sourceRunId: forkedFrom.runId,
  });
  const markerItem = makeForkMarkerTurnItem({
    targetProjection: input.projection,
    sourceThreadId: forkedFrom.threadId,
    sourceRunId: forkedFrom.runId,
  });
  const local = activeLocalTurnItems(input.projection).map((row) => ({
    visibility: "local" as const,
    sourceThreadId: row.sourceThreadId,
    sourceItemId: row.sourceItemId,
    item: row.item,
  }));

  return renumberVisibleTurnItems([
    ...inherited,
    {
      visibility: "synthetic",
      sourceThreadId: forkedFrom.threadId,
      sourceItemId: markerItem.id,
      item: markerItem,
    },
    ...local,
  ]);
}

export function threadShellFromProjection(
  projection: OrchestrationV2ThreadProjection,
): OrchestrationV2ThreadShell {
  const latestRun = projection.runs.at(-1) ?? null;
  const activeRun =
    projection.runs
      .filter(isInterruptibleRunForShell)
      .toSorted((left, right) => right.ordinal - left.ordinal)[0] ?? null;
  const activityRun =
    projection.runs
      .filter(isActivityRunForShell)
      .toSorted((left, right) => right.ordinal - left.ordinal)[0] ?? null;
  const pendingRuntimeRequest =
    projection.runtimeRequests
      .filter((request) => request.status === "pending")
      .toSorted(
        (left, right) =>
          DateTime.toEpochMillis(right.createdAt) - DateTime.toEpochMillis(left.createdAt),
      )[0] ?? null;
  const latestUserMessage =
    projection.messages
      .filter((message) => message.role === "user")
      .toSorted(
        (left, right) =>
          DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
      )[0] ?? null;
  const providerSession =
    projection.providerSessions
      .filter((session) => session.providerInstanceId === projection.thread.providerInstanceId)
      .toSorted(
        (left, right) =>
          DateTime.toEpochMillis(right.updatedAt) - DateTime.toEpochMillis(left.updatedAt),
      )[0] ?? null;
  const pendingBackgroundTasks = derivePendingBackgroundWork({
    latestRun,
    providerThreads: projection.providerThreads,
    turnItems: projection.turnItems,
    activeProviderThreadId: projection.thread.activeProviderThreadId,
    runs: projection.runs,
  });
  return {
    createdBy: projection.thread.createdBy,
    creationSource: projection.thread.creationSource,
    id: projection.thread.id,
    projectId: projection.thread.projectId,
    title: projection.thread.title,
    providerInstanceId: projection.thread.providerInstanceId,
    modelSelection: projection.thread.modelSelection,
    runtimeMode: projection.thread.runtimeMode,
    interactionMode: projection.thread.interactionMode,
    branch: projection.thread.branch,
    worktreePath: projection.thread.worktreePath,
    ...(projection.thread.linkedPullRequest === undefined
      ? {}
      : { linkedPullRequest: projection.thread.linkedPullRequest }),
    lineage: projection.thread.lineage,
    forkedFrom: projection.thread.forkedFrom,
    activeProviderThreadId: projection.thread.activeProviderThreadId,
    ...(projection.thread.historyOrigin === undefined
      ? {}
      : { historyOrigin: projection.thread.historyOrigin }),
    latestRunId: latestRun?.id ?? null,
    latestRunRequestedAt: latestRun?.requestedAt ?? null,
    latestRunStartedAt: latestRun?.startedAt ?? null,
    latestRunCompletedAt: latestRun?.completedAt ?? null,
    activeRunId: activeRun?.id ?? null,
    activityRunStatus: activityRun?.status ?? null,
    status: latestRun?.status ?? "idle",
    lastError: providerSession?.lastError ?? null,
    pendingRuntimeRequest:
      pendingRuntimeRequest === null
        ? null
        : {
            id: pendingRuntimeRequest.id,
            kind: pendingRuntimeRequest.kind,
            createdAt: pendingRuntimeRequest.createdAt,
          },
    // Thread detail owns message bodies. Keeping them out of shell rows makes
    // initial hydration and streaming updates independent of transcript size.
    latestVisibleMessage: null,
    latestUserMessageAt: latestUserMessage?.updatedAt ?? null,
    hasActionableProposedPlan: projection.plans.some(
      (plan) => plan.kind === "proposed_plan" && plan.status === "active",
    ),
    pendingBackgroundTasks: [...pendingBackgroundTasks],
    itemCount: activeLocalTurnItems(projection).length,
    visibleItemCount: projection.visibleTurnItems.length,
    createdAt: projection.thread.createdAt,
    updatedAt: projection.updatedAt,
    archivedAt: projection.thread.archivedAt,
    settledOverride: projection.thread.settledOverride,
    settledAt: projection.thread.settledAt,
    unsettledAt: projection.thread.unsettledAt ?? null,
    snoozedUntil: projection.thread.snoozedUntil ?? null,
    snoozedAt: projection.thread.snoozedAt ?? null,
    pinnedAt: projection.thread.pinnedAt ?? null,
    pinOrderKey: projection.thread.pinOrderKey ?? null,
    lastVisitedAt: projection.thread.lastVisitedAt,
    titleRegeneration: projection.thread.titleRegeneration ?? null,
    deletedAt: projection.thread.deletedAt,
  };
}

function isInterruptibleRunForShell(run: OrchestrationV2ThreadProjection["runs"][number]): boolean {
  return run.status === "preparing" || run.status === "starting" || run.status === "running";
}

type ShellActivityRunStatus = "preparing" | "running" | "starting" | "waiting";

function isActivityRunForShell(
  run: OrchestrationV2ThreadProjection["runs"][number],
): run is OrchestrationV2ThreadProjection["runs"][number] & {
  readonly status: ShellActivityRunStatus;
} {
  return isInterruptibleRunForShell(run) || run.status === "waiting";
}

type ShellThreadState = {
  readonly thread: OrchestrationV2ThreadProjection["thread"];
  readonly latestRunId: RunId | null;
  readonly latestRunStatus: OrchestrationV2ShellThreadStatus;
  readonly latestRunRequestedAt: DateTime.Utc | null;
  readonly latestRunStartedAt: DateTime.Utc | null;
  readonly latestRunCompletedAt: DateTime.Utc | null;
  readonly activeRunId: RunId | null;
  readonly activityRunStatus: ShellActivityRunStatus | null;
  readonly lastError: string | null;
  readonly pendingRuntimeRequest: OrchestrationV2ThreadProjection["runtimeRequests"][number] | null;
  readonly latestUserMessageAt: DateTime.Utc | null;
  readonly hasActionableProposedPlan: boolean;
  readonly pendingBackgroundTasks: OrchestrationV2ThreadShell["pendingBackgroundTasks"];
  readonly itemCount: number;
  readonly runlessItemCount: number;
  readonly updatedAt: OrchestrationV2ThreadProjection["updatedAt"];
  readonly runOrdinalById: ReadonlyMap<RunId, number>;
  readonly itemCountByRunId: ReadonlyMap<RunId, number>;
};

function shellStatusFromStoredRunStatus(status: string | null): OrchestrationV2ShellThreadStatus {
  switch (status) {
    case null:
      return "idle";
    case "preparing":
    case "queued":
    case "starting":
    case "running":
    case "waiting":
    case "completed":
    case "interrupted":
    case "failed":
    case "cancelled":
    case "rolled_back":
      return status;
    default:
      return "failed";
  }
}

function itemCountThroughRun(input: {
  readonly state: ShellThreadState;
  readonly runId: RunId;
}): number {
  const runOrdinal = input.state.runOrdinalById.get(input.runId);
  if (runOrdinal === undefined) {
    return 0;
  }

  let count = input.state.thread.historyOrigin === "v1_import" ? input.state.runlessItemCount : 0;
  for (const [runId, itemCount] of input.state.itemCountByRunId) {
    const itemRunOrdinal = input.state.runOrdinalById.get(runId);
    if (itemRunOrdinal !== undefined && itemRunOrdinal <= runOrdinal) {
      count += itemCount;
    }
  }
  return count;
}

function visibleItemCountForShell(input: {
  readonly threadId: ThreadId;
  readonly statesByThreadId: ReadonlyMap<ThreadId, ShellThreadState>;
  readonly seenThreadIds?: ReadonlySet<ThreadId>;
}): number {
  const state = input.statesByThreadId.get(input.threadId);
  if (state === undefined) {
    return 0;
  }

  const forkedFrom = state.thread.forkedFrom;
  if (forkedFrom?.type !== "run") {
    return state.itemCount;
  }

  const seenThreadIds = input.seenThreadIds ?? new Set<ThreadId>();
  if (seenThreadIds.has(state.thread.id)) {
    return state.itemCount;
  }

  const sourceState = input.statesByThreadId.get(forkedFrom.threadId);
  if (sourceState === undefined) {
    return state.itemCount;
  }

  const sourceForkedFrom = sourceState.thread.forkedFrom;
  const inheritedPrefixCount =
    sourceForkedFrom?.type === "run"
      ? visibleItemCountForShell({
          threadId: sourceState.thread.id,
          statesByThreadId: input.statesByThreadId,
          seenThreadIds: new Set([...seenThreadIds, state.thread.id]),
        }) - sourceState.itemCount
      : 0;

  return (
    inheritedPrefixCount +
    itemCountThroughRun({ state: sourceState, runId: forkedFrom.runId }) +
    1 +
    state.itemCount
  );
}

function shellFromState(input: {
  readonly state: ShellThreadState;
  readonly visibleItemCount: number;
}): OrchestrationV2ThreadShell {
  return {
    createdBy: input.state.thread.createdBy,
    creationSource: input.state.thread.creationSource,
    id: input.state.thread.id,
    projectId: input.state.thread.projectId,
    title: input.state.thread.title,
    providerInstanceId: input.state.thread.providerInstanceId,
    modelSelection: input.state.thread.modelSelection,
    runtimeMode: input.state.thread.runtimeMode,
    interactionMode: input.state.thread.interactionMode,
    branch: input.state.thread.branch,
    worktreePath: input.state.thread.worktreePath,
    ...(input.state.thread.linkedPullRequest === undefined
      ? {}
      : { linkedPullRequest: input.state.thread.linkedPullRequest }),
    lineage: input.state.thread.lineage,
    forkedFrom: input.state.thread.forkedFrom,
    activeProviderThreadId: input.state.thread.activeProviderThreadId,
    ...(input.state.thread.historyOrigin === undefined
      ? {}
      : { historyOrigin: input.state.thread.historyOrigin }),
    latestRunId: input.state.latestRunId,
    latestRunRequestedAt: input.state.latestRunRequestedAt,
    latestRunStartedAt: input.state.latestRunStartedAt,
    latestRunCompletedAt: input.state.latestRunCompletedAt,
    activeRunId: input.state.activeRunId,
    activityRunStatus: input.state.activityRunStatus,
    status: input.state.latestRunStatus,
    lastError: input.state.lastError,
    pendingRuntimeRequest:
      input.state.pendingRuntimeRequest === null
        ? null
        : {
            id: input.state.pendingRuntimeRequest.id,
            kind: input.state.pendingRuntimeRequest.kind,
            createdAt: input.state.pendingRuntimeRequest.createdAt,
          },
    latestVisibleMessage: null,
    latestUserMessageAt: input.state.latestUserMessageAt,
    hasActionableProposedPlan: input.state.hasActionableProposedPlan,
    pendingBackgroundTasks: input.state.pendingBackgroundTasks,
    itemCount: input.state.itemCount,
    visibleItemCount: input.visibleItemCount,
    createdAt: input.state.thread.createdAt,
    updatedAt: input.state.updatedAt,
    archivedAt: input.state.thread.archivedAt,
    settledOverride: input.state.thread.settledOverride,
    settledAt: input.state.thread.settledAt,
    unsettledAt: input.state.thread.unsettledAt ?? null,
    snoozedUntil: input.state.thread.snoozedUntil ?? null,
    snoozedAt: input.state.thread.snoozedAt ?? null,
    pinnedAt: input.state.thread.pinnedAt ?? null,
    pinOrderKey: input.state.thread.pinOrderKey ?? null,
    lastVisitedAt: input.state.thread.lastVisitedAt,
    titleRegeneration: input.state.thread.titleRegeneration ?? null,
    deletedAt: input.state.thread.deletedAt,
  };
}

export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> = Layer.effect(
  ProjectionStoreV2,
  Effect.gen(function* () {
    const sql = yield* SqlClient.SqlClient;

    const apply: ProjectionStoreV2Shape["apply"] = (event) =>
      Effect.gen(function* () {
        switch (event.type) {
          case "thread.created":
          case "thread.archived":
          case "thread.unarchived":
          case "thread.deleted":
          case "thread.settled":
          case "thread.unsettled":
          case "thread.snoozed":
          case "thread.unsnoozed":
          case "thread.pinned":
          case "thread.unpinned":
          case "thread.pin-reordered":
          case "thread.visited":
          case "thread.marked-unread":
          case "thread.metadata-updated":
          case "thread.runtime-mode-updated":
          case "thread.interaction-mode-updated":
          case "thread.model-selection-updated":
          case "thread.provider-switched": {
            const payloadJson = yield* encodeThreadPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_threads (
                thread_id,
                project_id,
                title,
                default_provider,
                provider_instance_id,
                runtime_mode,
                interaction_mode,
                active_provider_thread_id,
                created_at,
                updated_at,
                archived_at,
                deleted_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.projectId},
                ${event.payload.title},
                ${event.payload.providerInstanceId},
                ${event.payload.providerInstanceId},
                ${event.payload.runtimeMode},
                ${event.payload.interactionMode},
                ${event.payload.activeProviderThreadId},
                ${stringField(payload, "createdAt")},
                ${stringField(payload, "updatedAt")},
                ${nullableStringField(payload, "archivedAt")},
                ${nullableStringField(payload, "deletedAt")},
                ${payloadJson}
              )
              ON CONFLICT(thread_id)
              DO UPDATE SET
                project_id = excluded.project_id,
                title = excluded.title,
                default_provider = excluded.default_provider,
                provider_instance_id = excluded.provider_instance_id,
                runtime_mode = excluded.runtime_mode,
                interaction_mode = excluded.interaction_mode,
                active_provider_thread_id = excluded.active_provider_thread_id,
                created_at = excluded.created_at,
                updated_at = excluded.updated_at,
                archived_at = excluded.archived_at,
                deleted_at = excluded.deleted_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "run.created":
          case "run.updated": {
            const payloadJson = yield* encodeRunPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_runs (
                run_id,
                thread_id,
                ordinal,
                provider,
                provider_instance_id,
                provider_thread_id,
                status,
                requested_at,
                completed_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.ordinal},
                ${event.payload.providerInstanceId},
                ${event.payload.providerInstanceId},
                ${event.payload.providerThreadId},
                ${event.payload.status},
                ${stringField(payload, "requestedAt")},
                ${nullableStringField(payload, "completedAt")},
                ${payloadJson}
              )
              ON CONFLICT(run_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                ordinal = excluded.ordinal,
                provider = excluded.provider,
                provider_instance_id = excluded.provider_instance_id,
                provider_thread_id = excluded.provider_thread_id,
                status = excluded.status,
                requested_at = excluded.requested_at,
                completed_at = excluded.completed_at,
                payload_json = CASE
                  WHEN json_type(excluded.payload_json, '$.delegatedCompletion') IS NULL
                    AND json_type(orchestration_v2_projection_runs.payload_json, '$.delegatedCompletion') IS NOT NULL
                  THEN json_set(
                    excluded.payload_json,
                    '$.delegatedCompletion',
                    json_extract(
                      orchestration_v2_projection_runs.payload_json,
                      '$.delegatedCompletion'
                    )
                  )
                  ELSE excluded.payload_json
                END
            `;
            break;
          }
          case "run-attempt.created":
          case "run-attempt.updated": {
            const payloadJson = yield* encodeRunAttemptPayload(event.payload);
            yield* sql`
              INSERT INTO orchestration_v2_projection_run_attempts (
                attempt_id,
                thread_id,
                run_id,
                attempt_ordinal,
                root_node_id,
                provider,
                provider_instance_id,
                provider_thread_id,
                provider_turn_id,
                status,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.threadId},
                ${event.payload.runId},
                ${event.payload.attemptOrdinal},
                ${event.payload.rootNodeId},
                ${event.payload.providerInstanceId},
                ${event.payload.providerInstanceId},
                ${event.payload.providerThreadId},
                ${event.payload.providerTurnId},
                ${event.payload.status},
                ${payloadJson}
              )
              ON CONFLICT(attempt_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                attempt_ordinal = excluded.attempt_ordinal,
                root_node_id = excluded.root_node_id,
                provider = excluded.provider,
                provider_instance_id = excluded.provider_instance_id,
                provider_thread_id = excluded.provider_thread_id,
                provider_turn_id = excluded.provider_turn_id,
                status = excluded.status,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "node.updated": {
            const payloadJson = yield* encodeNodePayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_nodes (
                node_id,
                thread_id,
                run_id,
                parent_node_id,
                root_node_id,
                kind,
                status,
                provider_thread_id,
                provider_turn_id,
                runtime_request_id,
                checkpoint_scope_id,
                started_at,
                completed_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.runId},
                ${event.payload.parentNodeId},
                ${event.payload.rootNodeId},
                ${event.payload.kind},
                ${event.payload.status},
                ${event.payload.providerThreadId},
                ${event.payload.providerTurnId},
                ${event.payload.runtimeRequestId},
                ${event.payload.checkpointScopeId},
                ${nullableStringField(payload, "startedAt")},
                ${nullableStringField(payload, "completedAt")},
                ${payloadJson}
              )
              ON CONFLICT(node_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                parent_node_id = excluded.parent_node_id,
                root_node_id = excluded.root_node_id,
                kind = excluded.kind,
                status = excluded.status,
                provider_thread_id = excluded.provider_thread_id,
                provider_turn_id = excluded.provider_turn_id,
                runtime_request_id = excluded.runtime_request_id,
                checkpoint_scope_id = excluded.checkpoint_scope_id,
                started_at = excluded.started_at,
                completed_at = excluded.completed_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "subagent.updated": {
            const payloadJson = yield* encodeSubagentPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_subagents (
                subagent_id,
                thread_id,
                run_id,
                parent_node_id,
                provider,
                driver,
                provider_instance_id,
                provider_thread_id,
                child_thread_id,
                origin,
                status,
                started_at,
                completed_at,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.runId},
                ${event.payload.parentNodeId},
                ${event.payload.providerInstanceId},
                ${event.payload.driver},
                ${event.payload.providerInstanceId},
                ${event.payload.providerThreadId},
                ${event.payload.childThreadId},
                ${event.payload.origin},
                ${event.payload.status},
                ${nullableStringField(payload, "startedAt")},
                ${nullableStringField(payload, "completedAt")},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(subagent_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                parent_node_id = excluded.parent_node_id,
                provider = excluded.provider,
                driver = excluded.driver,
                provider_instance_id = excluded.provider_instance_id,
                provider_thread_id = excluded.provider_thread_id,
                child_thread_id = excluded.child_thread_id,
                origin = excluded.origin,
                status = excluded.status,
                started_at = excluded.started_at,
                completed_at = excluded.completed_at,
                updated_at = excluded.updated_at,
                payload_json = CASE
                  WHEN json_type(excluded.payload_json, '$.completionDelivery') IS NULL
                    AND json_type(orchestration_v2_projection_subagents.payload_json, '$.completionDelivery') IS NOT NULL
                  THEN json_set(
                    excluded.payload_json,
                    '$.completionDelivery',
                    json_extract(
                      orchestration_v2_projection_subagents.payload_json,
                      '$.completionDelivery'
                    )
                  )
                  ELSE excluded.payload_json
                END
            `;
            break;
          }
          case "provider-session.attached":
          case "provider-session.updated": {
            const payloadJson = yield* encodeProviderSessionPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_provider_sessions (
                provider_session_id,
                thread_id,
                provider,
                driver,
                provider_instance_id,
                status,
                model,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.threadId},
                ${event.payload.providerInstanceId},
                ${event.payload.driver},
                ${event.payload.providerInstanceId},
                ${event.payload.status},
                ${event.payload.model},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(provider_session_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                provider = excluded.provider,
                driver = excluded.driver,
                provider_instance_id = excluded.provider_instance_id,
                status = excluded.status,
                model = excluded.model,
                updated_at = excluded.updated_at,
                payload_json = excluded.payload_json
            `;
            if (event.type === "provider-session.attached") {
              yield* sql`
                INSERT OR IGNORE INTO orchestration_v2_projection_provider_session_bindings (
                  provider_session_id,
                  thread_id
                )
                VALUES (${event.payload.id}, ${event.threadId})
              `;
            }
            break;
          }
          case "provider-session.detached": {
            yield* sql`
              DELETE FROM orchestration_v2_projection_provider_session_bindings
              WHERE provider_session_id = ${event.payload.providerSessionId}
                AND thread_id = ${event.threadId}
            `;
            break;
          }
          case "provider-thread.updated": {
            const payloadJson = yield* encodeProviderThreadPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_provider_threads (
                provider_thread_id,
                thread_id,
                owner_node_id,
                provider,
                driver,
                provider_instance_id,
                provider_session_id,
                status,
                first_run_ordinal,
                last_run_ordinal,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.appThreadId},
                ${event.payload.ownerNodeId},
                ${event.payload.providerInstanceId},
                ${event.payload.driver},
                ${event.payload.providerInstanceId},
                ${event.payload.providerSessionId},
                ${event.payload.status},
                ${event.payload.firstRunOrdinal},
                ${event.payload.lastRunOrdinal},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(provider_thread_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                owner_node_id = excluded.owner_node_id,
                provider = excluded.provider,
                driver = excluded.driver,
                provider_instance_id = excluded.provider_instance_id,
                provider_session_id = excluded.provider_session_id,
                status = excluded.status,
                first_run_ordinal = excluded.first_run_ordinal,
                last_run_ordinal = excluded.last_run_ordinal,
                updated_at = excluded.updated_at,
                payload_json = excluded.payload_json
            `;
            if (event.payload.appThreadId !== null) {
              const threadRows = yield* sql<PayloadRow>`
                SELECT payload_json
                FROM orchestration_v2_projection_threads
                WHERE thread_id = ${event.payload.appThreadId}
                LIMIT 1
              `;
              const threadRow = threadRows[0];
              if (threadRow !== undefined) {
                const thread = yield* decodeThreadPayload(threadRow.payload_json);
                const updatedThread = {
                  ...thread,
                  activeProviderThreadId: event.payload.id,
                  updatedAt: event.payload.updatedAt,
                };
                const updatedThreadPayloadJson = yield* encodeThreadPayload(updatedThread);
                yield* sql`
                  UPDATE orchestration_v2_projection_threads
                  SET
                    active_provider_thread_id = ${event.payload.id},
                    updated_at = ${stringField(parseEncodedPayload(updatedThreadPayloadJson), "updatedAt")},
                    payload_json = ${updatedThreadPayloadJson}
                  WHERE thread_id = ${event.payload.appThreadId}
                `;
              }
            }
            break;
          }
          case "provider-turn.updated": {
            const existingRows =
              event.payload.tokenUsage === undefined
                ? yield* sql<PayloadRow>`
                    SELECT payload_json
                    FROM orchestration_v2_projection_provider_turns
                    WHERE provider_turn_id = ${event.payload.id}
                    LIMIT 1
                  `
                : [];
            const existing = existingRows[0];
            const providerTurn =
              existing === undefined
                ? event.payload
                : upsertProviderTurn(
                    [yield* decodeProviderTurnPayload(existing.payload_json)],
                    event.payload,
                  )[0]!;
            const payloadJson = yield* encodeProviderTurnPayload(providerTurn);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_provider_turns (
                provider_turn_id,
                thread_id,
                provider_thread_id,
                node_id,
                run_attempt_id,
                ordinal,
                status,
                started_at,
                completed_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.threadId},
                ${providerTurn.providerThreadId},
                ${providerTurn.nodeId},
                ${providerTurn.runAttemptId},
                ${providerTurn.ordinal},
                ${providerTurn.status},
                ${nullableStringField(payload, "startedAt")},
                ${nullableStringField(payload, "completedAt")},
                ${payloadJson}
              )
              ON CONFLICT(provider_turn_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                provider_thread_id = excluded.provider_thread_id,
                node_id = excluded.node_id,
                run_attempt_id = excluded.run_attempt_id,
                ordinal = excluded.ordinal,
                status = excluded.status,
                started_at = excluded.started_at,
                completed_at = excluded.completed_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "runtime-request.updated": {
            const payloadJson = yield* encodeRuntimeRequestPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_runtime_requests (
                runtime_request_id,
                thread_id,
                node_id,
                provider_turn_id,
                kind,
                status,
                created_at,
                resolved_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.threadId},
                ${event.payload.nodeId},
                ${event.payload.providerTurnId},
                ${event.payload.kind},
                ${event.payload.status},
                ${stringField(payload, "createdAt")},
                ${nullableStringField(payload, "resolvedAt")},
                ${payloadJson}
              )
              ON CONFLICT(runtime_request_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                node_id = excluded.node_id,
                provider_turn_id = excluded.provider_turn_id,
                kind = excluded.kind,
                status = excluded.status,
                created_at = excluded.created_at,
                resolved_at = excluded.resolved_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "message.updated": {
            const payloadJson = yield* encodeMessagePayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_messages (
                message_id,
                thread_id,
                run_id,
                node_id,
                role,
                streaming,
                created_at,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.runId},
                ${event.payload.nodeId},
                ${event.payload.role},
                ${booleanInt(event.payload.streaming)},
                ${stringField(payload, "createdAt")},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(message_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                node_id = excluded.node_id,
                role = excluded.role,
                streaming = excluded.streaming,
                created_at = excluded.created_at,
                updated_at = excluded.updated_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "plan.updated": {
            const payloadJson = yield* encodePlanPayload(event.payload);
            yield* sql`
              INSERT INTO orchestration_v2_projection_plans (
                plan_id,
                thread_id,
                run_id,
                node_id,
                kind,
                status,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.runId},
                ${event.payload.nodeId},
                ${event.payload.kind},
                ${event.payload.status},
                ${payloadJson}
              )
              ON CONFLICT(plan_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                node_id = excluded.node_id,
                kind = excluded.kind,
                status = excluded.status,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "turn-item.updated": {
            const payloadJson = yield* encodeTurnItemPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_turn_items (
                turn_item_id,
                thread_id,
                run_id,
                node_id,
                provider_thread_id,
                provider_turn_id,
                parent_item_id,
                ordinal,
                type,
                status,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.runId},
                ${event.payload.nodeId},
                ${event.payload.providerThreadId},
                ${event.payload.providerTurnId},
                ${event.payload.parentItemId},
                ${event.payload.ordinal},
                ${event.payload.type},
                ${event.payload.status},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(turn_item_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                node_id = excluded.node_id,
                provider_thread_id = excluded.provider_thread_id,
                provider_turn_id = excluded.provider_turn_id,
                parent_item_id = excluded.parent_item_id,
                ordinal = excluded.ordinal,
                type = excluded.type,
                status = excluded.status,
                updated_at = excluded.updated_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "checkpoint-scope.created": {
            const payloadJson = yield* encodeCheckpointScopePayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_checkpoint_scopes (
                scope_id,
                thread_id,
                run_id,
                node_id,
                parent_scope_id,
                provider_thread_id,
                kind,
                ordinal_within_parent,
                advances_app_run_count,
                created_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.runId},
                ${event.payload.nodeId},
                ${event.payload.parentScopeId},
                ${event.payload.providerThreadId},
                ${event.payload.kind},
                ${event.payload.ordinalWithinParent},
                ${booleanInt(event.payload.advancesAppRunCount)},
                ${stringField(payload, "createdAt")},
                ${payloadJson}
              )
              ON CONFLICT(scope_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                run_id = excluded.run_id,
                node_id = excluded.node_id,
                parent_scope_id = excluded.parent_scope_id,
                provider_thread_id = excluded.provider_thread_id,
                kind = excluded.kind,
                ordinal_within_parent = excluded.ordinal_within_parent,
                advances_app_run_count = excluded.advances_app_run_count,
                created_at = excluded.created_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "checkpoint.captured": {
            const payloadJson = yield* encodeCheckpointPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_checkpoints (
                checkpoint_id,
                thread_id,
                scope_id,
                run_id,
                node_id,
                parent_checkpoint_id,
                ordinal_within_scope,
                app_run_ordinal,
                status,
                captured_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.scopeId},
                ${event.payload.runId},
                ${event.payload.nodeId},
                ${event.payload.parentCheckpointId},
                ${event.payload.ordinalWithinScope},
                ${event.payload.appRunOrdinal},
                ${event.payload.status},
                ${stringField(payload, "capturedAt")},
                ${payloadJson}
              )
              ON CONFLICT(checkpoint_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                scope_id = excluded.scope_id,
                run_id = excluded.run_id,
                node_id = excluded.node_id,
                parent_checkpoint_id = excluded.parent_checkpoint_id,
                ordinal_within_scope = excluded.ordinal_within_scope,
                app_run_ordinal = excluded.app_run_ordinal,
                status = excluded.status,
                captured_at = excluded.captured_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "checkpoint.rollback-requested":
            break;
          case "context-handoff.updated": {
            const payloadJson = yield* encodeContextHandoffPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_context_handoffs (
                context_handoff_id,
                thread_id,
                target_run_id,
                to_provider_thread_id,
                strategy,
                status,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.threadId},
                ${event.payload.targetRunId},
                ${event.payload.toProviderThreadId},
                ${event.payload.strategy},
                ${event.payload.status},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(context_handoff_id)
              DO UPDATE SET
                thread_id = excluded.thread_id,
                target_run_id = excluded.target_run_id,
                to_provider_thread_id = excluded.to_provider_thread_id,
                strategy = excluded.strategy,
                status = excluded.status,
                updated_at = excluded.updated_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
          case "context-transfer.created":
          case "context-transfer.updated": {
            const payloadJson = yield* encodeContextTransferPayload(event.payload);
            const payload = parseEncodedPayload(payloadJson);
            yield* sql`
              INSERT INTO orchestration_v2_projection_context_transfers (
                context_transfer_id,
                source_thread_id,
                target_thread_id,
                target_run_id,
                type,
                status,
                source_provider,
                target_provider,
                source_provider_instance_id,
                target_provider_instance_id,
                updated_at,
                payload_json
              )
              VALUES (
                ${event.payload.id},
                ${event.payload.sourceThreadId},
                ${event.payload.targetThreadId},
                ${event.payload.targetRunId},
                ${event.payload.type},
                ${event.payload.status},
                ${event.payload.sourceProviderInstanceId},
                ${event.payload.targetProviderInstanceId},
                ${event.payload.sourceProviderInstanceId},
                ${event.payload.targetProviderInstanceId},
                ${stringField(payload, "updatedAt")},
                ${payloadJson}
              )
              ON CONFLICT(context_transfer_id)
              DO UPDATE SET
                source_thread_id = excluded.source_thread_id,
                target_thread_id = excluded.target_thread_id,
                target_run_id = excluded.target_run_id,
                type = excluded.type,
                status = excluded.status,
                source_provider = excluded.source_provider,
                target_provider = excluded.target_provider,
                source_provider_instance_id = excluded.source_provider_instance_id,
                target_provider_instance_id = excluded.target_provider_instance_id,
                updated_at = excluded.updated_at,
                payload_json = excluded.payload_json
            `;
            break;
          }
        }

        if (
          event.type !== "thread.created" &&
          event.type !== "thread.archived" &&
          event.type !== "thread.unarchived" &&
          event.type !== "thread.deleted" &&
          event.type !== "thread.settled" &&
          event.type !== "thread.unsettled" &&
          event.type !== "thread.snoozed" &&
          event.type !== "thread.unsnoozed" &&
          event.type !== "thread.pinned" &&
          event.type !== "thread.unpinned" &&
          event.type !== "thread.pin-reordered" &&
          event.type !== "thread.visited" &&
          event.type !== "thread.marked-unread" &&
          event.type !== "thread.metadata-updated" &&
          event.type !== "thread.runtime-mode-updated" &&
          event.type !== "thread.interaction-mode-updated" &&
          event.type !== "thread.model-selection-updated" &&
          event.type !== "thread.provider-switched"
        ) {
          const rows = yield* sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_threads
            WHERE thread_id = ${event.threadId}
            LIMIT 1
          `;
          const row = rows[0];
          if (row !== undefined) {
            const thread = yield* decodeThreadPayload(row.payload_json);
            const updatedThread = { ...thread, updatedAt: event.occurredAt };
            const payloadJson = yield* encodeThreadPayload(updatedThread);
            yield* sql`
              UPDATE orchestration_v2_projection_threads
              SET
                updated_at = ${stringField(parseEncodedPayload(payloadJson), "updatedAt")},
                payload_json = ${payloadJson}
              WHERE thread_id = ${event.threadId}
            `;
          }
        }
      }).pipe(
        Effect.mapError(
          (cause) =>
            new ProjectionStoreApplyEventError({
              eventType: event.type,
              cause,
            }),
        ),
      );

    const readCanonicalProjection = (
      threadId: ThreadId,
      window?: {
        readonly rowLimit: number;
        readonly anchorItemId?: TurnItemId | undefined;
        readonly requiredRunId?: RunId | undefined;
        readonly suppressLocal?: boolean | undefined;
      },
    ) =>
      Effect.gen(function* () {
        const threadRows = yield* sql<PayloadRow>`
          SELECT payload_json
          FROM orchestration_v2_projection_threads
          WHERE thread_id = ${threadId}
          LIMIT 1
        `;
        const threadRow = threadRows[0];
        if (!threadRow) {
          return yield* new ProjectionStoreThreadNotFoundError({ threadId });
        }

        const boundedTurnItemRows =
          window === undefined
            ? yield* sql<PayloadRow>`
                SELECT payload_json
                FROM orchestration_v2_projection_turn_items
                WHERE thread_id = ${threadId}
                ORDER BY ordinal ASC, turn_item_id ASC
              `
            : yield* sql<PayloadRow>`
                WITH eligible AS (
                  SELECT item.payload_json, item.ordinal, item.turn_item_id,
                    item.run_id, item.node_id, item.type
                  FROM orchestration_v2_projection_turn_items AS item
                  LEFT JOIN orchestration_v2_projection_runs AS run
                    ON run.run_id = item.run_id
                  WHERE item.thread_id = ${threadId}
                    AND item.ordinal <= COALESCE(
                      (
                        SELECT ordinal
                        FROM orchestration_v2_projection_turn_items
                        WHERE thread_id = ${threadId}
                          AND turn_item_id = ${window.anchorItemId ?? null}
                        LIMIT 1
                      ),
                      9223372036854775807
                    )
                    AND (
                      ${window.requiredRunId ?? null} IS NOT NULL
                      OR (
                        (run.status IS NULL OR run.status <> 'rolled_back')
                        AND NOT (
                          item.type = 'user_message'
                          AND json_extract(item.payload_json, '$.inputIntent') = 'queued_turn'
                          AND run.status IS 'cancelled'
                        )
                      )
                    )
                    AND (
                      ${window.requiredRunId ?? null} IS NULL
                      OR (
                        item.run_id IS NULL
                        AND EXISTS (
                          SELECT 1
                          FROM orchestration_v2_projection_threads AS source_thread
                          WHERE source_thread.thread_id = item.thread_id
                            AND json_extract(source_thread.payload_json, '$.historyOrigin') = 'v1_import'
                        )
                      )
                      OR run.ordinal <= (
                        SELECT required_run.ordinal
                        FROM orchestration_v2_projection_runs AS required_run
                        WHERE required_run.thread_id = item.thread_id
                          AND required_run.run_id = ${window.requiredRunId ?? null}
                        LIMIT 1
                      )
                    )
                    AND NOT (
                      item.type = 'run_interrupt_result'
                      AND EXISTS (
                        SELECT 1
                        FROM orchestration_v2_projection_run_attempts AS attempt
                        WHERE attempt.run_id = item.run_id
                          AND attempt.root_node_id = item.node_id
                          AND attempt.status = 'superseded'
                      )
                      AND NOT EXISTS (
                        SELECT 1
                        FROM orchestration_v2_projection_turn_items AS request
                        WHERE request.thread_id = item.thread_id
                          AND request.run_id = item.run_id
                          AND request.type = 'run_interrupt_request'
                      )
                    )
                ), selected AS (
                  SELECT payload_json, ordinal, turn_item_id, run_id, type
                  FROM eligible
                  ORDER BY ordinal DESC, turn_item_id DESC
                  LIMIT ${window.rowLimit}
                ), retained AS (
                  SELECT payload_json, ordinal, turn_item_id FROM selected
                  UNION
                  SELECT request.payload_json, request.ordinal, request.turn_item_id
                  FROM orchestration_v2_projection_turn_items AS request
                  WHERE request.run_id IN (
                      SELECT run_id FROM selected
                      WHERE type = 'run_interrupt_result' AND run_id IS NOT NULL
                    )
                    AND request.type = 'run_interrupt_request'
                  UNION
                  SELECT latest.payload_json, latest.ordinal, latest.turn_item_id
                  FROM (
                    SELECT payload_json, ordinal, turn_item_id
                    FROM orchestration_v2_projection_turn_items
                    WHERE thread_id = ${threadId}
                      AND ${window.anchorItemId ?? null} IS NULL
                      AND ${window.requiredRunId ?? null} IS NULL
                    ORDER BY ordinal DESC, turn_item_id DESC
                    LIMIT 1
                  ) AS latest
                )
                SELECT payload_json
                FROM retained
                ORDER BY ordinal ASC, turn_item_id ASC
              `;
        const cohortPayloads = boundedTurnItemRows.map((row) =>
          parseEncodedPayload(row.payload_json),
        );
        const cohortJson = (field: string) =>
          JSON.stringify(
            cohortPayloads.flatMap((payload) => {
              const value = nullableStringField(payload, field);
              return value === null ? [] : [value];
            }),
          );
        const cohortRunIds = cohortJson("runId");
        const cohortNodeIds = cohortJson("nodeId");
        const cohortProviderThreadIds = cohortJson("providerThreadId");
        const cohortProviderTurnIds = cohortJson("providerTurnId");
        const cohortMessageIds = cohortJson("messageId");
        const cohortPlanIds = cohortJson("planId");
        const cohortCheckpointIds = cohortJson("checkpointId");
        const cohortHandoffIds = cohortJson("contextHandoffId");

        const [
          thread,
          runRows,
          attemptRows,
          nodeRows,
          subagentRows,
          providerSessionRows,
          providerThreadRows,
          providerTurnRows,
          runtimeRequestRows,
          messageRows,
          planRows,
          turnItemRows,
          checkpointScopeRows,
          checkpointRows,
          contextHandoffRows,
          contextTransferRows,
        ] = yield* Effect.all([
          decodeThreadPayload(threadRow.payload_json),
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_runs
            WHERE thread_id = ${threadId}
            ORDER BY ordinal ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_runs
            WHERE thread_id = ${threadId}
              AND (status IN ('queued','preparing','starting','running','waiting')
                OR run_id IN (SELECT value FROM json_each(${cohortRunIds}))
                OR run_id = ${window.requiredRunId ?? null})
            ORDER BY ordinal ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_run_attempts
            WHERE thread_id = ${threadId}
            ORDER BY run_id ASC, attempt_ordinal ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_run_attempts
            WHERE thread_id = ${threadId}
              AND (status IN ('pending','starting','running','waiting')
                OR run_id IN (SELECT value FROM json_each(${cohortRunIds})))
            ORDER BY run_id ASC, attempt_ordinal ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_nodes
            WHERE thread_id = ${threadId}
            ORDER BY COALESCE(started_at, ''), node_id ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_nodes
            WHERE thread_id = ${threadId}
              AND (status IN ('pending','starting','running','waiting')
                OR run_id IN (SELECT value FROM json_each(${cohortRunIds}))
                OR node_id IN (SELECT value FROM json_each(${cohortNodeIds})))
            ORDER BY COALESCE(started_at, ''), node_id ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_subagents
            WHERE thread_id = ${threadId}
            ORDER BY COALESCE(started_at, ''), subagent_id ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_subagents
            WHERE thread_id = ${threadId}
              AND (status IN ('pending','starting','running','waiting')
                OR run_id IN (SELECT value FROM json_each(${cohortRunIds}))
                OR parent_node_id IN (SELECT value FROM json_each(${cohortNodeIds})))
            ORDER BY COALESCE(started_at, ''), subagent_id ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT sessions.payload_json
            FROM orchestration_v2_projection_provider_sessions AS sessions
            INNER JOIN orchestration_v2_projection_provider_session_bindings AS bindings
              ON bindings.provider_session_id = sessions.provider_session_id
            WHERE bindings.thread_id = ${threadId}
            ORDER BY sessions.updated_at ASC, sessions.provider_session_id ASC
          `
            : sql<PayloadRow>`
            SELECT DISTINCT sessions.payload_json
            FROM orchestration_v2_projection_provider_sessions AS sessions
            INNER JOIN orchestration_v2_projection_provider_session_bindings AS bindings
              ON bindings.provider_session_id = sessions.provider_session_id
            LEFT JOIN orchestration_v2_projection_provider_threads AS threads
              ON threads.provider_session_id = sessions.provider_session_id
            WHERE bindings.thread_id = ${threadId}
              AND (sessions.status IN ('starting','running','waiting')
                OR threads.provider_thread_id IN (SELECT value FROM json_each(${cohortProviderThreadIds})))
            ORDER BY sessions.updated_at ASC, sessions.provider_session_id ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_provider_threads
            WHERE thread_id = ${threadId}
               OR owner_node_id IN (
                 SELECT node_id
                 FROM orchestration_v2_projection_nodes
                 WHERE thread_id = ${threadId}
               )
               OR provider_thread_id IN (
                 SELECT provider_thread_id
                 FROM orchestration_v2_projection_subagents
                 WHERE thread_id = ${threadId}
                   AND provider_thread_id IS NOT NULL
               )
            ORDER BY COALESCE(first_run_ordinal, 0), provider_thread_id ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_provider_threads
            WHERE (thread_id = ${threadId} AND status = 'active')
              OR provider_thread_id IN (SELECT value FROM json_each(${cohortProviderThreadIds}))
              OR owner_node_id IN (SELECT value FROM json_each(${cohortNodeIds}))
            ORDER BY COALESCE(first_run_ordinal, 0), provider_thread_id ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_provider_turns
            WHERE thread_id = ${threadId}
            ORDER BY provider_thread_id ASC, ordinal ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_provider_turns
            WHERE thread_id = ${threadId}
              AND (status IN ('starting','running','waiting')
                OR provider_turn_id IN (SELECT value FROM json_each(${cohortProviderTurnIds}))
                OR node_id IN (SELECT value FROM json_each(${cohortNodeIds})))
            ORDER BY provider_thread_id ASC, ordinal ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_runtime_requests
            WHERE thread_id = ${threadId}
            ORDER BY created_at ASC, runtime_request_id ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_runtime_requests
            WHERE thread_id = ${threadId}
              AND (status IN ('pending','waiting')
                OR node_id IN (SELECT value FROM json_each(${cohortNodeIds}))
                OR provider_turn_id IN (SELECT value FROM json_each(${cohortProviderTurnIds})))
            ORDER BY created_at ASC, runtime_request_id ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
                SELECT payload_json FROM orchestration_v2_projection_messages
                WHERE thread_id = ${threadId} ORDER BY created_at ASC, message_id ASC
              `
            : sql<PayloadRow>`
                SELECT payload_json FROM orchestration_v2_projection_messages AS message
                WHERE message.thread_id = ${threadId}
                  AND (
                    message.message_id IN (
                      SELECT value FROM json_each(${cohortMessageIds})
                    )
                    OR message.run_id IN (
                      SELECT run_id FROM orchestration_v2_projection_runs
                      WHERE thread_id = ${threadId}
                        AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting')
                    )
                  )
                ORDER BY created_at ASC, message_id ASC
              `,
          window === undefined
            ? sql<PayloadRow>`
                SELECT payload_json FROM orchestration_v2_projection_plans
                WHERE thread_id = ${threadId} ORDER BY plan_id ASC
              `
            : sql<PayloadRow>`
                SELECT payload_json FROM orchestration_v2_projection_plans AS plan
                WHERE plan.thread_id = ${threadId}
                  AND (
                    plan.status = 'active'
                    OR plan.plan_id IN (
                      SELECT value FROM json_each(${cohortPlanIds})
                    )
                  )
                ORDER BY plan_id ASC
              `,
          Effect.succeed(boundedTurnItemRows),
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_checkpoint_scopes
            WHERE thread_id = ${threadId}
            ORDER BY ordinal_within_parent ASC, scope_id ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_checkpoint_scopes
            WHERE thread_id = ${threadId}
              AND (run_id IN (SELECT value FROM json_each(${cohortRunIds}))
                OR node_id IN (SELECT value FROM json_each(${cohortNodeIds})))
            ORDER BY ordinal_within_parent ASC, scope_id ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_checkpoints
            WHERE thread_id = ${threadId}
            ORDER BY scope_id ASC, ordinal_within_scope ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_checkpoints
            WHERE thread_id = ${threadId}
              AND (status IN ('pending','capturing')
                OR checkpoint_id IN (SELECT value FROM json_each(${cohortCheckpointIds}))
                OR run_id IN (SELECT value FROM json_each(${cohortRunIds}))
                OR node_id IN (SELECT value FROM json_each(${cohortNodeIds})))
            ORDER BY scope_id ASC, ordinal_within_scope ASC
          `,
          window === undefined
            ? sql<PayloadRow>`
                SELECT payload_json FROM orchestration_v2_projection_context_handoffs
                WHERE thread_id = ${threadId} ORDER BY rowid ASC
              `
            : sql<PayloadRow>`
                SELECT payload_json FROM orchestration_v2_projection_context_handoffs AS handoff
                WHERE handoff.thread_id = ${threadId}
                  AND (
                    handoff.status IN ('pending', 'ready')
                    OR handoff.context_handoff_id IN (
                      SELECT value FROM json_each(${cohortHandoffIds})
                    )
                  )
                ORDER BY rowid ASC
              `,
          window === undefined
            ? sql<PayloadRow>`
            SELECT payload_json
            FROM orchestration_v2_projection_context_transfers
            WHERE source_thread_id = ${threadId} OR target_thread_id = ${threadId}
            ORDER BY rowid ASC
          `
            : sql<PayloadRow>`
            SELECT payload_json FROM orchestration_v2_projection_context_transfers
            WHERE (source_thread_id = ${threadId} OR target_thread_id = ${threadId})
              AND (status IN ('pending','running','waiting')
                OR target_run_id IN (SELECT value FROM json_each(${cohortRunIds}))
                OR json_extract(payload_json, '$.contextHandoffId') IN
                  (SELECT value FROM json_each(${cohortHandoffIds})))
            ORDER BY rowid ASC
          `,
        ]);

        const [
          runs,
          attempts,
          nodes,
          subagents,
          providerSessions,
          providerThreads,
          providerTurns,
          runtimeRequests,
          messages,
          plans,
          turnItems,
          checkpointScopes,
          checkpoints,
          contextHandoffs,
          contextTransfers,
        ] = yield* Effect.all([
          decodeRows(decodeRunPayload, threadId)(runRows),
          decodeRows(decodeRunAttemptPayload, threadId)(attemptRows),
          decodeRows(decodeNodePayload, threadId)(nodeRows),
          decodeRows(decodeSubagentPayload, threadId)(subagentRows),
          decodeRows(decodeProviderSessionPayload, threadId)(providerSessionRows),
          decodeRows(decodeProviderThreadPayload, threadId)(providerThreadRows),
          decodeRows(decodeProviderTurnPayload, threadId)(providerTurnRows),
          decodeRows(decodeRuntimeRequestPayload, threadId)(runtimeRequestRows),
          decodeRows(decodeMessagePayload, threadId)(messageRows),
          decodeRows(decodePlanPayload, threadId)(planRows),
          decodeRows(decodeTurnItemPayload, threadId)(turnItemRows),
          decodeRows(decodeCheckpointScopePayload, threadId)(checkpointScopeRows),
          decodeRows(decodeCheckpointPayload, threadId)(checkpointRows),
          decodeRows(decodeContextHandoffPayload, threadId)(contextHandoffRows),
          decodeRows(decodeContextTransferPayload, threadId)(contextTransferRows),
        ]);
        const orderedMessages = sortMessagesByTurnItemOrder(messages, turnItems);
        const projection = {
          thread,
          runs,
          attempts,
          nodes,
          subagents,
          providerSessions,
          providerThreads,
          providerTurns,
          runtimeRequests,
          messages: orderedMessages,
          plans,
          turnItems,
          checkpointScopes,
          checkpoints,
          contextHandoffs,
          contextTransfers,
          visibleTurnItems: [],
          updatedAt: thread.updatedAt,
        } satisfies OrchestrationV2ThreadProjection;
        return withLocalVisibleTurnItems(projection);
      }).pipe(
        Effect.mapError((cause) =>
          isProjectionStoreThreadNotFoundError(cause)
            ? cause
            : new ProjectionStoreReadError({
                threadId,
                cause,
              }),
        ),
      );

    const readProjection = (
      threadId: ThreadId,
      seenThreadIds: ReadonlySet<ThreadId>,
      window?: {
        readonly rowLimit: number;
        readonly anchorItemId?: TurnItemId | undefined;
        readonly requiredRunId?: RunId | undefined;
        readonly suppressLocal?: boolean | undefined;
        readonly historyAnchor?:
          | { readonly threadId: ThreadId; readonly itemId: TurnItemId }
          | undefined;
      },
    ): Effect.Effect<OrchestrationV2ThreadProjection, ProjectionStoreV2Error> =>
      Effect.gen(function* () {
        const localWindow =
          window?.suppressLocal === true ||
          (window?.historyAnchor !== undefined && window.historyAnchor.threadId !== threadId)
            ? { ...window, rowLimit: 0, anchorItemId: undefined }
            : window;
        const projection = yield* readCanonicalProjection(threadId, localWindow);
        const forkedFrom = projection.thread.forkedFrom;
        if (forkedFrom?.type !== "run" || seenThreadIds.has(forkedFrom.threadId)) {
          return withLocalVisibleTurnItems(projection);
        }

        const sourceWindow =
          window === undefined
            ? undefined
            : yield* Effect.gen(function* () {
                const historyAnchor = window.historyAnchor;
                const anchorBelongsToSource = historyAnchor?.threadId === forkedFrom.threadId;
                const rows = yield* sql<{ readonly turn_item_id: string }>`
                  WITH fork_run AS (
                    SELECT ordinal
                    FROM orchestration_v2_projection_runs
                    WHERE thread_id = ${forkedFrom.threadId}
                      AND run_id = ${forkedFrom.runId}
                    LIMIT 1
                  ), fork_boundary AS (
                    SELECT (
                      SELECT item.ordinal
                      FROM orchestration_v2_projection_turn_items AS item
                      WHERE item.run_id = run.run_id
                      ORDER BY item.ordinal DESC
                      LIMIT 1
                    ) AS ordinal
                    FROM orchestration_v2_projection_runs AS run
                    WHERE run.thread_id = ${forkedFrom.threadId}
                      AND run.ordinal <= (SELECT ordinal FROM fork_run)
                      AND EXISTS (
                        SELECT 1
                        FROM orchestration_v2_projection_turn_items AS item
                        WHERE item.run_id = run.run_id
                        LIMIT 1
                      )
                    ORDER BY run.ordinal DESC
                    LIMIT 1
                  ), effective_boundary AS (
                    SELECT COALESCE(
                      (SELECT ordinal FROM fork_boundary),
                      (
                        SELECT ordinal
                        FROM orchestration_v2_projection_turn_items
                        WHERE thread_id = ${forkedFrom.threadId}
                          AND run_id IS NULL
                          AND json_extract(payload_json, '$.historyOrigin') = 'v1_import'
                        ORDER BY ordinal DESC, turn_item_id DESC
                        LIMIT 1
                      ),
                      -1
                    ) AS ordinal
                  )
                  SELECT turn_item_id
                  FROM orchestration_v2_projection_turn_items
                  WHERE thread_id = ${forkedFrom.threadId}
                    AND ordinal <= COALESCE(
                      (SELECT ordinal FROM effective_boundary),
                      -1
                    )
                    AND (
                      ${anchorBelongsToSource ? 1 : 0} = 0
                      OR ordinal <= COALESCE(
                        (SELECT ordinal FROM orchestration_v2_projection_turn_items
                         WHERE thread_id = ${forkedFrom.threadId}
                           AND turn_item_id = ${historyAnchor?.itemId ?? null}
                         LIMIT 1),
                        (SELECT ordinal FROM effective_boundary),
                        -1
                      )
                    )
                  ORDER BY ordinal DESC, turn_item_id DESC
                  LIMIT 1
                `;
                const anchor = rows[0]?.turn_item_id;
                const anchorIsInDescendant =
                  historyAnchor !== undefined &&
                  historyAnchor.threadId !== threadId &&
                  historyAnchor.threadId !== forkedFrom.threadId;
                return {
                  rowLimit: window.rowLimit,
                  requiredRunId: forkedFrom.runId,
                  suppressLocal: anchor === undefined || anchorIsInDescendant,
                  ...(anchor === undefined || anchorIsInDescendant
                    ? {}
                    : { anchorItemId: TurnItemId.make(anchor) }),
                  ...(historyAnchor === undefined || historyAnchor.threadId === threadId
                    ? {}
                    : { historyAnchor }),
                };
              });

        const sourceProjection = yield* readProjection(
          forkedFrom.threadId,
          new Set([...seenThreadIds, threadId]),
          sourceWindow,
        );
        return {
          ...projection,
          visibleTurnItems: buildVisibleTurnItems({
            projection,
            sourceProjection,
          }),
        };
      }).pipe(
        Effect.mapError((cause) =>
          isProjectionStoreThreadNotFoundError(cause) || isProjectionStoreReadError(cause)
            ? cause
            : new ProjectionStoreReadError({ threadId, cause }),
        ),
      );

    const getRecoveryThreadIds = Effect.fn("ProjectionStore.getRecoveryThreadIds")(
      function* (kind: ProjectionRecoveryKind) {
        const candidates = (() => {
          switch (kind) {
            case "queued-runs":
              return sql`
                SELECT thread_id FROM orchestration_v2_projection_runs
                WHERE status = 'queued'
                  AND NOT EXISTS (
                    SELECT 1 FROM orchestration_v2_projection_runs AS active
                    WHERE active.thread_id = orchestration_v2_projection_runs.thread_id
                      AND active.status IN ('preparing', 'starting', 'running', 'waiting')
                  )
              `;
            case "delegated-completions":
              return sql`
                SELECT thread_id FROM orchestration_v2_projection_runs
                WHERE CASE WHEN json_valid(payload_json)
                  THEN json_type(payload_json, '$.delegatedCompletion.delivery') = 'object'
                  ELSE 0 END
              `;
            case "subagent-results":
              return sql`
                SELECT child.thread_id FROM orchestration_v2_projection_threads AS child
                WHERE CASE WHEN json_valid(child.payload_json) THEN
                  json_extract(child.payload_json, '$.lineage.relationshipToParent') = 'subagent'
                  AND json_extract(child.payload_json, '$.lineage.parentThreadId') IS NOT NULL
                  AND json_extract(child.payload_json, '$.forkedFrom.type') = 'node'
                  AND (
                    SELECT status FROM orchestration_v2_projection_runs
                    WHERE thread_id = child.thread_id
                    ORDER BY ordinal DESC LIMIT 1
                  ) IN ('completed', 'interrupted', 'failed', 'cancelled', 'rolled_back')
                  AND NOT EXISTS (
                    SELECT 1 FROM orchestration_v2_projection_context_transfers
                    WHERE source_thread_id = child.thread_id
                      AND target_thread_id = json_extract(child.payload_json, '$.lineage.parentThreadId')
                      AND type = 'subagent_result'
                  )
                  ELSE 0 END
              `;
            case "runtime":
              return sql`
                WITH pending_provider_threads AS (
                  SELECT provider_thread_id, thread_id, owner_node_id
                  FROM orchestration_v2_projection_provider_threads
                  WHERE status = 'active'
                    OR CASE WHEN json_valid(payload_json)
                      THEN json_array_length(payload_json, '$.pendingBackgroundTasks') > 0
                      ELSE 0 END
                )
                SELECT thread_id FROM orchestration_v2_projection_runs
                WHERE status IN ('queued', 'preparing', 'starting', 'running', 'waiting')
                UNION
                SELECT thread_id FROM orchestration_v2_projection_runtime_requests
                WHERE status = 'pending'
                UNION
                SELECT bindings.thread_id
                FROM orchestration_v2_projection_provider_session_bindings AS bindings
                JOIN orchestration_v2_projection_provider_sessions AS sessions
                  ON sessions.provider_session_id = bindings.provider_session_id
                WHERE sessions.status NOT IN ('stopped', 'error')
                UNION
                SELECT thread_id FROM pending_provider_threads
                UNION
                SELECT nodes.thread_id FROM orchestration_v2_projection_nodes AS nodes
                JOIN pending_provider_threads ON pending_provider_threads.owner_node_id = nodes.node_id
                UNION
                SELECT subagents.thread_id FROM orchestration_v2_projection_subagents AS subagents
                JOIN pending_provider_threads
                  ON pending_provider_threads.provider_thread_id = subagents.provider_thread_id
                UNION
                SELECT thread_id FROM orchestration_v2_projection_turn_items
                WHERE type IN ('command_execution', 'dynamic_tool', 'subagent')
                  AND status IN ('pending', 'running', 'waiting')
                UNION
                SELECT thread_id FROM orchestration_v2_effect_outbox
                WHERE status IN ('pending', 'running')
              `;
          }
        })();
        const rows = yield* sql<{ readonly thread_id: string }>`
          SELECT thread_id FROM orchestration_v2_projection_threads
          WHERE deleted_at IS NULL
            AND thread_id IN (${candidates})
            ${kind === "queued-runs" ? sql`AND archived_at IS NULL` : sql``}
          ORDER BY updated_at ASC, thread_id ASC
        `;
        return rows.map((row) => ThreadId.make(row.thread_id));
      },
      Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })),
    );

    // Decode every canonical row once. Full thread reads repeat shared sessions,
    // provider threads, transfers, and inherited fork histories for each owner.
    const getUnreadableThreadIds = Effect.fn("ProjectionStore.getUnreadableThreadIds")(
      function* () {
        const actualThreadIds = new Set<ThreadId>();
        const unreadable = new Set<ThreadId>();
        const forkChildren = new Map<ThreadId, Array<ThreadId>>();
        const tables: ReadonlyArray<{
          readonly name: string;
          readonly id: string;
          readonly decode: (payload: string, entityId: string) => Effect.Effect<unknown, unknown>;
        }> = [
          {
            name: "threads",
            id: "thread_id",
            decode: (payload: string, entityId: string) =>
              decodeThreadPayload(payload).pipe(
                Effect.tap((thread) =>
                  Effect.sync(() => {
                    if (thread.forkedFrom?.type !== "run") return;
                    const sourceId = thread.forkedFrom.threadId;
                    const children = forkChildren.get(sourceId) ?? [];
                    children.push(ThreadId.make(entityId));
                    forkChildren.set(sourceId, children);
                  }),
                ),
              ),
          },
          { name: "runs", id: "run_id", decode: (payload) => decodeRunPayload(payload) },
          {
            name: "run_attempts",
            id: "attempt_id",
            decode: (payload) => decodeRunAttemptPayload(payload),
          },
          { name: "nodes", id: "node_id", decode: (payload) => decodeNodePayload(payload) },
          {
            name: "subagents",
            id: "subagent_id",
            decode: (payload) => decodeSubagentPayload(payload),
          },
          {
            name: "provider_sessions",
            id: "provider_session_id",
            decode: (payload) => decodeProviderSessionPayload(payload),
          },
          {
            name: "provider_threads",
            id: "provider_thread_id",
            decode: (payload) => decodeProviderThreadPayload(payload),
          },
          {
            name: "provider_turns",
            id: "provider_turn_id",
            decode: (payload) => decodeProviderTurnPayload(payload),
          },
          {
            name: "runtime_requests",
            id: "runtime_request_id",
            decode: (payload) => decodeRuntimeRequestPayload(payload),
          },
          {
            name: "messages",
            id: "message_id",
            decode: (payload) => decodeMessagePayload(payload),
          },
          { name: "plans", id: "plan_id", decode: decodePlanPayload },
          {
            name: "turn_items",
            id: "turn_item_id",
            decode: (payload) => decodeTurnItemPayload(payload),
          },
          {
            name: "checkpoint_scopes",
            id: "scope_id",
            decode: (payload) => decodeCheckpointScopePayload(payload),
          },
          {
            name: "checkpoints",
            id: "checkpoint_id",
            decode: (payload) => decodeCheckpointPayload(payload),
          },
          {
            name: "context_handoffs",
            id: "context_handoff_id",
            decode: (payload) => decodeContextHandoffPayload(payload),
          },
          {
            name: "context_transfers",
            id: "context_transfer_id",
            decode: (payload) => decodeContextTransferPayload(payload),
          },
        ];
        const pageSize = 500;
        for (const table of tables) {
          let afterRowId = 0;
          while (true) {
            const rows = yield* sql<{
              readonly row_id: number;
              readonly entity_id: string;
              readonly thread_id: string | null;
              readonly target_thread_id: string | null;
              readonly payload_json: string;
            }>`
              SELECT rowid AS row_id, ${sql(table.id)} AS entity_id,
                ${sql(table.name === "context_transfers" ? "source_thread_id" : "thread_id")} AS thread_id,
                ${table.name === "context_transfers" ? sql`target_thread_id` : sql`NULL`} AS target_thread_id,
                payload_json
              FROM ${sql(`orchestration_v2_projection_${table.name}`)}
              WHERE rowid > ${afterRowId}
              ORDER BY rowid ASC LIMIT ${pageSize}
            `;
            for (const row of rows) {
              if (table.name === "threads") actualThreadIds.add(ThreadId.make(row.entity_id));
              const decoded = yield* Effect.exit(
                Effect.suspend(() => table.decode(row.payload_json, row.entity_id)),
              );
              if (decoded._tag === "Success") continue;
              if (table.name === "provider_sessions") {
                const bindings = yield* sql<{ readonly thread_id: string }>`
                  SELECT thread_id FROM orchestration_v2_projection_provider_session_bindings
                  WHERE provider_session_id = ${row.entity_id}
                `;
                for (const binding of bindings) unreadable.add(ThreadId.make(binding.thread_id));
              } else {
                if (row.thread_id !== null) unreadable.add(ThreadId.make(row.thread_id));
                if (row.target_thread_id !== null)
                  unreadable.add(ThreadId.make(row.target_thread_id));
              }
              if (table.name === "provider_threads") {
                const owners = yield* sql<{ readonly thread_id: string }>`
                  SELECT thread_id FROM orchestration_v2_projection_nodes
                  WHERE node_id = (
                    SELECT owner_node_id FROM orchestration_v2_projection_provider_threads
                    WHERE provider_thread_id = ${row.entity_id}
                  )
                  UNION
                  SELECT thread_id FROM orchestration_v2_projection_subagents
                  WHERE provider_thread_id = ${row.entity_id}
                `;
                for (const owner of owners) unreadable.add(ThreadId.make(owner.thread_id));
              }
            }
            if (rows.length < pageSize) break;
            afterRowId = rows[rows.length - 1]!.row_id;
            yield* Effect.yieldNow;
          }
        }
        // A full fork projection also reads its source. Propagate decode failures
        // and missing sources through the same ancestry without loading history.
        const pending = [
          ...unreadable,
          ...[...forkChildren.keys()].filter((sourceId) => !actualThreadIds.has(sourceId)),
        ];
        for (let index = 0; index < pending.length; index += 1) {
          for (const childId of forkChildren.get(pending[index]!) ?? []) {
            if (unreadable.has(childId)) continue;
            unreadable.add(childId);
            pending.push(childId);
          }
        }
        return [...unreadable].filter((threadId) => actualThreadIds.has(threadId)).sort();
      },
      Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })),
    );

    const getThreadProjection: ProjectionStoreV2Shape["getThreadProjection"] = (threadId) =>
      readProjection(threadId, new Set());

    const getThread: ProjectionStoreV2Shape["getThread"] = (threadId) =>
      Effect.gen(function* () {
        const rows = yield* sql<PayloadRow>`
          SELECT payload_json FROM orchestration_v2_projection_threads
          WHERE thread_id = ${threadId}
        `;
        if (rows[0] === undefined) {
          return yield* new ProjectionStoreThreadNotFoundError({ threadId });
        }
        return yield* decodeThreadPayload(rows[0].payload_json);
      }).pipe(
        Effect.mapError((cause) =>
          isProjectionStoreThreadNotFoundError(cause)
            ? cause
            : new ProjectionStoreReadError({ threadId, cause }),
        ),
      );

    const requireThread = (threadId: ThreadId) =>
      Effect.gen(function* () {
        const rows = yield* sql<{ readonly thread_id: string }>`
        SELECT thread_id FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId} LIMIT 1
      `;
        if (rows.length === 0) return yield* new ProjectionStoreThreadNotFoundError({ threadId });
      });
    const controlReadError = (threadId: ThreadId) => (cause: unknown) =>
      isProjectionStoreThreadNotFoundError(cause)
        ? cause
        : new ProjectionStoreReadError({ threadId, cause });

    const getPendingNativeUserInputs: ProjectionStoreV2Shape["getPendingNativeUserInputs"] = (
      threadId,
      providerTurnId,
    ) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            const rows = yield* sql<PayloadRow>`
          SELECT payload_json FROM orchestration_v2_projection_runtime_requests
          WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId}
            AND kind = 'user_input' AND status = 'pending'
        `;
            const runtimeRequests = (yield* decodeRows(
              decodeRuntimeRequestPayload,
              threadId,
            )(rows)).filter((request) => request.responseCapability.type !== "message");
            if (runtimeRequests.length === 0) return { runtimeRequests, nodes: [], turnItems: [] };
            const nodeRows = yield* sql<PayloadRow>`
          SELECT payload_json FROM orchestration_v2_projection_nodes
          WHERE thread_id = ${threadId} AND ${sql.in(
            "node_id",
            runtimeRequests.map((request) => request.nodeId),
          )}
        `;
            const itemRows = yield* sql<PayloadRow>`
          SELECT payload_json FROM orchestration_v2_projection_turn_items
          WHERE thread_id = ${threadId} AND provider_turn_id = ${providerTurnId}
            AND type = 'user_input_request'
        `;
            const requestIds = new Set(runtimeRequests.map((request) => request.id));
            return {
              runtimeRequests,
              nodes: yield* decodeRows(decodeNodePayload, threadId)(nodeRows),
              turnItems: (yield* decodeRows(decodeTurnItemPayload, threadId)(itemRows)).filter(
                (item) => item.type === "user_input_request" && requestIds.has(item.requestId),
              ),
            };
          }),
        )
        .pipe(Effect.mapError(controlReadError(threadId)));

    const getRuntimeRequest: ProjectionStoreV2Shape["getRuntimeRequest"] = (threadId, requestId) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            yield* requireThread(threadId);
            const rows = yield* sql<PayloadRow>`
          SELECT payload_json FROM orchestration_v2_projection_runtime_requests
          WHERE thread_id = ${threadId} AND runtime_request_id = ${requestId}
        `;
            return rows[0] === undefined
              ? undefined
              : yield* decodeRuntimeRequestPayload(rows[0].payload_json);
          }),
        )
        .pipe(Effect.mapError(controlReadError(threadId)));

    const getProviderControlContext: ProjectionStoreV2Shape["getProviderControlContext"] = (
      threadId,
      target,
    ) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            yield* requireThread(threadId);
            // Match the full projection's ownership scope, but fetch only the target
            // records. Transcript items and fork ancestors are never read here.
            const [threadRows, turnRows, attemptRows, messageRows] = yield* Effect.all([
              sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_provider_threads
            WHERE provider_thread_id = ${target.providerThreadId} AND (
              thread_id = ${threadId}
              OR owner_node_id IN (SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId})
              OR provider_thread_id IN (SELECT provider_thread_id FROM orchestration_v2_projection_subagents WHERE thread_id = ${threadId})
            )`,
              sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_provider_turns
            WHERE thread_id = ${threadId} AND provider_turn_id = ${target.providerTurnId}`,
              target.attemptId === undefined
                ? Effect.succeed([] as ReadonlyArray<PayloadRow>)
                : sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_run_attempts
              WHERE thread_id = ${threadId} AND attempt_id = ${target.attemptId}`,
              target.messageId === undefined
                ? Effect.succeed([] as ReadonlyArray<PayloadRow>)
                : sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_messages
              WHERE thread_id = ${threadId} AND message_id = ${target.messageId}`,
            ]);
            const providerThread =
              threadRows[0] === undefined
                ? undefined
                : yield* decodeProviderThreadPayload(threadRows[0].payload_json);
            const providerTurn =
              turnRows[0] === undefined
                ? undefined
                : yield* decodeProviderTurnPayload(turnRows[0].payload_json);
            const attempt =
              attemptRows[0] === undefined
                ? undefined
                : yield* decodeRunAttemptPayload(attemptRows[0].payload_json);
            const message =
              messageRows[0] === undefined
                ? undefined
                : yield* decodeMessagePayload(messageRows[0].payload_json);
            let run: OrchestrationV2Run | undefined;
            if (target.messageId !== undefined && providerTurn !== undefined) {
              const rows =
                yield* sql<PayloadRow>`SELECT payload_json FROM orchestration_v2_projection_runs
            WHERE thread_id = ${threadId} AND json_extract(payload_json, '$.activeAttemptId') IS ${providerTurn.runAttemptId}
            ORDER BY ordinal ASC LIMIT 1`;
              if (rows[0] !== undefined) run = yield* decodeRunPayload(rows[0].payload_json);
            }
            return { providerThread, providerTurn, attempt, message, run };
          }),
        )
        .pipe(Effect.mapError(controlReadError(threadId)));

    const getCheckpointContext: ProjectionStoreV2Shape["getCheckpointContext"] = (threadId) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            const threads = yield* sql<{ readonly thread_id: string }>`
            SELECT thread_id FROM orchestration_v2_projection_threads
            WHERE thread_id = ${threadId} LIMIT 1
          `;
            if (threads.length === 0) {
              return yield* new ProjectionStoreThreadNotFoundError({ threadId });
            }
            // Checkpoint diffs need no transcript or fork ancestry. Select just the
            // metadata columns so large run/checkpoint JSON payloads stay in SQLite.
            const [runs, checkpointScopes, checkpoints] = yield* Effect.all([
              sql`
              SELECT run_id AS id, ordinal, status
              FROM orchestration_v2_projection_runs
              WHERE thread_id = ${threadId}
              ORDER BY ordinal ASC
            `,
              sql`
              SELECT scope_id AS id, run_id AS "runId", kind,
                json_extract(payload_json, '$.cwd') AS cwd
              FROM orchestration_v2_projection_checkpoint_scopes
              WHERE thread_id = ${threadId}
              ORDER BY ordinal_within_parent ASC, scope_id ASC
            `,
              sql`
              SELECT scope_id AS "scopeId", run_id AS "runId",
                app_run_ordinal AS "appRunOrdinal", status,
                json_extract(payload_json, '$.ref') AS ref
              FROM orchestration_v2_projection_checkpoints
              WHERE thread_id = ${threadId}
              ORDER BY scope_id ASC, ordinal_within_scope ASC
            `,
            ]);
            return yield* decodeCheckpointContext({ runs, checkpointScopes, checkpoints });
          }),
        )
        .pipe(
          Effect.mapError((cause) =>
            isProjectionStoreThreadNotFoundError(cause)
              ? cause
              : new ProjectionStoreReadError({ threadId, cause }),
          ),
        );

    const getThreadSnapshot: ProjectionStoreV2Shape["getThreadSnapshot"] = (threadId) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            const projection = yield* getThreadProjection(threadId);
            const rows = yield* sql<{ readonly snapshot_sequence: number | null }>`
            SELECT MAX(sequence) AS snapshot_sequence
            FROM orchestration_events
            WHERE application_event_version = 2
              AND aggregate_kind = 'thread'
              AND stream_id = ${threadId}
          `;
            return {
              schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
              snapshotSequence: rows[0]?.snapshot_sequence ?? 0,
              projection,
            };
          }),
        )
        .pipe(
          Effect.mapError((cause) =>
            isProjectionStoreThreadNotFoundError(cause) || isProjectionStoreReadError(cause)
              ? cause
              : new ProjectionStoreReadError({ threadId, cause }),
          ),
        );

    const getThreadSnapshotWindow: ProjectionStoreV2Shape["getThreadSnapshotWindow"] = (
      threadId,
      options,
    ) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            const historyAnchor =
              options.anchorItemId === undefined
                ? undefined
                : {
                    itemId: options.anchorItemId,
                    threadId:
                      options.anchorThreadId ??
                      (yield* sql<{ readonly thread_id: string }>`
                        SELECT thread_id
                        FROM orchestration_v2_projection_turn_items
                        WHERE turn_item_id = ${options.anchorItemId}
                        LIMIT 1
                      `).map((row) => ThreadId.make(row.thread_id))[0] ??
                      threadId,
                  };
            const projection = yield* readProjection(threadId, new Set(), {
              rowLimit: options.rowLimit,
              ...(historyAnchor?.threadId === threadId
                ? { anchorItemId: historyAnchor.itemId }
                : {}),
              ...(historyAnchor === undefined ? {} : { historyAnchor }),
            });
            const rows = yield* sql<{ readonly snapshot_sequence: number | null }>`
              SELECT MAX(sequence) AS snapshot_sequence
              FROM orchestration_events
              WHERE application_event_version = 2
                AND aggregate_kind = 'thread'
                AND stream_id = ${threadId}
            `;
            return {
              schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
              snapshotSequence: rows[0]?.snapshot_sequence ?? 0,
              projection,
            };
          }),
        )
        .pipe(
          Effect.mapError((cause) =>
            isProjectionStoreThreadNotFoundError(cause) || isProjectionStoreReadError(cause)
              ? cause
              : new ProjectionStoreReadError({ threadId, cause }),
          ),
        );

    const selectShellThreadRows = (threadId?: ThreadId, location?: "active" | "archive") =>
      sql<ShellThreadRow>`
            SELECT
              t.thread_id,
              t.payload_json,
              CASE
                WHEN json_extract(t.payload_json, '$.forkedFrom.type') = 'run'
                  THEN json_extract(t.payload_json, '$.forkedFrom.threadId')
                ELSE NULL
              END AS forked_from_run_source_thread_id,
              (
                SELECT r.run_id
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS latest_run_id,
              (
                SELECT r.status
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS latest_run_status,
              (
                SELECT r.requested_at
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS latest_run_requested_at,
              (
                SELECT json_extract(r.payload_json, '$.startedAt')
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS latest_run_started_at,
              (
                SELECT r.completed_at
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS latest_run_completed_at,
              (
                SELECT r.run_id
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                  AND r.status IN ('preparing', 'starting', 'running')
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS active_run_id,
              (
                SELECT r.status
                FROM orchestration_v2_projection_runs r
                WHERE r.thread_id = t.thread_id
                  AND r.status IN ('preparing', 'starting', 'running', 'waiting')
                ORDER BY r.ordinal DESC, r.run_id DESC
                LIMIT 1
              ) AS activity_run_status,
              (
                SELECT json_extract(session.payload_json, '$.lastError')
                FROM orchestration_v2_projection_provider_sessions session
                INNER JOIN orchestration_v2_projection_provider_session_bindings binding
                  ON binding.provider_session_id = session.provider_session_id
                WHERE binding.thread_id = t.thread_id
                  AND session.provider_instance_id = t.provider_instance_id
                ORDER BY session.updated_at DESC, session.provider_session_id DESC
                LIMIT 1
              ) AS last_error,
              (
                SELECT request.payload_json
                FROM orchestration_v2_projection_runtime_requests request
                WHERE request.thread_id = t.thread_id
                  AND request.status = 'pending'
                ORDER BY request.created_at DESC, request.runtime_request_id DESC
                LIMIT 1
              ) AS pending_request_payload_json,
              (
                SELECT message.updated_at
                FROM orchestration_v2_projection_messages message
                WHERE message.thread_id = t.thread_id
                  AND message.role = 'user'
                ORDER BY message.updated_at DESC, message.message_id DESC
                LIMIT 1
              ) AS latest_user_message_at,
              EXISTS (
                SELECT 1
                FROM orchestration_v2_projection_plans plan
                WHERE plan.thread_id = t.thread_id
                  AND plan.kind = 'proposed_plan'
                  AND plan.status = 'active'
              ) AS has_actionable_proposed_plan,
              (
                SELECT COUNT(*)
                FROM orchestration_v2_projection_turn_items i
                LEFT JOIN orchestration_v2_projection_runs r
                  ON r.run_id = i.run_id
                WHERE i.thread_id = t.thread_id
                  AND (i.run_id IS NULL OR r.status <> 'rolled_back')
              ) AS item_count,
              (
                SELECT COUNT(*)
                FROM orchestration_v2_projection_turn_items i
                WHERE i.thread_id = t.thread_id
                  AND i.run_id IS NULL
              ) AS runless_item_count
            FROM orchestration_v2_projection_threads t
            WHERE t.deleted_at IS NULL${threadId === undefined ? sql`` : sql` AND t.thread_id = ${threadId}`}${
              location === "active"
                ? sql` AND json_extract(t.payload_json, '$.archivedAt') IS NULL`
                : location === "archive"
                  ? sql` AND json_extract(t.payload_json, '$.archivedAt') IS NOT NULL`
                  : sql``
            }
            ORDER BY t.updated_at ASC, t.thread_id ASC
          `;

    const selectShellRunRows = (threadIds?: ReadonlyArray<ThreadId>) =>
      threadIds === undefined
        ? sql<ShellRunRow>`
            SELECT thread_id, run_id, ordinal
            FROM orchestration_v2_projection_runs
          `
        : sql<ShellRunRow>`
            SELECT thread_id, run_id, ordinal
            FROM orchestration_v2_projection_runs
            WHERE thread_id IN ${sql.in(threadIds)}
          `;

    const selectShellRunItemCounts = (threadIds?: ReadonlyArray<ThreadId>) =>
      threadIds === undefined
        ? sql<ShellRunItemCountRow>`
            SELECT thread_id, run_id, COUNT(*) AS item_count
            FROM orchestration_v2_projection_turn_items
            WHERE run_id IS NOT NULL
            GROUP BY thread_id, run_id
          `
        : sql<ShellRunItemCountRow>`
            SELECT thread_id, run_id, COUNT(*) AS item_count
            FROM orchestration_v2_projection_turn_items
            WHERE run_id IS NOT NULL
              AND thread_id IN ${sql.in(threadIds)}
            GROUP BY thread_id, run_id
          `;

    const selectShellProviderThreadRows = (threadIds?: ReadonlyArray<ThreadId>) =>
      threadIds === undefined
        ? sql<PayloadRow & { readonly thread_id: string }>`
            SELECT thread_id, payload_json
            FROM orchestration_v2_projection_provider_threads
            WHERE thread_id IS NOT NULL
          `
        : sql<PayloadRow & { readonly thread_id: string }>`
            SELECT thread_id, payload_json
            FROM orchestration_v2_projection_provider_threads
            WHERE thread_id IN ${sql.in(threadIds)}
          `;

    const selectShellPendingTurnItemRows = (threadIds?: ReadonlyArray<ThreadId>) =>
      threadIds === undefined
        ? sql<PayloadRow & { readonly thread_id: string }>`
            SELECT i.thread_id, i.payload_json
            FROM orchestration_v2_projection_turn_items i
            LEFT JOIN orchestration_v2_projection_runs r
              ON r.run_id = i.run_id
            WHERE i.type IN ('command_execution', 'dynamic_tool', 'subagent')
              AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled')
              -- A rolled-back run's items are abandoned, not pending. Without
              -- this the shell reports Waiting for work no one will finish,
              -- matching the item_count query's exclusion above.
              AND (i.run_id IS NULL OR r.status <> 'rolled_back')
          `
        : sql<PayloadRow & { readonly thread_id: string }>`
            SELECT i.thread_id, i.payload_json
            FROM orchestration_v2_projection_turn_items i
            LEFT JOIN orchestration_v2_projection_runs r
              ON r.run_id = i.run_id
            WHERE i.type IN ('command_execution', 'dynamic_tool', 'subagent')
              AND i.status NOT IN ('completed', 'interrupted', 'failed', 'cancelled')
              AND (i.run_id IS NULL OR r.status <> 'rolled_back')
              AND i.thread_id IN ${sql.in(threadIds)}
          `;

    const runMapsByThreadId = (input: {
      readonly runRows: ReadonlyArray<ShellRunRow>;
      readonly itemCountRows: ReadonlyArray<ShellRunItemCountRow>;
    }) => {
      const runOrdinalsByThreadId = new Map<ThreadId, Map<RunId, number>>();
      for (const row of input.runRows) {
        const threadId = ThreadId.make(row.thread_id);
        const runId = RunId.make(row.run_id);
        const existing = runOrdinalsByThreadId.get(threadId) ?? new Map<RunId, number>();
        existing.set(runId, row.ordinal);
        runOrdinalsByThreadId.set(threadId, existing);
      }
      const itemCountsByThreadId = new Map<ThreadId, Map<RunId, number>>();
      for (const row of input.itemCountRows) {
        const threadId = ThreadId.make(row.thread_id);
        const runId = RunId.make(row.run_id);
        const existing = itemCountsByThreadId.get(threadId) ?? new Map<RunId, number>();
        existing.set(runId, row.item_count);
        itemCountsByThreadId.set(threadId, existing);
      }
      return { runOrdinalsByThreadId, itemCountsByThreadId };
    };

    const pendingBackgroundDataByThreadId = (input: {
      readonly providerThreadRows: ReadonlyArray<PayloadRow & { readonly thread_id: string }>;
      readonly pendingTurnItemRows: ReadonlyArray<PayloadRow & { readonly thread_id: string }>;
    }) =>
      Effect.gen(function* () {
        const providerThreadsByThreadId = new Map<
          ThreadId,
          Array<OrchestrationV2ThreadProjection["providerThreads"][number]>
        >();
        for (const row of input.providerThreadRows) {
          const providerThread = yield* decodeProviderThreadPayload(row.payload_json);
          const threadId =
            row.thread_id.length > 0 ? ThreadId.make(row.thread_id) : providerThread.appThreadId;
          if (threadId === null) {
            continue;
          }
          const existing = providerThreadsByThreadId.get(threadId) ?? [];
          existing.push(providerThread);
          providerThreadsByThreadId.set(threadId, existing);
        }

        const pendingTurnItemsByThreadId = new Map<ThreadId, Array<OrchestrationV2TurnItem>>();
        for (const row of input.pendingTurnItemRows) {
          const turnItem = yield* decodeTurnItemPayload(row.payload_json);
          const threadId = ThreadId.make(row.thread_id);
          const existing = pendingTurnItemsByThreadId.get(threadId) ?? [];
          existing.push(turnItem);
          pendingTurnItemsByThreadId.set(threadId, existing);
        }

        return { providerThreadsByThreadId, pendingTurnItemsByThreadId };
      });

    const getSettlementCandidates: ProjectionStoreV2Shape["getSettlementCandidates"] = () =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            // Settlement needs activity timestamps, not item counts, run history,
            // provider sessions or inherited fork rows. Reject parked and busy
            // threads before loading the remaining candidates' background work.
            const rows = yield* sql<SettlementThreadRow>`
            SELECT t.thread_id, t.payload_json,
              r.run_id AS latest_run_id,
              r.status AS latest_run_status,
              r.requested_at AS latest_run_requested_at,
              json_extract(r.payload_json, '$.startedAt') AS latest_run_started_at,
              r.completed_at AS latest_run_completed_at,
              (
                SELECT message.updated_at
                FROM orchestration_v2_projection_messages message
                WHERE message.thread_id = t.thread_id AND message.role = 'user'
                ORDER BY message.updated_at DESC, message.message_id DESC
                LIMIT 1
              ) AS latest_user_message_at
            FROM orchestration_v2_projection_threads t
            LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = (
              SELECT latest.run_id FROM orchestration_v2_projection_runs latest
              WHERE latest.thread_id = t.thread_id
              ORDER BY latest.ordinal DESC, latest.run_id DESC
              LIMIT 1
            )
            WHERE t.deleted_at IS NULL
              AND json_extract(t.payload_json, '$.archivedAt') IS NULL
              AND json_extract(t.payload_json, '$.settledOverride') IS NULL
              AND json_extract(t.payload_json, '$.pinnedAt') IS NULL
              AND NOT EXISTS (
                SELECT 1 FROM orchestration_v2_projection_runs active
                WHERE active.thread_id = t.thread_id
                  AND active.status IN ('preparing', 'starting', 'running', 'waiting')
              )
              AND NOT EXISTS (
                SELECT 1 FROM orchestration_v2_projection_runtime_requests request
                WHERE request.thread_id = t.thread_id AND request.status = 'pending'
              )
            ORDER BY t.updated_at ASC, t.thread_id ASC
          `;
            if (rows.length === 0) return [];
            const threadIds = rows.map((row) => ThreadId.make(row.thread_id));
            const { providerThreadsByThreadId, pendingTurnItemsByThreadId } =
              yield* pendingBackgroundDataByThreadId({
                providerThreadRows: yield* selectShellProviderThreadRows(threadIds),
                pendingTurnItemRows: yield* selectShellPendingTurnItemRows(threadIds),
              });
            return yield* Effect.forEach(rows, (row) =>
              Effect.gen(function* () {
                const thread = yield* decodeThreadPayload(row.payload_json);
                const status = shellStatusFromStoredRunStatus(row.latest_run_status);
                const latestRunId =
                  row.latest_run_id === null ? null : RunId.make(row.latest_run_id);
                return {
                  ...thread,
                  pinnedAt: thread.pinnedAt ?? null,
                  snoozedUntil: thread.snoozedUntil ?? null,
                  snoozedAt: thread.snoozedAt ?? null,
                  status,
                  latestRunId,
                  latestRunRequestedAt:
                    row.latest_run_requested_at === null
                      ? null
                      : DateTime.makeUnsafe(row.latest_run_requested_at),
                  latestRunStartedAt:
                    row.latest_run_started_at === null
                      ? null
                      : DateTime.makeUnsafe(row.latest_run_started_at),
                  latestRunCompletedAt:
                    row.latest_run_completed_at === null
                      ? null
                      : DateTime.makeUnsafe(row.latest_run_completed_at),
                  latestUserMessageAt:
                    row.latest_user_message_at === null
                      ? null
                      : DateTime.makeUnsafe(row.latest_user_message_at),
                  activityRunStatus: null,
                  pendingRuntimeRequest: null,
                  pendingBackgroundTasks: derivePendingBackgroundWork({
                    latestRun:
                      latestRunId === null || status === "idle"
                        ? null
                        : { id: latestRunId, ordinal: 0, status },
                    providerThreads: providerThreadsByThreadId.get(thread.id) ?? [],
                    turnItems: pendingTurnItemsByThreadId.get(thread.id) ?? [],
                    activeProviderThreadId: thread.activeProviderThreadId,
                    hasActiveRun: false,
                  }),
                } satisfies ProjectionSettlementCandidate;
              }),
            );
          }),
        )
        .pipe(Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })));

    const shellThreadStateFromRow = (input: {
      readonly row: ShellThreadRow;
      readonly runOrdinalsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;
      readonly itemCountsByThreadId: ReadonlyMap<ThreadId, Map<RunId, number>>;
      readonly providerThreadsByThreadId: ReadonlyMap<
        ThreadId,
        ReadonlyArray<OrchestrationV2ThreadProjection["providerThreads"][number]>
      >;
      readonly pendingTurnItemsByThreadId: ReadonlyMap<
        ThreadId,
        ReadonlyArray<OrchestrationV2TurnItem>
      >;
    }) =>
      Effect.gen(function* () {
        const {
          row,
          runOrdinalsByThreadId,
          itemCountsByThreadId,
          providerThreadsByThreadId,
          pendingTurnItemsByThreadId,
        } = input;
        const thread = yield* decodeThreadPayload(row.payload_json);
        const pendingRuntimeRequest =
          row.pending_request_payload_json === null
            ? null
            : yield* decodeRuntimeRequestPayload(row.pending_request_payload_json);
        const latestRunId = row.latest_run_id === null ? null : RunId.make(row.latest_run_id);
        const latestRunStatus = shellStatusFromStoredRunStatus(row.latest_run_status);
        const pendingBackgroundTasks = [
          ...derivePendingBackgroundWork({
            latestRun:
              latestRunId === null || latestRunStatus === "idle"
                ? null
                : {
                    id: latestRunId,
                    ordinal: 0,
                    status: latestRunStatus,
                  },
            providerThreads: providerThreadsByThreadId.get(thread.id) ?? [],
            turnItems: pendingTurnItemsByThreadId.get(thread.id) ?? [],
            activeProviderThreadId: thread.activeProviderThreadId,
            hasActiveRun: row.active_run_id !== null,
          }),
        ];
        return {
          thread,
          latestRunId,
          latestRunStatus,
          latestRunRequestedAt:
            row.latest_run_requested_at === null
              ? null
              : DateTime.makeUnsafe(row.latest_run_requested_at),
          latestRunStartedAt:
            row.latest_run_started_at === null
              ? null
              : DateTime.makeUnsafe(row.latest_run_started_at),
          latestRunCompletedAt:
            row.latest_run_completed_at === null
              ? null
              : DateTime.makeUnsafe(row.latest_run_completed_at),
          activeRunId: row.active_run_id === null ? null : RunId.make(row.active_run_id),
          activityRunStatus:
            row.activity_run_status === "preparing" ||
            row.activity_run_status === "starting" ||
            row.activity_run_status === "running" ||
            row.activity_run_status === "waiting"
              ? row.activity_run_status
              : null,
          lastError: row.last_error,
          pendingRuntimeRequest,
          latestUserMessageAt:
            row.latest_user_message_at === null
              ? null
              : DateTime.makeUnsafe(row.latest_user_message_at),
          hasActionableProposedPlan: row.has_actionable_proposed_plan === 1,
          pendingBackgroundTasks,
          itemCount: row.item_count,
          runlessItemCount: row.runless_item_count,
          updatedAt: thread.updatedAt,
          runOrdinalById: runOrdinalsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
          itemCountByRunId: itemCountsByThreadId.get(ThreadId.make(row.thread_id)) ?? new Map(),
        } satisfies ShellThreadState;
      });

    const getShellSnapshot: ProjectionStoreV2Shape["getShellSnapshot"] = (options) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            const targetThreadRows = yield* selectShellThreadRows(undefined, options?.location);
            const targetThreadIds = new Set(
              targetThreadRows.map((row) => ThreadId.make(row.thread_id)),
            );
            const rowsByThreadId = new Map(
              targetThreadRows.map((row) => [ThreadId.make(row.thread_id), row] as const),
            );
            const pendingSourceIds = targetThreadRows.flatMap((row) =>
              row.forked_from_run_source_thread_id === null
                ? []
                : [ThreadId.make(row.forked_from_run_source_thread_id)],
            );
            while (pendingSourceIds.length > 0) {
              const sourceId = pendingSourceIds.pop();
              if (sourceId === undefined || rowsByThreadId.has(sourceId)) continue;
              const source = (yield* selectShellThreadRows(sourceId))[0];
              if (source === undefined) continue;
              rowsByThreadId.set(sourceId, source);
              if (source.forked_from_run_source_thread_id !== null) {
                pendingSourceIds.push(ThreadId.make(source.forked_from_run_source_thread_id));
              }
            }
            const threadRows = [...rowsByThreadId.values()];
            const threadIds = [...rowsByThreadId.keys()];
            const readForThreadIds = <A>(
              read: (ids: ReadonlyArray<ThreadId>) => Effect.Effect<ReadonlyArray<A>, unknown>,
            ) =>
              threadIds.length === 0 ? Effect.succeed([] as ReadonlyArray<A>) : read(threadIds);
            const [runRows, itemCountRows, sequenceRows, providerThreadRows, pendingTurnItemRows] =
              yield* Effect.all([
                readForThreadIds(selectShellRunRows),
                readForThreadIds(selectShellRunItemCounts),
                sql<{ readonly snapshot_sequence: number | null }>`
            SELECT MAX(sequence) AS snapshot_sequence
            FROM orchestration_events
            WHERE application_event_version = 2
              AND aggregate_kind = 'thread'
          `,
                readForThreadIds(selectShellProviderThreadRows),
                readForThreadIds(selectShellPendingTurnItemRows),
              ]);

            const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({
              runRows,
              itemCountRows,
            });
            const { providerThreadsByThreadId, pendingTurnItemsByThreadId } =
              yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows });
            const states = yield* Effect.forEach(threadRows, (row) =>
              shellThreadStateFromRow({
                row,
                runOrdinalsByThreadId,
                itemCountsByThreadId,
                providerThreadsByThreadId,
                pendingTurnItemsByThreadId,
              }),
            );
            const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));

            const shells = states
              .filter((state) => targetThreadIds.has(state.thread.id))
              .map((state) =>
                shellFromState({
                  state,
                  visibleItemCount: visibleItemCountForShell({
                    threadId: state.thread.id,
                    statesByThreadId,
                  }),
                }),
              );

            return {
              schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
              snapshotSequence: sequenceRows[0]?.snapshot_sequence ?? 0,
              threads: shells.filter((thread) => thread.archivedAt === null),
              archivedThreads: shells.filter((thread) => thread.archivedAt !== null),
            };
          }),
        )
        .pipe(
          Effect.mapError(
            (cause) =>
              new ProjectionStoreReadError({
                threadId: ThreadId.make("thread:shell"),
                cause,
              }),
          ),
        );

    // Per-thread shell for the live shell streams: reads only the target thread
    // plus its fork-source chain instead of materializing every thread. Returns
    // null when the thread is deleted or unknown.
    const getThreadShell: ProjectionStoreV2Shape["getThreadShell"] = (threadId) =>
      sql
        .withTransaction(
          Effect.gen(function* () {
            const rowsByThreadId = new Map<ThreadId, ShellThreadRow>();
            const pending: Array<ThreadId> = [threadId];
            while (pending.length > 0) {
              const nextId = pending.pop();
              if (nextId === undefined || rowsByThreadId.has(nextId)) {
                continue;
              }
              const rows = yield* selectShellThreadRows(nextId);
              const row = rows[0];
              if (row === undefined) {
                continue;
              }
              rowsByThreadId.set(nextId, row);
              if (row.forked_from_run_source_thread_id !== null) {
                pending.push(ThreadId.make(row.forked_from_run_source_thread_id));
              }
            }
            if (!rowsByThreadId.has(threadId)) {
              return null;
            }

            const threadIds = [...rowsByThreadId.keys()];
            const [runRows, itemCountRows, providerThreadRows, pendingTurnItemRows] =
              yield* Effect.all([
                selectShellRunRows(threadIds),
                selectShellRunItemCounts(threadIds),
                selectShellProviderThreadRows(threadIds),
                selectShellPendingTurnItemRows(threadIds),
              ]);
            const { runOrdinalsByThreadId, itemCountsByThreadId } = runMapsByThreadId({
              runRows,
              itemCountRows,
            });
            const { providerThreadsByThreadId, pendingTurnItemsByThreadId } =
              yield* pendingBackgroundDataByThreadId({ providerThreadRows, pendingTurnItemRows });

            const states = yield* Effect.forEach([...rowsByThreadId.values()], (row) =>
              shellThreadStateFromRow({
                row,
                runOrdinalsByThreadId,
                itemCountsByThreadId,
                providerThreadsByThreadId,
                pendingTurnItemsByThreadId,
              }),
            );
            const statesByThreadId = new Map(states.map((state) => [state.thread.id, state]));
            const state = statesByThreadId.get(threadId);
            if (state === undefined) {
              return null;
            }
            return shellFromState({
              state,
              visibleItemCount: visibleItemCountForShell({ threadId, statesByThreadId }),
            });
          }),
        )
        .pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })));

    return {
      apply,
      getShellSnapshot,
      getThreadShell,
      getThread,
      getSettlementCandidates,
      getThreadProjection,
      getCheckpointContext,
      getPendingNativeUserInputs,
      getRuntimeRequest,
      getProviderControlContext,
      getRecoveryThreadIds,
      getUnreadableThreadIds,
      getThreadSnapshot,
      getThreadSnapshotWindow,
    } satisfies ProjectionStoreV2Shape;
  }),
);

export const layerMemory: Layer.Layer<ProjectionStoreV2> = Layer.effect(
  ProjectionStoreV2,
  Effect.gen(function* () {
    const replayState = yield* Ref.make(makeProjectionReplayState());
    const sequence = yield* Ref.make(0);

    const service: ProjectionStoreV2Shape = {
      apply: (event) =>
        Effect.gen(function* () {
          const result = yield* Ref.modify(replayState, (existing) => {
            const next: ProjectionReplayState = {
              projections: new Map(existing.projections),
              providerSessionThreadIds: new Map(existing.providerSessionThreadIds),
            };
            if (!applyToProjectionReplayState(next, event)) {
              return [
                new ProjectionStoreThreadNotFoundError({ threadId: event.threadId }),
                existing,
              ] as const;
            }
            return [undefined, next] as const;
          });

          if (result) {
            return yield* result;
          }
          yield* Ref.update(sequence, (current) => current + 1);
        }),
      getShellSnapshot: (options) =>
        Effect.gen(function* () {
          const existing = (yield* Ref.get(replayState)).projections;
          const selectedThreadIds = [...existing.entries()]
            .filter(([, projection]) => {
              if (options?.location === "active") return projection.thread.archivedAt === null;
              if (options?.location === "archive") return projection.thread.archivedAt !== null;
              return true;
            })
            .map(([threadId]) => threadId);
          const shells = yield* Effect.forEach(
            selectedThreadIds.toSorted((left, right) => String(left).localeCompare(String(right))),
            (threadId) =>
              service.getThreadProjection(threadId).pipe(Effect.map(threadShellFromProjection)),
          );
          const visible = shells.filter((thread) => thread.deletedAt === null);
          return {
            schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
            snapshotSequence: yield* Ref.get(sequence),
            threads: visible.filter((thread) => thread.archivedAt === null),
            archivedThreads: visible.filter((thread) => thread.archivedAt !== null),
          };
        }),
      getThreadShell: (threadId) =>
        Effect.gen(function* () {
          const existing = (yield* Ref.get(replayState)).projections;
          if (!existing.has(threadId)) {
            return null;
          }
          const shell = yield* service
            .getThreadProjection(threadId)
            .pipe(Effect.map(threadShellFromProjection));
          return shell.deletedAt === null ? shell : null;
        }),
      getThread: (threadId) =>
        Effect.gen(function* () {
          const projection = (yield* Ref.get(replayState)).projections.get(threadId);
          if (projection === undefined) {
            return yield* new ProjectionStoreThreadNotFoundError({ threadId });
          }
          return projection.thread;
        }),
      getSettlementCandidates: () =>
        Effect.gen(function* () {
          const projections = (yield* Ref.get(replayState)).projections;
          return [...projections.values()]
            .filter(
              ({ thread, runs, runtimeRequests }) =>
                thread.deletedAt === null &&
                thread.archivedAt === null &&
                thread.settledOverride === null &&
                thread.pinnedAt == null &&
                !runs.some(isActivityRunForShell) &&
                !runtimeRequests.some((request) => request.status === "pending"),
            )
            .map(threadShellFromProjection)
            .toSorted(
              (left, right) =>
                DateTime.toEpochMillis(left.updatedAt) - DateTime.toEpochMillis(right.updatedAt) ||
                left.id.localeCompare(right.id),
            );
        }),
      getRecoveryThreadIds: (kind) =>
        Effect.gen(function* () {
          const projections = (yield* Ref.get(replayState)).projections;
          return [...projections.values()]
            .filter((projection) => needsRecovery(projection, kind))
            .toSorted(
              (left, right) =>
                DateTime.toEpochMillis(left.thread.updatedAt) -
                  DateTime.toEpochMillis(right.thread.updatedAt) ||
                left.thread.id.localeCompare(right.thread.id),
            )
            .map((projection) => projection.thread.id);
        }),
      getUnreadableThreadIds: () => Effect.succeed([]),
      getPendingNativeUserInputs: (threadId, providerTurnId) =>
        Effect.gen(function* () {
          const projection = (yield* Ref.get(replayState)).projections.get(threadId);
          if (projection === undefined) return { runtimeRequests: [], nodes: [], turnItems: [] };
          const runtimeRequests = projection.runtimeRequests.filter(
            (request) =>
              request.providerTurnId === providerTurnId &&
              request.kind === "user_input" &&
              request.status === "pending" &&
              request.responseCapability.type !== "message",
          );
          const requestIds = new Set(runtimeRequests.map((request) => request.id));
          const nodeIds = new Set(runtimeRequests.map((request) => request.nodeId));
          return {
            runtimeRequests,
            nodes: projection.nodes.filter((node) => nodeIds.has(node.id)),
            turnItems: projection.turnItems.filter(
              (item) => item.type === "user_input_request" && requestIds.has(item.requestId),
            ),
          };
        }),
      getRuntimeRequest: (threadId, requestId) =>
        Effect.gen(function* () {
          const projection = (yield* Ref.get(replayState)).projections.get(threadId);
          if (projection === undefined)
            return yield* new ProjectionStoreThreadNotFoundError({ threadId });
          return projection.runtimeRequests.find((request) => request.id === requestId);
        }),
      getProviderControlContext: (threadId, target) =>
        Effect.gen(function* () {
          const projection = (yield* Ref.get(replayState)).projections.get(threadId);
          if (projection === undefined)
            return yield* new ProjectionStoreThreadNotFoundError({ threadId });
          const providerTurn = projection.providerTurns.find(
            (turn) => turn.id === target.providerTurnId,
          );
          return {
            providerThread: projection.providerThreads.find(
              (thread) => thread.id === target.providerThreadId,
            ),
            providerTurn,
            attempt:
              target.attemptId === undefined
                ? undefined
                : projection.attempts.find((attempt) => attempt.id === target.attemptId),
            message:
              target.messageId === undefined
                ? undefined
                : projection.messages.find((message) => message.id === target.messageId),
            run:
              target.messageId === undefined || providerTurn === undefined
                ? undefined
                : projection.runs.find((run) => run.activeAttemptId === providerTurn.runAttemptId),
          };
        }),
      getCheckpointContext: (threadId) =>
        Effect.gen(function* () {
          const projection = (yield* Ref.get(replayState)).projections.get(threadId);
          if (projection === undefined) {
            return yield* new ProjectionStoreThreadNotFoundError({ threadId });
          }
          return {
            runs: projection.runs.map(({ id, ordinal, status }) => ({ id, ordinal, status })),
            checkpointScopes: projection.checkpointScopes.map(({ id, runId, kind, cwd }) => ({
              id,
              runId,
              kind,
              cwd,
            })),
            checkpoints: projection.checkpoints.map(
              ({ scopeId, runId, appRunOrdinal, status, ref }) => ({
                scopeId,
                runId,
                appRunOrdinal,
                status,
                ref,
              }),
            ),
          };
        }),
      getThreadProjection: (threadId) =>
        Effect.gen(function* () {
          const existing = (yield* Ref.get(replayState)).projections;
          const readProjection = (
            targetThreadId: ThreadId,
            seenThreadIds: ReadonlySet<ThreadId>,
          ): OrchestrationV2ThreadProjection | null => {
            const projection = existing.get(targetThreadId);
            if (!projection) {
              return null;
            }
            const forkedFrom = projection.thread.forkedFrom;
            if (forkedFrom?.type !== "run" || seenThreadIds.has(forkedFrom.threadId)) {
              return withLocalVisibleTurnItems(projection);
            }
            const sourceProjection = readProjection(
              forkedFrom.threadId,
              new Set([...seenThreadIds, targetThreadId]),
            );
            return {
              ...projection,
              visibleTurnItems: buildVisibleTurnItems({
                projection,
                sourceProjection,
              }),
            };
          };
          const projection = readProjection(threadId, new Set());
          if (!projection) {
            return yield* new ProjectionStoreThreadNotFoundError({ threadId });
          }
          return projection;
        }),
      getThreadSnapshot: (threadId) =>
        service.getThreadProjection(threadId).pipe(
          Effect.flatMap((projection) =>
            Ref.get(sequence).pipe(
              Effect.map((snapshotSequence) => ({
                schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
                snapshotSequence,
                projection,
              })),
            ),
          ),
        ),
      getThreadSnapshotWindow: (threadId, options) =>
        service.getThreadSnapshot(threadId).pipe(
          Effect.map((snapshot) => {
            const anchorIndex =
              options.anchorItemId === undefined
                ? snapshot.projection.visibleTurnItems.length
                : snapshot.projection.visibleTurnItems.findIndex(
                    (row) => row.sourceItemId === options.anchorItemId,
                  ) + 1;
            const visibleTurnItems = snapshot.projection.visibleTurnItems.slice(
              Math.max(0, anchorIndex - options.rowLimit),
              anchorIndex,
            );
            return {
              ...snapshot,
              projection: { ...snapshot.projection, visibleTurnItems },
            };
          }),
        ),
    };

    return service;
  }),
);