diff --git a/README.md b/README.md
index 1019a28..7290519 100644
--- a/README.md
+++ b/README.md
@@ -461,6 +461,39 @@ The one-time fleet setup this pairs with — the GitHub OAuth App, the global
 **and** switchboard environments — is ePHPm node config, not switchboard's; it is
 described in [`docs/preview-access-gate.md`](docs/preview-access-gate.md).
 
+### Reconcile (level-triggered convergence)
+
+The drain/queue path is **edge-triggered** on the `switchboard:gen` counter: a
+node only re-walks desired state when the counter advances past its own cursor.
+Because `gen` is a *separate* gossip key from the content it guards, an increment
+that is lost or not-yet-replicated leaves a node "current" at a generation whose
+teardown it never materialized — so the torn-down preview stays served, its
+database and `<key>.toml` override on disk, every health check green. This is the
+class of fault behind override counts drifting between nodes and orphaned
+overrides lingering as 404s.
+
+The reconcile pass is the **level-triggered** backstop. On its own interval,
+independent of `gen`, it reads the KV desired state *directly* and converges this
+node's on-disk previews to it. Pruning classifies each preview directory from its
+**own** `switchboard:preview:<label>` key (a plain `set`, immune to both the
+`gen` propagation race and the index's lost-update race), so it never trusts the
+index and never false-positives a live preview into an orphan.
+
+| Flag | Env | Default | Meaning |
+|---|---|---|---|
+| `--reconcile-interval-secs` | `SWITCHBOARD_RECONCILE_INTERVAL_SECS` | `0` | Seconds between reconcile passes. **`0` disables it.** Requires `--kv-secret-file` and a resolvable API site key; a backstop, not the hot path, so a cadence well above the drain tick (e.g. 30–60s) is right. |
+| `--reconcile-prune` | `SWITCHBOARD_RECONCILE_PRUNE` | `false` | Actually remove orphaned previews. **Off by default**: until set, the pass is observability-only and logs each orphan it *would* prune at `WARN`. |
+| `--reconcile-deploy-missing` | `SWITCHBOARD_RECONCILE_DEPLOY_MISSING` | `false` | Enqueue a deploy for a desired preview whose directory is absent on this node (recovers a deploy a node never saw). Off by default — re-provisioning runs composer and a checkout. |
+| `--reconcile-keep-sites` | `SWITCHBOARD_RECONCILE_KEEP_SITES` | *(empty)* | Comma-separated vhost directory names never pruned — the node's infra/test sites (e.g. `switchboard,site-a,site-b`). The API's own site key is always protected in addition. |
+| `--reconcile-max-prunes-per-cycle` | `SWITCHBOARD_RECONCILE_MAX_PRUNES_PER_CYCLE` | `8` | Blast-radius guard: at most this many orphans are removed per pass, the excess deferred to later passes so a misconfiguration cannot wipe the fleet in one tick. |
+| `--reconcile-api-site` | `SWITCHBOARD_RECONCILE_API_SITE` | *(derived from `--drain-host`)* | The switchboard-api vhost's canonical site key, whose keyspace holds the `switchboard:*` desired-state keys. Set only if it differs from the drain host's site key. |
+
+Rollout is fail-safe: enable the interval first and read the dry-run `WOULD
+prune` logs; turn on `--reconcile-prune` once they look right; add
+`--reconcile-deploy-missing` last. A KV read failure aborts the whole pass — the
+reconcile prunes **nothing** rather than act against an authority it could not
+read.
+
 ### GitHub reporting (optional)
 
 | Flag | Env | Default | Meaning |
diff --git a/src/config.rs b/src/config.rs
index 87b6fb0..38a98ab 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -271,6 +271,76 @@ pub struct Config {
     #[arg(long, default_value = "127.0.0.1:6379", env = "SWITCHBOARD_KV_ADDR")]
     pub kv_addr: String,
 
+    // ── level-triggered reconcile ──────────────────────────────────────
+    /// Seconds between **reconcile** passes. **Zero disables it** (the default).
+    ///
+    /// The drain/queue path is edge-triggered on `switchboard:gen`: a node only
+    /// re-walks desired state when the counter advances, so a `gen` increment
+    /// that is lost or not-yet-gossiped strands a teardown (the preview stays
+    /// served, its database and override on disk) with every health check green
+    /// (switchboard#24). This pass is the level-triggered safety net: every
+    /// interval it reads the KV desired state *directly* (independent of `gen`)
+    /// and converges this node's on-disk previews to it — pruning orphans and,
+    /// with `--reconcile-deploy-missing`, enqueuing deploys it never saw.
+    ///
+    /// Requires `--kv-secret-file` (the pass reads desired state over RESP) and a
+    /// resolvable API site key (`--reconcile-api-site`, else derived from
+    /// `--drain-host`); the daemon refuses to start with a non-zero interval and
+    /// neither. A sane cadence is well above the two-second drain tick — it is a
+    /// backstop, not the hot path — e.g. 30–60s.
+    #[arg(long, default_value_t = 0, env = "SWITCHBOARD_RECONCILE_INTERVAL_SECS")]
+    pub reconcile_interval_secs: u64,
+
+    /// Actually remove orphaned previews the reconcile finds. **Off by default**:
+    /// until it is set the pass is observability-only, logging each orphan it
+    /// *would* prune at WARN (which alone surfaces the drift the original
+    /// incident was found by hand-diffing three nodes for). Turn it on once the
+    /// dry-run logs look right.
+    #[arg(long, default_value_t = false, env = "SWITCHBOARD_RECONCILE_PRUNE")]
+    pub reconcile_prune: bool,
+
+    /// Enqueue a deploy for a desired preview whose directory is absent on this
+    /// node. **Off by default**: re-provisioning runs composer and a checkout,
+    /// so recovering a missed *deploy* is a heavier, separate opt-in from
+    /// removing a stranded *teardown*. When on, the reconcile writes the exact
+    /// job document into the queue and the normal deploy path runs it.
+    #[arg(
+        long,
+        default_value_t = false,
+        env = "SWITCHBOARD_RECONCILE_DEPLOY_MISSING"
+    )]
+    pub reconcile_deploy_missing: bool,
+
+    /// Comma-separated vhost directory names the reconcile must never prune —
+    /// the node's infra/test sites that are not PR previews (e.g.
+    /// `switchboard,site-a,site-b`). The API's own site key is always protected
+    /// in addition to this list. A preview directory is only ever a prune
+    /// candidate when it is absent from this set *and* its own KV preview key is
+    /// gone or torn down.
+    #[arg(long, default_value = "", env = "SWITCHBOARD_RECONCILE_KEEP_SITES")]
+    pub reconcile_keep_sites: String,
+
+    /// Upper bound on orphans pruned in a single pass — a blast-radius guard.
+    /// A misconfiguration (wrong keep-list, a flushed KV) cannot then wipe the
+    /// fleet in one tick; the excess is deferred to later passes, giving the
+    /// WARN/INFO logs time to be noticed. Floored at one.
+    #[arg(
+        long,
+        default_value_t = 8,
+        env = "SWITCHBOARD_RECONCILE_MAX_PRUNES_PER_CYCLE"
+    )]
+    pub reconcile_max_prunes_per_cycle: usize,
+
+    /// The switchboard-api vhost's canonical **site key**, whose gossip-replicated
+    /// keyspace holds the `switchboard:*` desired-state keys the reconcile reads.
+    ///
+    /// Defaults to the site key derived from `--drain-host` (the same host the
+    /// drain kick addresses), which is correct whenever the API is reached at its
+    /// own vhost. Set it explicitly only if the API's KV site key differs from
+    /// its drain host.
+    #[arg(long, env = "SWITCHBOARD_RECONCILE_API_SITE")]
+    pub reconcile_api_site: Option<String>,
+
     // ── pre-serve static-analysis gate ─────────────────────────────────
     /// Path to the operator-controlled `ephpm analyze` policy file (YAML), used
     /// to screen a preview's checkout before it is published.
@@ -445,6 +515,52 @@ impl Config {
         }
     }
 
+    /// Whether the level-triggered reconcile pass is enabled.
+    #[must_use]
+    pub fn reconcile_enabled(&self) -> bool {
+        self.reconcile_interval_secs > 0
+    }
+
+    /// The switchboard-api vhost's canonical site key, whose keyspace the
+    /// reconcile reads: the explicit `--reconcile-api-site` if set, otherwise
+    /// derived from `--drain-host` with this node's suffix rule.
+    ///
+    /// `Ok(None)` when neither an override nor a drain host is available.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if a drain host is present but does not normalize to a
+    /// valid site key.
+    pub fn reconcile_api_site(&self) -> anyhow::Result<Option<String>> {
+        if let Some(explicit) = &self.reconcile_api_site {
+            return Ok(Some(explicit.clone()));
+        }
+        let Some(host) = &self.drain_host else {
+            return Ok(None);
+        };
+        let suffix = self.effective_sites_domain_suffix();
+        crate::site_key::site_key(host, suffix.as_deref())
+            .map(Some)
+            .with_context(|| {
+                format!("cannot derive the switchboard-api site key from --drain-host {host:?}")
+            })
+    }
+
+    /// The reconcile keep-list — vhost names never pruned — from
+    /// `--reconcile-keep-sites`, with the API site key always added.
+    #[must_use]
+    pub fn reconcile_keep_set(&self, api_site: &str) -> std::collections::HashSet<String> {
+        let mut set: std::collections::HashSet<String> = self
+            .reconcile_keep_sites
+            .split(',')
+            .map(str::trim)
+            .filter(|s| !s.is_empty())
+            .map(str::to_string)
+            .collect();
+        set.insert(api_site.to_string());
+        set
+    }
+
     /// Check the combinations clap cannot express.
     ///
     /// # Errors
@@ -483,6 +599,24 @@ impl Config {
             "--fork-secrets has no effect without --allow-fork-deploy — set \
              both to build forks with operator secrets, or neither"
         );
+        // The reconcile reads desired state over RESP and addresses the API's
+        // KV keyspace by site key; without a secret or a resolvable site key it
+        // is silent dead configuration, so refuse to start rather than run a
+        // pass that can do nothing.
+        if self.reconcile_enabled() {
+            anyhow::ensure!(
+                self.kv_secret_file.is_some(),
+                "--reconcile-interval-secs is non-zero but --kv-secret-file is unset — \
+                 the reconcile reads cluster desired state over RESP and cannot without \
+                 ePHPm's [kv] secret (set it to 0 to disable the reconcile)"
+            );
+            anyhow::ensure!(
+                self.reconcile_api_site()?.is_some(),
+                "--reconcile-interval-secs is non-zero but the switchboard-api site key \
+                 could not be resolved — set --reconcile-api-site, or --drain-host to \
+                 derive it from"
+            );
+        }
         // A daemon that cannot remove a tenant's database when its PR closes is
         // a data-retention problem, and the live preview cluster hit it exactly
         // this way: a hand-provisioned systemd unit passed neither path, so
diff --git a/src/kv.rs b/src/kv.rs
index 50c283d..afbc485 100644
--- a/src/kv.rs
+++ b/src/kv.rs
@@ -305,6 +305,177 @@ impl VerdictCache {
     }
 }
 
+/// A reader for switchboard-api's **cluster desired-state** keys in ePHPm's KV.
+///
+/// The webhook path publishes desired state into the switchboard-api vhost's own
+/// gossip-replicated keyspace ([`crate::reconcile`] and switchboard-api's
+/// `ClusterState`): `switchboard:index` (a JSON array of every live label) and
+/// `switchboard:preview:<label>` (the full schema-1 job document). ePHPm's
+/// `ephpm_kv_*` surface has **no SCAN**, which is the whole reason the index is
+/// hand-rolled — so enumerating desired state means reading the index, then the
+/// per-label preview keys it points at.
+///
+/// This reader is what makes the daemon's reconcile **level-triggered**. The PHP
+/// `DrainHandler` only walks the index when `switchboard:gen` has advanced past
+/// the node's `last_gen`; because `gen` is a *separate* gossip key from the
+/// content it guards, an increment that is lost or not-yet-replicated leaves a
+/// node's cursor "current" at a generation whose teardown it never materialized,
+/// and every later drain short-circuits (switchboard#24, and the
+/// gen-vs-key-content propagation race). Reading the content keys **directly**,
+/// every interval, removes that dependency: the daemon converges on the KV's
+/// actual contents regardless of the counter.
+///
+/// # Scope
+///
+/// The keys live in the switchboard-api vhost's keyspace, reached with the same
+/// `AUTH <site> <derived>` scoping [`KvRevoker`] uses — here the site is the API
+/// vhost's own canonical site key (`api_site`), derived from the daemon's
+/// `--drain-host`. The password is `HMAC-SHA256(kv_secret, api_site)`.
+///
+/// # Fail-safe
+///
+/// Every read is all-or-nothing: a transport/auth failure returns `Err`, and the
+/// reconcile treats that as "KV unavailable this cycle" and does **nothing**
+/// (never prunes on an unreadable authority). Only a definitive reply — a parsed
+/// index, a bulk value, or a genuine nil — drives a decision.
+#[derive(Debug, Clone)]
+pub struct ClusterReader {
+    addr: String,
+    kv_secret: String,
+    api_site: String,
+}
+
+/// One consistent read of the cluster desired state: the index, plus the raw
+/// preview document for every label asked about (index labels ∪ the caller's
+/// on-disk labels). A `None` document is a genuine KV nil (the key is absent —
+/// a retired preview whose TTL has elapsed), distinct from a label simply not
+/// being present in the map.
+#[derive(Debug, Clone)]
+pub struct ClusterSnapshot {
+    /// Every label listed in `switchboard:index` at read time.
+    pub index: Vec<String>,
+    /// `label -> Some(raw job JSON)` when the preview key is present,
+    /// `label -> None` when it is a genuine nil.
+    pub docs: std::collections::HashMap<String, Option<String>>,
+}
+
+impl ClusterReader {
+    /// Build a reader for the switchboard-api keyspace.
+    ///
+    /// `addr` is ePHPm's RESP listener, `kv_secret` its `[kv] secret`, and
+    /// `api_site` the API vhost's canonical site key (whose keyspace holds the
+    /// `switchboard:*` desired-state keys).
+    #[must_use]
+    pub fn new(
+        addr: impl Into<String>,
+        kv_secret: impl Into<String>,
+        api_site: impl Into<String>,
+    ) -> Self {
+        Self {
+            addr: addr.into(),
+            kv_secret: kv_secret.into(),
+            api_site: api_site.into(),
+        }
+    }
+
+    /// Read the index and every requested preview document under one connection.
+    ///
+    /// `extra_labels` are labels the caller wants documents for that may not be
+    /// in the index — the labels reconstructed from what is actually on disk, so
+    /// an orphan whose index entry was already pruned is still classified from
+    /// its own (now-nil) preview key rather than guessed at.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the connection, AUTH, index read, or any preview read
+    /// fails. A partial read is never returned — the caller must be able to
+    /// trust that a missing document means a genuine nil, not a dropped frame.
+    pub async fn snapshot(&self, extra_labels: &[String]) -> anyhow::Result<ClusterSnapshot> {
+        tokio::time::timeout(OP_TIMEOUT, self.snapshot_inner(extra_labels))
+            .await
+            .with_context(|| format!("KV desired-state read from {} timed out", self.addr))?
+    }
+
+    async fn snapshot_inner(&self, extra_labels: &[String]) -> anyhow::Result<ClusterSnapshot> {
+        let password = derive_site_kv_password(&self.kv_secret, &self.api_site);
+
+        let stream = TcpStream::connect(&self.addr)
+            .await
+            .with_context(|| format!("cannot connect to ePHPm KV listener at {}", self.addr))?;
+        let (read_half, mut write_half) = stream.into_split();
+        let mut reader = BufReader::new(read_half);
+
+        write_half
+            .write_all(&encode_command(&["AUTH", &self.api_site, &password]))
+            .await
+            .context("failed to send KV AUTH")?;
+        read_reply(&mut reader)
+            .await
+            .context("KV AUTH was rejected")?;
+
+        // The index first, then the union of its labels and the caller's.
+        write_half
+            .write_all(&encode_command(&["GET", INDEX_KEY]))
+            .await
+            .context("failed to send KV GET for the index")?;
+        let index_raw = read_bulk_reply(&mut reader)
+            .await
+            .context("reading switchboard:index")?;
+        let index = parse_index(index_raw.as_deref());
+
+        let mut labels: Vec<String> = index.clone();
+        for label in extra_labels {
+            if !labels.contains(label) {
+                labels.push(label.clone());
+            }
+        }
+
+        let mut docs = std::collections::HashMap::with_capacity(labels.len());
+        for label in labels {
+            let key = format!("{PREVIEW_PREFIX}{label}");
+            write_half
+                .write_all(&encode_command(&["GET", &key]))
+                .await
+                .with_context(|| format!("failed to send KV GET for {key}"))?;
+            let value = read_bulk_reply(&mut reader)
+                .await
+                .with_context(|| format!("reading {key}"))?;
+            docs.insert(label, value);
+        }
+
+        let _ = write_half.write_all(&encode_command(&["QUIT"])).await;
+        Ok(ClusterSnapshot { index, docs })
+    }
+}
+
+/// `switchboard:index` — the JSON array of live labels.
+const INDEX_KEY: &str = "switchboard:index";
+/// Prefix of a per-label preview key: `switchboard:preview:<label>`.
+const PREVIEW_PREFIX: &str = "switchboard:preview:";
+
+/// Parse `switchboard:index` — a JSON array — into a list of string labels,
+/// mirroring `ClusterState::index()` (non-strings filtered out, nil ⇒ empty).
+///
+/// A malformed or non-array value is treated as an empty index rather than an
+/// error: the index is a best-effort hint for the *add* direction, and the prune
+/// direction never trusts it (it classifies each on-disk site from that site's
+/// own preview key). An empty index therefore withdraws nothing.
+fn parse_index(raw: Option<&str>) -> Vec<String> {
+    let Some(raw) = raw else {
+        return Vec::new();
+    };
+    let Ok(serde_json::Value::Array(items)) = serde_json::from_str::<serde_json::Value>(raw) else {
+        return Vec::new();
+    };
+    items
+        .into_iter()
+        .filter_map(|v| match v {
+            serde_json::Value::String(s) => Some(s),
+            _ => None,
+        })
+        .collect()
+}
+
 /// Encode a command as a RESP2 array of bulk strings — the dialect ePHPm's KV
 /// server parses (`*<n>\r\n` then `$<len>\r\n<arg>\r\n` per argument).
 fn encode_command(args: &[&str]) -> Vec<u8> {
@@ -620,6 +791,98 @@ mod tests {
         }
     }
 
+    // ── the cluster desired-state reader ────────────────────────────────
+
+    #[test]
+    fn parse_index_reads_a_json_array_of_strings() {
+        assert_eq!(
+            parse_index(Some(r#"["a","b","c"]"#)),
+            vec!["a".to_string(), "b".to_string(), "c".to_string()]
+        );
+        // Non-strings are filtered (mirrors ClusterState::index()).
+        assert_eq!(parse_index(Some(r#"["a",1,null,"b"]"#)), vec!["a", "b"]);
+        // A nil, a non-array, or malformed JSON is an empty index — the prune
+        // path never trusts the index, so "empty" withdraws nothing.
+        assert!(parse_index(None).is_empty());
+        assert!(parse_index(Some("null")).is_empty());
+        assert!(parse_index(Some("{}")).is_empty());
+        assert!(parse_index(Some("{not json")).is_empty());
+    }
+
+    /// Build one RESP bulk-string frame (`$<len>\r\n<payload>\r\n`).
+    fn bulk(payload: &str) -> Vec<u8> {
+        format!("${}\r\n{payload}\r\n", payload.len()).into_bytes()
+    }
+
+    /// **The end-to-end desired-state read.** A mock RESP server replies `+OK` to
+    /// AUTH, the index as a bulk array, then one reply per preview key in request
+    /// order (index labels ∪ the caller's extra labels). Assert the snapshot
+    /// classifies a present doc, a genuine nil, and an extra (on-disk-only) label.
+    #[tokio::test]
+    async fn snapshot_reads_index_and_previews_under_one_connection() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap().to_string();
+
+        let server = tokio::spawn(async move {
+            let (mut sock, _) = listener.accept().await.unwrap();
+            let mut out = Vec::new();
+            out.extend_from_slice(b"+OK\r\n"); // AUTH
+            out.extend_from_slice(&bulk(r#"["live","torn"]"#)); // GET index
+            out.extend_from_slice(&bulk(r#"{"intent":"deploy"}"#)); // preview live
+            out.extend_from_slice(b"$-1\r\n"); // preview torn — nil
+            out.extend_from_slice(&bulk(r#"{"intent":"deploy"}"#)); // preview ghost (extra)
+            sock.write_all(&out).await.unwrap();
+            let mut buf = Vec::new();
+            sock.read_to_end(&mut buf).await.unwrap();
+            String::from_utf8_lossy(&buf).into_owned()
+        });
+
+        let reader = ClusterReader::new(addr, "master-secret", "switchboard");
+        let snap = reader
+            .snapshot(&["ghost".to_string()])
+            .await
+            .expect("a well-formed desired-state read must succeed");
+
+        assert_eq!(snap.index, vec!["live".to_string(), "torn".to_string()]);
+        assert_eq!(
+            snap.docs.get("live"),
+            Some(&Some(r#"{"intent":"deploy"}"#.to_string()))
+        );
+        assert_eq!(
+            snap.docs.get("torn"),
+            Some(&None),
+            "a nil is a genuine miss"
+        );
+        assert_eq!(
+            snap.docs.get("ghost"),
+            Some(&Some(r#"{"intent":"deploy"}"#.to_string())),
+            "an on-disk-only label is still fetched"
+        );
+
+        let received = server.await.unwrap();
+        assert!(received.contains("AUTH"), "must AUTH: {received:?}");
+        assert!(
+            received.contains("switchboard"),
+            "AUTH must name the API site: {received:?}"
+        );
+        assert!(
+            received.contains("switchboard:index"),
+            "must read the index: {received:?}"
+        );
+        assert!(
+            received.contains("switchboard:preview:ghost"),
+            "must read the extra label's preview key: {received:?}"
+        );
+    }
+
+    /// A read against a dead listener is an `Err`, which the reconcile treats as
+    /// "KV unavailable" and prunes nothing.
+    #[tokio::test]
+    async fn snapshot_on_a_dead_addr_errors() {
+        let reader = ClusterReader::new("127.0.0.1:1", "s", "switchboard");
+        assert!(reader.snapshot(&[]).await.is_err());
+    }
+
     /// Every preview's verdict lives in the **one** switchboard-owned namespace,
     /// so the cluster-wide dedup works across the whole fleet — the cache is not
     /// scoped per preview (there is no preview-`site` parameter), only the
diff --git a/src/main.rs b/src/main.rs
index dbf8ad4..a9daca3 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -27,6 +27,7 @@ mod kv;
 mod manifest;
 mod preview_auth;
 mod queue;
+mod reconcile;
 mod secrets;
 mod site_key;
 mod site_override;
@@ -51,6 +52,7 @@ use deployer::PreviewRequest;
 use drain::DrainKicker;
 use job::Intent;
 use queue::{ClaimedJob, Queue};
+use reconcile::ReconcileContext;
 use secrets::Secrets;
 use validate::Verdict;
 
@@ -239,6 +241,21 @@ async fn main() -> anyhow::Result<()> {
     let webhook_server_enabled = config.webhook_server_enabled;
     let listen = config.listen.clone();
 
+    // Resolve the level-triggered reconcile's settings before `config` moves
+    // into the shared state. `validate()` has already guaranteed a KV secret and
+    // a resolvable API site key whenever the interval is non-zero.
+    let reconcile_setup = if config.reconcile_enabled() {
+        let api_site = config
+            .reconcile_api_site()?
+            .expect("validate() guarantees a reconcile API site key when enabled");
+        let keep = config.reconcile_keep_set(&api_site);
+        let interval = Duration::from_secs(config.reconcile_interval_secs.max(1));
+        Some((api_site, keep, interval))
+    } else {
+        info!("level-triggered reconcile disabled (--reconcile-interval-secs 0)");
+        None
+    };
+
     let state = Arc::new(AppState {
         config,
         secrets,
@@ -249,6 +266,39 @@ async fn main() -> anyhow::Result<()> {
         tokio::spawn(drain_loop(kicker, drain_interval));
     }
 
+    if let Some((api_site, keep, interval)) = reconcile_setup {
+        // The reconcile reads desired state over RESP; without a KV secret it
+        // can do nothing, and validate() only reaches here with one configured.
+        if state.kv_secret.is_some() {
+            info!(
+                interval_s = interval.as_secs(),
+                prune = state.config.reconcile_prune,
+                deploy_missing = state.config.reconcile_deploy_missing,
+                api_site = %api_site,
+                keep = keep.len(),
+                max_prunes_per_cycle = state.config.reconcile_max_prunes_per_cycle,
+                "level-triggered reconcile enabled"
+            );
+            if !state.config.reconcile_prune {
+                tracing::warn!(
+                    "reconcile is in dry-run: orphans will be logged as WOULD-prune but \
+                     NOT removed. Set --reconcile-prune once the logs look right"
+                );
+            }
+            tokio::spawn(reconcile_loop(
+                Arc::clone(&state),
+                queue.clone(),
+                api_site,
+                keep,
+                interval,
+            ));
+        } else {
+            tracing::warn!(
+                "reconcile is enabled but no KV secret resolved — reconcile pass skipped"
+            );
+        }
+    }
+
     if webhook_server_enabled {
         tracing::warn!(
             %listen,
@@ -277,6 +327,61 @@ async fn drain_loop(kicker: DrainKicker, interval: Duration) {
     }
 }
 
+// ── the level-triggered reconcile ──────────────────────────────────────
+
+/// Reconcile this node's on-disk previews to the cluster's KV desired state on a
+/// timer, independent of the drain/queue path's `switchboard:gen` cursor. Every
+/// failure mode is transient (KV briefly unreachable, a single wedged site), so
+/// nothing here is fatal — a failed pass is logged and the next tick retries.
+async fn reconcile_loop(
+    state: Arc<AppState>,
+    queue: Queue,
+    api_site: String,
+    keep: std::collections::HashSet<String>,
+    interval: Duration,
+) {
+    // Guarded at the spawn site; kept here so a future refactor cannot start the
+    // loop without the secret the reconcile reads desired state with.
+    let Some(kv_secret) = state.kv_secret.clone() else {
+        tracing::warn!("reconcile loop started without a KV secret — not running");
+        return;
+    };
+
+    let mut ticker = tokio::time::interval(interval);
+    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    loop {
+        ticker.tick().await;
+
+        let cfg = &state.config;
+        let suffix = cfg.effective_sites_domain_suffix();
+        let ctx = ReconcileContext {
+            sites_dir: &cfg.sites_dir,
+            sqlite_dir: cfg.sqlite_dir.as_deref(),
+            site_overrides_dir: cfg.site_overrides_dir.as_deref(),
+            vhost_temp_base: cfg.vhost_temp_base.as_deref(),
+            state_dir: &cfg.state_dir,
+            allow_incomplete: cfg.allow_incomplete_teardown,
+            kv_addr: &cfg.kv_addr,
+            kv_secret: &kv_secret,
+            api_site: &api_site,
+            preview_domain: &cfg.preview_domain,
+            sites_domain_suffix: suffix.as_deref(),
+            keep: &keep,
+            prune: cfg.reconcile_prune,
+            deploy_missing: cfg.reconcile_deploy_missing,
+            max_prunes_per_cycle: cfg.reconcile_max_prunes_per_cycle.max(1),
+            queue: &queue,
+        };
+
+        if let Err(e) = reconcile::reconcile_once(&ctx).await {
+            tracing::warn!(
+                error = %format!("{e:#}"),
+                "reconcile pass failed (KV unreadable?) — nothing pruned; will retry"
+            );
+        }
+    }
+}
+
 // ── the queue consumer ─────────────────────────────────────────────────
 
 /// Scan, claim, coalesce and process, forever.
diff --git a/src/queue.rs b/src/queue.rs
index d0046aa..9bfb7e2 100644
--- a/src/queue.rs
+++ b/src/queue.rs
@@ -248,6 +248,92 @@ impl Queue {
         Ok(claimed)
     }
 
+    /// Whether a job for `label` is already waiting in `queue/` or in flight in
+    /// `claimed/`.
+    ///
+    /// The reconcile's *add* direction uses this to avoid re-enqueuing a deploy
+    /// for a preview that is already being provisioned: the daemon's coalescing
+    /// would collapse the duplicates anyway, but not writing them keeps the
+    /// queue readable and a slow deploy from being restarted every interval.
+    ///
+    /// Best-effort and cheap: it parses only enough of each pending/claimed file
+    /// to read its label, and a file it cannot read or parse is simply not a
+    /// match (the caller then enqueues, which is safe — worst case a coalesced
+    /// duplicate).
+    ///
+    /// # Errors
+    ///
+    /// Returns an error only if `queue/` exists but cannot be listed.
+    pub fn has_job_for_label(&self, label: &str) -> anyhow::Result<bool> {
+        for dir in [&self.queue_dir, &self.claimed_dir] {
+            let entries = match std::fs::read_dir(dir) {
+                Ok(e) => e,
+                Err(e) if e.kind() == ErrorKind::NotFound => continue,
+                Err(e) => {
+                    return Err(e).with_context(|| format!("failed to read {}", dir.display()));
+                }
+            };
+            for entry in entries.flatten() {
+                if !entry.file_type().is_ok_and(|t| t.is_file()) {
+                    continue;
+                }
+                let name = entry.file_name().to_string_lossy().into_owned();
+                if !name.ends_with(".json") {
+                    continue;
+                }
+                let Ok(bytes) = std::fs::read(entry.path()) else {
+                    continue;
+                };
+                if let Ok(job) = Job::parse(&bytes) {
+                    if job.label() == label {
+                        return Ok(true);
+                    }
+                }
+            }
+        }
+        Ok(false)
+    }
+
+    /// Write a job document into `queue/` under a fresh, contract-shaped
+    /// filename (`<13-digit millis>-<16 hex>.json`), atomically.
+    ///
+    /// This is the mirror of the switchboard-api `DrainHandler`'s materialize
+    /// step, on the daemon side: the reconcile writes the exact job document it
+    /// read from cluster desired state so the existing [`Queue::claim_pending`]
+    /// → coalesce → validate → deploy path runs it, rather than the reconcile
+    /// re-implementing provisioning. A materialized job carries a fresh filename
+    /// on every node, so there is nothing to deduplicate against here (per-node
+    /// materialization of shared state, not a webhook redelivery).
+    ///
+    /// The write is atomic — a temp file in the same directory, then `rename` —
+    /// so a scanner never sees a half-written job.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the queue directory cannot be created or written.
+    pub fn enqueue(&self, job_json: &[u8]) -> anyhow::Result<String> {
+        std::fs::create_dir_all(&self.queue_dir).with_context(|| {
+            format!(
+                "failed to create queue directory {}",
+                self.queue_dir.display()
+            )
+        })?;
+
+        let name = format!("{}-{}.json", now_millis_13(), random_16_hex());
+        let final_path = self.queue_dir.join(&name);
+        let tmp_path = self.queue_dir.join(format!(".{name}.tmp"));
+
+        std::fs::write(&tmp_path, job_json)
+            .with_context(|| format!("failed to write {}", tmp_path.display()))?;
+        match std::fs::rename(&tmp_path, &final_path) {
+            Ok(()) => Ok(name),
+            Err(e) => {
+                let _ = std::fs::remove_file(&tmp_path);
+                Err(e).with_context(|| format!("failed to publish job {}", final_path.display()))
+            }
+        }
+    }
+
     /// Remove a finished job from `claimed/`.
     ///
     /// # Errors
@@ -287,6 +373,35 @@ pub fn enqueued_at_ms(name: &str) -> Option<u64> {
     millis.parse().ok()
 }
 
+/// The current Unix time as a 13-digit millisecond string — the timestamp half
+/// of a queue filename. Fixed-width until 2286, which is what keeps a
+/// lexicographic sort of `queue/` chronological (see [`Queue::pending`]).
+fn now_millis_13() -> String {
+    let millis = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map_or(0, |d| d.as_millis());
+    // 13 digits covers 1970-01-01 … 2286; `{:013}` pads the (implausible) case
+    // of a clock set before ~2001 so the field never shrinks below 13.
+    format!("{millis:013}")
+}
+
+/// 16 lowercase hex chars from 8 CSPRNG bytes — the random half of a queue
+/// filename, making a collision within a millisecond effectively impossible.
+fn random_16_hex() -> String {
+    let mut bytes = [0u8; 8];
+    // A CSPRNG failure is not fatal here: the filename only needs to be unique,
+    // not unguessable, and the millisecond prefix plus the process's own
+    // scheduling already separates concurrent writers. Fall back to the low
+    // bits of the nanosecond clock rather than aborting a reconcile.
+    if getrandom::fill(&mut bytes).is_err() {
+        let nanos = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .map_or(0, |d| d.subsec_nanos());
+        bytes[..4].copy_from_slice(&nanos.to_le_bytes());
+    }
+    hex::encode(bytes)
+}
+
 /// A file's mtime in Unix milliseconds — the fallback enqueue time.
 ///
 /// The claim is a `link()`, which preserves mtime, so the claimed file still
@@ -559,6 +674,72 @@ mod tests {
         );
     }
 
+    #[test]
+    fn enqueue_writes_a_contract_shaped_claimable_job() {
+        let (_d, q) = queue();
+        let name = q
+            .enqueue(sample_json("recon-pr-1", "deploy").as_bytes())
+            .unwrap();
+
+        // The filename is the queue contract's shape, so it sorts chronologically
+        // and carries a parseable enqueue time.
+        assert!(name.ends_with(".json"));
+        assert!(
+            enqueued_at_ms(&name).is_some(),
+            "{name} must parse as an enqueue time"
+        );
+
+        // And it round-trips through the normal consume path.
+        let claimed = q.claim_pending().unwrap();
+        assert_eq!(claimed.len(), 1);
+        assert_eq!(claimed[0].job.label(), "recon-pr-1");
+        assert_eq!(claimed[0].job.intent().unwrap(), crate::job::Intent::Deploy);
+    }
+
+    #[test]
+    fn enqueue_leaves_no_temp_file_behind() {
+        let (_d, q) = queue();
+        q.enqueue(sample_json("recon-pr-1", "deploy").as_bytes())
+            .unwrap();
+        // The atomic write uses a dotfile temp then renames; none must linger.
+        for entry in std::fs::read_dir(q.queue_dir()).unwrap().flatten() {
+            let name = entry.file_name().to_string_lossy().into_owned();
+            assert!(
+                !name.ends_with(".tmp"),
+                "a staging temp file leaked: {name}"
+            );
+        }
+    }
+
+    #[test]
+    fn has_job_for_label_sees_pending_and_claimed() {
+        let (_d, q) = queue();
+        assert!(!q.has_job_for_label("recon-pr-1").unwrap());
+
+        write_job(
+            &q,
+            "1787456737243-bbbbbbbbbbbbbbbb.json",
+            "recon-pr-1",
+            "deploy",
+        );
+        assert!(
+            q.has_job_for_label("recon-pr-1").unwrap(),
+            "a pending job for the label is seen"
+        );
+        assert!(
+            !q.has_job_for_label("other-pr-2").unwrap(),
+            "a different label is not a match"
+        );
+
+        // Once claimed (in flight) it still counts, so the reconcile does not
+        // re-enqueue a deploy that is already running.
+        q.claim("1787456737243-bbbbbbbbbbbbbbbb.json").unwrap();
+        assert!(
+            q.has_job_for_label("recon-pr-1").unwrap(),
+            "a claimed job for the label is still seen"
+        );
+    }
+
     #[test]
     fn complete_removes_the_claim_and_is_idempotent() {
         let (_d, q) = queue();
diff --git a/src/reconcile.rs b/src/reconcile.rs
new file mode 100644
index 0000000..a81311d
--- /dev/null
+++ b/src/reconcile.rs
@@ -0,0 +1,787 @@
+//! Level-triggered convergence of this node's on-disk previews to the cluster's
+//! desired state.
+//!
+//! # The bug this closes
+//!
+//! Preview fan-out is edge-triggered. switchboard-api publishes desired state
+//! into gossip-replicated KV and bumps a `switchboard:gen` counter; each node's
+//! PHP `DrainHandler` only walks `switchboard:index` when `gen` has advanced
+//! past its own `last_gen`, and once it records itself current at a generation
+//! it never re-walks. That makes correctness depend on the counter, and the
+//! counter is a *separate* gossip key from the content it guards:
+//!
+//! * a `gen` increment that is lost or not-yet-replicated leaves a node's cursor
+//!   "current" at a generation whose teardown it never materialized — so every
+//!   later drain short-circuits and the torn-down preview stays served, with its
+//!   tenant database and docroot override on disk, every health check green
+//!   (switchboard#24, and the gen-vs-key-content propagation race);
+//! * the index itself is a read-modify-write with an acknowledged lost-update
+//!   window, so a label can transiently vanish from it.
+//!
+//! Both are the same class of fault: an edge trigger that misses a state it
+//! never observed an *event* for. The symptom on the live cluster is override
+//! `.toml` counts drifting between nodes and orphaned overrides for torn-down
+//! previews lingering as 404s.
+//!
+//! # The fix: reconcile the level, not the edge
+//!
+//! This module runs on its own interval, independent of `gen`, and converges the
+//! node's actual on-disk state to what the KV *currently says*:
+//!
+//! * **Prune.** For every preview directory under `sites_dir`, it reads that
+//!   site's **own** `switchboard:preview:<label>` key ([`crate::kv::ClusterReader`]).
+//!   A key that is present with `intent: deploy` is live and kept; a key that is
+//!   a genuine nil (retired — its TTL elapsed) or carries `intent: teardown` is
+//!   an orphan and is removed with the same KEEP-guarded, exact-path
+//!   [`crate::teardown::teardown_preview`] the webhook teardown path uses. The
+//!   per-site key is a plain `set` (only teardown ever gives it a TTL), so it is
+//!   immune to *both* the `gen` propagation race and the index lost-update race —
+//!   pruning never trusts the index.
+//! * **Add.** For every label the index lists whose preview key says `deploy`
+//!   but whose directory is absent, it enqueues the exact job document into the
+//!   node's queue, so the existing claim → coalesce → validate → deploy path
+//!   provisions it. This recovers a deploy a node simply never saw.
+//!
+//! # Why this is safe under concurrent drains
+//!
+//! Two nodes reconciling the same desired state converge rather than fight:
+//! teardown and enqueue are both idempotent (teardown tolerates already-absent
+//! artifacts; a materialized job carries a fresh per-node filename and is
+//! coalesced). The gen counter is kept as the PHP fast-path's optimization hint;
+//! nothing here depends on it.
+//!
+//! # Fail-safe posture
+//!
+//! * A KV read failure aborts the whole cycle — the reconcile does **nothing**
+//!   rather than prune against an unreadable authority.
+//! * Infra/non-preview vhosts (`switchboard`, `site-a`, …) are protected by an
+//!   explicit keep-list plus the API's own site key.
+//! * Pruning is opt-in (`--reconcile-prune`); until it is set the reconcile logs
+//!   what it *would* remove at WARN, which alone would have surfaced the original
+//!   incident that was only found by hand-diffing three nodes.
+//! * A per-cycle prune cap bounds the blast radius of a misconfiguration to a few
+//!   removals per interval, leaving the loud logs time to be noticed.
+
+use std::collections::{HashMap, HashSet};
+use std::path::Path;
+
+use crate::job::{Intent, Job};
+use crate::kv::ClusterReader;
+use crate::queue::Queue;
+use crate::site_key;
+use crate::teardown::{self, Preview, TeardownContext};
+
+/// Everything one reconcile pass needs. Mirrors the teardown/deploy context the
+/// job path already assembles from [`crate::config::Config`], plus the reconcile
+/// policy knobs.
+pub struct ReconcileContext<'a> {
+    // ── where the artifacts live (same as `TeardownContext`) ────────────
+    pub sites_dir: &'a Path,
+    pub sqlite_dir: Option<&'a Path>,
+    pub site_overrides_dir: Option<&'a Path>,
+    pub vhost_temp_base: Option<&'a Path>,
+    pub state_dir: &'a Path,
+    pub allow_incomplete: bool,
+
+    // ── KV desired-state source ─────────────────────────────────────────
+    /// ePHPm's RESP listener (`[kv.redis_compat] listen`).
+    pub kv_addr: &'a str,
+    /// ePHPm's `[kv] secret`. Required — the reconcile reads desired state over
+    /// RESP, so a node without it cannot reconcile and the loop is not started.
+    pub kv_secret: &'a str,
+    /// The switchboard-api vhost's canonical site key, whose keyspace holds the
+    /// `switchboard:*` desired-state keys.
+    pub api_site: &'a str,
+
+    // ── host → site-key derivation (same rule as the deploy path) ────────
+    pub preview_domain: &'a str,
+    pub sites_domain_suffix: Option<&'a str>,
+
+    // ── policy ──────────────────────────────────────────────────────────
+    /// Vhost directory names never touched by pruning (infra/test sites). The
+    /// API site key is always added to this set by the caller.
+    pub keep: &'a HashSet<String>,
+    /// Actually remove orphans. When `false` the reconcile is observability-only:
+    /// it logs each orphan it *would* prune at WARN and removes nothing.
+    pub prune: bool,
+    /// Enqueue deploys for desired previews whose directory is absent.
+    pub deploy_missing: bool,
+    /// Upper bound on prunes performed in a single pass (blast-radius guard).
+    pub max_prunes_per_cycle: usize,
+
+    /// The node's job queue (for the add direction).
+    pub queue: &'a Queue,
+}
+
+/// What one reconcile pass observed and did. Returned for logging and tests.
+#[derive(Debug, Default, PartialEq, Eq)]
+pub struct ReconcileReport {
+    /// Preview directories found under `sites_dir` (excluding keep-list and
+    /// staging `.tmp` dirs).
+    pub on_disk: usize,
+    /// Directories whose own preview key says `deploy` — live, kept.
+    pub live: usize,
+    /// Orphans removed (0 when `--reconcile-prune` is unset).
+    pub pruned: usize,
+    /// Orphans that would be removed but for `--reconcile-prune` being unset.
+    pub would_prune: usize,
+    /// Orphans whose removal returned an error.
+    pub prune_failed: usize,
+    /// Orphans left for a later pass because the per-cycle cap was hit.
+    pub prune_deferred: usize,
+    /// Deploys enqueued for desired-but-absent previews.
+    pub deployed: usize,
+    /// Desired-but-absent previews skipped because a job for them is already
+    /// queued or in flight.
+    pub deploy_skipped: usize,
+}
+
+/// A single orphan to remove: its site key (names every ePHPm artifact) and its
+/// reconstructed label (names the API's `applied/<label>` marker).
+#[derive(Debug, Clone, PartialEq, Eq)]
+struct PrunePlan {
+    site_key: String,
+    label: String,
+}
+
+/// The decisions a pass reached, separated from executing them so the logic is
+/// unit-testable without a filesystem or a KV server.
+#[derive(Debug, Default, PartialEq, Eq)]
+struct Plan {
+    prune: Vec<PrunePlan>,
+    deploy: Vec<String>,
+}
+
+/// Run one reconcile pass.
+///
+/// # Errors
+///
+/// Returns an error only if the KV desired-state read fails — the caller logs it
+/// and waits for the next interval. Individual prune failures are counted in the
+/// [`ReconcileReport`], not surfaced as an error, so one wedged site does not
+/// stop the rest converging.
+pub async fn reconcile_once(ctx: &ReconcileContext<'_>) -> anyhow::Result<ReconcileReport> {
+    let suffix = ctx.sites_domain_suffix;
+    let domain = ctx.preview_domain;
+
+    // (1) What is actually on disk.
+    let on_disk = list_preview_dirs(ctx.sites_dir, ctx.keep)?;
+
+    // (2) The labels those directories reconstruct to, so the snapshot can
+    // classify an orphan whose index entry was already pruned from its own key.
+    let mut disk_labels: Vec<String> = Vec::with_capacity(on_disk.len());
+    for site in &on_disk {
+        let label = label_for_site_key(site, suffix, domain);
+        if !disk_labels.contains(&label) {
+            disk_labels.push(label);
+        }
+    }
+
+    // (3) One consistent read of desired state. A failure aborts the pass — we
+    // never prune against an authority we could not read.
+    let reader = ClusterReader::new(ctx.kv_addr, ctx.kv_secret, ctx.api_site);
+    let snapshot = reader.snapshot(&disk_labels).await?;
+
+    // (4) Decide.
+    let plan = build_plan(
+        &on_disk,
+        &snapshot.index,
+        &snapshot.docs,
+        suffix,
+        domain,
+        ctx.deploy_missing,
+    );
+
+    // (5) Execute.
+    let mut report = ReconcileReport {
+        on_disk: on_disk.len(),
+        live: on_disk.len().saturating_sub(plan.prune.len()),
+        ..ReconcileReport::default()
+    };
+    execute_prune(ctx, &plan, &mut report).await;
+    execute_deploy(ctx, &plan, &snapshot.docs, &mut report);
+
+    tracing::info!(
+        on_disk = report.on_disk,
+        live = report.live,
+        pruned = report.pruned,
+        would_prune = report.would_prune,
+        prune_failed = report.prune_failed,
+        prune_deferred = report.prune_deferred,
+        deployed = report.deployed,
+        deploy_skipped = report.deploy_skipped,
+        prune_enabled = ctx.prune,
+        deploy_missing_enabled = ctx.deploy_missing,
+        "reconcile pass complete"
+    );
+    Ok(report)
+}
+
+/// Remove (or, in dry-run, report) each planned orphan.
+async fn execute_prune(ctx: &ReconcileContext<'_>, plan: &Plan, report: &mut ReconcileReport) {
+    for target in &plan.prune {
+        if report.pruned + report.prune_failed >= ctx.max_prunes_per_cycle {
+            // Bound the blast radius: the rest wait for the next pass, so a
+            // misconfiguration cannot wipe the fleet in one tick and the WARN
+            // logs above have time to be noticed.
+            report.prune_deferred += 1;
+            continue;
+        }
+
+        if !ctx.prune {
+            report.would_prune += 1;
+            tracing::warn!(
+                site_key = %target.site_key,
+                label = %target.label,
+                "reconcile: orphan preview on disk with no live desired state — \
+                 WOULD prune (set --reconcile-prune to remove it)"
+            );
+            continue;
+        }
+
+        let tctx = TeardownContext {
+            sites_dir: ctx.sites_dir,
+            sqlite_dir: ctx.sqlite_dir,
+            site_overrides_dir: ctx.site_overrides_dir,
+            vhost_temp_base: ctx.vhost_temp_base,
+            state_dir: ctx.state_dir,
+            allow_incomplete: ctx.allow_incomplete,
+            kv_secret: Some(ctx.kv_secret),
+            kv_addr: ctx.kv_addr,
+        };
+        let preview = Preview {
+            site_key: &target.site_key,
+            label: &target.label,
+        };
+        match teardown::teardown_preview(&preview, &tctx).await {
+            Ok(()) => {
+                report.pruned += 1;
+                tracing::info!(
+                    site_key = %target.site_key,
+                    "reconcile: pruned an orphan preview (no live desired state)"
+                );
+            }
+            Err(e) => {
+                report.prune_failed += 1;
+                tracing::error!(
+                    site_key = %target.site_key,
+                    error = %format!("{e:#}"),
+                    "reconcile: failed to prune an orphan preview"
+                );
+            }
+        }
+    }
+}
+
+/// Enqueue a deploy for each planned add whose job is not already queued.
+fn execute_deploy(
+    ctx: &ReconcileContext<'_>,
+    plan: &Plan,
+    docs: &HashMap<String, Option<String>>,
+    report: &mut ReconcileReport,
+) {
+    for label in &plan.deploy {
+        // A job for an absent-but-desired preview is only worth writing if one
+        // is not already in flight — the daemon's coalescing would collapse a
+        // duplicate anyway, but not writing it keeps a slow deploy from being
+        // restarted every interval.
+        match ctx.queue.has_job_for_label(label) {
+            Ok(true) => {
+                report.deploy_skipped += 1;
+                continue;
+            }
+            Ok(false) => {}
+            Err(e) => {
+                tracing::warn!(
+                    %label,
+                    error = %format!("{e:#}"),
+                    "reconcile: could not check the queue for an existing job — enqueuing anyway"
+                );
+            }
+        }
+
+        let Some(Some(raw)) = docs.get(label) else {
+            // Classified deploy above, so the document was present; this only
+            // trips on a race where it vanished between read and here.
+            continue;
+        };
+        match ctx.queue.enqueue(raw.as_bytes()) {
+            Ok(name) => {
+                report.deployed += 1;
+                tracing::info!(
+                    %label,
+                    job = %name,
+                    "reconcile: enqueued a deploy for a desired preview missing from disk"
+                );
+            }
+            Err(e) => tracing::error!(
+                %label,
+                error = %format!("{e:#}"),
+                "reconcile: failed to enqueue a missing deploy"
+            ),
+        }
+    }
+}
+
+/// Decide what to prune and what to deploy. Pure — no IO — so it can be tested
+/// against hand-built desired-state maps.
+fn build_plan(
+    on_disk: &[String],
+    index: &[String],
+    docs: &HashMap<String, Option<String>>,
+    suffix: Option<&str>,
+    preview_domain: &str,
+    deploy_missing: bool,
+) -> Plan {
+    let mut plan = Plan::default();
+
+    // Prune: classify each on-disk site from its OWN preview key. The index is
+    // deliberately not consulted here — a per-site key is immune to the index's
+    // lost-update race, so this is the authority that cannot false-positive a
+    // live preview into an orphan.
+    for site in on_disk {
+        let label = label_for_site_key(site, suffix, preview_domain);
+        if matches!(classify(docs.get(&label)), Desired::Live) {
+            continue;
+        }
+        plan.prune.push(PrunePlan {
+            site_key: site.clone(),
+            label,
+        });
+    }
+
+    if deploy_missing {
+        let present: HashSet<&str> = on_disk.iter().map(String::as_str).collect();
+        for label in index {
+            if !matches!(classify(docs.get(label)), Desired::Live) {
+                continue;
+            }
+            // The directory name a deploy for this label would create.
+            let Ok(site) = site_key_for_label(label, suffix, preview_domain) else {
+                continue;
+            };
+            if !present.contains(site.as_str()) && !plan.deploy.contains(label) {
+                plan.deploy.push(label.clone());
+            }
+        }
+    }
+
+    plan
+}
+
+/// Is a preview's desired state a live deploy, or is it gone?
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Desired {
+    /// Present with `intent: deploy` — should be on disk.
+    Live,
+    /// A genuine nil (retired, TTL elapsed), an `intent: teardown`, or an
+    /// unparseable document — none of which is a live preview.
+    Gone,
+}
+
+/// Classify a preview document read from KV.
+///
+/// `raw` is the map entry: `None` = the label was not in the snapshot (treated
+/// as gone), `Some(None)` = a genuine KV nil, `Some(Some(json))` = a document.
+/// An unparseable document is `Gone`: desired state the daemon cannot understand
+/// is not something it will keep a preview alive for.
+fn classify(raw: Option<&Option<String>>) -> Desired {
+    let Some(Some(json)) = raw else {
+        return Desired::Gone;
+    };
+    match Job::parse(json.as_bytes()).and_then(|job| job.intent()) {
+        Ok(Intent::Deploy) => Desired::Live,
+        _ => Desired::Gone,
+    }
+}
+
+/// The canonical site key (directory name) a deploy for `label` creates —
+/// `site_key(<label>.<preview_domain>)`, the exact forward rule the deploy path
+/// uses ([`crate::site_key`]).
+///
+/// # Errors
+///
+/// Propagates [`crate::site_key::site_key`]'s error when the host does not
+/// normalize to a valid key.
+fn site_key_for_label(
+    label: &str,
+    suffix: Option<&str>,
+    preview_domain: &str,
+) -> anyhow::Result<String> {
+    let host = format!("{label}.{}", preview_domain.trim_matches('.'));
+    site_key::site_key(&host, suffix)
+}
+
+/// Reconstruct the preview *label* from an on-disk site key — the inverse of
+/// [`site_key_for_label`].
+///
+/// With a suffix configured the key *is* the label (the deploy stripped the
+/// suffix); without one the key is the full FQDN, so the label is the key with
+/// `.<preview_domain>` removed. The label only ever names the API's
+/// `applied/<label>` marker and the KV preview key, both of which fail safe if
+/// the reconstruction is off (a nil lookup, a no-op marker removal).
+fn label_for_site_key(site_key: &str, suffix: Option<&str>, preview_domain: &str) -> String {
+    if suffix.is_some() {
+        return site_key.to_string();
+    }
+    let dot_domain = format!(".{}", preview_domain.trim_matches('.'));
+    site_key
+        .strip_suffix(&dot_domain)
+        .unwrap_or(site_key)
+        .to_string()
+}
+
+/// List preview directories directly under `sites_dir`.
+///
+/// Keeps only entries that are directories named by a valid site key, excluding
+/// the keep-list (infra/test vhosts) and `.tmp` staging directories a deploy in
+/// flight owns. A missing `sites_dir` is an empty list, not an error.
+///
+/// # Errors
+///
+/// Returns an error if `sites_dir` exists but cannot be listed.
+fn list_preview_dirs(sites_dir: &Path, keep: &HashSet<String>) -> anyhow::Result<Vec<String>> {
+    use anyhow::Context as _;
+
+    let entries = match std::fs::read_dir(sites_dir) {
+        Ok(e) => e,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+        Err(e) => {
+            return Err(e).with_context(|| format!("failed to read {}", sites_dir.display()));
+        }
+    };
+
+    let mut out = Vec::new();
+    for entry in entries.flatten() {
+        // Follow symlinks: `file_type()` on the dirent does not, and a preview
+        // dir is a plain directory anyway, but a symlinked docroot must still
+        // count as present.
+        let is_dir = entry.metadata().map(|m| m.is_dir()).unwrap_or(false);
+        if !is_dir {
+            continue;
+        }
+        let name = entry.file_name().to_string_lossy().into_owned();
+        if name.ends_with(".tmp") {
+            // A staging directory owned by a deploy mid-swap — not a preview.
+            continue;
+        }
+        if !site_key::is_valid_site_key(&name) {
+            continue;
+        }
+        if keep.contains(&name) {
+            continue;
+        }
+        out.push(name);
+    }
+    out.sort();
+    Ok(out)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::job::sample_json;
+
+    const DOMAIN: &str = "preview.ephpm.dev";
+    const SUFFIX: Option<&str> = Some(".preview.ephpm.dev");
+
+    /// A `Some(Some(deploy job))` document for a label.
+    fn deploy_doc(label: &str) -> Option<String> {
+        Some(sample_json(label, "deploy"))
+    }
+
+    fn docs(pairs: &[(&str, Option<String>)]) -> HashMap<String, Option<String>> {
+        pairs
+            .iter()
+            .map(|(k, v)| ((*k).to_string(), v.clone()))
+            .collect()
+    }
+
+    #[test]
+    fn classify_reads_the_intent() {
+        assert_eq!(classify(Some(&deploy_doc("a"))), Desired::Live);
+        assert_eq!(
+            classify(Some(&Some(sample_json("a", "teardown")))),
+            Desired::Gone
+        );
+        // A genuine nil (retired) and a totally absent label both read as gone.
+        assert_eq!(classify(Some(&None)), Desired::Gone);
+        assert_eq!(classify(None), Desired::Gone);
+        // Unparseable desired state is not something to keep a preview alive for.
+        assert_eq!(
+            classify(Some(&Some("{not json".to_string()))),
+            Desired::Gone
+        );
+    }
+
+    #[test]
+    fn site_key_and_label_are_inverses_with_a_suffix() {
+        // On the live cluster (suffix configured) the key is the bare label.
+        let key = site_key_for_label("ephpm-wp-pr-8", SUFFIX, DOMAIN).unwrap();
+        assert_eq!(key, "ephpm-wp-pr-8");
+        assert_eq!(label_for_site_key(&key, SUFFIX, DOMAIN), "ephpm-wp-pr-8");
+    }
+
+    #[test]
+    fn site_key_and_label_are_inverses_without_a_suffix() {
+        // A node with no suffix names the directory by the full FQDN.
+        let key = site_key_for_label("ephpm-wp-pr-8", None, DOMAIN).unwrap();
+        assert_eq!(key, "ephpm-wp-pr-8.preview.ephpm.dev");
+        assert_eq!(label_for_site_key(&key, None, DOMAIN), "ephpm-wp-pr-8");
+    }
+
+    #[test]
+    fn a_live_deploy_is_kept_and_an_orphan_is_pruned() {
+        let on_disk = vec!["live-pr-1".to_string(), "orphan-pr-2".to_string()];
+        let docs = docs(&[
+            ("live-pr-1", deploy_doc("live-pr-1")),
+            // orphan's key is a genuine nil (retired).
+            ("orphan-pr-2", None),
+        ]);
+        let plan = build_plan(&on_disk, &[], &docs, SUFFIX, DOMAIN, false);
+        assert_eq!(plan.prune.len(), 1, "only the orphan is pruned");
+        assert_eq!(plan.prune[0].site_key, "orphan-pr-2");
+        assert_eq!(plan.prune[0].label, "orphan-pr-2");
+        assert!(plan.deploy.is_empty());
+    }
+
+    #[test]
+    fn a_teardown_intent_prunes_even_while_still_in_the_index() {
+        // The stranded-teardown case: the preview key still exists but says
+        // teardown (retired-with-TTL, not yet expired), and the label is still
+        // in the index. The node must prune it, not keep it.
+        let on_disk = vec!["torn-pr-3".to_string()];
+        let docs = docs(&[("torn-pr-3", Some(sample_json("torn-pr-3", "teardown")))]);
+        let index = vec!["torn-pr-3".to_string()];
+        let plan = build_plan(&on_disk, &index, &docs, SUFFIX, DOMAIN, true);
+        assert_eq!(plan.prune.len(), 1);
+        assert_eq!(plan.prune[0].site_key, "torn-pr-3");
+        // A teardown-intent label is never re-deployed.
+        assert!(plan.deploy.is_empty());
+    }
+
+    #[test]
+    fn an_orphan_not_in_the_index_is_still_pruned() {
+        // The lingering-override symptom: the index entry was already pruned, so
+        // the label is absent from the index AND its key is nil. Classified from
+        // its own (nil) key, it is still an orphan.
+        let on_disk = vec!["ghost-pr-4".to_string()];
+        let docs = docs(&[("ghost-pr-4", None)]);
+        let plan = build_plan(&on_disk, &[], &docs, SUFFIX, DOMAIN, true);
+        assert_eq!(plan.prune.len(), 1);
+        assert_eq!(plan.prune[0].site_key, "ghost-pr-4");
+    }
+
+    #[test]
+    fn a_live_preview_missing_from_the_index_is_not_pruned() {
+        // The index lost-update race: a live preview transiently fell out of the
+        // index. Because pruning classifies from the per-site key (not the
+        // index), it is kept — this is the false-positive the direct-key
+        // authority exists to prevent.
+        let on_disk = vec!["live-pr-5".to_string()];
+        let docs = docs(&[("live-pr-5", deploy_doc("live-pr-5"))]);
+        let plan = build_plan(&on_disk, &[], &docs, SUFFIX, DOMAIN, true);
+        assert!(
+            plan.prune.is_empty(),
+            "a live preview absent from the index must not be pruned"
+        );
+    }
+
+    #[test]
+    fn a_desired_preview_missing_from_disk_is_enqueued() {
+        let on_disk: Vec<String> = vec![];
+        let docs = docs(&[("wanted-pr-6", deploy_doc("wanted-pr-6"))]);
+        let index = vec!["wanted-pr-6".to_string()];
+        let plan = build_plan(&on_disk, &index, &docs, SUFFIX, DOMAIN, true);
+        assert_eq!(plan.deploy, vec!["wanted-pr-6".to_string()]);
+        assert!(plan.prune.is_empty());
+    }
+
+    #[test]
+    fn adds_are_skipped_when_deploy_missing_is_off() {
+        let on_disk: Vec<String> = vec![];
+        let docs = docs(&[("wanted-pr-6", deploy_doc("wanted-pr-6"))]);
+        let index = vec!["wanted-pr-6".to_string()];
+        let plan = build_plan(&on_disk, &index, &docs, SUFFIX, DOMAIN, false);
+        assert!(plan.deploy.is_empty(), "add direction is opt-in");
+    }
+
+    #[test]
+    fn a_present_desired_preview_is_not_re_enqueued() {
+        let on_disk = vec!["here-pr-7".to_string()];
+        let docs = docs(&[("here-pr-7", deploy_doc("here-pr-7"))]);
+        let index = vec!["here-pr-7".to_string()];
+        let plan = build_plan(&on_disk, &index, &docs, SUFFIX, DOMAIN, true);
+        assert!(plan.deploy.is_empty());
+        assert!(plan.prune.is_empty());
+    }
+
+    #[test]
+    fn list_preview_dirs_filters_infra_staging_and_files() {
+        let root = tempfile::tempdir().unwrap();
+        let sites = root.path();
+        for dir in ["live-pr-1", "orphan-pr-2", "switchboard", "live-pr-1.tmp"] {
+            std::fs::create_dir_all(sites.join(dir)).unwrap();
+        }
+        // A stray file must never be taken for a preview directory.
+        std::fs::write(sites.join("README"), "x").unwrap();
+
+        let mut keep = HashSet::new();
+        keep.insert("switchboard".to_string());
+
+        let found = list_preview_dirs(sites, &keep).unwrap();
+        assert_eq!(
+            found,
+            vec!["live-pr-1".to_string(), "orphan-pr-2".to_string()],
+            "keep-list, .tmp staging and plain files are all excluded"
+        );
+    }
+
+    #[test]
+    fn a_missing_sites_dir_is_an_empty_list() {
+        let root = tempfile::tempdir().unwrap();
+        let found = list_preview_dirs(&root.path().join("nope"), &HashSet::new()).unwrap();
+        assert!(found.is_empty());
+    }
+
+    #[tokio::test]
+    async fn dry_run_reports_would_prune_and_removes_nothing() {
+        let root = tempfile::tempdir().unwrap();
+        let sites = root.path().join("sites");
+        let state = root.path().join(".switchboard");
+        std::fs::create_dir_all(sites.join("orphan-pr-2")).unwrap();
+        std::fs::create_dir_all(&state).unwrap();
+        let queue = Queue::new(&state);
+        queue.ensure_dirs().unwrap();
+        let keep = HashSet::new();
+
+        let ctx = ReconcileContext {
+            sites_dir: &sites,
+            sqlite_dir: None,
+            site_overrides_dir: None,
+            vhost_temp_base: None,
+            state_dir: &state,
+            allow_incomplete: true,
+            kv_addr: "unused",
+            kv_secret: "s",
+            api_site: "switchboard",
+            preview_domain: DOMAIN,
+            sites_domain_suffix: SUFFIX,
+            keep: &keep,
+            prune: false,
+            deploy_missing: false,
+            max_prunes_per_cycle: 8,
+            queue: &queue,
+        };
+        let plan = Plan {
+            prune: vec![PrunePlan {
+                site_key: "orphan-pr-2".to_string(),
+                label: "orphan-pr-2".to_string(),
+            }],
+            deploy: vec![],
+        };
+        let mut report = ReconcileReport::default();
+        execute_prune(&ctx, &plan, &mut report).await;
+
+        assert_eq!(report.would_prune, 1);
+        assert_eq!(report.pruned, 0);
+        assert!(
+            sites.join("orphan-pr-2").exists(),
+            "dry-run must not remove the directory"
+        );
+    }
+
+    #[tokio::test]
+    async fn prune_removes_the_orphan_directory() {
+        let root = tempfile::tempdir().unwrap();
+        let sites = root.path().join("sites");
+        let state = root.path().join(".switchboard");
+        std::fs::create_dir_all(sites.join("orphan-pr-2")).unwrap();
+        std::fs::create_dir_all(&state).unwrap();
+        let queue = Queue::new(&state);
+        queue.ensure_dirs().unwrap();
+        let keep = HashSet::new();
+
+        let ctx = ReconcileContext {
+            sites_dir: &sites,
+            sqlite_dir: None,
+            site_overrides_dir: None,
+            vhost_temp_base: None,
+            state_dir: &state,
+            allow_incomplete: true, // no sqlite/overrides dirs configured in the test
+            kv_addr: "127.0.0.1:1", // revocation write fails, best-effort — must not fail teardown
+            kv_secret: "s",
+            api_site: "switchboard",
+            preview_domain: DOMAIN,
+            sites_domain_suffix: SUFFIX,
+            keep: &keep,
+            prune: true,
+            deploy_missing: false,
+            max_prunes_per_cycle: 8,
+            queue: &queue,
+        };
+        let plan = Plan {
+            prune: vec![PrunePlan {
+                site_key: "orphan-pr-2".to_string(),
+                label: "orphan-pr-2".to_string(),
+            }],
+            deploy: vec![],
+        };
+        let mut report = ReconcileReport::default();
+        execute_prune(&ctx, &plan, &mut report).await;
+
+        assert_eq!(report.pruned, 1, "the orphan is removed");
+        assert!(!sites.join("orphan-pr-2").exists());
+    }
+
+    #[tokio::test]
+    async fn the_per_cycle_cap_defers_the_rest() {
+        let root = tempfile::tempdir().unwrap();
+        let sites = root.path().join("sites");
+        let state = root.path().join(".switchboard");
+        for dir in ["a-pr-1", "b-pr-2", "c-pr-3"] {
+            std::fs::create_dir_all(sites.join(dir)).unwrap();
+        }
+        std::fs::create_dir_all(&state).unwrap();
+        let queue = Queue::new(&state);
+        queue.ensure_dirs().unwrap();
+        let keep = HashSet::new();
+
+        let ctx = ReconcileContext {
+            sites_dir: &sites,
+            sqlite_dir: None,
+            site_overrides_dir: None,
+            vhost_temp_base: None,
+            state_dir: &state,
+            allow_incomplete: true,
+            kv_addr: "127.0.0.1:1",
+            kv_secret: "s",
+            api_site: "switchboard",
+            preview_domain: DOMAIN,
+            sites_domain_suffix: SUFFIX,
+            keep: &keep,
+            prune: true,
+            deploy_missing: false,
+            max_prunes_per_cycle: 2,
+            queue: &queue,
+        };
+        let plan = Plan {
+            prune: ["a-pr-1", "b-pr-2", "c-pr-3"]
+                .iter()
+                .map(|s| PrunePlan {
+                    site_key: (*s).to_string(),
+                    label: (*s).to_string(),
+                })
+                .collect(),
+            deploy: vec![],
+        };
+        let mut report = ReconcileReport::default();
+        execute_prune(&ctx, &plan, &mut report).await;
+
+        assert_eq!(report.pruned, 2, "cap honoured");
+        assert_eq!(
+            report.prune_deferred, 1,
+            "the third waits for the next pass"
+        );
+    }
+}