import { findMountedWindowStart } from "@/agent-stream/history-window";
import { z } from "zod";
import {
  AgentStatusSchema,
  AgentTimelineItemPayloadSchema,
  WorkspaceGitHubRuntimePayloadSchema,
} from "@getpaseo/protocol/messages";
import { AgentProviderSchema } from "@getpaseo/protocol/provider-manifest";
import type { PluginTimelineData } from "@getpaseo/plugin";
import {
  normalizeProjectDescriptor,
  normalizeWorkspaceDescriptor,
  type Agent,
  type AgentTimelineCursorState,
  type ProjectDescriptor,
  type WorkspaceDescriptor,
} from "@/stores/session-store";
import {
  isUnreconciledLocalUserMessage,
  type AgentToolCallData,
  type StreamItem,
} from "@/types/stream";
import { normalizeAgentSnapshot } from "@/utils/agent-snapshots";
import { clearLegacyReplicaCache } from "./legacy-cleanup";
import {
  REPLICA_SINGLETON_ROW_ID,
  type ReplicaRow,
  type ReplicaRowChanges,
  type ReplicaRowKey,
  type ReplicaRowKind,
  type ReplicaRowStore,
} from "./row-store";

export type DirectoryReplicaMutation =
  | { kind: "agent"; type: "upsert"; id: string; value: Agent }
  | { kind: "agent"; type: "delete"; id: string }
  | { kind: "workspace"; type: "upsert"; id: string; value: WorkspaceDescriptor }
  | { kind: "workspace"; type: "delete"; id: string }
  | { kind: "project"; type: "upsert"; id: string; value: ProjectDescriptor }
  | { kind: "project"; type: "delete"; id: string };

export interface CachedDirectory {
  agents: Map<string, Agent>;
  workspaces: Map<string, WorkspaceDescriptor>;
  projects: Map<string, ProjectDescriptor>;
  checkpoint?: DirectoryCheckpoint;
}

export interface CachedWorkspace {
  workspace: WorkspaceDescriptor;
  project?: ProjectDescriptor;
}

export interface DirectoryCursor {
  generation: string;
  afterSeq: number;
}

export interface DirectoryCheckpoint {
  projects?: DirectoryCursor;
  workspaces?: DirectoryCursor;
  agents?: DirectoryCursor;
}

export interface CachedTimeline {
  agentId: string;
  items: StreamItem[];
  range: AgentTimelineCursorState | null;
  hasOlder: boolean;
}

const PERSIST_DELAY_MS = 1_000;
const MAX_TIMELINE_ITEMS = 50;
const MAX_CACHE_BYTES = 32 * 1024 * 1024;
const IsoDateSchema = z.iso.datetime();
const TimelinePositionSchema = z.strictObject({
  epoch: z.string(),
  seq: z.number().int().nonnegative(),
});
const PluginTimelineDataSchema: z.ZodType<PluginTimelineData> = z.lazy(() =>
  z.union([
    z.null(),
    z.boolean(),
    z.number(),
    z.string(),
    z.array(PluginTimelineDataSchema),
    z.record(z.string(), PluginTimelineDataSchema),
  ]),
);

const TimelineItemBaseShape = {
  id: z.string(),
  timelineCursor: TimelinePositionSchema.optional(),
  // COMPAT(active-turn-membership): absent on caches written before turn membership.
  turnId: z.string().optional(),
  timestamp: IsoDateSchema,
};

const TodoEntrySchema = z.strictObject({
  text: z.string(),
  completed: z.boolean(),
  id: z.string().optional(),
  status: z.enum(["pending", "in_progress", "completed"]).optional(),
  activeForm: z.string().optional(),
});

const TaskActivitySchema = z.discriminatedUnion("type", [
  z.strictObject({ type: z.literal("created"), count: z.number().int().nonnegative() }),
  z.strictObject({
    type: z.enum(["added", "started", "completed"]),
    task: z.string(),
  }),
]);

const StoredTimelineItemSchema = z.discriminatedUnion("kind", [
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("user_message"),
    clientMessageId: z.string().optional(),
    messageId: z.string().optional(),
    text: z.string(),
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("assistant_message"),
    messageId: z.string().optional(),
    text: z.string(),
    // Reject old caches containing display fragments; refetch the complete source text.
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("thought"),
    text: z.string(),
    status: z.enum(["loading", "ready"]),
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("todo_list"),
    provider: AgentProviderSchema,
    items: z.array(TodoEntrySchema),
    activity: TaskActivitySchema,
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("notification"),
    sourceType: z.enum(["error", "notification"]),
    level: z.enum(["info", "warning", "error"]),
    message: z.string(),
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("compaction"),
    status: z.enum(["loading", "completed"]),
    trigger: z.enum(["auto", "manual"]).optional(),
    preTokens: z.number().nonnegative().optional(),
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("tool_call"),
    provider: AgentProviderSchema,
    item: AgentTimelineItemPayloadSchema.refine((item) => item.type === "tool_call"),
  }),
  z.strictObject({
    ...TimelineItemBaseShape,
    kind: z.literal("plugin"),
    pluginId: z.string(),
    pluginItemId: z.string(),
    itemKind: z.string(),
    version: z.number().int().positive(),
    data: PluginTimelineDataSchema,
  }),
]);

const AgentCapabilitiesSchema = z.strictObject({
  supportsStreaming: z.boolean(),
  supportsSessionPersistence: z.boolean(),
  supportsSessionListing: z.boolean().optional(),
  supportsDynamicModes: z.boolean(),
  supportsMcpServers: z.boolean(),
  supportsReasoningStream: z.boolean(),
  supportsToolInvocations: z.boolean(),
  supportsRewindConversation: z.boolean().optional(),
  supportsRewindFiles: z.boolean().optional(),
  supportsRewindBoth: z.boolean().optional(),
});

const StoredProjectCheckoutSchema = z.union([
  z.strictObject({
    cwd: z.string(),
    isGit: z.literal(false),
    currentBranch: z.null(),
    remoteUrl: z.null(),
    worktreeRoot: z.null(),
    isPaseoOwnedWorktree: z.literal(false),
    mainRepoRoot: z.null(),
  }),
  z.strictObject({
    cwd: z.string(),
    isGit: z.literal(true),
    currentBranch: z.string().nullable(),
    remoteUrl: z.string().nullable(),
    worktreeRoot: z.string(),
    isPaseoOwnedWorktree: z.literal(false),
    mainRepoRoot: z.string().nullable(),
  }),
  z.strictObject({
    cwd: z.string(),
    isGit: z.literal(true),
    currentBranch: z.string().nullable(),
    remoteUrl: z.string().nullable(),
    worktreeRoot: z.string(),
    isPaseoOwnedWorktree: z.literal(true),
    mainRepoRoot: z.string(),
  }),
]);

const StoredProjectPlacementSchema = z.strictObject({
  projectKey: z.string(),
  projectName: z.string(),
  workspaceName: z.string().nullable().optional(),
  checkout: StoredProjectCheckoutSchema,
});

const StoredAgentSnapshotSchema = z.strictObject({
  id: z.string(),
  provider: AgentProviderSchema,
  cwd: z.string(),
  workspaceId: z.string().optional(),
  model: z.string().nullable(),
  thinkingOptionId: z.string().nullable().optional(),
  createdAt: IsoDateSchema,
  updatedAt: IsoDateSchema,
  lastUserMessageAt: IsoDateSchema.nullable(),
  status: AgentStatusSchema,
  activeTurn: z
    .strictObject({
      turnId: z.string(),
      startedAt: IsoDateSchema.nullable(),
    })
    .nullable()
    .optional(),
  capabilities: AgentCapabilitiesSchema,
  currentModeId: z.string().nullable(),
  availableModes: z.array(z.never()).max(0),
  pendingPermissions: z.array(z.never()).max(0),
  persistence: z.null(),
  lastError: z.string().optional(),
  title: z.string().nullable(),
  labels: z.record(z.string(), z.string()),
  // Optional: clients without background_agents never received a background agent to cache.
  background: z.boolean().optional(),
  requiresAttention: z.boolean().optional(),
  attentionReason: z.enum(["finished", "error", "permission"]).nullable().optional(),
  attentionTimestamp: IsoDateSchema.nullable().optional(),
  archivedAt: IsoDateSchema.nullable().optional(),
});

const StoredAgentSchema = z.strictObject({
  snapshot: StoredAgentSnapshotSchema,
  turn: z
    .discriminatedUnion("phase", [
      z.strictObject({ phase: z.literal("idle") }),
      z.strictObject({
        phase: z.literal("open"),
        turnId: z.string().nullable(),
        startedAt: IsoDateSchema.nullable(),
      }),
    ])
    .optional(),
  projectPlacement: StoredProjectPlacementSchema.nullable(),
  lastActivityAt: IsoDateSchema,
});

const WorkspaceScriptSchema = z.strictObject({
  scriptName: z.string(),
  type: z.enum(["script", "service"]),
  hostname: z.string(),
  port: z.number().int().positive().nullable(),
  localProxyUrl: z.string().nullable().optional(),
  publicProxyUrl: z.string().nullable().optional(),
  proxyUrl: z.string().nullable(),
  lifecycle: z.enum(["running", "stopped"]),
  health: z.enum(["healthy", "unhealthy"]).nullable(),
  exitCode: z.number().nullable(),
  terminalId: z.string().nullable(),
});

const WorkspaceGitRuntimeSchema = z
  .strictObject({
    currentBranch: z.string().nullable().optional(),
    remoteUrl: z.string().nullable().optional(),
    isPaseoOwnedWorktree: z.boolean().optional(),
    isDirty: z.boolean().nullable().optional(),
    aheadBehind: z.strictObject({ ahead: z.number(), behind: z.number() }).nullable().optional(),
    aheadOfOrigin: z.number().nullable().optional(),
    behindOfOrigin: z.number().nullable().optional(),
  })
  .nullable()
  .optional();

const StoredWorkspaceSchema = z.strictObject({
  // Require both keys: a cursor without these values would preserve the wrong sort on reconnect.
  createdAt: IsoDateSchema.nullable(),
  lastUserMessageAt: IsoDateSchema.nullable(),
  id: z.string(),
  projectId: z.string(),
  projectDisplayName: z.string(),
  projectCustomName: z.string().nullable(),
  projectCustomIconRevision: z.string().nullable(),
  projectRootPath: z.string(),
  workspaceDirectory: z.string(),
  worktreeSlug: z.string().optional(),
  projectKind: z.enum(["git", "non_git", "directory"]),
  workspaceKind: z.enum(["directory", "local_checkout", "checkout", "worktree"]),
  // Require the flag so older rows that lost background visibility invalidate the workspace cursor.
  background: z.boolean(),
  name: z.string(),
  title: z.string().nullable(),
  pinnedAt: z.string().nullable(),
  // Optional because entries written before labels existed have none. A cached workspace that
  // dropped them painted its row without its chips and stayed that way: the directory cursor is
  // current on reconnect, so the daemon has nothing newer to send back.
  labels: z.array(z.string()).optional(),
  status: z.enum(["needs_input", "failed", "running", "attention", "done"]),
  statusEnteredAt: IsoDateSchema.nullable(),
  activityAt: z.null(),
  archivingAt: z.string().nullable(),
  diffStat: z.strictObject({ additions: z.number(), deletions: z.number() }).nullable(),
  scripts: z.array(WorkspaceScriptSchema),
  gitRuntime: WorkspaceGitRuntimeSchema,
  githubRuntime: WorkspaceGitHubRuntimePayloadSchema,
  forge: z.string().optional(),
});

const StoredProjectSchema = z.strictObject({
  projectId: z.string(),
  projectKey: z.string().optional(),
  projectDisplayName: z.string(),
  projectCustomName: z.string().nullable(),
  projectCustomIconRevision: z.string().nullable(),
  projectIconRevision: z.string().optional(),
  projectRootPath: z.string(),
  projectKind: z.enum(["git", "non_git", "directory"]),
});

const StoredTimelineSchema = z.strictObject({
  agentId: z.string(),
  items: z.array(StoredTimelineItemSchema),
  range: z
    .strictObject({
      epoch: z.string(),
      startSeq: z.number().int().nonnegative(),
      endSeq: z.number().int().nonnegative(),
    })
    .nullable(),
  hasOlder: z.boolean(),
});

type StoredAgent = z.infer<typeof StoredAgentSchema>;
type StoredTimeline = z.infer<typeof StoredTimelineSchema>;
type StoredTimelineItem = z.infer<typeof StoredTimelineItemSchema>;
type StoredToolCall = Extract<StoredTimelineItem, { kind: "tool_call" }>["item"];
type StoredWorkspace = z.infer<typeof StoredWorkspaceSchema>;
type StoredProject = z.infer<typeof StoredProjectSchema>;

interface ReplicaCacheOptions {
  maxBytes?: number;
  clearLegacyCache?: () => Promise<void>;
}

type StructuredReplicaUpsert =
  | { serverId: string; kind: "agent"; id: string; value: Agent }
  | { serverId: string; kind: "workspace"; id: string; value: WorkspaceDescriptor }
  | { serverId: string; kind: "project"; id: string; value: ProjectDescriptor }
  | { serverId: string; kind: "timeline"; id: string; value: CachedTimeline }
  | { serverId: string; kind: "checkpoint"; id: string; value: DirectoryCheckpoint };

const DirectoryCursorSchema = z.strictObject({
  generation: z.string(),
  afterSeq: z.number().int().nonnegative(),
});

const DirectoryCheckpointSchema = z.strictObject({
  projects: DirectoryCursorSchema.optional(),
  workspaces: DirectoryCursorSchema.optional(),
  agents: DirectoryCursorSchema.optional(),
});

// Old checkpoints could describe a baseline that an overlapping refresh had discarded.
// Revalidate directory contents once without evicting cached rows or timelines.
const StoredDirectoryCheckpointSchema = z.strictObject({
  version: z.literal(1),
  cursors: DirectoryCheckpointSchema,
});

function serializeDirectoryCheckpoint(cursors: DirectoryCheckpoint) {
  return { version: 1 as const, cursors };
}

function deserializeDirectoryCheckpoint(payload: string): DirectoryCheckpoint {
  return parseStoredPayload(StoredDirectoryCheckpointSchema, payload).cursors;
}

function deserializeTimeline(stored: StoredTimeline | null): CachedTimeline | null {
  if (!stored) {
    return null;
  }
  return {
    agentId: stored.agentId,
    items: stored.items.map(deserializeTimelineItem),
    range: stored.range,
    hasOlder: stored.hasOlder,
  };
}

function timelineBase(item: StreamItem) {
  return {
    id: item.id,
    ...(item.timelineCursor ? { timelineCursor: item.timelineCursor } : {}),
    ...(item.turnId ? { turnId: item.turnId } : {}),
    timestamp: item.timestamp.toISOString(),
  };
}

function serializeAgentToolCall(data: AgentToolCallData): StoredToolCall {
  const base = {
    type: "tool_call" as const,
    callId: data.callId,
    name: data.name,
    detail: data.detail,
    ...(data.metadata ? { metadata: data.metadata } : {}),
    ...(data.agentMessage ? { agentMessage: data.agentMessage } : {}),
  };
  switch (data.status) {
    case "running":
    case "completed":
    case "canceled":
      return { ...base, status: data.status, error: null };
    case "failed":
      return { ...base, status: data.status, error: data.error };
  }
}

function serializeTimelineItem(item: StreamItem): StoredTimelineItem | null {
  const base = timelineBase(item);
  switch (item.kind) {
    case "user_message":
      return {
        ...base,
        kind: item.kind,
        ...(item.clientMessageId ? { clientMessageId: item.clientMessageId } : {}),
        ...(item.messageId ? { messageId: item.messageId } : {}),
        text: item.text,
      };
    case "assistant_message":
      return {
        ...base,
        kind: item.kind,
        ...(item.messageId ? { messageId: item.messageId } : {}),
        text: item.text,
      };
    case "thought":
      return { ...base, kind: item.kind, text: item.text, status: item.status };
    case "todo_list":
      return {
        ...base,
        kind: item.kind,
        provider: item.provider,
        items: item.items,
        activity: item.activity,
      };
    case "notification":
      return {
        ...base,
        kind: item.kind,
        sourceType: item.sourceType,
        level: item.level,
        message: item.message,
      };
    case "compaction":
      return {
        ...base,
        kind: item.kind,
        status: item.status,
        ...(item.trigger ? { trigger: item.trigger } : {}),
        ...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
      };
    case "tool_call":
      if (item.payload.source !== "agent") return null;
      return {
        ...base,
        kind: item.kind,
        provider: item.payload.data.provider,
        item: serializeAgentToolCall(item.payload.data),
      };
    case "plugin":
      return {
        ...base,
        kind: item.kind,
        pluginId: item.pluginId,
        pluginItemId: item.pluginItemId,
        itemKind: item.itemKind,
        version: item.version,
        data: item.data,
      };
  }
}

function deserializeTimelineItem(item: StoredTimelineItem): StreamItem {
  if (item.kind === "plugin") {
    return {
      id: item.id,
      ...(item.timelineCursor ? { timelineCursor: item.timelineCursor } : {}),
      ...(item.turnId ? { turnId: item.turnId } : {}),
      timestamp: new Date(item.timestamp),
      kind: item.kind,
      pluginId: item.pluginId,
      pluginItemId: item.pluginItemId,
      itemKind: item.itemKind,
      version: item.version,
      data: item.data,
    };
  }
  return deserializeBuiltinTimelineItem(item);
}

function deserializeBuiltinTimelineItem(
  item: Exclude<StoredTimelineItem, { kind: "plugin" }>,
): StreamItem {
  const base = {
    id: item.id,
    ...(item.timelineCursor ? { timelineCursor: item.timelineCursor } : {}),
    ...(item.turnId ? { turnId: item.turnId } : {}),
    timestamp: new Date(item.timestamp),
  };
  switch (item.kind) {
    case "user_message":
      return {
        ...base,
        kind: item.kind,
        ...(item.clientMessageId ? { clientMessageId: item.clientMessageId } : {}),
        ...(item.messageId ? { messageId: item.messageId } : {}),
        text: item.text,
      };
    case "assistant_message":
      return {
        ...base,
        kind: item.kind,
        ...(item.messageId ? { messageId: item.messageId } : {}),
        text: item.text,
      };
    case "thought":
      return { ...base, kind: item.kind, text: item.text, status: item.status };
    case "todo_list":
      return {
        ...base,
        kind: item.kind,
        provider: item.provider,
        items: item.items,
        activity: item.activity,
      };
    case "notification":
      return {
        ...base,
        kind: item.kind,
        sourceType: item.sourceType,
        level: item.level,
        message: item.message,
      };
    case "compaction":
      return {
        ...base,
        kind: item.kind,
        status: item.status,
        ...(item.trigger ? { trigger: item.trigger } : {}),
        ...(item.preTokens !== undefined ? { preTokens: item.preTokens } : {}),
      };
    case "tool_call": {
      const tool = item.item;
      if (tool.type !== "tool_call") {
        throw new Error("Stored tool call contains a non-tool timeline item");
      }
      return {
        ...base,
        kind: item.kind,
        payload: {
          source: "agent",
          data: {
            provider: item.provider,
            callId: tool.callId,
            name: tool.name,
            status: tool.status,
            error: tool.error,
            detail: tool.detail,
            ...(tool.metadata ? { metadata: tool.metadata } : {}),
            ...(tool.agentMessage ? { agentMessage: tool.agentMessage } : {}),
          },
        },
      };
    }
  }
}

function serializeProjectPlacement(agent: Agent): StoredAgent["projectPlacement"] {
  return agent.projectPlacement ?? null;
}

function serializeAgentTurn(agent: Agent): NonNullable<StoredAgent["turn"]> {
  if (agent.turn.phase === "idle") return { phase: "idle" };
  return {
    phase: "open",
    turnId: agent.turn.turnId,
    startedAt: agent.turn.startedAt?.toISOString() ?? null,
  };
}

function serializeAgentCapabilities(
  capabilities: Agent["capabilities"],
): StoredAgent["snapshot"]["capabilities"] {
  return {
    supportsStreaming: capabilities.supportsStreaming,
    supportsSessionPersistence: capabilities.supportsSessionPersistence,
    ...(capabilities.supportsSessionListing !== undefined
      ? { supportsSessionListing: capabilities.supportsSessionListing }
      : {}),
    supportsDynamicModes: capabilities.supportsDynamicModes,
    supportsMcpServers: capabilities.supportsMcpServers,
    supportsReasoningStream: capabilities.supportsReasoningStream,
    supportsToolInvocations: capabilities.supportsToolInvocations,
    ...(capabilities.supportsRewindConversation !== undefined
      ? { supportsRewindConversation: capabilities.supportsRewindConversation }
      : {}),
    ...(capabilities.supportsRewindFiles !== undefined
      ? { supportsRewindFiles: capabilities.supportsRewindFiles }
      : {}),
    ...(capabilities.supportsRewindBoth !== undefined
      ? { supportsRewindBoth: capabilities.supportsRewindBoth }
      : {}),
  };
}

function serializeAgent(agent: Agent): StoredAgent {
  const snapshot = {
    id: agent.id,
    provider: agent.provider,
    cwd: agent.cwd,
    ...(agent.workspaceId ? { workspaceId: agent.workspaceId } : {}),
    model: agent.model,
    thinkingOptionId: agent.thinkingOptionId ?? null,
    createdAt: agent.createdAt.toISOString(),
    updatedAt: agent.updatedAt.toISOString(),
    lastUserMessageAt: agent.lastUserMessageAt?.toISOString() ?? null,
    status: agent.status,
    ...(agent.turn.phase === "open" && agent.turn.turnId
      ? {
          activeTurn: {
            turnId: agent.turn.turnId,
            startedAt: agent.turn.startedAt?.toISOString() ?? null,
          },
        }
      : {}),
    capabilities: serializeAgentCapabilities(agent.capabilities),
    currentModeId: agent.currentModeId,
    availableModes: [],
    pendingPermissions: [],
    persistence: null,
    ...(agent.lastError ? { lastError: agent.lastError } : {}),
    title: agent.title,
    labels: agent.labels,
    ...(agent.background ? { background: true } : {}),
    requiresAttention: agent.requiresAttention ?? false,
    attentionReason: agent.attentionReason ?? null,
    attentionTimestamp: agent.attentionTimestamp?.toISOString() ?? null,
    archivedAt: agent.archivedAt?.toISOString() ?? null,
  };
  return {
    snapshot,
    turn: serializeAgentTurn(agent),
    projectPlacement: serializeProjectPlacement(agent),
    lastActivityAt: agent.lastActivityAt.toISOString(),
  };
}

function deserializeAgent(serverId: string, stored: StoredAgent): Agent {
  const normalized = normalizeAgentSnapshot(stored.snapshot, serverId);
  let turn = normalized.turn;
  if (stored.turn?.phase === "idle") turn = { phase: "idle", cancellationRequestId: null };
  if (stored.turn?.phase === "open") {
    turn = {
      phase: "open",
      turnId: stored.turn.turnId,
      startedAt: stored.turn.startedAt ? new Date(stored.turn.startedAt) : null,
      cancellationRequestId: null,
    };
  }
  return {
    ...normalized,
    turn,
    lastActivityAt: new Date(stored.lastActivityAt),
    projectPlacement: stored.projectPlacement,
  };
}

function restoreWorkspace(stored: StoredWorkspace): WorkspaceDescriptor {
  return normalizeWorkspaceDescriptor({ ...stored, createdAt: stored.createdAt ?? undefined });
}

function serializeWorkspace(workspace: WorkspaceDescriptor): StoredWorkspace {
  return {
    createdAt: workspace.createdAt?.toISOString() ?? null,
    lastUserMessageAt: workspace.lastUserMessageAt?.toISOString() ?? null,
    id: workspace.id,
    projectId: workspace.projectId,
    projectDisplayName: workspace.projectDisplayName,
    projectCustomName: workspace.projectCustomName ?? null,
    projectCustomIconRevision: workspace.projectCustomIconRevision ?? null,
    projectRootPath: workspace.projectRootPath,
    workspaceDirectory: workspace.workspaceDirectory,
    worktreeSlug: workspace.worktreeSlug,
    projectKind: workspace.projectKind,
    workspaceKind: workspace.workspaceKind,
    background: workspace.background ?? false,
    name: workspace.name,
    title: workspace.title ?? null,
    pinnedAt: workspace.pinnedAt ?? null,
    labels: workspace.labels,
    status: workspace.status,
    statusEnteredAt: workspace.statusEnteredAt?.toISOString() ?? null,
    activityAt: null,
    archivingAt: workspace.archivingAt,
    diffStat: workspace.diffStat,
    scripts: workspace.scripts.map((script) => ({
      scriptName: script.scriptName,
      type: script.type,
      hostname: script.hostname,
      port: script.port,
      ...(script.localProxyUrl !== undefined ? { localProxyUrl: script.localProxyUrl } : {}),
      ...(script.publicProxyUrl !== undefined ? { publicProxyUrl: script.publicProxyUrl } : {}),
      proxyUrl: script.proxyUrl,
      lifecycle: script.lifecycle,
      health: script.health,
      exitCode: script.exitCode,
      terminalId: script.terminalId,
    })),
    gitRuntime: workspace.gitRuntime,
    githubRuntime: workspace.githubRuntime,
    forge: workspace.forge,
  };
}

function serializeProject(project: ProjectDescriptor): StoredProject {
  return {
    projectId: project.projectId,
    ...(project.projectKey ? { projectKey: project.projectKey } : {}),
    projectDisplayName: project.projectDisplayName,
    projectCustomName: project.projectCustomName,
    projectCustomIconRevision: project.projectCustomIconRevision ?? null,
    projectIconRevision: project.projectIconRevision,
    projectRootPath: project.projectRootPath,
    projectKind: project.projectKind,
  };
}

function isTimelineItemStoredLosslessly(item: StreamItem): boolean {
  switch (item.kind) {
    case "user_message":
      return (item.images?.length ?? 0) === 0 && (item.attachments?.length ?? 0) === 0;
    case "tool_call":
      return item.payload.source === "agent";
    default:
      return true;
  }
}

function serializeTimeline(timeline: CachedTimeline): StoredTimeline | null {
  const canonicalItems = timeline.items.filter(
    (item) => item.kind !== "user_message" || !isUnreconciledLocalUserMessage(item),
  );
  const start = findMountedWindowStart({
    items: canonicalItems,
    minMountedCount: MAX_TIMELINE_ITEMS,
  });
  const items = canonicalItems
    .slice(start)
    .map(serializeTimelineItem)
    .filter((item) => item !== null);
  const range = timeline.range;
  const canPersistCoverage =
    range !== null &&
    range.retainedRanges === undefined &&
    canonicalItems.length <= MAX_TIMELINE_ITEMS &&
    items.length === canonicalItems.length &&
    canonicalItems.every(
      (item) =>
        isTimelineItemStoredLosslessly(item) &&
        item.timelineCursor?.epoch === range.epoch &&
        item.timelineCursor.seq >= range.startSeq &&
        item.timelineCursor.seq <= range.endSeq,
    ) &&
    canonicalItems.some((item) => item.timelineCursor?.seq === range.endSeq);
  return {
    agentId: timeline.agentId,
    items,
    range: canPersistCoverage
      ? { epoch: range.epoch, startSeq: range.startSeq, endSeq: range.endSeq }
      : null,
    hasOlder: canPersistCoverage ? timeline.hasOlder : false,
  };
}

function rowKey(key: Pick<ReplicaRowKey, "kind" | "id">): string {
  return `${key.kind}\u0000${key.id}`;
}

function pendingRowKey(key: ReplicaRowKey): string {
  return `${key.serverId}\u0000${rowKey(key)}`;
}

// Budget accounting runs over every stored row of a touched host on each persist. Rows are
// immutable once stored, so their size is computed once. The count itself avoids the JS Buffer
// polyfill, which materialises the whole byte array just to measure it.
const rowBytesCache = new WeakMap<ReplicaRow, number>();

function utf8ByteLength(text: string): number {
  let bytes = 0;
  for (let index = 0; index < text.length; index += 1) {
    const code = text.charCodeAt(index);
    if (code < 0x80) bytes += 1;
    else if (code < 0x800) bytes += 2;
    else if (code >= 0xd800 && code <= 0xdbff) {
      bytes += 4;
      index += 1;
    } else bytes += 3;
  }
  return bytes;
}

function rowBytes(row: ReplicaRow): number {
  let bytes = rowBytesCache.get(row);
  if (bytes === undefined) {
    bytes = utf8ByteLength(row.payload);
    rowBytesCache.set(row, bytes);
  }
  return bytes;
}

function parseJsonPayload(payload: string): unknown {
  return JSON.parse(payload);
}

function parseStoredPayload<Value>(schema: z.ZodType<Value>, payload: string): Value {
  const parsed = schema.safeParse(parseJsonPayload(payload));
  if (!parsed.success) throw new Error("Invalid replica row payload");
  return parsed.data;
}

interface DirectoryReadAccumulator {
  agents: Map<string, Agent>;
  workspaces: Map<string, WorkspaceDescriptor>;
  projects: Map<string, ProjectDescriptor>;
  checkpoint?: DirectoryCheckpoint;
}

interface PendingReplicaChanges {
  upserts: StructuredReplicaUpsert[];
  deletes: ReplicaRowKey[];
  baselines: Set<string>;
}

function applyDirectoryRow(
  serverId: string,
  row: ReplicaRow,
  result: DirectoryReadAccumulator,
): void {
  switch (row.kind) {
    case "agent": {
      const stored = parseStoredPayload(StoredAgentSchema, row.payload);
      if (stored.snapshot.id !== row.id) throw new Error("Replica agent row id mismatch");
      result.agents.set(row.id, deserializeAgent(serverId, stored));
      return;
    }
    case "workspace": {
      const stored = parseStoredPayload(StoredWorkspaceSchema, row.payload);
      if (stored.id !== row.id) throw new Error("Replica workspace row id mismatch");
      result.workspaces.set(row.id, restoreWorkspace(stored));
      return;
    }
    case "project": {
      const stored = parseStoredPayload(StoredProjectSchema, row.payload);
      if (stored.projectId !== row.id) throw new Error("Replica project row id mismatch");
      result.projects.set(row.id, normalizeProjectDescriptor(stored));
      return;
    }
    case "checkpoint":
      if (row.id !== REPLICA_SINGLETON_ROW_ID) {
        throw new Error("Replica checkpoint row id mismatch");
      }
      result.checkpoint = deserializeDirectoryCheckpoint(row.payload);
      return;
    default:
      return;
  }
}

function directoryEntityForRow(row: ReplicaRow): keyof DirectoryCheckpoint | undefined {
  if (row.kind === "agent") return "agents";
  if (row.kind === "workspace") return "workspaces";
  if (row.kind === "project") return "projects";
  return undefined;
}

interface ReplicaRead {
  serverId: string;
  kinds: readonly ReplicaRowKind[];
  ids?: readonly string[];
  invalidated: boolean;
}

function matchesReplicaRead(read: ReplicaRead, row: ReplicaRowKey): boolean {
  return (
    read.serverId === row.serverId &&
    read.kinds.includes(row.kind) &&
    (!read.ids || read.ids.includes(row.id))
  );
}

export class ReplicaCache {
  private readonly activeServerIds = new Set<string>();
  private readonly activeReads = new Set<ReplicaRead>();
  private readonly storedRows = new Map<string, Map<string, ReplicaRow>>();
  private readonly hostBytes = new Map<string, number>();
  private readonly hostWriteOrder = new Map<string, true>();
  private pendingUpserts = new Map<string, StructuredReplicaUpsert>();
  private pendingDeletes = new Map<string, ReplicaRowKey>();
  private pendingBaselines = new Set<string>();
  private readonly invalidatedHosts = new Set<string>();
  private readonly maxBytes: number;
  private totalBytes = 0;
  private persistTimer: ReturnType<typeof setTimeout> | null = null;
  private writeQueue: Promise<void> = Promise.resolve();
  private preparePromise: Promise<void> | null = null;
  private storedIndexPromise: Promise<void> | null = null;

  constructor(
    private readonly rowStore: ReplicaRowStore,
    options: ReplicaCacheOptions = {},
  ) {
    this.maxBytes = Math.max(options.maxBytes ?? MAX_CACHE_BYTES, 0);
    this.clearLegacyCache = options.clearLegacyCache ?? clearLegacyReplicaCache;
  }

  private readonly clearLegacyCache: () => Promise<void>;

  async readAgent(serverId: string, agentId: string): Promise<Agent | undefined> {
    const rows = await this.readRows(serverId, ["agent"], [agentId]);
    const row = rows[0];
    if (!row) return undefined;
    try {
      const stored = parseStoredPayload(StoredAgentSchema, row.payload);
      if (stored.snapshot.id !== row.id) throw new Error("Replica agent row id mismatch");
      return deserializeAgent(serverId, stored);
    } catch {
      await this.deleteInvalidRow(row);
      return undefined;
    }
  }

  async readWorkspace(serverId: string, workspaceId: string): Promise<CachedWorkspace | undefined> {
    const workspaceRow = (await this.readRows(serverId, ["workspace"], [workspaceId]))[0];
    if (!workspaceRow) return undefined;
    let workspace: WorkspaceDescriptor;
    try {
      const stored = parseStoredPayload(StoredWorkspaceSchema, workspaceRow.payload);
      if (stored.id !== workspaceRow.id) throw new Error("Replica workspace row id mismatch");
      workspace = restoreWorkspace(stored);
    } catch {
      await this.deleteInvalidRow(workspaceRow);
      return undefined;
    }

    let project: ProjectDescriptor | undefined;
    const projectRow = (await this.readRows(serverId, ["project"], [workspace.projectId]))[0];
    if (projectRow) {
      try {
        const stored = parseStoredPayload(StoredProjectSchema, projectRow.payload);
        if (stored.projectId !== projectRow.id) throw new Error("Replica project row id mismatch");
        project = normalizeProjectDescriptor(stored);
      } catch {
        await this.deleteInvalidRow(projectRow);
      }
    }

    return { workspace, ...(project ? { project } : {}) };
  }

  async readDirectory(serverId: string): Promise<CachedDirectory> {
    const rows = await this.readRows(serverId, ["agent", "workspace", "project", "checkpoint"]);
    const result: DirectoryReadAccumulator = {
      agents: new Map(),
      workspaces: new Map(),
      projects: new Map(),
    };
    const invalidEntities = new Set<keyof DirectoryCheckpoint>();
    const invalidRows: ReplicaRow[] = [];
    for (const row of rows) {
      try {
        applyDirectoryRow(serverId, row, result);
      } catch {
        const invalidEntity = directoryEntityForRow(row);
        if (invalidEntity) invalidEntities.add(invalidEntity);
        invalidRows.push(row);
      }
    }
    if (result.checkpoint && invalidEntities.size > 0) {
      result.checkpoint = { ...result.checkpoint };
      for (const entity of invalidEntities) delete result.checkpoint[entity];
    }
    if (invalidRows.length > 0) {
      await this.repairInvalidDirectoryRows(serverId, invalidRows, result.checkpoint);
    }
    return result;
  }

  async readTimeline(serverId: string, agentId: string): Promise<CachedTimeline | undefined> {
    const rows = await this.readRows(serverId, ["timeline"], [agentId]);
    const row = rows[0];
    if (!row) return undefined;
    try {
      const stored = parseStoredPayload(StoredTimelineSchema, row.payload);
      if (stored.agentId !== agentId) return undefined;
      return deserializeTimeline(stored) ?? undefined;
    } catch {
      await this.deleteInvalidRow(row);
      return undefined;
    }
  }

  private async readRows(
    serverId: string,
    kinds: readonly ReplicaRowKind[],
    ids?: readonly string[],
  ): Promise<ReplicaRow[]> {
    if (!this.activeServerIds.has(serverId)) return [];
    try {
      await this.prepareStore();
      while (this.activeServerIds.has(serverId)) {
        // A read may only answer from rows the store already holds, so it waits for this host's
        // accepted commits to land. A store that rejects them will keep rejecting them: fail closed
        // rather than re-attempt the write on every pass. A write rejected for another host leaves
        // this host's stored rows readable.
        const read: ReplicaRead = { serverId, kinds, ids, invalidated: false };
        this.activeReads.add(read);
        try {
          if (!(await this.syncPending()) && this.hasPendingHostChanges(serverId)) return [];
          if (read.invalidated || this.hasPendingReadChanges(read)) continue;
          const rows = await this.rowStore.read(serverId, kinds, ids);
          if (
            this.activeServerIds.has(serverId) &&
            !read.invalidated &&
            !this.hasPendingReadChanges(read)
          ) {
            return rows;
          }
        } finally {
          this.activeReads.delete(read);
        }
      }
      return [];
    } catch {
      return [];
    }
  }

  private async deleteInvalidRow(row: ReplicaRow): Promise<void> {
    const invalidEntity = directoryEntityForRow(row);
    if (!invalidEntity) {
      const changes = { upserts: [], deletes: [row] } satisfies ReplicaRowChanges;
      await this.queueOperation(async () => {
        await this.rowStore.apply(changes);
        this.applyStoredChanges(changes);
      });
      return;
    }

    const checkpointRow = (
      await this.readRows(row.serverId, ["checkpoint"], [REPLICA_SINGLETON_ROW_ID])
    )[0];
    let checkpoint: DirectoryCheckpoint | undefined;
    if (checkpointRow) {
      try {
        checkpoint = deserializeDirectoryCheckpoint(checkpointRow.payload);
        checkpoint = { ...checkpoint };
        delete checkpoint[invalidEntity];
      } catch {
        checkpoint = undefined;
      }
    }
    const changes: ReplicaRowChanges = {
      deletes: [row, ...(checkpointRow && !checkpoint ? [checkpointRow] : [])],
      upserts:
        checkpointRow && checkpoint
          ? [
              {
                serverId: row.serverId,
                kind: "checkpoint",
                id: REPLICA_SINGLETON_ROW_ID,
                payload: JSON.stringify(serializeDirectoryCheckpoint(checkpoint)),
              },
            ]
          : [],
    };
    await this.queueOperation(async () => {
      await this.rowStore.apply(changes);
      this.applyStoredChanges(changes);
    });
  }

  private async repairInvalidDirectoryRows(
    serverId: string,
    invalidRows: ReplicaRow[],
    checkpoint: DirectoryCheckpoint | undefined,
  ): Promise<void> {
    const checkpointWasInvalid = invalidRows.some((row) => row.kind === "checkpoint");
    const changes: ReplicaRowChanges = {
      deletes: invalidRows,
      upserts:
        checkpoint !== undefined && !checkpointWasInvalid
          ? [
              {
                serverId,
                kind: "checkpoint",
                id: REPLICA_SINGLETON_ROW_ID,
                payload: JSON.stringify(serializeDirectoryCheckpoint(checkpoint)),
              },
            ]
          : [],
    };
    await this.queueOperation(async () => {
      await this.rowStore.apply(changes);
      this.applyStoredChanges(changes);
    });
  }

  commitDirectoryMutations(
    serverId: string,
    mutations: readonly DirectoryReplicaMutation[],
    checkpoint?: DirectoryCheckpoint,
  ): void {
    if (!this.activeServerIds.has(serverId)) return;
    if (this.invalidatedHosts.has(serverId)) return;
    for (const mutation of mutations) {
      if (mutation.type === "delete") {
        this.queueEntityDelete(serverId, mutation.kind, mutation.id);
      } else {
        this.queueUpsert({
          serverId,
          kind: mutation.kind,
          id: mutation.id,
          value: mutation.value,
        } as StructuredReplicaUpsert);
      }
    }
    if (checkpoint) {
      this.queueUpsert({
        serverId,
        kind: "checkpoint",
        id: REPLICA_SINGLETON_ROW_ID,
        value: checkpoint,
      });
    }
    this.schedulePersist();
  }

  replaceDirectoryBaseline(serverId: string, directory: CachedDirectory): void {
    if (!this.activeServerIds.has(serverId)) return;
    this.invalidateHostReads(serverId, ["agent", "workspace", "project", "checkpoint"]);
    // Replacing the directory must not discard independently accepted timeline changes.
    for (const [key, row] of this.pendingUpserts) {
      if (row.serverId === serverId && row.kind !== "timeline") this.pendingUpserts.delete(key);
    }
    for (const [key, row] of this.pendingDeletes) {
      if (row.serverId === serverId && row.kind !== "timeline") this.pendingDeletes.delete(key);
    }
    this.pendingBaselines.add(serverId);
    for (const [id, value] of directory.agents) {
      this.queueUpsert({ serverId, kind: "agent", id, value });
    }
    for (const [id, value] of directory.workspaces) {
      this.queueUpsert({ serverId, kind: "workspace", id, value });
    }
    for (const [id, value] of directory.projects) {
      this.queueUpsert({ serverId, kind: "project", id, value });
    }
    if (directory.checkpoint) {
      this.queueUpsert({
        serverId,
        kind: "checkpoint",
        id: REPLICA_SINGLETON_ROW_ID,
        value: directory.checkpoint,
      });
    }
    this.schedulePersist();
  }

  commitTimeline(serverId: string, agentId: string, timeline: CachedTimeline): void {
    if (!this.activeServerIds.has(serverId)) return;
    if (timeline.agentId !== agentId) throw new Error("Timeline cache key does not match payload");
    this.queueUpsert({ serverId, kind: "timeline", id: agentId, value: timeline });
    this.schedulePersist();
  }

  setHosts(serverIds: Iterable<string>): void {
    const next = new Set(serverIds);
    const removed = [...this.activeServerIds].filter((serverId) => !next.has(serverId));
    const added = [...next].filter((serverId) => !this.activeServerIds.has(serverId));
    for (const serverId of [...removed, ...added]) this.invalidateHostReads(serverId);
    this.activeServerIds.clear();
    for (const serverId of next) this.activeServerIds.add(serverId);
    for (const serverId of removed) {
      this.removeStoredHost(serverId);
      this.dropPendingHostChanges(serverId);
      this.invalidatedHosts.delete(serverId);
      this.queueOperation(() => this.rowStore.deleteHost(serverId));
    }
  }

  reconcileServerId(oldServerId: string, newServerId: string): void {
    this.invalidateHostReads(oldServerId);
    this.invalidateHostReads(newServerId);
    const rows = this.storedRows.get(oldServerId);
    if (rows) {
      const newRows = this.storedRows.get(newServerId) ?? new Map<string, ReplicaRow>();
      this.totalBytes -=
        (this.hostBytes.get(oldServerId) ?? 0) + (this.hostBytes.get(newServerId) ?? 0);
      this.storedRows.delete(oldServerId);
      for (const [key, row] of rows) newRows.set(key, { ...row, serverId: newServerId });
      this.storedRows.set(newServerId, newRows);
      const bytes = [...newRows.values()].reduce((sum, row) => sum + rowBytes(row), 0);
      this.hostBytes.delete(oldServerId);
      this.hostBytes.set(newServerId, bytes);
      this.totalBytes += bytes;
      this.hostWriteOrder.delete(oldServerId);
      this.touchHost(newServerId);
    }
    this.renamePendingHostChanges(oldServerId, newServerId);
    if (this.activeServerIds.delete(oldServerId)) this.activeServerIds.add(newServerId);
    this.queueOperation(() => this.rowStore.renameHost(oldServerId, newServerId));
  }

  async flush(): Promise<void> {
    await this.syncPending();
  }

  /** Resolves to whether every pending change reached the store. */
  private async syncPending(): Promise<boolean> {
    const persisted = await this.persist();
    await this.writeQueue.catch(() => undefined);
    return persisted;
  }

  private async flushPending(): Promise<void> {
    await this.persist();
  }

  private async persist(): Promise<boolean> {
    if (this.persistTimer) {
      clearTimeout(this.persistTimer);
      this.persistTimer = null;
    }
    if (!this.hasPendingChanges()) return true;
    const write = this.writeQueue
      .catch(() => undefined)
      .then(async () => {
        const pending = this.drainPendingChanges();
        try {
          await this.prepareStore();
          await this.ensureStoredIndex();
          const changes = this.materializePendingChanges(pending);
          const { changes: boundedChanges, evicted } = await this.fitChangesToBudget(changes);
          if (boundedChanges.upserts.length > 0 || boundedChanges.deletes.length > 0) {
            await this.rowStore.apply(boundedChanges);
            this.applyStoredChanges(boundedChanges);
          }
          for (const serverId of pending.baselines) {
            if (!evicted.has(serverId)) this.invalidatedHosts.delete(serverId);
          }
        } catch {
          this.restorePendingChanges(pending);
          if (this.hasPendingChanges()) this.schedulePersist();
          return false;
        }
        return true;
      });
    // The queue only sequences writes; every consumer decides for itself what a failure means.
    this.writeQueue = write.then(
      () => undefined,
      () => undefined,
    );
    return write;
  }

  private queueEntityDelete(serverId: string, kind: ReplicaRowKind, id: string): void {
    this.queueDelete({ serverId, kind, id });
  }

  private queueUpsert(row: StructuredReplicaUpsert): void {
    this.invalidateRowReads(row);
    const key = pendingRowKey(row);
    this.pendingDeletes.delete(key);
    this.pendingUpserts.set(key, row);
  }

  private queueDelete(key: ReplicaRowKey): void {
    this.invalidateRowReads(key);
    const pendingKey = pendingRowKey(key);
    this.pendingUpserts.delete(pendingKey);
    this.pendingDeletes.set(pendingKey, key);
  }

  private hasPendingChanges(): boolean {
    return (
      this.pendingUpserts.size > 0 || this.pendingDeletes.size > 0 || this.pendingBaselines.size > 0
    );
  }

  private hasPendingReadChanges(read: ReplicaRead): boolean {
    if (
      this.pendingBaselines.has(read.serverId) &&
      read.kinds.some((kind) => kind !== "timeline")
    ) {
      return true;
    }
    return (
      [...this.pendingUpserts.values()].some((row) => matchesReplicaRead(read, row)) ||
      [...this.pendingDeletes.values()].some((row) => matchesReplicaRead(read, row))
    );
  }

  private invalidateRowReads(row: ReplicaRowKey): void {
    for (const read of this.activeReads) {
      if (matchesReplicaRead(read, row)) read.invalidated = true;
    }
  }

  private invalidateHostReads(serverId: string, kinds?: readonly ReplicaRowKind[]): void {
    for (const read of this.activeReads) {
      if (
        read.serverId === serverId &&
        (!kinds || kinds.some((kind) => read.kinds.includes(kind)))
      ) {
        read.invalidated = true;
      }
    }
  }

  private hasPendingHostChanges(serverId: string): boolean {
    if (this.pendingBaselines.has(serverId)) return true;
    for (const row of this.pendingUpserts.values()) {
      if (row.serverId === serverId) return true;
    }
    for (const row of this.pendingDeletes.values()) {
      if (row.serverId === serverId) return true;
    }
    return false;
  }

  private drainPendingChanges(): PendingReplicaChanges {
    const changes = {
      upserts: [...this.pendingUpserts.values()],
      deletes: [...this.pendingDeletes.values()],
      baselines: this.pendingBaselines,
    };
    this.pendingUpserts = new Map();
    this.pendingDeletes = new Map();
    this.pendingBaselines = new Set();
    return changes;
  }

  private materializePendingChanges(pending: PendingReplicaChanges): ReplicaRowChanges {
    const deletes = new Map(pending.deletes.map((row) => [pendingRowKey(row), row]));
    const upserts: ReplicaRow[] = [];
    for (const upsert of pending.upserts) {
      const row = this.materializeUpsert(upsert);
      if (row) upserts.push(row);
      else deletes.set(pendingRowKey(upsert), upsert);
    }
    for (const serverId of pending.baselines) {
      const replacementKeys = new Set(
        upserts.filter((row) => row.serverId === serverId && row.kind !== "timeline").map(rowKey),
      );
      for (const row of this.storedRows.get(serverId)?.values() ?? []) {
        if (row.kind !== "timeline" && !replacementKeys.has(rowKey(row))) {
          deletes.set(pendingRowKey(row), row);
        }
      }
    }
    return { upserts, deletes: [...deletes.values()] };
  }

  private materializeUpsert(upsert: StructuredReplicaUpsert): ReplicaRow | null {
    let value: unknown;
    switch (upsert.kind) {
      case "agent":
        value = serializeAgent(upsert.value);
        break;
      case "workspace":
        value = serializeWorkspace(upsert.value);
        break;
      case "project":
        value = serializeProject(upsert.value);
        break;
      case "timeline":
        value = serializeTimeline(upsert.value);
        if (!value) return null;
        break;
      case "checkpoint":
        value = serializeDirectoryCheckpoint(upsert.value);
        break;
    }
    return {
      serverId: upsert.serverId,
      kind: upsert.kind,
      id: upsert.id,
      payload: JSON.stringify(value),
    };
  }

  private restorePendingChanges(changes: PendingReplicaChanges): void {
    for (const serverId of changes.baselines) {
      if (this.activeServerIds.has(serverId)) this.pendingBaselines.add(serverId);
    }
    for (const key of changes.deletes) {
      const pendingKey = pendingRowKey(key);
      if (
        this.activeServerIds.has(key.serverId) &&
        !this.pendingUpserts.has(pendingKey) &&
        !this.pendingDeletes.has(pendingKey)
      ) {
        this.queueDelete(key);
      }
    }
    for (const row of changes.upserts) {
      const pendingKey = pendingRowKey(row);
      if (
        this.activeServerIds.has(row.serverId) &&
        !this.pendingUpserts.has(pendingKey) &&
        !this.pendingDeletes.has(pendingKey)
      ) {
        this.queueUpsert(row);
      }
    }
  }

  private applyStoredChanges(changes: ReplicaRowChanges): void {
    const touchedServerIds = new Set<string>();
    for (const key of changes.deletes) {
      const rows = this.storedRows.get(key.serverId);
      const previous = rows?.get(rowKey(key));
      if (previous) {
        rows?.delete(rowKey(key));
        this.adjustHostBytes(key.serverId, -rowBytes(previous));
      }
      touchedServerIds.add(key.serverId);
    }
    for (const row of changes.upserts) {
      const rows = this.storedRows.get(row.serverId) ?? new Map<string, ReplicaRow>();
      const previous = rows.get(rowKey(row));
      const previousBytes = previous ? rowBytes(previous) : 0;
      rows.set(rowKey(row), row);
      this.storedRows.set(row.serverId, rows);
      this.adjustHostBytes(row.serverId, rowBytes(row) - previousBytes);
      touchedServerIds.add(row.serverId);
    }
    for (const serverId of touchedServerIds) {
      if ((this.storedRows.get(serverId)?.size ?? 0) === 0) {
        this.removeStoredHost(serverId);
      } else {
        this.touchHost(serverId);
      }
    }
  }

  private adjustHostBytes(serverId: string, delta: number): void {
    this.hostBytes.set(serverId, (this.hostBytes.get(serverId) ?? 0) + delta);
    this.totalBytes += delta;
  }

  private touchHost(serverId: string): void {
    this.hostWriteOrder.delete(serverId);
    this.hostWriteOrder.set(serverId, true);
  }

  private async fitChangesToBudget(
    changes: ReplicaRowChanges,
  ): Promise<{ changes: ReplicaRowChanges; evicted: Set<string> }> {
    const touchedServerIds = new Set<string>();
    for (const key of changes.deletes) touchedServerIds.add(key.serverId);
    for (const row of changes.upserts) touchedServerIds.add(row.serverId);

    const projectedRows = new Map<string, Map<string, ReplicaRow>>();
    const projectedBytes = new Map(this.hostBytes);
    for (const serverId of touchedServerIds) {
      projectedRows.set(serverId, new Map(this.storedRows.get(serverId)));
    }
    for (const key of changes.deletes) {
      projectedRows.get(key.serverId)?.delete(rowKey(key));
    }
    for (const row of changes.upserts) {
      projectedRows.get(row.serverId)?.set(rowKey(row), row);
    }
    for (const [serverId, rows] of projectedRows) {
      projectedBytes.set(
        serverId,
        [...rows.values()].reduce((sum, row) => sum + rowBytes(row), 0),
      );
    }

    const writeOrder = [...this.hostWriteOrder.keys()].filter(
      (serverId) => !touchedServerIds.has(serverId),
    );
    writeOrder.push(...touchedServerIds);
    let projectedTotal = [...projectedBytes.values()].reduce((sum, bytes) => sum + bytes, 0);
    const evicted = new Set<string>();
    while (projectedTotal > this.maxBytes) {
      const serverId = writeOrder.shift();
      if (serverId === undefined) break;
      const bytes = projectedBytes.get(serverId) ?? 0;
      projectedTotal -= bytes;
      projectedBytes.delete(serverId);
      evicted.add(serverId);
      this.invalidatedHosts.add(serverId);
      if (this.storedRows.has(serverId)) await this.rowStore.deleteHost(serverId);
      this.removeStoredHost(serverId);
    }

    return {
      changes: {
        upserts: changes.upserts.filter((row) => !evicted.has(row.serverId)),
        deletes: changes.deletes.filter((key) => !evicted.has(key.serverId)),
      },
      evicted,
    };
  }

  private removeStoredHost(serverId: string): void {
    this.totalBytes -= this.hostBytes.get(serverId) ?? 0;
    this.storedRows.delete(serverId);
    this.hostBytes.delete(serverId);
    this.hostWriteOrder.delete(serverId);
  }

  private dropPendingHostChanges(serverId: string): void {
    this.pendingBaselines.delete(serverId);
    for (const [key, row] of this.pendingUpserts) {
      if (row.serverId === serverId) this.pendingUpserts.delete(key);
    }
    for (const [key, row] of this.pendingDeletes) {
      if (row.serverId === serverId) this.pendingDeletes.delete(key);
    }
  }

  private renamePendingHostChanges(oldServerId: string, newServerId: string): void {
    const changes = this.drainPendingChanges();
    for (const key of changes.deletes) {
      this.queueDelete(key.serverId === oldServerId ? { ...key, serverId: newServerId } : key);
    }
    for (const row of changes.upserts) {
      this.queueUpsert(row.serverId === oldServerId ? { ...row, serverId: newServerId } : row);
    }
    for (const serverId of changes.baselines) {
      this.pendingBaselines.add(serverId === oldServerId ? newServerId : serverId);
    }
    if (this.invalidatedHosts.delete(oldServerId)) this.invalidatedHosts.add(newServerId);
  }

  private prepareStore(): Promise<void> {
    this.preparePromise ??= (async () => {
      await this.rowStore.open();
      // COMPAT(replica-blob-cache): remove after 2026-11
      await this.clearLegacyCache().catch(() => undefined);
    })();
    return this.preparePromise;
  }

  private ensureStoredIndex(): Promise<void> {
    this.storedIndexPromise ??= this.rowStore.readAll().then(async (hosts) => {
      this.storedRows.clear();
      this.hostBytes.clear();
      this.hostWriteOrder.clear();
      this.totalBytes = 0;
      for (const host of hosts) {
        if (!this.activeServerIds.has(host.serverId)) {
          await this.rowStore.deleteHost(host.serverId);
          continue;
        }
        const rows = new Map(host.rows.map((row) => [rowKey(row), row]));
        const bytes = host.rows.reduce((sum, row) => sum + rowBytes(row), 0);
        this.storedRows.set(host.serverId, rows);
        this.hostBytes.set(host.serverId, bytes);
        this.totalBytes += bytes;
        this.touchHost(host.serverId);
      }
      return undefined;
    });
    return this.storedIndexPromise;
  }

  private queueOperation(operation: () => Promise<void>): Promise<void> {
    this.writeQueue = this.writeQueue
      .catch(() => undefined)
      .then(async () => {
        await this.prepareStore();
        await operation();
        return undefined;
      })
      .catch(() => undefined);
    return this.writeQueue;
  }

  private schedulePersist(): void {
    if (this.persistTimer) return;
    this.persistTimer = setTimeout(() => {
      this.persistTimer = null;
      void this.flushPending();
    }, PERSIST_DELAY_MS);
  }
}