rename compaction to remake

10 files changed, 3493 insertions, 0 deletions

diff --git a/src/remake/api.rs b/src/remake/api.rs
new file mode 100644
index 0000000..9b490f4
--- /dev/null
+++ b/src/remake/api.rs
@@ -0,0 +1,474 @@
+//! `/admin/remakes` handlers.
+//!
+//! * `POST /admin/remakes` proposes a remake of the current epoch (`POST
+//!   /admin/compaction` is a shortcut, kept from before). The receiving
+//!   node becomes its coordinator. `202` with the report, or with `?wait=true` the final
+//!   report once decided. A dropped request does not abort anything. `?force=true`
+//!   accepts a `settleSecs` below the dormancy threshold (tests, single-site clusters).
+//! * `GET /admin/remakes` lists all remakes.
+//! * `GET /admin/remakes/{id}` reports one; `?wait=true` long-polls until decided
+//!   (bounded by the proposal's own expiry).
+//! * `GET /admin/epochs` shows the local epoch, the archived previous ones and whether
+//!   the node is catching up with a newer epoch from a peer.
+//! * `DELETE /admin/remakes/{id}` asks the coordinator to abort. Only possible while
+//!   proposed; `409` once committed (a switch can't be undone) or already over.
+//!   `DELETE /admin/remakes` does the same for the remake of the local epoch that
+//!   is in progress, without having to know its ID.
+//!
+//! Request body for POST (all optional):
+//! `{ "replayWindow": {"epochs": 1, "seconds": 604800},
+//!    "archiveRetention": {"epochs": 2, "seconds": 2592000},
+//!    "settleSecs": 480 }`
+
+use std::time::Duration;
+
+use axum::{
+    body::Bytes,
+    extract::{Path, Query, State},
+    http::{header::LOCATION, StatusCode},
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value as JsonValue};
+use tokio::sync::broadcast;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use super::coordinator;
+use super::local::LocalStatus;
+use super::records::{
+    self, unix_millis, AbortRequest, EpochRecord, Phase, Policy, Proposal, Record, Retention,
+};
+use super::Host;
+use crate::document::store::StoreError;
+
+#[derive(Debug, Default, Deserialize)]
+pub(crate) struct WaitQuery {
+    #[serde(default)]
+    wait: bool,
+}
+
+#[derive(Debug, Default, Deserialize)]
+pub(crate) struct StartQuery {
+    #[serde(default)]
+    wait: bool,
+    /// Allows a settle period shorter than the dormancy threshold.
+    #[serde(default)]
+    force: bool,
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+struct StartRequest {
+    replay_window: Option<Retention>,
+    archive_retention: Option<Retention>,
+    settle_secs: Option<u64>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct Report {
+    id: String,
+    phase: Phase,
+    #[serde(flatten)]
+    record: Record,
+    /// What happened to this remake on the node answering the request.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    local: Option<LocalStatus>,
+}
+
+#[derive(Debug, thiserror::Error)]
+enum StartError {
+    #[error("remake {0} is already open for this epoch")]
+    Open(String),
+    #[error(transparent)]
+    Store(#[from] StoreError),
+}
+
+pub(super) fn error_response(
+    status: StatusCode,
+    message: impl Into<String>,
+    details: Option<JsonValue>,
+) -> Response {
+    let mut body = json!({ "error": message.into() });
+    if let (Some(JsonValue::Object(extra)), Some(object)) = (details, body.as_object_mut()) {
+        object.extend(extra);
+    }
+    (status, Json(body)).into_response()
+}
+
+fn grace<H: Host>(host: &H) -> u64 {
+    host.dormant_after().as_millis() as u64
+}
+
+fn report<H: Host>(host: &H, id: String, record: Record) -> Report {
+    Report {
+        phase: record.phase(unix_millis(), grace(host)),
+        local: host.remake().status(&id),
+        id,
+        record,
+    }
+}
+
+async fn load_report<H: Host>(host: &H, id: &str) -> Option<Report> {
+    let wanted = id.to_owned();
+    let record = host
+        .read_doc(move |doc| records::load(doc, &wanted))
+        .await?;
+    Some(report(host, id.to_owned(), record))
+}
+
+/// Waits until the remake leaves `proposed`. Bounded by the proposal's own expiry,
+/// at which point it is reported as `expired`.
+async fn wait_until_final<H: Host>(host: &H, id: &str) -> Option<Report> {
+    // A record that was readable may be briefly unreadable or absent while concurrent
+    // changes merge; give up only if that persists.
+    const MAX_MISSES: u32 = 5;
+
+    let mut changes = host.subscribe_changes();
+    let mut seen = false;
+    let mut misses = 0;
+    loop {
+        let Some(report) = load_report(host, id).await else {
+            misses += 1;
+            warn!(remake = %id, seen, misses, "Remake record not readable while waiting");
+            if !seen || misses >= MAX_MISSES {
+                return None;
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+            continue;
+        };
+        seen = true;
+        misses = 0;
+        if report.phase.is_final() {
+            return Some(report);
+        }
+        // Re-read on every published change, and at least once a second: expiry is a
+        // matter of time, not of document changes.
+        if let Ok(Err(broadcast::error::RecvError::Closed)) =
+            tokio::time::timeout(Duration::from_secs(1), changes.recv()).await
+        {
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    }
+}
+
+pub(crate) async fn start<H: Host>(
+    State(host): State<H>,
+    Query(query): Query<StartQuery>,
+    body: Bytes,
+) -> Response {
+    if !host.is_node_ready().await {
+        return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
+    }
+
+    let request: StartRequest = if body.is_empty() {
+        StartRequest::default()
+    } else {
+        match serde_json::from_slice(&body) {
+            Ok(request) => request,
+            Err(error) => {
+                return error_response(
+                    StatusCode::BAD_REQUEST,
+                    format!("invalid request: {error}"),
+                    None,
+                )
+            }
+        }
+    };
+
+    let dormant_after = host.dormant_after();
+    let settle = request
+        .settle_secs
+        .map(Duration::from_secs)
+        .unwrap_or(dormant_after * 2);
+    if settle < dormant_after && query.force {
+        warn!(
+            settle_secs = settle.as_secs_f64(),
+            dormant_after_secs = dormant_after.as_secs(),
+            "Forced settle period shorter than the dormancy threshold; a partition may go unnoticed"
+        );
+    } else if settle < dormant_after {
+        return error_response(
+            StatusCode::BAD_REQUEST,
+            format!(
+                "settleSecs must be at least {} (the dormancy threshold), so a partition is detected before deciding; add ?force=true to override",
+                dormant_after.as_secs()
+            ),
+            None,
+        );
+    }
+
+    // Fail fast instead of proposing something that will abort anyway.
+    let peers = host.peer_summaries().await;
+    let dormant: Vec<String> = peers
+        .iter()
+        .filter(|peer| !peer.active)
+        .map(|peer| peer.id.clone())
+        .collect();
+    if !dormant.is_empty() {
+        return error_response(
+            StatusCode::CONFLICT,
+            "dormant peers block remake; forget them first (DELETE /peers/{pk})",
+            Some(json!({ "dormantPeers": dormant })),
+        );
+    }
+
+    let mut peers_considered: Vec<String> = peers.into_iter().map(|peer| peer.id).collect();
+    peers_considered.sort();
+
+    let now = unix_millis();
+    let settle_ms = settle.as_millis() as u64;
+    let id = Uuid::new_v4().to_string();
+    let grace = grace(&host);
+
+    let proposal = Proposal {
+        coordinator: host.node_id(),
+        from_epoch: host.current_epoch().into(),
+        base_heads: Vec::new(),
+        policy: Policy {
+            replay_window: request
+                .replay_window
+                .unwrap_or_else(Policy::default_replay_window),
+            archive_retention: request
+                .archive_retention
+                .unwrap_or_else(Policy::default_archive_retention),
+        },
+        settle_ms,
+        proposed_at: now,
+        settle_until: now + settle_ms,
+        peers_considered,
+    };
+
+    // From here on, nothing depends on the client: writing the proposal and starting the
+    // coordinator run in their own task, so a client that disconnects mid-request can't
+    // leave a proposal without a coordinator. Dropping the handle doesn't cancel it.
+    let proposing = tokio::spawn(propose(host.clone(), id.clone(), proposal, now, grace));
+    let written = match proposing.await {
+        Ok(written) => written,
+        Err(error) => {
+            error!(error = %error, "Proposal task failed");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to write the proposal",
+                None,
+            );
+        }
+    };
+
+    let record = match written {
+        Ok(record) => record,
+        Err(StartError::Open(open)) => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "another remake of this epoch is still open",
+                Some(json!({ "remake": open })),
+            );
+        }
+        Err(StartError::Store(error)) => {
+            error!(error = %format!("{error:#}"), "Failed to write remake proposal");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to write the proposal",
+                None,
+            );
+        }
+    };
+
+    info!(
+        remake = %id,
+        settle_secs = settle.as_secs(),
+        "Remake proposed"
+    );
+
+    if query.wait {
+        return match wait_until_final(&host, &id).await {
+            Some(report) => Json(report).into_response(),
+            None => error_response(StatusCode::NOT_FOUND, "remake record vanished", None),
+        };
+    }
+
+    let location = format!("/admin/remakes/{id}");
+    (
+        StatusCode::ACCEPTED,
+        [(LOCATION, location)],
+        Json(report(&host, id, record)),
+    )
+        .into_response()
+}
+
+/// Writes the proposal (unless another one is open for the epoch) and starts its
+/// coordinator. Runs as its own task, independent of the HTTP request.
+async fn propose<H: Host>(
+    host: H,
+    id: String,
+    mut proposal: Proposal,
+    now: u64,
+    grace: u64,
+) -> Result<Record, StartError> {
+    let record_id = id.clone();
+    let record = host
+        .commit_doc(move |doc| -> Result<Record, StartError> {
+            if let Some((open, _)) = records::load_all(doc).into_iter().find(|(_, record)| {
+                record.proposal.from_epoch == proposal.from_epoch
+                    && record.phase(now, grace).is_open()
+            }) {
+                return Err(StartError::Open(open));
+            }
+
+            // The base is everything before the proposal; the proposal itself is not in it.
+            let mut heads: Vec<String> = doc.get_heads().iter().map(|h| h.to_string()).collect();
+            heads.sort();
+            proposal.base_heads = heads;
+
+            records::write_proposal(doc, &record_id, &proposal)?;
+            Ok(Record {
+                proposal,
+                decision: None,
+                abort_requested: None,
+            })
+        })
+        .await?;
+
+    tokio::spawn(coordinator::run(host, id));
+    Ok(record)
+}
+
+pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
+    let all = host.read_doc(records::load_all).await;
+    let mut reports: Vec<Report> = all
+        .into_iter()
+        .map(|(id, record)| report(&host, id, record))
+        .collect();
+    reports.sort_by_key(|report| report.record.proposal.proposed_at);
+    Json(reports).into_response()
+}
+
+pub(crate) async fn get<H: Host>(
+    State(host): State<H>,
+    Path(id): Path<String>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    let found = if query.wait {
+        wait_until_final(&host, &id).await
+    } else {
+        load_report(&host, &id).await
+    };
+    match found {
+        Some(report) => Json(report).into_response(),
+        None => error_response(StatusCode::NOT_FOUND, "no such remake", None),
+    }
+}
+
+/// `DELETE /admin/remakes/{id}`.
+pub(crate) async fn abort<H: Host>(
+    State(host): State<H>,
+    Path(id): Path<String>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    abort_by_id(host, id, query.wait).await
+}
+
+/// `DELETE /admin/remakes`: aborts whichever remake of the local epoch is still
+/// proposed. There is at most one, unless two coordinators raced (then both abort anyway
+/// and this picks the oldest).
+pub(crate) async fn abort_current<H: Host>(
+    State(host): State<H>,
+    Query(query): Query<WaitQuery>,
+) -> Response {
+    let local = EpochRecord::from(host.current_epoch());
+    let now = unix_millis();
+    let grace = grace(&host);
+    let open = host
+        .read_doc(move |doc| {
+            records::load_all(doc)
+                .into_iter()
+                .filter(|(_, record)| {
+                    record.proposal.from_epoch == local
+                        && record.phase(now, grace) == Phase::Proposed
+                })
+                .min_by_key(|(_, record)| record.proposal.proposed_at)
+                .map(|(id, _)| id)
+        })
+        .await;
+
+    match open {
+        Some(id) => abort_by_id(host, id, query.wait).await,
+        None => error_response(
+            StatusCode::NOT_FOUND,
+            "no remake of the local epoch is in progress",
+            None,
+        ),
+    }
+}
+
+async fn abort_by_id<H: Host>(host: H, id: String, wait: bool) -> Response {
+    let Some(current) = load_report(&host, &id).await else {
+        return error_response(StatusCode::NOT_FOUND, "no such remake", None);
+    };
+
+    match current.phase {
+        Phase::Proposed => {}
+        Phase::Committed => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "already committed; the switch cannot be undone",
+                Some(json!({ "remake": current })),
+            )
+        }
+        Phase::Aborted | Phase::Expired => {
+            return error_response(
+                StatusCode::CONFLICT,
+                "already over",
+                Some(json!({ "remake": current })),
+            )
+        }
+    }
+
+    // The coordinator applies the request on its next check (within about a second).
+    // If it is gone, the proposal expires anyway.
+    if current.record.abort_requested.is_none() {
+        let request = AbortRequest {
+            by: host.node_id(),
+            at: unix_millis(),
+        };
+        let record_id = id.clone();
+        if let Err(error) = host
+            .commit_doc(move |doc| records::write_abort_request(doc, &record_id, &request))
+            .await
+        {
+            error!(remake = %id, error = %format!("{error:#}"), "Failed to write abort request");
+            return error_response(
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "failed to record the abort request",
+                None,
+            );
+        }
+        info!(remake = %id, "Abort requested");
+    }
+
+    let report = if wait {
+        wait_until_final(&host, &id).await
+    } else {
+        load_report(&host, &id).await
+    };
+    match report {
+        Some(report) if report.phase.is_final() => Json(report).into_response(),
+        Some(report) => (StatusCode::ACCEPTED, Json(report)).into_response(),
+        None => error_response(StatusCode::NOT_FOUND, "no such remake", None),
+    }
+}
+
+/// `GET /admin/epochs`: the local epoch, the previous epochs this node still keeps, and
+/// an ongoing catch-up with a peer's newer epoch (`null` if none).
+pub(crate) async fn epochs<H: Host>(State(host): State<H>) -> Response {
+    let current = EpochRecord::from(host.current_epoch());
+    Json(json!({
+        "current": current,
+        "archived": host.remake().archived_epochs(),
+        "catchUp": host.remake().catch_up_report(),
+        "quarantine": host.remake().list_quarantine(|entry| entry.id.clone()),
+    }))
+    .into_response()
+}
diff --git a/src/remake/catchup.rs b/src/remake/catchup.rs
new file mode 100644
index 0000000..22aa0f8
--- /dev/null
+++ b/src/remake/catchup.rs
@@ -0,0 +1,506 @@
+//! Catching up with a newer epoch: the node rejoins.
+//!
+//! Peers on another epoch don't sync, so a node that missed a switch (new, late,
+//! returning, or one whose own build failed) can't follow through ordinary sync. Once
+//! an active peer advertises a newer generation, the node
+//!
+//! 1. stops being ready and keeps its document aside; reads are answered from a copy
+//!    of it, marked stale (`X-Continuum-Stale`), writes get `503`;
+//! 2. switches to an empty document labelled with the newer epoch: an empty history
+//!    conflicts with nothing, so ordinary sync fills it from peers on that epoch,
+//!    exactly like a new node joining;
+//! 3. once caught up with a peer, checks that the history starts with the epoch's
+//!    snapshot change;
+//! 4. decides about its own changes the cluster never saw (none, for a new node):
+//!    * the metadata of the epoch that followed the node's own records which changes
+//!      went into it (per-actor clock); the rest are this node's missing changes;
+//!    * within that epoch's replay window they are replayed (state per path, as in a
+//!      switch), otherwise they are quarantined for an operator
+//!      (`/admin/quarantine`), as are changes that can't be told apart;
+//! 5. archives the old document and serves requests again. A node that was never
+//!    ready (a new node) is made ready by bootstrap, which waits for the rejoin.
+//!
+//! If no peer completes a sync in time, or the check fails, the node restores its old
+//! document and epoch and tries again later.
+
+use std::collections::BTreeMap;
+use std::convert::Infallible;
+use std::time::Duration;
+
+use automerge::{AutoCommit, ChangeHash};
+use tokio::time::Instant;
+use tracing::{error, info, warn};
+use uuid::Uuid;
+
+use super::delta::{self, ReplayStats};
+use super::local::{CatchUp, LocalStatus, Quarantined, Rejoin};
+use super::records::unix_millis;
+use super::snapshot::{self, EpochMetadata};
+use super::Host;
+use crate::document::replica::Epoch;
+
+/// Pause before trying the same target again after a failed rejoin.
+const RETRY: Duration = Duration::from_secs(30);
+/// Lower bound for how long a rejoin may take to complete a sync.
+const MIN_REJOIN_TIMEOUT: Duration = Duration::from_secs(60);
+
+/// Called by the switch worker when there is nothing to switch to locally, and on every
+/// tick while a rejoin is in progress.
+pub(super) async fn check<H: Host>(host: &H) {
+    if let Some((target, started)) = host.remake().rejoin_progress() {
+        let timeout = host.dormant_after().max(MIN_REJOIN_TIMEOUT);
+        if host.initial_sync_done() {
+            finish(host).await;
+        } else if started.elapsed() > timeout {
+            if let Some(rejoin) = host.remake().take_rejoin() {
+                let reason = format!(
+                    "no peer on {target} completed a sync within {}s",
+                    timeout.as_secs()
+                );
+                abandon(host, rejoin, reason).await;
+            }
+        }
+        return;
+    }
+
+    let local = host.current_epoch();
+    let mut ahead: Vec<(String, Epoch)> = host
+        .peer_summaries()
+        .await
+        .into_iter()
+        .filter(|peer| peer.active)
+        .filter_map(|peer| peer.epoch.map(|epoch| (peer.id, epoch)))
+        .filter(|(_, epoch)| epoch.generation > local.generation)
+        .collect();
+    ahead.sort_by(|a, b| a.0.cmp(&b.0));
+
+    let Some(generation) = ahead.iter().map(|(_, epoch)| epoch.generation).max() else {
+        if host.remake().catch_up().is_some() {
+            host.remake().set_catch_up(None);
+        }
+        return;
+    };
+
+    let mut targets: Vec<Epoch> = ahead
+        .iter()
+        .filter(|(_, epoch)| epoch.generation == generation)
+        .map(|(_, epoch)| *epoch)
+        .collect();
+    targets.sort_by_key(|epoch| epoch.id);
+    targets.dedup();
+    let [target] = targets.as_slice() else {
+        let listed: Vec<String> = targets.iter().map(ToString::to_string).collect();
+        warn!(epochs = ?listed, "Peers disagree about the newest epoch; not catching up");
+        return;
+    };
+    let target = *target;
+
+    // Back off between attempts for the same target.
+    let previous = host
+        .remake()
+        .catch_up()
+        .filter(|catch_up| catch_up.target == target);
+    if previous
+        .as_ref()
+        .is_some_and(|catch_up| catch_up.last_attempt.elapsed() < RETRY)
+    {
+        return;
+    }
+    let attempts = previous.map_or(0, |catch_up| catch_up.attempts) + 1;
+
+    start(host, local, target, attempts).await;
+}
+
+async fn start<H: Host>(host: &H, local: Epoch, target: Epoch, attempts: u32) {
+    host.remake().set_catch_up(Some(CatchUp {
+        target,
+        attempts,
+        last_attempt: Instant::now(),
+        error: None,
+    }));
+
+    // Keep answering reads, from a copy of the current document marked stale, before
+    // the node stops being ready: no window in which reads get 503.
+    if host.is_node_ready().await {
+        let copy = host.read_doc(|doc| doc.clone()).await;
+        host.set_stale_view(Some((local, copy)));
+    }
+    let was_ready = host.suspend().await;
+    let old = match host
+        .switch_epoch(target, |_live| Ok::<_, Infallible>(AutoCommit::new()))
+        .await
+    {
+        Ok(old) => old,
+        Err(never) => match never {},
+    };
+    // After the swap: messages of the old epoch are dropped from now on, so nothing
+    // can mark the new, empty document as caught up by mistake. A stale "done" from
+    // just before is caught in `finish` (the snapshot change is missing then).
+    host.reset_initial_sync();
+
+    info!(from = %local, to = %target, attempt = attempts, "Rejoining newer epoch");
+    host.remake().start_rejoin(Rejoin {
+        previous: local,
+        target,
+        old,
+        started: Instant::now(),
+        was_ready,
+    });
+}
+
+/// What to do with this node's own changes the new epoch lacks.
+enum Plan {
+    Nothing,
+    Replay(Vec<Vec<String>>),
+    Quarantine {
+        paths: Vec<Vec<String>>,
+        reason: String,
+    },
+}
+
+enum Prepared {
+    /// Caught up with someone, but not with the target epoch yet: keep waiting.
+    NotYet(AutoCommit),
+    Failed(AutoCommit, String),
+    Ready {
+        old: AutoCommit,
+        metadata: EpochMetadata,
+        plan: Plan,
+        own_missing: usize,
+        had_history: bool,
+    },
+}
+
+async fn finish<H: Host>(host: &H) {
+    let Some(rejoin) = host.remake().take_rejoin() else {
+        return;
+    };
+    let Rejoin {
+        previous,
+        target,
+        old,
+        started,
+        was_ready,
+    } = rejoin;
+
+    // Everything heavy runs on a copy, without a lock.
+    let copy = host.read_doc(|doc| doc.clone()).await;
+    let prepared = tokio::task::spawn_blocking(move || prepare(copy, old, previous, target)).await;
+
+    let (old, metadata, plan, own_missing, had_history) = match prepared {
+        Ok(Prepared::Ready {
+            old,
+            metadata,
+            plan,
+            own_missing,
+            had_history,
+        }) => (old, metadata, plan, own_missing, had_history),
+        Ok(Prepared::NotYet(old)) => {
+            host.reset_initial_sync();
+            host.remake().start_rejoin(Rejoin {
+                previous,
+                target,
+                old,
+                started,
+                was_ready,
+            });
+            return;
+        }
+        Ok(Prepared::Failed(old, reason)) => {
+            let rejoin = Rejoin {
+                previous,
+                target,
+                old,
+                started,
+                was_ready,
+            };
+            return abandon(host, rejoin, reason).await;
+        }
+        Err(join_error) => {
+            // The old document is gone with the panicked task; nothing to restore.
+            error!(error = %join_error, "Rejoin task failed; staying on the new epoch");
+            if was_ready {
+                host.resume().await;
+            }
+            host.set_stale_view(None);
+            return;
+        }
+    };
+
+    let mut stats = ReplayStats::default();
+    let mut quarantine_id = None;
+    let mut keep_archive = true;
+    let mut old = Some(old);
+
+    match plan {
+        Plan::Nothing => {}
+        Plan::Replay(paths) => {
+            let document = old.take().expect("old document");
+            let (replayed, document) = host
+                .commit_doc(move |doc| {
+                    let stats = delta::replay_paths(&document, doc, &paths);
+                    Ok::<_, Infallible>((stats, document))
+                })
+                .await
+                .unwrap_or_else(|never| match never {});
+            stats = replayed;
+            old = Some(document);
+        }
+        Plan::Quarantine { paths, reason } => {
+            let id = Uuid::new_v4().to_string();
+            warn!(quarantine = %id, epoch = %previous, paths = paths.len(), reason = %reason, "Quarantined local changes");
+            host.remake().add_quarantine(Quarantined {
+                id: id.clone(),
+                epoch: previous,
+                target,
+                created_at: unix_millis(),
+                reason,
+                paths,
+                document: old.take().expect("old document"),
+            });
+            quarantine_id = Some(id);
+            keep_archive = false; // the quarantine holds the document
+        }
+    }
+
+    if let (true, true, Some(document)) = (had_history, keep_archive, old) {
+        host.remake().archive(
+            previous,
+            document,
+            metadata.policy.archive_retention.clone(),
+            target.generation,
+        );
+    }
+
+    host.remake().set_status(
+        &metadata.remake_id,
+        LocalStatus {
+            switched_at: Some(unix_millis()),
+            replayed_paths: Some(stats.replayed),
+            failed_paths: Some(stats.failed),
+            quarantine: quarantine_id.clone(),
+            error: None,
+        },
+    );
+    host.remake().set_catch_up(None);
+    if was_ready {
+        host.resume().await;
+    }
+    // After resuming, so reads move from the stale view to the live document directly.
+    host.set_stale_view(None);
+
+    info!(
+        from = %previous,
+        to = %target,
+        own_changes = own_missing,
+        replayed = stats.replayed,
+        failed = stats.failed,
+        quarantine = ?quarantine_id,
+        secs = started.elapsed().as_secs(),
+        "Rejoined newer epoch"
+    );
+}
+
+/// Verifies the synced document and decides about the old document's own changes.
+fn prepare(mut copy: AutoCommit, mut old: AutoCommit, previous: Epoch, target: Epoch) -> Prepared {
+    if copy.get_change_by_hash(&ChangeHash(target.id)).is_none() {
+        return Prepared::NotYet(old);
+    }
+    let metadata = match snapshot::verify_document(&mut copy, target) {
+        Ok(metadata) => metadata,
+        Err(reason) => return Prepared::Failed(old, reason),
+    };
+
+    let old_heads = old.get_heads();
+    if old_heads.is_empty() {
+        return Prepared::Ready {
+            old,
+            metadata,
+            plan: Plan::Nothing,
+            own_missing: 0,
+            had_history: false,
+        };
+    }
+
+    // The epoch that followed ours records (per actor) which of our changes it took.
+    // Snapshots carry earlier epochs' metadata forward, so this works across any gap.
+    let next = snapshot::read_epoch_metadata(&copy, previous.generation + 1)
+        .ok()
+        .filter(|next| next.previous == previous);
+
+    let own = old.get_actor().clone();
+    let clock = next
+        .as_ref()
+        .map_or_else(BTreeMap::new, |next| next.clock.clone());
+    let missing = delta::missing_from_snapshot(&mut old, &clock, &own);
+    if missing.own == 0 {
+        return Prepared::Ready {
+            old,
+            metadata,
+            plan: Plan::Nothing,
+            own_missing: 0,
+            had_history: true,
+        };
+    }
+
+    let paths = delta::changed_paths(&mut old, &missing.before, &old_heads);
+    let gap = target.generation - previous.generation;
+
+    let plan = match &next {
+        None => Plan::Quarantine {
+            paths,
+            reason: format!(
+                "the metadata of the epoch after {previous} is missing, so it is unknown which of {} local changes the cluster has",
+                missing.own
+            ),
+        },
+        Some(next) => {
+            let away_ms = unix_millis().saturating_sub(next.proposed_at);
+            if next.policy.replay_window.covers(gap, away_ms) {
+                Plan::Replay(paths)
+            } else {
+                Plan::Quarantine {
+                    paths,
+                    reason: format!(
+                        "{} local changes are outside the replay window ({gap} epochs behind, cluster moved on {}s ago)",
+                        missing.own,
+                        away_ms / 1000
+                    ),
+                }
+            }
+        }
+    };
+
+    Prepared::Ready {
+        old,
+        metadata,
+        plan,
+        own_missing: missing.own,
+        had_history: true,
+    }
+}
+
+/// Restores the old document and epoch, and retries later.
+async fn abandon<H: Host>(host: &H, rejoin: Rejoin, reason: String) {
+    let Rejoin {
+        previous,
+        target,
+        old,
+        was_ready,
+        ..
+    } = rejoin;
+    warn!(target = %target, reason = %reason, "Rejoin failed; restoring the previous epoch");
+
+    if let Err(never) = host
+        .switch_epoch(previous, move |_live| Ok::<_, Infallible>(old))
+        .await
+    {
+        match never {}
+    }
+    // A node that was never ready must sync again before bootstrap may finish.
+    host.reset_initial_sync();
+    if was_ready {
+        host.resume().await;
+    }
+    host.set_stale_view(None);
+
+    let attempts = host
+        .remake()
+        .catch_up()
+        .filter(|catch_up| catch_up.target == target)
+        .map_or(1, |catch_up| catch_up.attempts);
+    host.remake().set_catch_up(Some(CatchUp {
+        target,
+        attempts,
+        last_attempt: Instant::now(),
+        error: Some(reason),
+    }));
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::remake::snapshot::{build, SnapshotParams};
+    use automerge::{transaction::Transactable, ActorId, ROOT};
+    use serde_json::{json, Value as JsonValue};
+
+    /// An old-epoch document with one own change after the base, and the cluster's
+    /// document of the next epoch (snapshot plus one foreign change).
+    fn scenario(policy: &JsonValue) -> (AutoCommit, AutoCommit, Epoch) {
+        let mut old = AutoCommit::new().with_actor(ActorId::from(&b"late-node"[..]));
+        old.put(ROOT, "base", 1_i64).unwrap();
+        old.commit();
+        let base = old.get_heads();
+
+        let snapshot = build(
+            &mut old,
+            &SnapshotParams {
+                remake_id: "c-1",
+                previous: Epoch::GENESIS,
+                base_heads: &base,
+                policy,
+                proposed_at: 0,
+            },
+        )
+        .unwrap();
+        let target = snapshot.epoch;
+
+        let mut cluster = snapshot.doc.with_actor(ActorId::from(&b"cluster-node"[..]));
+        cluster.put(ROOT, "theirs", 1_i64).unwrap();
+        cluster.commit();
+
+        old.put(ROOT, "mine", 1_i64).unwrap();
+        old.commit();
+        (old, cluster, target)
+    }
+
+    #[test]
+    fn replays_within_the_window() {
+        let (old, cluster, target) = scenario(&json!({}));
+        match prepare(cluster, old, Epoch::GENESIS, target) {
+            Prepared::Ready {
+                plan: Plan::Replay(paths),
+                own_missing,
+                ..
+            } => {
+                assert_eq!(own_missing, 1);
+                assert_eq!(paths, vec![vec!["mine".to_owned()]]);
+            }
+            _ => panic!("expected a replay"),
+        }
+    }
+
+    #[test]
+    fn quarantines_outside_the_window() {
+        let (old, cluster, target) = scenario(&json!({ "replayWindow": { "epochs": 0 } }));
+        match prepare(cluster, old, Epoch::GENESIS, target) {
+            Prepared::Ready {
+                plan: Plan::Quarantine { paths, .. },
+                ..
+            } => assert_eq!(paths, vec![vec!["mine".to_owned()]]),
+            _ => panic!("expected a quarantine"),
+        }
+    }
+
+    #[test]
+    fn new_node_has_nothing_to_replay() {
+        let (_, cluster, target) = scenario(&json!({}));
+        assert!(matches!(
+            prepare(cluster, AutoCommit::new(), Epoch::GENESIS, target),
+            Prepared::Ready {
+                plan: Plan::Nothing,
+                had_history: false,
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn waits_for_the_target_epoch() {
+        let (old, _, target) = scenario(&json!({}));
+        assert!(matches!(
+            prepare(AutoCommit::new(), old, Epoch::GENESIS, target),
+            Prepared::NotYet(_)
+        ));
+    }
+}
diff --git a/src/remake/coordinator.rs b/src/remake/coordinator.rs
new file mode 100644
index 0000000..b84b0a9
--- /dev/null
+++ b/src/remake/coordinator.rs
@@ -0,0 +1,183 @@
+//! The coordinator: the node that received the proposal watches the cluster during the
+//! settle period and then writes the decision.
+//!
+//! It aborts if, at any point before deciding:
+//! * an operator requested an abort,
+//! * a competing proposal for the same epoch appeared (both coordinators abort),
+//! * a peer it considered went dormant (a partition may hide a competitor),
+//! * the local epoch changed, or
+//! * the decision would come later than the settle period plus the grace margin
+//!   (other nodes report the proposal as expired by then).
+//!
+//! Otherwise it builds the snapshot at the base heads and commits, recording the new
+//! epoch so every node can verify its own build against it.
+//!
+//! The task lives only in memory. If the coordinator dies, nobody decides and the
+//! proposal expires.
+
+use std::collections::HashMap;
+use std::time::Duration;
+
+use tracing::{info, warn};
+
+use super::records::{self, Decision, Record};
+use super::snapshot;
+use super::Host;
+use crate::document::store::StoreError;
+
+const TICK: Duration = Duration::from_secs(1);
+
+pub(crate) async fn run<H: Host>(host: H, id: String) {
+    info!(remake = %id, "Coordinating remake");
+
+    loop {
+        tokio::time::sleep(TICK).await;
+
+        match step(&host, &id).await {
+            Ok(true) => return,
+            Ok(false) => {}
+            Err(error) => {
+                warn!(remake = %id, error = %format!("{error:#}"), "Coordinator step failed; retrying");
+            }
+        }
+    }
+}
+
+/// One evaluation. Returns true once the coordinator is done.
+async fn step<H: Host>(host: &H, id: &str) -> Result<bool, StoreError> {
+    let wanted = id.to_owned();
+    let (record, all) = host
+        .read_doc(move |doc| (records::load(doc, &wanted), records::load_all(doc)))
+        .await;
+
+    let Some(record) = record else {
+        warn!(remake = %id, "Remake record vanished; stopping coordinator");
+        return Ok(true);
+    };
+    if record.decision.is_some() {
+        return Ok(true);
+    }
+
+    let now = records::unix_millis();
+    let grace = host.dormant_after().as_millis() as u64;
+
+    if let Some(reason) = abort_reason(host, id, &record, &all, now, grace).await {
+        decide_abort(host, id, reason).await?;
+        return Ok(true);
+    }
+
+    if now < record.proposal.settle_until {
+        return Ok(false);
+    }
+
+    commit(host, id, &record).await?;
+    Ok(true)
+}
+
+async fn abort_reason<H: Host>(
+    host: &H,
+    id: &str,
+    record: &Record,
+    all: &[(String, Record)],
+    now: u64,
+    grace: u64,
+) -> Option<String> {
+    let proposal = &record.proposal;
+
+    if let Some(request) = &record.abort_requested {
+        return Some(format!("aborted by operator via {}", request.by));
+    }
+
+    if records::EpochRecord::from(host.current_epoch()) != proposal.from_epoch {
+        return Some("local epoch changed during settle".to_owned());
+    }
+
+    if let Some((other, _)) = all.iter().find(|(other, candidate)| {
+        other != id
+            && candidate.proposal.from_epoch == proposal.from_epoch
+            && candidate.phase(now, grace).is_open()
+    }) {
+        return Some(format!("competing remake {other} for the same epoch"));
+    }
+
+    // Peers forgotten in the meantime are ignored: an operator decided they are gone.
+    let peers: HashMap<String, bool> = host
+        .peer_summaries()
+        .await
+        .into_iter()
+        .map(|peer| (peer.id, peer.active))
+        .collect();
+    let dormant: Vec<&str> = proposal
+        .peers_considered
+        .iter()
+        .filter(|peer| peers.get(peer.as_str()) == Some(&false))
+        .map(String::as_str)
+        .collect();
+    if !dormant.is_empty() {
+        return Some(format!(
+            "peers went dormant during settle: {}",
+            dormant.join(", ")
+        ));
+    }
+
+    if now > proposal.settle_until.saturating_add(grace) {
+        return Some("decision deadline passed".to_owned());
+    }
+
+    None
+}
+
+async fn commit<H: Host>(host: &H, id: &str, record: &Record) -> Result<(), StoreError> {
+    let Some(request) = record.snapshot_request(id) else {
+        return decide_abort(
+            host,
+            id,
+            "unreadable epoch or base heads in proposal".into(),
+        )
+        .await;
+    };
+
+    // Clone under the read lock, build without any lock.
+    let copy = host.read_doc(|doc| doc.clone()).await;
+    let mut snapshot = match snapshot::build_detached(copy, request).await {
+        Ok((snapshot, _copy)) => snapshot,
+        Err(error) => {
+            return decide_abort(host, id, format!("snapshot build failed: {error}")).await;
+        }
+    };
+
+    let to_epoch = snapshot.epoch;
+    let snapshot_bytes = snapshot.doc.save().len() as u64;
+
+    let decision = Decision::Committed {
+        to_epoch: to_epoch.into(),
+        snapshot_bytes,
+        decided_at: records::unix_millis(),
+    };
+    write_decision(host, id, decision).await?;
+
+    // The local switch reuses it instead of building again.
+    host.remake().store_prebuilt(id, snapshot);
+
+    info!(remake = %id, to_epoch = %to_epoch, snapshot_bytes, "Remake committed");
+    Ok(())
+}
+
+async fn decide_abort<H: Host>(host: &H, id: &str, reason: String) -> Result<(), StoreError> {
+    warn!(remake = %id, reason = %reason, "Remake aborted");
+    write_decision(
+        host,
+        id,
+        Decision::Aborted {
+            reason,
+            decided_at: records::unix_millis(),
+        },
+    )
+    .await
+}
+
+async fn write_decision<H: Host>(host: &H, id: &str, decision: Decision) -> Result<(), StoreError> {
+    let id = id.to_owned();
+    host.commit_doc(move |doc| records::write_decision(doc, &id, &decision))
+        .await
+}
diff --git a/src/remake/delta.rs b/src/remake/delta.rs
new file mode 100644
index 0000000..b07c862
--- /dev/null
+++ b/src/remake/delta.rs
@@ -0,0 +1,387 @@
+//! Replaying a node's changes after the base into the new epoch.
+//!
+//! Changes from the old epoch can't be applied to the new document (their history is
+//! unrelated), so they are replayed as state: for every path that changed between two
+//! sets of heads, the new document gets whatever the old document's merged view holds
+//! there now (or a delete if nothing). Replayed values are fresh operations, so they win
+//! over what the snapshot holds at the same path.
+//!
+//! Granularity:
+//! * Map entries are replayed per key, so changes to sibling keys made on different
+//!   nodes don't overwrite each other.
+//! * Lists and text are replayed whole: concurrent edits of the same list or text on two
+//!   nodes around a remake resolve as last-writer-wins.
+//! * Edits inside a map that lost a merge conflict produce no patch in Automerge and are
+//!   not replayed. Such maps only exist briefly between concurrent creations.
+//!
+//! The protocol's own metadata (`_continuum/remakes`, `_continuum/epochs`) is never
+//! replayed: the snapshot owns it.
+
+use std::collections::{BTreeMap, HashSet};
+
+use automerge::{ActorId, AutoCommit, ChangeHash, PatchAction, Prop};
+use tracing::warn;
+
+use super::snapshot::{EPOCHS, META_ROOT, REMAKES};
+use crate::document::store::{self, StoreError};
+
+#[derive(Debug, Default, Clone, Copy)]
+pub(crate) struct ReplayStats {
+    pub(crate) replayed: usize,
+    pub(crate) failed: usize,
+}
+
+/// Replays into `dst` everything that changed in `src` between `before` and `after`,
+/// as one commit. Both head sets must be in `src`.
+pub(crate) fn replay(
+    src: &mut AutoCommit,
+    before: &[ChangeHash],
+    after: &[ChangeHash],
+    dst: &mut AutoCommit,
+) -> ReplayStats {
+    if before == after {
+        return ReplayStats::default();
+    }
+
+    let paths = changed_paths(src, before, after);
+    replay_paths(src, dst, &paths)
+}
+
+/// Replays the state of `paths` (from [`changed_paths`]) from `src` into `dst`, as one
+/// commit. Lets callers find the paths outside a lock and copy them inside it.
+pub(crate) fn replay_paths(
+    src: &AutoCommit,
+    dst: &mut AutoCommit,
+    paths: &[Vec<String>],
+) -> ReplayStats {
+    let mut stats = ReplayStats::default();
+    for path in paths {
+        replay_path(src, dst, path, &mut stats);
+    }
+    let _ = dst.commit();
+    stats
+}
+
+fn replay_path(src: &AutoCommit, dst: &mut AutoCommit, path: &[String], stats: &mut ReplayStats) {
+    if is_protocol_path(path) {
+        return;
+    }
+
+    // Replacing an ancestor of the protocol metadata as a whole would clobber it; replay
+    // its children one by one instead.
+    if contains_protocol_path(path) {
+        let mut keys = store::keys_at(src, path).unwrap_or_default();
+        keys.extend(store::keys_at(dst, path).unwrap_or_default());
+        keys.sort();
+        keys.dedup();
+        for key in keys {
+            let mut child = path.to_vec();
+            child.push(key);
+            replay_path(src, dst, &child, stats);
+        }
+        return;
+    }
+
+    match store::copy_path(src, dst, path) {
+        Ok(()) => stats.replayed += 1,
+        Err(error) => {
+            log_failure(path, &error);
+            stats.failed += 1;
+        }
+    }
+}
+
+fn log_failure(path: &[String], error: &StoreError) {
+    warn!(path = ?path, error = %error, "Could not replay change into the new epoch");
+}
+
+fn is_protocol_path(path: &[String]) -> bool {
+    matches!(path, [root, part, ..] if root == META_ROOT && (part == REMAKES || part == EPOCHS))
+}
+
+/// True for the root and for `_continuum` itself.
+fn contains_protocol_path(path: &[String]) -> bool {
+    matches!(path, [] | [_]) && path.first().is_none_or(|root| root == META_ROOT)
+}
+
+/// What a late node has that a snapshot lacks.
+#[derive(Debug)]
+pub(crate) struct Missing {
+    /// Heads of everything except `own` changes the snapshot lacks (and changes built on
+    /// them). Replaying from these heads to the document's heads replays exactly those.
+    pub(crate) before: Vec<ChangeHash>,
+    /// Changes by `own` that the snapshot lacks.
+    pub(crate) own: usize,
+}
+
+/// Splits `doc`'s history against a snapshot `clock`: a change (actor, seq) is in the
+/// snapshot iff seq <= clock[actor].
+///
+/// Only changes this node wrote itself (actor `own`) are worth replaying: other nodes'
+/// changes after the base were replayed by those nodes when they switched, and replaying
+/// them again would overwrite newer writes in the new epoch with stale ones.
+pub(crate) fn missing_from_snapshot(
+    doc: &mut AutoCommit,
+    clock: &BTreeMap<String, u64>,
+    own: &ActorId,
+) -> Missing {
+    let own_hex = own.to_string();
+    let changes: Vec<(ChangeHash, Vec<ChangeHash>, bool)> = doc
+        .get_changes(&[])
+        .iter()
+        .map(|change| {
+            let actor = change.actor_id().to_string();
+            let in_snapshot = clock.get(&actor).is_some_and(|&seq| change.seq() <= seq);
+            let own_missing = actor == own_hex && !in_snapshot;
+            (change.hash(), change.deps().to_vec(), own_missing)
+        })
+        .collect();
+
+    // Excluded: own missing changes and everything that (transitively) depends on them.
+    // A fixpoint, so it doesn't rely on the order `get_changes` returns.
+    let mut excluded: HashSet<ChangeHash> = changes
+        .iter()
+        .filter(|(_, _, own_missing)| *own_missing)
+        .map(|(hash, _, _)| *hash)
+        .collect();
+    let own_count = excluded.len();
+    loop {
+        let before = excluded.len();
+        for (hash, deps, _) in &changes {
+            if !excluded.contains(hash) && deps.iter().any(|dep| excluded.contains(dep)) {
+                excluded.insert(*hash);
+            }
+        }
+        if excluded.len() == before {
+            break;
+        }
+    }
+
+    // Heads of the kept part: kept changes no kept change depends on.
+    let kept: Vec<&(ChangeHash, Vec<ChangeHash>, bool)> = changes
+        .iter()
+        .filter(|(hash, _, _)| !excluded.contains(hash))
+        .collect();
+    let depended_on: HashSet<ChangeHash> = kept
+        .iter()
+        .flat_map(|(_, deps, _)| deps.iter().copied())
+        .collect();
+    let mut before: Vec<ChangeHash> = kept
+        .iter()
+        .map(|(hash, _, _)| *hash)
+        .filter(|hash| !depended_on.contains(hash))
+        .collect();
+    before.sort();
+
+    Missing {
+        before,
+        own: own_count,
+    }
+}
+
+/// The minimal set of store paths covering every change between `before` and `after`:
+/// sorted, and no path is a prefix of another.
+pub(crate) fn changed_paths(
+    doc: &mut AutoCommit,
+    before: &[ChangeHash],
+    after: &[ChangeHash],
+) -> Vec<Vec<String>> {
+    let mut paths: Vec<Vec<String>> = doc.diff(before, after).iter().map(patch_path).collect();
+    paths.sort();
+    paths.dedup();
+
+    // Sorted order puts a prefix right before its extensions, so comparing with the
+    // last kept path is enough.
+    let mut minimal: Vec<Vec<String>> = Vec::new();
+    for path in paths {
+        if minimal.last().is_some_and(|kept| path.starts_with(kept)) {
+            continue;
+        }
+        minimal.push(path);
+    }
+    minimal
+}
+
+fn patch_path(patch: &automerge::Patch) -> Vec<String> {
+    let mut path = Vec::new();
+    for (_, prop) in &patch.path {
+        match prop {
+            Prop::Map(key) => path.push(key.clone()),
+            // Inside a list: the list is replayed as a whole.
+            Prop::Seq(_) => return path,
+        }
+    }
+
+    match &patch.action {
+        PatchAction::PutMap { key, .. } | PatchAction::DeleteMap { key, .. } => {
+            path.push(key.clone())
+        }
+        PatchAction::Increment {
+            prop: Prop::Map(key),
+            ..
+        } => path.push(key.clone()),
+        // List and text edits: the containing object.
+        _ => {}
+    }
+    path
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, ROOT};
+    use serde_json::json;
+
+    fn view(doc: &AutoCommit) -> serde_json::Value {
+        store::read_path(doc, &[]).unwrap().unwrap()
+    }
+
+    #[test]
+    fn minimal_paths() {
+        let mut doc = AutoCommit::new();
+        doc.put(ROOT, "untouched", 1_i64).unwrap();
+        let before = doc.get_heads();
+
+        let m = doc.put_object(ROOT, "m", ObjType::Map).unwrap();
+        doc.put(&m, "a", 1_i64).unwrap();
+        doc.put(ROOT, "x", 1_i64).unwrap();
+        doc.put(ROOT, "c", ScalarValue::counter(0)).unwrap();
+        doc.commit();
+        let middle = doc.get_heads();
+        doc.increment(ROOT, "c", 2).unwrap();
+        let list = doc.put_object(ROOT, "l", ObjType::List).unwrap();
+        doc.insert(&list, 0, "e").unwrap();
+        doc.commit();
+        let after = doc.get_heads();
+
+        let all = changed_paths(&mut doc, &before, &after);
+        assert_eq!(
+            all,
+            vec![
+                vec!["c".to_owned()],
+                vec!["l".to_owned()],
+                vec!["m".to_owned()],
+                vec!["x".to_owned()],
+            ]
+        );
+
+        let later = changed_paths(&mut doc, &middle, &after);
+        assert_eq!(later, vec![vec!["c".to_owned()], vec!["l".to_owned()]]);
+    }
+
+    #[test]
+    fn replays_state_per_key() {
+        // Base shared by the old document and the snapshot.
+        let mut old = AutoCommit::new();
+        let m = old.put_object(ROOT, "m", ObjType::Map).unwrap();
+        old.put(&m, "keep", "base").unwrap();
+        old.put(&m, "change", "base").unwrap();
+        old.put(ROOT, "gone", true).unwrap();
+        old.commit();
+        let base = old.get_heads();
+
+        let mut new = AutoCommit::new();
+        store::materialize_into(&old, &mut new, &|_| false).unwrap();
+        // Something the new epoch already holds that the delta must not clobber.
+        let Some((_, m_new)) = new.get(ROOT, "m").unwrap() else {
+            panic!("m missing");
+        };
+        new.put(&m_new, "keep", "new epoch").unwrap();
+        new.commit();
+
+        // The delta: one key changed, one key deleted.
+        old.put(&m, "change", "delta").unwrap();
+        old.delete(ROOT, "gone").unwrap();
+        old.commit();
+        let after = old.get_heads();
+
+        let stats = replay(&mut old, &base, &after, &mut new);
+        assert_eq!(stats.failed, 0);
+        assert_eq!(
+            view(&new),
+            json!({ "m": { "keep": "new epoch", "change": "delta" } })
+        );
+    }
+
+    #[test]
+    fn missing_counts_only_own_changes() {
+        let own = ActorId::from(&b"own-actor"[..]);
+        let other = ActorId::from(&b"other-actor"[..]);
+
+        let mut doc = AutoCommit::new().with_actor(own.clone());
+        doc.put(ROOT, "base", 1_i64).unwrap();
+        doc.commit();
+        let base = doc.get_heads();
+
+        // The snapshot contains the base only.
+        let clock: BTreeMap<String, u64> = [(own.to_string(), 1)].into_iter().collect();
+
+        // A foreign change after the base (replayed by its own node) ...
+        let mut foreign = doc.fork().with_actor(other);
+        foreign.put(ROOT, "theirs", 1_i64).unwrap();
+        foreign.commit();
+        doc.merge(&mut foreign).unwrap();
+        let with_foreign = doc.get_heads();
+
+        // ... and one of ours.
+        doc.put(ROOT, "mine", 1_i64).unwrap();
+        doc.commit();
+        let after = doc.get_heads();
+
+        let missing = missing_from_snapshot(&mut doc, &clock, &own);
+        assert_eq!(missing.own, 1);
+        assert_eq!(missing.before, with_foreign);
+        assert_ne!(missing.before, base);
+        assert_eq!(
+            changed_paths(&mut doc, &missing.before, &after),
+            vec![vec!["mine".to_owned()]]
+        );
+
+        // Nothing of ours missing: nothing to replay.
+        let full: BTreeMap<String, u64> = [(own.to_string(), 2)].into_iter().collect();
+        let none = missing_from_snapshot(&mut doc, &full, &own);
+        assert_eq!(none.own, 0);
+        let mut heads = after.clone();
+        heads.sort();
+        assert_eq!(none.before, heads);
+    }
+
+    #[test]
+    fn protocol_metadata_is_never_replayed() {
+        let mut old = AutoCommit::new();
+        old.put(ROOT, "x", 1_i64).unwrap();
+        old.commit();
+        let base = old.get_heads();
+
+        let mut new = AutoCommit::new();
+        store::write_path(
+            &mut new,
+            &[META_ROOT.into(), EPOCHS.into(), "1".into()],
+            &json!({ "generation": 1 }),
+        )
+        .unwrap();
+        new.commit();
+
+        // The old epoch replaces `_continuum` wholesale and writes a remake record.
+        store::write_path(
+            &mut old,
+            &[META_ROOT.into()],
+            &json!({ "forgotten": { "pk": 1 } }),
+        )
+        .unwrap();
+        store::write_path(
+            &mut old,
+            &[META_ROOT.into(), REMAKES.into(), "c".into()],
+            &json!({ "x": 1 }),
+        )
+        .unwrap();
+        old.commit();
+        let after = old.get_heads();
+
+        replay(&mut old, &base, &after, &mut new);
+        let meta = &view(&new)[META_ROOT];
+        assert_eq!(meta[EPOCHS]["1"]["generation"], json!(1));
+        assert_eq!(meta["forgotten"]["pk"], json!(1));
+        assert!(meta.get(REMAKES).is_none());
+    }
+}
diff --git a/src/remake/local.rs b/src/remake/local.rs
new file mode 100644
index 0000000..35a4ff1
--- /dev/null
+++ b/src/remake/local.rs
@@ -0,0 +1,320 @@
+//! Per-node, in-memory remake state: snapshots built ahead of the switch, switch
+//! progress for reports, a rejoin in progress, quarantined changes, and archived
+//! documents of previous epochs.
+//!
+//! Nothing here is replicated or durable. After a restart the node simply has no
+//! archive or quarantine, and rebuilds snapshots when needed.
+
+use std::collections::HashMap;
+use std::sync::{Mutex, MutexGuard};
+
+use automerge::AutoCommit;
+use serde::Serialize;
+use tokio::time::Instant;
+
+use super::records::{unix_millis, Retention};
+use super::snapshot::Snapshot;
+use crate::document::replica::Epoch;
+
+#[derive(Default)]
+pub(crate) struct RemakeState {
+    // A std mutex: every access is short and never spans an await.
+    inner: Mutex<Inner>,
+}
+
+#[derive(Default)]
+struct Inner {
+    /// Built by the coordinator at commit time; consumed by the local switch.
+    prebuilt: HashMap<String, Snapshot>,
+    /// When this node first saw a remake committed.
+    first_seen: HashMap<String, Instant>,
+    status: HashMap<String, LocalStatus>,
+    archive: Vec<Archived>,
+    /// Last attempt to catch up with a newer epoch.
+    catch_up: Option<CatchUp>,
+    /// A rejoin in progress: the node syncs an empty document of a newer epoch.
+    rejoin: Option<Rejoin>,
+    quarantine: Vec<Quarantined>,
+}
+
+/// Catching up with a newer epoch: attempts and the last failure.
+#[derive(Clone, Debug)]
+pub(crate) struct CatchUp {
+    pub(crate) target: Epoch,
+    pub(crate) attempts: u32,
+    pub(crate) last_attempt: Instant,
+    pub(crate) error: Option<String>,
+}
+
+/// The node left `previous` and is syncing `target` from scratch.
+pub(crate) struct Rejoin {
+    pub(crate) previous: Epoch,
+    pub(crate) target: Epoch,
+    /// The document of `previous`, until it is replayed, quarantined or restored.
+    pub(crate) old: AutoCommit,
+    pub(crate) started: Instant,
+    /// The node served requests before; it does again once the rejoin is over.
+    pub(crate) was_ready: bool,
+}
+
+/// Changes of this node that were not replayed into the current epoch.
+pub(crate) struct Quarantined {
+    pub(crate) id: String,
+    /// The epoch the changes were made in, and the one the node rejoined.
+    pub(crate) epoch: Epoch,
+    pub(crate) target: Epoch,
+    pub(crate) created_at: u64,
+    pub(crate) reason: String,
+    /// Store paths the changes touched (minimal, prefix-free).
+    pub(crate) paths: Vec<Vec<String>>,
+    /// The old document; `paths` are read from its merged view.
+    pub(crate) document: AutoCommit,
+}
+
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct CatchUpReport {
+    pub(crate) target: String,
+    pub(crate) attempts: u32,
+    /// True while the node syncs the target epoch (not serving requests).
+    pub(crate) rejoining: bool,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) error: Option<String>,
+}
+
+/// What happened to a remake on this node.
+#[derive(Clone, Debug, Default, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct LocalStatus {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) switched_at: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) replayed_paths: Option<usize>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) failed_paths: Option<usize>,
+    /// ID of the quarantine entry holding this node's changes that weren't replayed.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) quarantine: Option<String>,
+    /// Why the switch did not happen (yet).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub(crate) error: Option<String>,
+}
+
+struct Archived {
+    epoch: Epoch,
+    // Kept for auditing and recovery by hand; not read by the protocol.
+    #[allow(dead_code)]
+    doc: AutoCommit,
+    archived_at: u64,
+    retention: Retention,
+}
+
+/// Summary of an archived epoch.
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct ArchivedEpoch {
+    pub(crate) generation: u64,
+    pub(crate) id: String,
+    pub(crate) archived_at: u64,
+}
+
+impl RemakeState {
+    fn lock(&self) -> MutexGuard<'_, Inner> {
+        self.inner
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner())
+    }
+
+    pub(crate) fn store_prebuilt(&self, id: &str, snapshot: Snapshot) {
+        self.lock().prebuilt.insert(id.to_owned(), snapshot);
+    }
+
+    pub(crate) fn take_prebuilt(&self, id: &str) -> Option<Snapshot> {
+        self.lock().prebuilt.remove(id)
+    }
+
+    /// When this node first saw the remake committed (now, on the first call).
+    pub(crate) fn first_seen(&self, id: &str) -> Instant {
+        *self
+            .lock()
+            .first_seen
+            .entry(id.to_owned())
+            .or_insert_with(Instant::now)
+    }
+
+    pub(crate) fn status(&self, id: &str) -> Option<LocalStatus> {
+        self.lock().status.get(id).cloned()
+    }
+
+    pub(crate) fn set_status(&self, id: &str, status: LocalStatus) {
+        self.lock().status.insert(id.to_owned(), status);
+    }
+
+    /// Keeps the previous epoch's document, then drops what has outlived its retention.
+    pub(crate) fn archive(
+        &self,
+        epoch: Epoch,
+        doc: AutoCommit,
+        retention: Retention,
+        current_generation: u64,
+    ) {
+        let mut inner = self.lock();
+        inner.archive.push(Archived {
+            epoch,
+            doc,
+            archived_at: unix_millis(),
+            retention,
+        });
+        prune(&mut inner.archive, current_generation);
+    }
+
+    pub(crate) fn prune_archive(&self, current_generation: u64) {
+        prune(&mut self.lock().archive, current_generation);
+    }
+
+    pub(crate) fn catch_up(&self) -> Option<CatchUp> {
+        self.lock().catch_up.clone()
+    }
+
+    pub(crate) fn set_catch_up(&self, catch_up: Option<CatchUp>) {
+        self.lock().catch_up = catch_up;
+    }
+
+    pub(crate) fn catch_up_report(&self) -> Option<CatchUpReport> {
+        let inner = self.lock();
+        let rejoining = inner.rejoin.is_some();
+        inner.catch_up.as_ref().map(|catch_up| CatchUpReport {
+            target: catch_up.target.to_string(),
+            attempts: catch_up.attempts,
+            rejoining,
+            error: catch_up.error.clone(),
+        })
+    }
+
+    pub(crate) fn is_rejoining(&self) -> bool {
+        self.lock().rejoin.is_some()
+    }
+
+    /// Progress of the rejoin in progress: (target, started).
+    pub(crate) fn rejoin_progress(&self) -> Option<(Epoch, Instant)> {
+        self.lock()
+            .rejoin
+            .as_ref()
+            .map(|rejoin| (rejoin.target, rejoin.started))
+    }
+
+    pub(crate) fn start_rejoin(&self, rejoin: Rejoin) {
+        self.lock().rejoin = Some(rejoin);
+    }
+
+    pub(crate) fn take_rejoin(&self) -> Option<Rejoin> {
+        self.lock().rejoin.take()
+    }
+
+    pub(crate) fn add_quarantine(&self, quarantined: Quarantined) {
+        self.lock().quarantine.push(quarantined);
+    }
+
+    /// Runs `f` on the quarantined entry with this ID.
+    pub(crate) fn with_quarantined<T>(
+        &self,
+        id: &str,
+        f: impl FnOnce(&Quarantined) -> T,
+    ) -> Option<T> {
+        self.lock()
+            .quarantine
+            .iter()
+            .find(|entry| entry.id == id)
+            .map(f)
+    }
+
+    pub(crate) fn list_quarantine<T>(&self, f: impl Fn(&Quarantined) -> T) -> Vec<T> {
+        self.lock().quarantine.iter().map(f).collect()
+    }
+
+    pub(crate) fn take_quarantined(&self, id: &str) -> Option<Quarantined> {
+        let mut inner = self.lock();
+        let index = inner.quarantine.iter().position(|entry| entry.id == id)?;
+        Some(inner.quarantine.remove(index))
+    }
+
+    /// Puts an entry back, e.g. after a failed replay.
+    pub(crate) fn restore_quarantined(&self, quarantined: Quarantined) {
+        self.lock().quarantine.push(quarantined);
+    }
+
+    pub(crate) fn archived_epochs(&self) -> Vec<ArchivedEpoch> {
+        self.lock()
+            .archive
+            .iter()
+            .map(|archived| ArchivedEpoch {
+                generation: archived.epoch.generation,
+                id: archived.epoch.id_hex(),
+                archived_at: archived.archived_at,
+            })
+            .collect()
+    }
+}
+
+/// An archive is kept until every limit of its retention has passed.
+fn prune(archive: &mut Vec<Archived>, current_generation: u64) {
+    let now = unix_millis();
+    archive.retain(|archived| {
+        archived.retention.covers(
+            current_generation.saturating_sub(archived.epoch.generation),
+            now.saturating_sub(archived.archived_at),
+        )
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn archive_retention() {
+        let remake = RemakeState::default();
+        let epoch = |generation| Epoch {
+            generation,
+            id: [generation as u8; 32],
+        };
+
+        // Kept by epoch count only.
+        remake.archive(
+            epoch(0),
+            AutoCommit::new(),
+            Retention {
+                epochs: Some(1),
+                seconds: None,
+            },
+            1,
+        );
+        assert_eq!(remake.archived_epochs().len(), 1);
+        remake.prune_archive(2);
+        assert!(remake.archived_epochs().is_empty());
+
+        // Kept by time while the epoch count has passed.
+        remake.archive(
+            epoch(1),
+            AutoCommit::new(),
+            Retention {
+                epochs: Some(0),
+                seconds: Some(3600),
+            },
+            5,
+        );
+        assert_eq!(remake.archived_epochs().len(), 1);
+
+        // No limits at all: not kept.
+        remake.archive(
+            epoch(2),
+            AutoCommit::new(),
+            Retention {
+                epochs: None,
+                seconds: None,
+            },
+            3,
+        );
+        assert_eq!(remake.archived_epochs().len(), 1);
+    }
+}
diff --git a/src/remake/mod.rs b/src/remake/mod.rs
new file mode 100644
index 0000000..d971fcf
--- /dev/null
+++ b/src/remake/mod.rs
@@ -0,0 +1,150 @@
+//! Remakes: coordinated moves of the whole cluster to a new document, a new epoch.
+//! Compaction is one: the new document has the old one's content without its history.
+//!
+//! * `snapshot`: deterministic build of the next epoch's initial document.
+//! * `records`: the replicated remake records under `_continuum/remakes/<id>`.
+//! * `coordinator`: proposal, settle period and decision on the node that was asked.
+//! * `switch`: switching this node to a committed epoch (worker).
+//! * `catchup`: rejoining a newer epoch from peers (late, new or failed nodes), then
+//!   replaying or quarantining this node's own missing changes.
+//! * `quarantine`: inspecting, replaying or discarding quarantined changes.
+//! * `delta`: replaying this node's changes since the base into the new epoch.
+//! * `local`: per-node in-memory state (prebuilt snapshots, switch status, archive).
+//! * `api`: `/admin/remakes` handlers.
+//!
+//! The rest of the node is reached exclusively through [`Host`], so this module never
+//! depends on `AppState`, peers or sync internals.
+
+mod api;
+mod catchup;
+mod coordinator;
+mod delta;
+mod local;
+mod quarantine;
+mod records;
+pub(crate) mod snapshot;
+mod switch;
+
+pub(crate) use local::RemakeState;
+pub(crate) use switch::run_worker;
+
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
+
+use automerge::AutoCommit;
+use axum::{
+    routing::{get, post},
+    Router,
+};
+use tokio::sync::broadcast;
+
+use crate::document::replica::Epoch;
+use crate::swarm::sync::ChangeSet;
+
+/// What remake needs from the node. Implemented by `AppState`.
+///
+/// Used generically (`H: Host`), never as `dyn Host`, so the generic methods are fine.
+pub(crate) trait Host: Clone + Send + Sync + 'static {
+    /// This node's in-memory remake state.
+    fn remake(&self) -> &RemakeState;
+
+    /// This node's peer ID.
+    fn node_id(&self) -> String;
+
+    /// The epoch of the local replica.
+    fn current_epoch(&self) -> Epoch;
+
+    /// Lower bound and default basis for the settle period.
+    fn dormant_after(&self) -> Duration;
+
+    fn is_node_ready(&self) -> impl Future<Output = bool> + Send;
+
+    /// Known (not forgotten) peers.
+    fn peer_summaries(&self) -> impl Future<Output = Vec<PeerSummary>> + Send;
+
+    /// Stops serving requests (back to `Joining`, keeping the cluster ID), as while a
+    /// rejoin syncs a fresh document. Returns whether the node was serving.
+    fn suspend(&self) -> impl Future<Output = bool> + Send;
+
+    /// Serves requests again after [`Host::suspend`].
+    fn resume(&self) -> impl Future<Output = ()> + Send;
+
+    /// Serves reads from this document (read-only, marked stale) while the node is not
+    /// ready; `None` ends it.
+    fn set_stale_view(&self, view: Option<(Epoch, AutoCommit)>);
+
+    /// Whether the document caught up with a peer since the last reset.
+    fn initial_sync_done(&self) -> bool;
+
+    /// Makes the next sync count as initial again.
+    fn reset_initial_sync(&self);
+
+    /// Store paths touched by each published document change.
+    fn subscribe_changes(&self) -> broadcast::Receiver<Arc<ChangeSet>>;
+
+    /// Reads the document on the blocking pool.
+    fn read_doc<T, F>(&self, f: F) -> impl Future<Output = T> + Send
+    where
+        F: FnOnce(&AutoCommit) -> T + Send + 'static,
+        T: Send + 'static;
+
+    /// One committed change (rolled back on error), published and replicated.
+    fn commit_doc<T, E, F>(&self, f: F) -> impl Future<Output = Result<T, E>> + Send
+    where
+        F: FnOnce(&mut AutoCommit) -> Result<T, E> + Send + 'static,
+        T: Send + 'static,
+        E: Send + 'static;
+
+    /// Replaces the document and its epoch atomically. Under the write lock, `f` gets
+    /// the live document and returns the new one. Afterwards the node resets every
+    /// peer's sync state and publishes the change. Returns the old document.
+    fn switch_epoch<E, F>(
+        &self,
+        epoch: Epoch,
+        f: F,
+    ) -> impl Future<Output = Result<AutoCommit, E>> + Send
+    where
+        F: FnOnce(&mut AutoCommit) -> Result<AutoCommit, E> + Send + 'static,
+        E: Send + 'static;
+}
+
+#[derive(Clone, Debug)]
+pub(crate) struct PeerSummary {
+    pub(crate) id: String,
+    /// Not dormant.
+    pub(crate) active: bool,
+    /// The epoch the peer last advertised.
+    pub(crate) epoch: Option<Epoch>,
+}
+
+/// Admin routes. Merged into the authenticated part of the node's router.
+pub(crate) fn router<H: Host>() -> Router<H> {
+    Router::new()
+        .route(
+            "/admin/remakes",
+            get(api::list::<H>)
+                .post(api::start::<H>)
+                .delete(api::abort_current::<H>),
+        )
+        // The original endpoint, kept as a shortcut for a compacting remake.
+        .route("/admin/compaction", post(api::start::<H>))
+        .route(
+            "/admin/remakes/{id}",
+            get(api::get::<H>).delete(api::abort::<H>),
+        )
+        .route("/admin/epochs", get(api::epochs::<H>))
+        .route("/admin/quarantine", get(quarantine::list::<H>))
+        .route(
+            "/admin/quarantine/{id}",
+            get(quarantine::get::<H>).delete(quarantine::discard::<H>),
+        )
+        .route(
+            "/admin/quarantine/{id}/document",
+            get(quarantine::document::<H>),
+        )
+        .route(
+            "/admin/quarantine/{id}/replay",
+            post(quarantine::replay::<H>),
+        )
+}
diff --git a/src/remake/quarantine.rs b/src/remake/quarantine.rs
new file mode 100644
index 0000000..5643549
--- /dev/null
+++ b/src/remake/quarantine.rs
@@ -0,0 +1,160 @@
+//! `/admin/quarantine`: this node's own changes that were not replayed when it rejoined
+//! a newer epoch (outside the replay window, or impossible to tell apart).
+//!
+//! * `GET /admin/quarantine` lists the entries.
+//! * `GET /admin/quarantine/{id}` shows the writes: every changed path with the value
+//!   the old document's merged view holds there (`deleted` if none).
+//! * `GET /admin/quarantine/{id}/document` downloads the old document
+//!   (`AutoCommit::save()`), for inspection with other tools.
+//! * `POST /admin/quarantine/{id}/replay` writes them into the current epoch anyway (as
+//!   fresh writes, so they win over what is there) and removes the entry.
+//! * `DELETE /admin/quarantine/{id}` discards the entry.
+//!
+//! Quarantine lives in memory on the node that holds it.
+
+use std::convert::Infallible;
+
+use axum::{
+    extract::{Path, State},
+    http::{header, StatusCode},
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::Serialize;
+use serde_json::{json, Value as JsonValue};
+use tracing::info;
+
+use super::api::error_response;
+use super::delta;
+use super::local::Quarantined;
+use super::Host;
+use crate::document::store;
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct Summary {
+    id: String,
+    /// The epoch the changes were made in.
+    epoch: String,
+    /// The epoch the node rejoined.
+    target: String,
+    created_at: u64,
+    reason: String,
+    paths: usize,
+}
+
+fn summary(entry: &Quarantined) -> Summary {
+    Summary {
+        id: entry.id.clone(),
+        epoch: entry.epoch.to_string(),
+        target: entry.target.to_string(),
+        created_at: entry.created_at,
+        reason: entry.reason.clone(),
+        paths: entry.paths.len(),
+    }
+}
+
+fn not_found() -> Response {
+    error_response(StatusCode::NOT_FOUND, "no such quarantine entry", None)
+}
+
+pub(crate) async fn list<H: Host>(State(host): State<H>) -> Response {
+    Json(host.remake().list_quarantine(summary)).into_response()
+}
+
+pub(crate) async fn get<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    let Some((summary, document, paths)) = host.remake().with_quarantined(&id, |entry| {
+        (summary(entry), entry.document.clone(), entry.paths.clone())
+    }) else {
+        return not_found();
+    };
+
+    let writes = tokio::task::spawn_blocking(move || {
+        paths
+            .into_iter()
+            .map(|path| match store::read_path(&document, &path) {
+                Ok(Some(value)) => json!({ "path": path, "value": value }),
+                Ok(None) => json!({ "path": path, "deleted": true }),
+                Err(error) => json!({ "path": path, "error": error.to_string() }),
+            })
+            .collect::<Vec<JsonValue>>()
+    })
+    .await;
+
+    match writes {
+        Ok(writes) => Json(json!({ "entry": summary, "writes": writes })).into_response(),
+        Err(_) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "reading the writes failed",
+            None,
+        ),
+    }
+}
+
+pub(crate) async fn document<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    let Some(mut document) = host
+        .remake()
+        .with_quarantined(&id, |entry| entry.document.clone())
+    else {
+        return not_found();
+    };
+
+    match tokio::task::spawn_blocking(move || document.save()).await {
+        Ok(bytes) => (
+            [
+                (header::CONTENT_TYPE, "application/octet-stream".to_owned()),
+                (
+                    header::CONTENT_DISPOSITION,
+                    format!("attachment; filename=\"quarantine-{id}.automerge\""),
+                ),
+            ],
+            bytes,
+        )
+            .into_response(),
+        Err(_) => error_response(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "saving the document failed",
+            None,
+        ),
+    }
+}
+
+pub(crate) async fn replay<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    if !host.is_node_ready().await {
+        return error_response(StatusCode::SERVICE_UNAVAILABLE, "node is not ready", None);
+    }
+    let Some(entry) = host.remake().take_quarantined(&id) else {
+        return not_found();
+    };
+
+    let (stats, entry) = host
+        .commit_doc(move |doc| {
+            let stats = delta::replay_paths(&entry.document, doc, &entry.paths);
+            Ok::<_, Infallible>((stats, entry))
+        })
+        .await
+        .unwrap_or_else(|never| match never {});
+
+    if stats.failed > 0 && stats.replayed == 0 {
+        // Nothing went in; keep the entry for another look.
+        host.remake().restore_quarantined(entry);
+        return error_response(
+            StatusCode::CONFLICT,
+            "none of the writes could be replayed",
+            Some(json!({ "failed": stats.failed })),
+        );
+    }
+
+    info!(quarantine = %id, replayed = stats.replayed, failed = stats.failed, "Replayed quarantined changes");
+    Json(json!({ "replayed": stats.replayed, "failed": stats.failed })).into_response()
+}
+
+pub(crate) async fn discard<H: Host>(State(host): State<H>, Path(id): Path<String>) -> Response {
+    match host.remake().take_quarantined(&id) {
+        Some(_) => {
+            info!(quarantine = %id, "Discarded quarantined changes");
+            StatusCode::NO_CONTENT.into_response()
+        }
+        None => not_found(),
+    }
+}
diff --git a/src/remake/records.rs b/src/remake/records.rs
new file mode 100644
index 0000000..219fc9b
--- /dev/null
+++ b/src/remake/records.rs
@@ -0,0 +1,428 @@
+//! Remake records: `_continuum/remakes/<id>` in the replicated document.
+//!
+//! A record has up to three parts, each with a single kind of writer, so concurrent
+//! writes never fight over the same key:
+//!
+//! * `proposal`: written once, by the coordinator.
+//! * `decision`: written once, by the coordinator only (committed or aborted).
+//! * `abortRequested`: written by any node on an operator's request. It is an input to
+//!   the coordinator's decision, not a decision itself.
+//!
+//! A proposal whose coordinator never decides (crash, partition) is reported as
+//! `expired` once its settle period plus a grace margin has passed. Expiry is derived,
+//! never written, so a late but valid commit can't collide with someone's abort.
+//!
+//! Records edited into an unreadable shape through the store API are skipped with a
+//! warning.
+
+use automerge::{AutoCommit, ChangeHash};
+use serde::{Deserialize, Serialize};
+use tracing::warn;
+
+use super::snapshot::{SnapshotRequest, META_ROOT, REMAKES};
+use crate::document::replica::Epoch;
+use crate::document::store::{self, StoreError};
+
+/// One day, for retention defaults.
+const DAY_SECS: u64 = 24 * 60 * 60;
+
+pub(crate) fn unix_millis() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.as_millis() as u64)
+        .unwrap_or(0)
+}
+
+// --- record types ------------------------------------------------------------
+
+/// An epoch in a record: generation plus hex-encoded snapshot hash.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct EpochRecord {
+    pub(crate) generation: u64,
+    pub(crate) id: String,
+}
+
+impl From<Epoch> for EpochRecord {
+    fn from(epoch: Epoch) -> Self {
+        Self {
+            generation: epoch.generation,
+            id: epoch.id_hex(),
+        }
+    }
+}
+
+impl EpochRecord {
+    pub(crate) fn to_epoch(&self) -> Option<Epoch> {
+        Some(Epoch {
+            generation: self.generation,
+            id: decode_hex32(&self.id)?,
+        })
+    }
+}
+
+/// How long something is kept. A value is kept until every given limit has passed.
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Retention {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) epochs: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) seconds: Option<u64>,
+}
+
+impl Retention {
+    /// True while at least one given limit has not passed: `epochs` epochs and
+    /// `age_ms` milliseconds after the fact.
+    pub(crate) fn covers(&self, epochs: u64, age_ms: u64) -> bool {
+        self.epochs.is_some_and(|limit| epochs <= limit)
+            || self
+                .seconds
+                .is_some_and(|limit| age_ms <= limit.saturating_mul(1000))
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Policy {
+    /// How stale a returning node's changes may be and still be replayed automatically;
+    /// beyond it they are quarantined.
+    #[serde(default = "Policy::default_replay_window")]
+    pub(crate) replay_window: Retention,
+    /// How long nodes keep the old epoch's document.
+    #[serde(default = "Policy::default_archive_retention")]
+    pub(crate) archive_retention: Retention,
+}
+
+impl Policy {
+    pub(crate) fn default_replay_window() -> Retention {
+        Retention {
+            epochs: Some(1),
+            seconds: Some(7 * DAY_SECS),
+        }
+    }
+
+    pub(crate) fn default_archive_retention() -> Retention {
+        Retention {
+            epochs: Some(2),
+            seconds: Some(30 * DAY_SECS),
+        }
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Proposal {
+    pub(crate) coordinator: String,
+    pub(crate) from_epoch: EpochRecord,
+    /// Hex change hashes, sorted. The proposal record itself is not part of the base.
+    pub(crate) base_heads: Vec<String>,
+    pub(crate) policy: Policy,
+    pub(crate) settle_ms: u64,
+    pub(crate) proposed_at: u64,
+    pub(crate) settle_until: u64,
+    /// Peers that must stay active during the settle period.
+    pub(crate) peers_considered: Vec<String>,
+}
+
+impl Proposal {
+    pub(crate) fn base_change_hashes(&self) -> Option<Vec<ChangeHash>> {
+        self.base_heads
+            .iter()
+            .map(|hex| decode_hex32(hex).map(ChangeHash))
+            .collect()
+    }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(tag = "outcome", rename_all = "camelCase")]
+pub(crate) enum Decision {
+    #[serde(rename_all = "camelCase")]
+    Committed {
+        to_epoch: EpochRecord,
+        snapshot_bytes: u64,
+        decided_at: u64,
+    },
+    #[serde(rename_all = "camelCase")]
+    Aborted { reason: String, decided_at: u64 },
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct AbortRequest {
+    pub(crate) by: String,
+    pub(crate) at: u64,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct Record {
+    pub(crate) proposal: Proposal,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) decision: Option<Decision>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub(crate) abort_requested: Option<AbortRequest>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) enum Phase {
+    /// Settling; the coordinator has not decided yet.
+    Proposed,
+    /// No decision long after the settle period: the coordinator is gone.
+    Expired,
+    Committed,
+    Aborted,
+}
+
+impl Phase {
+    /// Still able to become (or already) the next epoch.
+    pub(crate) fn is_open(self) -> bool {
+        matches!(self, Phase::Proposed | Phase::Committed)
+    }
+
+    pub(crate) fn is_final(self) -> bool {
+        !matches!(self, Phase::Proposed)
+    }
+}
+
+impl Record {
+    /// The epoch this remake committed to, if committed and readable.
+    pub(crate) fn committed_epoch(&self) -> Option<Epoch> {
+        match &self.decision {
+            Some(Decision::Committed { to_epoch, .. }) => to_epoch.to_epoch(),
+            _ => None,
+        }
+    }
+
+    /// Everything needed to (re)build this remake's snapshot; None if the proposal
+    /// is unreadable.
+    pub(crate) fn snapshot_request(&self, id: &str) -> Option<SnapshotRequest> {
+        Some(SnapshotRequest {
+            remake_id: id.to_owned(),
+            previous: self.proposal.from_epoch.to_epoch()?,
+            base_heads: self.proposal.base_change_hashes()?,
+            policy: serde_json::to_value(&self.proposal.policy).ok()?,
+            proposed_at: self.proposal.proposed_at,
+        })
+    }
+
+    pub(crate) fn phase(&self, now_ms: u64, grace_ms: u64) -> Phase {
+        match &self.decision {
+            Some(Decision::Committed { .. }) => Phase::Committed,
+            Some(Decision::Aborted { .. }) => Phase::Aborted,
+            None if now_ms > self.proposal.settle_until.saturating_add(grace_ms) => Phase::Expired,
+            None => Phase::Proposed,
+        }
+    }
+}
+
+// --- document access ---------------------------------------------------------
+
+fn records_dir() -> Vec<String> {
+    vec![META_ROOT.to_owned(), REMAKES.to_owned()]
+}
+
+pub(crate) fn record_path(id: &str, part: &str) -> Vec<String> {
+    let mut path = records_dir();
+    path.push(id.to_owned());
+    if !part.is_empty() {
+        path.push(part.to_owned());
+    }
+    path
+}
+
+/// The record with this ID, if it exists and is readable.
+pub(crate) fn load(doc: &AutoCommit, id: &str) -> Option<Record> {
+    let value = match store::read_path(doc, &record_path(id, "")) {
+        Ok(Some(value)) => value,
+        Ok(None) => return None,
+        Err(error) => {
+            warn!(remake = %id, error = %error, "Failed to read remake record");
+            return None;
+        }
+    };
+    match serde_json::from_value(value) {
+        Ok(record) => Some(record),
+        Err(error) => {
+            warn!(remake = %id, error = %error, "Skipping unreadable remake record");
+            None
+        }
+    }
+}
+
+/// All readable records.
+pub(crate) fn load_all(doc: &AutoCommit) -> Vec<(String, Record)> {
+    store::keys_at(doc, &records_dir())
+        .unwrap_or_default()
+        .into_iter()
+        .filter_map(|id| load(doc, &id).map(|record| (id, record)))
+        .collect()
+}
+
+fn write_part<T: Serialize>(
+    doc: &mut AutoCommit,
+    id: &str,
+    part: &str,
+    value: &T,
+) -> Result<(), StoreError> {
+    let json = serde_json::to_value(value).map_err(anyhow::Error::from)?;
+    store::write_path(doc, &record_path(id, part), &json)
+}
+
+pub(crate) fn write_proposal(
+    doc: &mut AutoCommit,
+    id: &str,
+    proposal: &Proposal,
+) -> Result<(), StoreError> {
+    write_part(doc, id, "proposal", proposal)
+}
+
+pub(crate) fn write_decision(
+    doc: &mut AutoCommit,
+    id: &str,
+    decision: &Decision,
+) -> Result<(), StoreError> {
+    write_part(doc, id, "decision", decision)
+}
+
+pub(crate) fn write_abort_request(
+    doc: &mut AutoCommit,
+    id: &str,
+    request: &AbortRequest,
+) -> Result<(), StoreError> {
+    write_part(doc, id, "abortRequested", request)
+}
+
+fn decode_hex32(hex: &str) -> Option<[u8; 32]> {
+    if hex.len() != 64 || !hex.is_ascii() {
+        return None;
+    }
+    let mut out = [0u8; 32];
+    for (i, byte) in out.iter_mut().enumerate() {
+        *byte = u8::from_str_radix(&hex[2 * i..2 * i + 2], 16).ok()?;
+    }
+    Some(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn retention_covers_until_every_limit_passed() {
+        let both = Retention {
+            epochs: Some(1),
+            seconds: Some(10),
+        };
+        assert!(both.covers(1, 60_000)); // epochs still within
+        assert!(both.covers(5, 5_000)); // time still within
+        assert!(!both.covers(2, 10_001));
+        let none = Retention {
+            epochs: None,
+            seconds: None,
+        };
+        assert!(!none.covers(0, 0));
+    }
+
+    fn proposal(settle_until: u64) -> Proposal {
+        Proposal {
+            coordinator: "node-a".into(),
+            from_epoch: Epoch::GENESIS.into(),
+            base_heads: vec!["ab".repeat(32)],
+            policy: Policy {
+                replay_window: Policy::default_replay_window(),
+                archive_retention: Policy::default_archive_retention(),
+            },
+            settle_ms: 1_000,
+            proposed_at: settle_until.saturating_sub(1_000),
+            settle_until,
+            peers_considered: vec!["node-b".into()],
+        }
+    }
+
+    #[test]
+    fn round_trip_through_the_document() {
+        let mut doc = AutoCommit::new();
+        let p = proposal(10_000);
+        write_proposal(&mut doc, "c-1", &p).unwrap();
+        write_abort_request(
+            &mut doc,
+            "c-1",
+            &AbortRequest {
+                by: "x".into(),
+                at: 5,
+            },
+        )
+        .unwrap();
+        let decision = Decision::Committed {
+            to_epoch: EpochRecord {
+                generation: 1,
+                id: "cd".repeat(32),
+            },
+            snapshot_bytes: 123,
+            decided_at: 11_000,
+        };
+        write_decision(&mut doc, "c-1", &decision).unwrap();
+        doc.commit();
+
+        let all = load_all(&doc);
+        assert_eq!(all.len(), 1);
+        let (id, record) = &all[0];
+        assert_eq!(id, "c-1");
+        assert_eq!(record.proposal.policy, p.policy);
+        assert_eq!(record.proposal.base_change_hashes().unwrap().len(), 1);
+        assert!(record.abort_requested.is_some());
+        assert_eq!(record.phase(0, 0), Phase::Committed);
+    }
+
+    #[test]
+    fn phases() {
+        let mut record = Record {
+            proposal: proposal(10_000),
+            decision: None,
+            abort_requested: None,
+        };
+        assert_eq!(record.phase(10_000, 500), Phase::Proposed);
+        assert_eq!(record.phase(10_500, 500), Phase::Proposed);
+        assert_eq!(record.phase(10_501, 500), Phase::Expired);
+        record.decision = Some(Decision::Aborted {
+            reason: "test".into(),
+            decided_at: 1,
+        });
+        assert_eq!(record.phase(0, 0), Phase::Aborted);
+    }
+
+    #[test]
+    fn unreadable_records_are_skipped() {
+        let mut doc = AutoCommit::new();
+        store::write_path(
+            &mut doc,
+            &record_path("broken", ""),
+            &serde_json::json!({ "proposal": "not an object" }),
+        )
+        .unwrap();
+        write_proposal(&mut doc, "fine", &proposal(1)).unwrap();
+        doc.commit();
+        let all = load_all(&doc);
+        assert_eq!(all.len(), 1);
+        assert_eq!(all[0].0, "fine");
+    }
+
+    #[test]
+    fn epoch_record_round_trip() {
+        let epoch = Epoch {
+            generation: 3,
+            id: [0x5a; 32],
+        };
+        assert_eq!(EpochRecord::from(epoch).to_epoch(), Some(epoch));
+        assert_eq!(
+            EpochRecord {
+                generation: 1,
+                id: "zz".into()
+            }
+            .to_epoch(),
+            None
+        );
+    }
+}
diff --git a/src/remake/snapshot.rs b/src/remake/snapshot.rs
new file mode 100644
index 0000000..da23fdf
--- /dev/null
+++ b/src/remake/snapshot.rs
@@ -0,0 +1,637 @@
+//! Deterministic snapshot of the document at agreed base heads.
+//!
+//! Every node builds the snapshot of a committed remake on its own. For that to
+//! work, the snapshot must be byte-identical everywhere: same operations, same order,
+//! same actor, same timestamp. Its change hash then identifies the new epoch, and a
+//! node whose build produces a different hash knows it must fetch the snapshot instead.
+//!
+//! Determinism rests on:
+//! * the content: the merged view at the base heads (`store::materialize_into`), which
+//!   depends only on the base, not on how a node's history got there;
+//! * a fixed actor derived from the remake ID and generation;
+//! * commit time 0 and no message;
+//! * canonical (sorted) JSON for the metadata;
+//! * the same Automerge version on every node (encodings may differ across versions).
+//!
+//! The snapshot carries its own metadata under `_continuum/epochs/<generation>`:
+//! previous epoch, base heads, per-actor clock and retention policy. Later steps read it
+//! at the snapshot change itself, so edits made afterwards through the store API can't
+//! change what the protocol sees.
+
+// `Snapshot::clock` is kept for returning nodes (step 7); late nodes read the clock from
+// the metadata instead.
+#![cfg_attr(not(test), allow(dead_code))]
+
+use std::collections::BTreeMap;
+
+use automerge::{transaction::CommitOptions, ActorId, AutoCommit, ChangeHash};
+use serde_json::{json, Map as JsonMap, Value as JsonValue};
+use sha2::{Digest, Sha256};
+
+use super::records::{EpochRecord, Policy};
+use crate::document::replica::Epoch;
+use crate::document::store::{self, StoreError};
+
+pub(crate) const META_ROOT: &str = "_continuum";
+pub(crate) const REMAKES: &str = "remakes";
+pub(crate) const EPOCHS: &str = "epochs";
+
+/// Everything that determines a snapshot besides the base content.
+pub(crate) struct SnapshotParams<'a> {
+    /// ID of the committed remake; seeds the snapshot actor.
+    pub(crate) remake_id: &'a str,
+    /// The epoch being compacted. The snapshot starts generation `previous + 1`.
+    pub(crate) previous: Epoch,
+    /// Heads of the base. Every node must hold all of these changes.
+    pub(crate) base_heads: &'a [ChangeHash],
+    /// Retention policy from the remake request; stored verbatim (canonicalized).
+    pub(crate) policy: &'a JsonValue,
+    /// When the remake was proposed (unix ms); recorded for replay windows.
+    pub(crate) proposed_at: u64,
+}
+
+pub(crate) struct Snapshot {
+    /// The new epoch: next generation, ID = hash of the snapshot change.
+    pub(crate) epoch: Epoch,
+    /// A document consisting of exactly that one change.
+    pub(crate) doc: AutoCommit,
+    /// Highest sequence number per actor (hex) contained in the base. A change
+    /// (actor, seq) is part of the snapshot iff seq <= clock[actor].
+    pub(crate) clock: BTreeMap<String, u64>,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub(crate) enum SnapshotError {
+    #[error("base head {0} is not in the local document")]
+    MissingBase(ChangeHash),
+    #[error("snapshot produced {0} heads instead of exactly one")]
+    UnexpectedHeads(usize),
+    #[error(transparent)]
+    Doc(#[from] automerge::AutomergeError),
+    #[error(transparent)]
+    Store(#[from] StoreError),
+    #[error("snapshot task failed: {0}")]
+    Task(String),
+}
+
+/// Owned version of [`SnapshotParams`], for moving into a blocking task.
+#[derive(Clone, Debug)]
+pub(crate) struct SnapshotRequest {
+    pub(crate) remake_id: String,
+    pub(crate) previous: Epoch,
+    pub(crate) base_heads: Vec<ChangeHash>,
+    pub(crate) policy: JsonValue,
+    pub(crate) proposed_at: u64,
+}
+
+impl SnapshotRequest {
+    pub(crate) fn params(&self) -> SnapshotParams<'_> {
+        SnapshotParams {
+            remake_id: &self.remake_id,
+            previous: self.previous,
+            base_heads: &self.base_heads,
+            policy: &self.policy,
+            proposed_at: self.proposed_at,
+        }
+    }
+}
+
+/// Builds the snapshot from `copy`, a clone of the live document, on the blocking pool.
+/// No lock is held while building; the caller only needs a lock for the clone.
+/// Returns the copy as well, for callers that continue working with it.
+pub(crate) async fn build_detached(
+    mut copy: AutoCommit,
+    request: SnapshotRequest,
+) -> Result<(Snapshot, AutoCommit), SnapshotError> {
+    tokio::task::spawn_blocking(move || {
+        let snapshot = build(&mut copy, &request.params())?;
+        Ok((snapshot, copy))
+    })
+    .await
+    .map_err(|error| SnapshotError::Task(error.to_string()))?
+}
+
+/// The protocol's view of an epoch, as recorded in its snapshot.
+#[derive(Clone, Debug)]
+pub(crate) struct EpochMetadata {
+    pub(crate) remake_id: String,
+    pub(crate) previous: Epoch,
+    /// Highest sequence number per actor of `previous` contained in the snapshot.
+    pub(crate) clock: BTreeMap<String, u64>,
+    pub(crate) policy: Policy,
+    /// When the remake was proposed (unix ms): the point the cluster moved on.
+    pub(crate) proposed_at: u64,
+}
+
+#[derive(serde::Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct MetadataRecord {
+    generation: u64,
+    remake_id: String,
+    previous: EpochRecord,
+    clock: BTreeMap<String, u64>,
+    policy: Policy,
+    #[serde(default)]
+    proposed_at: u64,
+}
+
+fn parse_metadata(doc: &AutoCommit, generation: u64) -> Result<EpochMetadata, String> {
+    let value = store::read_path(doc, &epoch_metadata_path(generation))
+        .map_err(|error| error.to_string())?
+        .ok_or_else(|| format!("no epoch metadata for generation {generation}"))?;
+    let record: MetadataRecord = serde_json::from_value(value)
+        .map_err(|error| format!("invalid epoch metadata: {error}"))?;
+    if record.generation != generation {
+        return Err(format!(
+            "epoch metadata names generation {} instead of {generation}",
+            record.generation
+        ));
+    }
+    let previous = record
+        .previous
+        .to_epoch()
+        .ok_or_else(|| "invalid previous epoch in metadata".to_owned())?;
+
+    Ok(EpochMetadata {
+        remake_id: record.remake_id,
+        previous,
+        clock: record.clock,
+        policy: record.policy,
+        proposed_at: record.proposed_at,
+    })
+}
+
+/// Checks that `doc` is a document of `epoch`: its history starts with exactly the
+/// snapshot change `epoch.id` (no dependencies), and the metadata at that change names
+/// the same generation. Returns the metadata.
+///
+/// This does not prove the snapshot's content matches the old epoch; the hash commits
+/// to it, and the peer is a cluster member trusted as much as for ordinary sync.
+pub(crate) fn verify_document(doc: &mut AutoCommit, epoch: Epoch) -> Result<EpochMetadata, String> {
+    let root = ChangeHash(epoch.id);
+    let Some(change) = doc.get_change_by_hash(&root) else {
+        return Err(format!(
+            "document does not contain the snapshot change of {epoch}"
+        ));
+    };
+    if !change.deps().is_empty() {
+        return Err(format!("snapshot change of {epoch} has dependencies"));
+    }
+
+    let at_snapshot = doc.fork_at(&[root]).map_err(|error| error.to_string())?;
+    parse_metadata(&at_snapshot, epoch.generation)
+}
+
+/// Metadata of an earlier generation, as the current document holds it. Snapshots
+/// carry the metadata of all earlier epochs forward (only remake records stay
+/// behind), so a node many epochs behind can still find the epoch that followed its own.
+pub(crate) fn read_epoch_metadata(
+    doc: &AutoCommit,
+    generation: u64,
+) -> Result<EpochMetadata, String> {
+    parse_metadata(doc, generation)
+}
+
+/// Path of the metadata record for an epoch generation.
+pub(crate) fn epoch_metadata_path(generation: u64) -> Vec<String> {
+    vec![
+        META_ROOT.to_owned(),
+        EPOCHS.to_owned(),
+        generation.to_string(),
+    ]
+}
+
+/// Builds the snapshot of `source` at `params.base_heads`.
+pub(crate) fn build(
+    source: &mut AutoCommit,
+    params: &SnapshotParams<'_>,
+) -> Result<Snapshot, SnapshotError> {
+    // fork_at with unknown heads would quietly build something else.
+    for head in params.base_heads {
+        if source.get_change_by_hash(head).is_none() {
+            return Err(SnapshotError::MissingBase(*head));
+        }
+    }
+
+    let mut base = source.fork_at(params.base_heads)?;
+    let clock = clock_of(&mut base);
+    let generation = params.previous.generation + 1;
+
+    let mut snapshot = AutoCommit::new().with_actor(snapshot_actor(params.remake_id, generation));
+
+    // Remake records describe the old epoch's protocol and stay behind.
+    store::materialize_into(&base, &mut snapshot, &is_remake_record)?;
+
+    let mut base_heads: Vec<String> = params.base_heads.iter().map(|h| h.to_string()).collect();
+    base_heads.sort();
+
+    let metadata = canonical(json!({
+        "generation": generation,
+        "remakeId": params.remake_id,
+        "previous": {
+            "generation": params.previous.generation,
+            "id": params.previous.id_hex(),
+        },
+        "baseHeads": base_heads,
+        "clock": clock,
+        "policy": params.policy,
+        "proposedAt": params.proposed_at,
+    }));
+    store::write_path(&mut snapshot, &epoch_metadata_path(generation), &metadata)?;
+
+    let _ = snapshot.commit_with(CommitOptions::default().with_time(0));
+
+    let heads = snapshot.get_heads();
+    let [id] = heads.as_slice() else {
+        return Err(SnapshotError::UnexpectedHeads(heads.len()));
+    };
+
+    Ok(Snapshot {
+        epoch: Epoch {
+            generation,
+            id: id.0,
+        },
+        doc: snapshot,
+        clock,
+    })
+}
+
+/// Highest sequence number per actor in `doc`'s history.
+fn clock_of(doc: &mut AutoCommit) -> BTreeMap<String, u64> {
+    let mut clock = BTreeMap::new();
+    for change in doc.get_changes(&[]) {
+        let seq = clock.entry(change.actor_id().to_string()).or_insert(0);
+        *seq = (*seq).max(change.seq());
+    }
+    clock
+}
+
+/// The same 16-byte actor on every node, unique per remake and generation.
+fn snapshot_actor(remake_id: &str, generation: u64) -> ActorId {
+    let digest = Sha256::new()
+        .chain_update(b"continuum.snapshot\0")
+        .chain_update(remake_id.as_bytes())
+        .chain_update(b"\0")
+        .chain_update(generation.to_be_bytes())
+        .finalize();
+    ActorId::from(&digest[..16])
+}
+
+fn is_remake_record(path: &[String]) -> bool {
+    matches!(path, [root, records] if root == META_ROOT && records == REMAKES)
+}
+
+/// Rebuilds every object with sorted keys, so the result is the same whichever key
+/// order serde_json happens to use (it keeps insertion order with `preserve_order`).
+fn canonical(value: JsonValue) -> JsonValue {
+    match value {
+        JsonValue::Object(map) => {
+            let sorted: BTreeMap<String, JsonValue> = map
+                .into_iter()
+                .map(|(key, value)| (key, canonical(value)))
+                .collect();
+            JsonValue::Object(sorted.into_iter().collect::<JsonMap<_, _>>())
+        }
+        JsonValue::Array(items) => JsonValue::Array(items.into_iter().map(canonical).collect()),
+        other => other,
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use automerge::{transaction::Transactable, ObjType, ReadDoc, ScalarValue, Value, ROOT};
+
+    fn doc(actor: &[u8]) -> AutoCommit {
+        AutoCommit::new().with_actor(ActorId::from(actor))
+    }
+
+    fn params<'a>(heads: &'a [ChangeHash], policy: &'a JsonValue) -> SnapshotParams<'a> {
+        SnapshotParams {
+            remake_id: "c-1",
+            previous: Epoch::GENESIS,
+            base_heads: heads,
+            policy,
+            proposed_at: 0,
+        }
+    }
+
+    fn snapshot_of(source: &mut AutoCommit) -> Snapshot {
+        let heads = source.get_heads();
+        let policy = json!({});
+        build(source, &params(&heads, &policy)).expect("snapshot")
+    }
+
+    /// The user-visible view, without our own metadata.
+    fn user_view(doc: &AutoCommit) -> JsonValue {
+        let mut view = store::read_path(doc, &[]).unwrap().unwrap();
+        view.as_object_mut().unwrap().remove(META_ROOT);
+        view
+    }
+
+    #[test]
+    fn same_view_different_history_same_snapshot() {
+        // One history: overwrites, increments, one-shot text.
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        a.commit();
+        a.put(ROOT, "x", 2_i64).unwrap();
+        a.put(ROOT, "c", ScalarValue::counter(0)).unwrap();
+        a.increment(ROOT, "c", 5).unwrap();
+        let text = a.put_object(ROOT, "t", ObjType::Text).unwrap();
+        a.splice_text(&text, 0, 0, "hello").unwrap();
+        let m = a.put_object(ROOT, "m", ObjType::Map).unwrap();
+        a.put(&m, "k", "v").unwrap();
+        a.commit();
+
+        // Another actor, other order, other steps, same result.
+        let mut b = doc(b"actor-b");
+        let m = b.put_object(ROOT, "m", ObjType::Map).unwrap();
+        b.put(&m, "k", "v").unwrap();
+        let text = b.put_object(ROOT, "t", ObjType::Text).unwrap();
+        b.splice_text(&text, 0, 0, "hel").unwrap();
+        b.commit();
+        b.splice_text(&text, 3, 0, "lo").unwrap();
+        b.put(ROOT, "c", ScalarValue::counter(5)).unwrap();
+        b.put(ROOT, "x", 2_i64).unwrap();
+        b.commit();
+
+        assert_eq!(user_view(&a), user_view(&b));
+
+        let snap_a = snapshot_of(&mut a);
+        let snap_b = snapshot_of(&mut b);
+
+        // Different histories mean different clocks, and the clock is part of the
+        // metadata. Equal views alone therefore give equal content but not equal IDs;
+        // equal bases do give equal IDs (see `rebuild_is_deterministic`).
+        assert_eq!(user_view(&snap_a.doc), user_view(&snap_b.doc));
+        assert_eq!(user_view(&snap_a.doc), user_view(&a));
+    }
+
+    #[test]
+    fn rebuild_is_deterministic() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        let list = a.put_object(ROOT, "l", ObjType::List).unwrap();
+        a.insert(&list, 0, "first").unwrap();
+        let inner = a.insert_object(&list, 1, ObjType::Map).unwrap();
+        a.put(&inner, "k", true).unwrap();
+        a.commit();
+
+        // A second node holding the same changes, received in a different way.
+        let mut b = doc(b"actor-b");
+        b.merge(&mut a.clone()).unwrap();
+
+        let first = snapshot_of(&mut a);
+        let again = snapshot_of(&mut a);
+        let elsewhere = snapshot_of(&mut b);
+
+        assert_eq!(first.epoch, again.epoch);
+        assert_eq!(first.epoch, elsewhere.epoch);
+        assert_eq!(first.epoch.generation, 1);
+    }
+
+    #[test]
+    fn conflicts_collapse_into_the_merged_view() {
+        let mut a = doc(b"actor-a");
+        let cfg = a.put_object(ROOT, "cfg", ObjType::Map).unwrap();
+        a.put(&cfg, "x", 1_i64).unwrap();
+        a.commit();
+
+        let mut b = doc(b"actor-b");
+        let cfg = b.put_object(ROOT, "cfg", ObjType::Map).unwrap();
+        b.put(&cfg, "y", 2_i64).unwrap();
+        b.commit();
+
+        a.merge(&mut b).unwrap();
+        let (_, conflicts) = store::read_path_with_conflicts(&a, &[]).unwrap().unwrap();
+        assert!(
+            !conflicts.is_empty(),
+            "test setup should produce a conflict"
+        );
+
+        let snap = snapshot_of(&mut a);
+        assert_eq!(user_view(&snap.doc), json!({ "cfg": { "x": 1, "y": 2 } }));
+
+        let (_, conflicts) = store::read_path_with_conflicts(&snap.doc, &[])
+            .unwrap()
+            .unwrap();
+        assert!(
+            conflicts.is_empty(),
+            "snapshot still has conflicts: {conflicts:?}"
+        );
+    }
+
+    #[test]
+    fn base_heads_bound_content_and_clock() {
+        let mut a = doc(b"actor-a");
+        for i in 0..3_i64 {
+            a.put(ROOT, format!("a{i}"), i).unwrap();
+            a.commit();
+        }
+
+        let mut b = doc(b"actor-b");
+        for i in 0..2_i64 {
+            b.put(ROOT, format!("b{i}"), i).unwrap();
+            b.commit();
+        }
+        a.merge(&mut b).unwrap();
+
+        let base = a.get_heads();
+
+        // After the base: must not appear in the snapshot or its clock.
+        a.put(ROOT, "late", true).unwrap();
+        a.commit();
+
+        let policy = json!({});
+        let snap = build(&mut a, &params(&base, &policy)).unwrap();
+
+        let view = user_view(&snap.doc);
+        assert!(view.get("late").is_none());
+        assert_eq!(view["a2"], json!(2));
+        assert_eq!(view["b1"], json!(1));
+
+        let actor_a = ActorId::from(b"actor-a".as_slice()).to_string();
+        let actor_b = ActorId::from(b"actor-b".as_slice()).to_string();
+        assert_eq!(snap.clock.get(&actor_a), Some(&3));
+        assert_eq!(snap.clock.get(&actor_b), Some(&2));
+    }
+
+    #[test]
+    fn metadata_written_and_remake_records_dropped() {
+        let mut a = doc(b"actor-a");
+        store::write_path(
+            &mut a,
+            &[META_ROOT.into(), REMAKES.into(), "c-1".into()],
+            &json!({ "state": "committed" }),
+        )
+        .unwrap();
+        store::write_path(
+            &mut a,
+            &[META_ROOT.into(), "forgotten".into(), "pk".into()],
+            &json!({ "at": 1 }),
+        )
+        .unwrap();
+        a.commit();
+
+        let heads = a.get_heads();
+        let policy = json!({ "replayWindow": { "epochs": 1 } });
+        let snap = build(&mut a, &params(&heads, &policy)).unwrap();
+
+        let meta = store::read_path(&snap.doc, &[META_ROOT.into()])
+            .unwrap()
+            .unwrap();
+        assert!(meta.get(REMAKES).is_none());
+        assert_eq!(meta["forgotten"]["pk"]["at"], json!(1));
+
+        let epoch = &meta[EPOCHS]["1"];
+        assert_eq!(epoch["generation"], json!(1));
+        assert_eq!(epoch["remakeId"], json!("c-1"));
+        assert_eq!(epoch["previous"]["generation"], json!(0));
+        assert_eq!(epoch["policy"], policy);
+        assert_eq!(epoch["baseHeads"].as_array().unwrap().len(), heads.len());
+    }
+
+    #[test]
+    fn automerge_types_survive() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "c", ScalarValue::counter(3)).unwrap();
+        a.increment(ROOT, "c", 4).unwrap();
+        let text = a.put_object(ROOT, "t", ObjType::Text).unwrap();
+        a.splice_text(&text, 0, 0, "abc").unwrap();
+        a.put(ROOT, "bytes", ScalarValue::Bytes(vec![0, 1, 255]))
+            .unwrap();
+        a.put(ROOT, "ts", ScalarValue::Timestamp(1_700_000_000_000))
+            .unwrap();
+        a.put(ROOT, "neg_zero", -0.0_f64).unwrap();
+        a.commit();
+
+        let snap = snapshot_of(&mut a);
+        let d = &snap.doc;
+
+        match d.get(ROOT, "c").unwrap() {
+            Some((Value::Scalar(s), _)) => match s.as_ref() {
+                ScalarValue::Counter(c) => assert_eq!(i64::from(c.clone()), 7),
+                other => panic!("counter became {other:?}"),
+            },
+            other => panic!("unexpected {other:?}"),
+        }
+
+        let (_, text_id) = d.get(ROOT, "t").unwrap().unwrap();
+        assert_eq!(d.object_type(&text_id).unwrap(), ObjType::Text);
+        assert_eq!(d.text(&text_id).unwrap(), "abc");
+
+        match d.get(ROOT, "bytes").unwrap() {
+            Some((Value::Scalar(s), _)) => {
+                assert_eq!(s.as_ref(), &ScalarValue::Bytes(vec![0, 1, 255]))
+            }
+            other => panic!("unexpected {other:?}"),
+        }
+        match d.get(ROOT, "ts").unwrap() {
+            Some((Value::Scalar(s), _)) => {
+                assert_eq!(s.as_ref(), &ScalarValue::Timestamp(1_700_000_000_000))
+            }
+            other => panic!("unexpected {other:?}"),
+        }
+        match d.get(ROOT, "neg_zero").unwrap() {
+            Some((Value::Scalar(s), _)) => match s.as_ref() {
+                ScalarValue::F64(f) => assert_eq!(f.to_bits(), (-0.0_f64).to_bits()),
+                other => panic!("float became {other:?}"),
+            },
+            other => panic!("unexpected {other:?}"),
+        }
+    }
+
+    #[test]
+    fn snapshot_is_a_single_change_with_the_epoch_id() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        a.commit();
+
+        let mut snap = snapshot_of(&mut a);
+        let heads = snap.doc.get_heads();
+        assert_eq!(heads.len(), 1);
+        assert_eq!(heads[0].0, snap.epoch.id);
+        assert_eq!(snap.doc.get_changes(&[]).len(), 1);
+    }
+
+    #[test]
+    fn missing_base_is_rejected() {
+        let mut a = doc(b"actor-a");
+        a.put(ROOT, "x", 1_i64).unwrap();
+        a.commit();
+
+        let mut other = doc(b"actor-b");
+        other.put(ROOT, "y", 1_i64).unwrap();
+        other.commit();
+        let foreign = other.get_heads();
+
+        let policy = json!({});
+        assert!(matches!(
+            build(&mut a, &params(&foreign, &policy)),
+            Err(SnapshotError::MissingBase(_))
+        ));
+    }
+
+    #[test]
+    fn empty_document() {
+        let mut a = doc(b"actor-a");
+        let snap = snapshot_of(&mut a);
+        assert_eq!(snap.epoch.generation, 1);
+        assert!(snap.clock.is_empty());
+        assert_eq!(user_view(&snap.doc), json!({}));
+    }
+
+    #[test]
+    fn fetched_document_is_verified_and_late_changes_replayed() {
+        use crate::remake::delta;
+
+        // Old epoch: the base, then a late write by this node.
+        let mut late = doc(b"late-node");
+        late.put(ROOT, "shared", "base").unwrap();
+        late.commit();
+        let base = late.get_heads();
+        let policy = json!({});
+        let mut snap = build(&mut late, &params(&base, &policy)).unwrap();
+
+        // Meanwhile the cluster moved on in the new epoch.
+        let mut cluster = snap
+            .doc
+            .fork()
+            .with_actor(ActorId::from(&b"cluster-node"[..]));
+        cluster.put(ROOT, "new", 1_i64).unwrap();
+        cluster.commit();
+        let bytes = cluster.save();
+
+        late.put(ROOT, "mine", "late").unwrap();
+        late.commit();
+
+        let mut fetched = AutoCommit::load(&bytes).unwrap();
+        let metadata = verify_document(&mut fetched, snap.epoch).unwrap();
+        assert_eq!(metadata.previous, Epoch::GENESIS);
+        assert_eq!(metadata.remake_id, "c-1");
+        assert_eq!(
+            metadata.policy.replay_window,
+            Policy::default_replay_window()
+        );
+
+        // Documents of other epochs are refused.
+        let mut wrong = snap.epoch;
+        wrong.generation = 2;
+        assert!(verify_document(&mut fetched, wrong).is_err());
+        let mut wrong = snap.epoch;
+        wrong.id = [0; 32];
+        assert!(verify_document(&mut fetched, wrong).is_err());
+
+        let own = late.get_actor().clone();
+        let missing = delta::missing_from_snapshot(&mut late, &metadata.clock, &own);
+        assert_eq!(missing.own, 1);
+        let heads = late.get_heads();
+        let mut adopted = fetched.with_actor(ActorId::from(&b"late-node-next"[..]));
+        delta::replay(&mut late, &missing.before, &heads, &mut adopted);
+        assert_eq!(
+            user_view(&adopted),
+            json!({ "shared": "base", "new": 1, "mine": "late" })
+        );
+    }
+}
diff --git a/src/remake/switch.rs b/src/remake/switch.rs
new file mode 100644
index 0000000..b4ab829
--- /dev/null
+++ b/src/remake/switch.rs
@@ -0,0 +1,248 @@
+//! Switching this node to a committed epoch.
+//!
+//! The worker watches the remake records. When a remake of the local epoch is
+//! committed, it waits one keepalive interval (so the commit record can reach peers
+//! through ordinary sync before this node stops syncing the old epoch), then:
+//!
+//! 1. clones the live document under the read lock;
+//! 2. without any lock: takes the coordinator's prebuilt snapshot or builds it, checks
+//!    that its hash is the committed epoch ID, and replays the clone's changes since
+//!    the base into it;
+//! 3. under the write lock: replays whatever arrived since the clone, swaps document and
+//!    epoch, and hands back the old document;
+//! 4. archives the old document according to the policy.
+//!
+//! The write lock is held only for step 3, which is proportional to the changes made
+//! during steps 1-2.
+//!
+//! A node whose build doesn't match the committed epoch, or that never saw the commit,
+//! rejoins the new epoch from peers instead (`catchup`).
+
+use std::time::Duration;
+
+use automerge::{ActorId, AutoCommit, ChangeHash};
+use tokio::sync::broadcast;
+use tracing::{error, info, warn};
+
+use super::catchup;
+use super::delta::{self, ReplayStats};
+use super::local::LocalStatus;
+use super::records::{self, unix_millis, EpochRecord, Record};
+use super::snapshot::{self, META_ROOT, REMAKES};
+use super::Host;
+use crate::document::replica::Epoch;
+use crate::document::store;
+use crate::swarm::sync::paths_related;
+
+const TICK: Duration = Duration::from_secs(1);
+
+pub(crate) async fn run_worker<H: Host>(host: H) {
+    info!("Starting remake worker");
+
+    let records_dir = vec![META_ROOT.to_owned(), REMAKES.to_owned()];
+    let mut changes = host.subscribe_changes();
+    let mut changes_closed = false;
+    let mut tick = tokio::time::interval(TICK);
+    tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+
+    loop {
+        // React promptly to changes of the records; the timer covers the switch delay.
+        let should_check = tokio::select! {
+            _ = tick.tick() => true,
+            received = changes.recv(), if !changes_closed => match received {
+                Ok(set) => set.paths.iter().any(|path| paths_related(path, &records_dir)),
+                Err(broadcast::error::RecvError::Lagged(_)) => true,
+                Err(broadcast::error::RecvError::Closed) => {
+                    warn!("Change events closed; remake worker continues on its timer only");
+                    changes_closed = true;
+                    false
+                }
+            }
+        };
+
+        if should_check {
+            check(&host).await;
+        }
+    }
+}
+
+async fn check<H: Host>(host: &H) {
+    let local = host.current_epoch();
+    host.remake().prune_archive(local.generation);
+
+    // While rejoining, the document is being filled from peers: its records are not
+    // complete enough to act on.
+    if host.remake().is_rejoining() {
+        return catchup::check(host).await;
+    }
+
+    let all = host.read_doc(records::load_all).await;
+    let local_record = EpochRecord::from(local);
+
+    let committed: Vec<(String, Record)> = all
+        .into_iter()
+        .filter(|(_, record)| {
+            record.proposal.from_epoch == local_record && record.committed_epoch().is_some()
+        })
+        .collect();
+
+    let (id, record) = match committed.as_slice() {
+        // Nothing committed here; maybe peers moved on without us.
+        [] => return catchup::check(host).await,
+        [(id, record)] => (id.clone(), record.clone()),
+        several => {
+            // The settle protocol prevents this; never guess which one is right. Peers
+            // that did switch can still be followed.
+            let ids: Vec<&str> = several.iter().map(|(id, _)| id.as_str()).collect();
+            error!(remakes = ?ids, "Several remakes committed for this epoch; not switching on my own");
+            return catchup::check(host).await;
+        }
+    };
+
+    if host
+        .remake()
+        .status(&id)
+        .is_some_and(|status| status.error.is_some())
+    {
+        // Our own switch failed (reported in the remake's status): fetch the epoch
+        // from a peer that made it.
+        return catchup::check(host).await;
+    }
+
+    let switch_delay = host.dormant_after() / 4; // one keepalive interval
+    if host.remake().first_seen(&id).elapsed() < switch_delay {
+        return;
+    }
+
+    match switch(host, &id, &record).await {
+        Ok(stats) => {
+            host.remake().set_status(
+                &id,
+                LocalStatus {
+                    switched_at: Some(unix_millis()),
+                    replayed_paths: Some(stats.replayed),
+                    failed_paths: Some(stats.failed),
+                    quarantine: None,
+                    error: None,
+                },
+            );
+        }
+        Err(reason) => {
+            error!(remake = %id, reason = %reason, "Epoch switch failed");
+            host.remake().set_status(
+                &id,
+                LocalStatus {
+                    error: Some(reason),
+                    ..LocalStatus::default()
+                },
+            );
+        }
+    }
+}
+
+async fn switch<H: Host>(host: &H, id: &str, record: &Record) -> Result<ReplayStats, String> {
+    let (Some(expected), Some(request)) = (record.committed_epoch(), record.snapshot_request(id))
+    else {
+        return Err("unreadable remake record".to_owned());
+    };
+    let previous = request.previous;
+    let base = request.base_heads.clone();
+    let retention = record.proposal.policy.archive_retention.clone();
+    // Only the coordinator carries the record into the new epoch: one writer, so the
+    // nodes don't all write the same path concurrently right after switching (a merge
+    // conflict, plus one change per node to sync). The epoch metadata in the snapshot
+    // keeps the essentials on every node anyway.
+    let carry_record = record.proposal.coordinator == host.node_id();
+    let record_path = records::record_path(id, "");
+
+    info!(remake = %id, from = %previous, to = %expected, "Switching epoch");
+
+    // 1. Consistent copy, under the read lock only.
+    let copy = host.read_doc(|doc| doc.clone()).await;
+    let prebuilt = host
+        .remake()
+        .take_prebuilt(id)
+        .filter(|snapshot| snapshot.epoch == expected);
+
+    // 2. Build (or reuse), verify and replay without holding any lock.
+    let (new_doc, copy_heads, stats) = tokio::task::spawn_blocking(
+        move || -> Result<(AutoCommit, Vec<ChangeHash>, ReplayStats), String> {
+            let mut copy = copy;
+            let snapshot = match prebuilt {
+                Some(snapshot) => snapshot,
+                None => snapshot::build(&mut copy, &request.params())
+                    .map_err(|error| format!("snapshot build failed: {error}"))?,
+            };
+            if snapshot.epoch != expected {
+                return Err(format!(
+                    "built snapshot {} does not match the committed epoch {expected}",
+                    snapshot.epoch
+                ));
+            }
+
+            // A fresh actor: this node's writes in the new epoch.
+            let mut doc = snapshot.doc.with_actor(ActorId::random());
+
+            // Keep this remake's record visible in the new epoch, for its history.
+            if carry_record {
+                if let Err(error) = store::copy_path(&copy, &mut doc, &record_path) {
+                    warn!(error = %error, "Could not carry the remake record into the new epoch");
+                }
+            }
+
+            let heads = copy.get_heads();
+            let stats = delta::replay(&mut copy, &base, &heads, &mut doc);
+            Ok((doc, heads, stats))
+        },
+    )
+    .await
+    .map_err(|error| format!("switch task failed: {error}"))??;
+
+    // 3. Swap, replaying what arrived since the copy.
+    let old = swap_in(host, expected, new_doc, copy_heads).await?;
+
+    // 4. Keep the old document as configured.
+    host.remake()
+        .archive(previous, old, retention, expected.generation);
+
+    info!(
+        remake = %id,
+        epoch = %expected,
+        replayed = stats.replayed,
+        failed = stats.failed,
+        "Switched to the new epoch"
+    );
+    Ok(stats)
+}
+
+/// Under the write lock: replays into `new_doc` whatever the live document received
+/// since `copy_heads` (the heads of the copy `new_doc` was prepared from), then swaps it
+/// in as `target`. Returns the old document.
+pub(super) async fn swap_in<H: Host>(
+    host: &H,
+    target: Epoch,
+    new_doc: AutoCommit,
+    copy_heads: Vec<ChangeHash>,
+) -> Result<AutoCommit, String> {
+    host.switch_epoch(target, move |live| {
+        // Only one worker switches, but be sure the copy still describes the live document.
+        if !copy_heads
+            .iter()
+            .all(|head| live.get_change_by_hash(head).is_some())
+        {
+            return Err("the document was replaced while preparing the switch".to_owned());
+        }
+        let mut new_doc = new_doc;
+        let live_heads = live.get_heads();
+        let late = delta::replay(live, &copy_heads, &live_heads, &mut new_doc);
+        if late.replayed + late.failed > 0 {
+            info!(
+                replayed = late.replayed,
+                failed = late.failed,
+                "Replayed late changes"
+            );
+        }
+        Ok(new_doc)
+    })
+    .await
+}