From a37a08572238162503cd8d8536b4260ee49fc868 Mon Sep 17 00:00:00 2001
From: Luther Monson <luther.monson@gmail.com>
Date: Sun, 20 Sep 2026 14:53:07 -0700
Subject: [PATCH] fix: level-triggered reconcile so previews converge across
 the cluster
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The drain/queue fan-out is edge-triggered on the switchboard:gen counter:
a node only re-walks switchboard:index when gen advances past its own
last_gen, and once it records itself current at a generation it never
re-walks. gen is a separate gossip key from the content it guards, so a
lost or not-yet-replicated increment leaves a node "current" at a
generation whose teardown it never materialized — the torn-down preview
stays served with its database and <key>.toml override on disk, every
health check green (switchboard#24, and the gen-vs-key-content
propagation race). The index itself has an acknowledged lost-update
window on top of that. The symptom on the live cluster: override .toml
counts drifting between nodes and orphaned overrides lingering as 404s.

Add a level-triggered reconcile pass in the daemon that converges this
node's on-disk previews to the KV desired state directly, independent of
gen:

- Prune: classify each preview directory from its OWN
  switchboard:preview:<label> key (a plain set — only teardown gives it a
  TTL — so it is immune to both the gen propagation race and the index
  lost-update race). A key that is a genuine nil (retired) or carries
  intent:teardown is an orphan and is removed with the existing
  KEEP-guarded, exact-path teardown_preview. Pruning never trusts the
  index, so a live preview that transiently fell out of it is not
  false-positived into an orphan.
- Add: for a label the index lists whose preview key says deploy but
  whose directory is absent, enqueue the exact job document so the
  existing claim -> coalesce -> validate -> deploy path provisions it.

Correctness no longer depends on gen; it stays as the PHP fast-path's
optimization hint. The pass is fail-safe: a KV read failure aborts it
(prunes nothing), infra vhosts are protected by a keep-list plus the API
site key, pruning is opt-in (--reconcile-prune; dry-run logs WOULD-prune
until then), and a per-cycle cap bounds the blast radius. Two nodes
reconciling the same desired state converge rather than fight (teardown
and enqueue are both idempotent).

New knobs (all off by default, so upgrade is a no-op):
--reconcile-interval-secs, --reconcile-prune, --reconcile-deploy-missing,
--reconcile-keep-sites, --reconcile-max-prunes-per-cycle,
--reconcile-api-site.
---
 README.md        |  33 ++
 src/config.rs    | 134 ++++++++
 src/kv.rs        | 263 ++++++++++++++++
 src/main.rs      | 105 +++++++
 src/queue.rs     | 181 +++++++++++
 src/reconcile.rs | 787 +++++++++++++++++++++++++++++++++++++++++++++++
 6 files changed, 1503 insertions(+)
 create mode 100644 src/reconcile.rs

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"
+        );
+    }
+}