From 9b98b6e654933c1dc698b3a6d115a9bba475d15c Mon Sep 17 00:00:00 2001
From: Luther Monson <luther.monson@gmail.com>
Date: Sun, 20 Sep 2026 16:30:36 -0700
Subject: [PATCH] fix: reconcile prunes by GitHub PR state, not the KV index
 (v0.3.0)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

v0.2.0's reconcile read desired state from the KV index over RESP. The
preview nodes keep ePHPm's KV RESP listener off (no [kv.redis_compat]
listener, no [kv] secret), so that reader was inert there — it could
never read desired state and every pass aborted.

Switch the prune authority to GitHub PR state, which needs no ePHPm-side
change and is more authoritative than the KV index (which has an
acknowledged lost-update window). Each pass:

- lists every preview directory under sites_dir;
- parses its canonical site key <owner>-<repo>-pr-<N> back to (repo, N);
- asks GitHub what that PR is now, under the switchboard App creds the
  daemon already mints reporting tokens with (org installation token,
  minted once per pass);
- open (or an unrecognised state) => keep; merged or closed => prune, via
  the existing KEEP-guarded, exact-path teardown::teardown_preview.

Prune-only by design: re-deploying a missing but still-open preview stays
on the webhook path.

Fail-safe in every direction — uncertainty always keeps:
- a site key that does not parse as <owner>-<repo>-pr-<N> (an infra vhost,
  or a hashed/overflow PreviewLabel) is skipped, never a candidate;
- a per-PR GitHub read error keeps that preview;
- a failure to mint the installation token aborts the whole pass, so a
  total GitHub outage prunes nothing;
- the keep-list still applies (the API vhost site key is auto-added).

Removes the RESP ClusterReader path and the add-direction plumbing
(Queue::enqueue/has_job_for_label) added in #37, so there is one
authority. Knobs: --reconcile-interval-secs, --reconcile-prune,
--reconcile-keep-sites (all default-off/empty), plus --reconcile-owner
(default "ephpm"). Requires --app-id/--app-key when enabled (validated).
No ePHPm-side, [kv], or restart change is needed.

Tests: merged/closed -> prune, open/unknown -> keep, API-error -> keep,
missing-lookup -> keep, unparseable/hashed key -> keep, hyphenated-repo
parsing, keep-list respected, dry-run removes nothing, real prune removes.
---
 Cargo.lock       |   2 +-
 Cargo.toml       |   2 +-
 README.md        |  36 +-
 src/config.rs    | 123 +++----
 src/kv.rs        | 263 --------------
 src/main.rs      | 141 +++++---
 src/queue.rs     | 181 ----------
 src/reconcile.rs | 895 +++++++++++++++++++++--------------------------
 8 files changed, 553 insertions(+), 1090 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 4cbd5a3..1956469 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1243,7 +1243,7 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
 
 [[package]]
 name = "switchboard"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
  "anyhow",
  "axum",
diff --git a/Cargo.toml b/Cargo.toml
index 327d966..056d0c8 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "switchboard"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2024"
 rust-version = "1.85"
 license = "MIT"
diff --git a/README.md b/README.md
index 7290519..11c0957 100644
--- a/README.md
+++ b/README.md
@@ -472,27 +472,31 @@ database and `<key>.toml` override on disk, every health check green. This is th
 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.
+The reconcile pass is the **level-triggered** backstop, and its authority is
+**GitHub PR state** — needing no KV and no ePHPm-side change (the preview nodes
+keep ePHPm's RESP listener off). On its own interval, independent of `gen`, it
+lists every preview directory, parses each site key `<owner>-<repo>-pr-<N>` back
+to its pull request, and asks GitHub whether that PR is still open. **Open (or an
+unrecognised state) ⇒ keep; merged or closed ⇒ prune.** GitHub is more
+authoritative than the KV index (which has a lost-update window), and a preview
+that has fallen off every other signal is still correctly classified by the one
+fact that actually decides whether it should exist. Re-*deploying* a missing but
+still-open preview is deliberately left to the webhook path; this pass only
+removes what GitHub says is gone.
 
 | 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. |
+| `--reconcile-interval-secs` | `SWITCHBOARD_RECONCILE_INTERVAL_SECS` | `0` | Seconds between reconcile passes. **`0` disables it.** Requires `--app-id`/`--app-key` (the pass queries GitHub); 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 (PRs merged/closed). **Off by default**: until set, the pass is observability-only and logs each orphan it *would* prune at `WARN`. |
+| `--reconcile-keep-sites` | `SWITCHBOARD_RECONCILE_KEEP_SITES` | *(empty)* | Comma-separated vhost directory names never pruned — the node's infra/test sites (e.g. `site-a,site-b,preview.ephpm.dev`). The API's own site key is always protected in addition. (Belt-and-braces: a non-`<owner>-<repo>-pr-<N>` directory never parses as a preview, so it is never a prune candidate regardless.) |
+| `--reconcile-owner` | `SWITCHBOARD_RECONCILE_OWNER` | `ephpm` | The GitHub owner/org previews belong to — the leading segment of `<owner>-<repo>-pr-<N>`, and the owner PR state is queried under. |
 
 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.
+prune` logs; turn on `--reconcile-prune` once they look right. Uncertainty always
+resolves to **keep** — a site key that does not parse as `<owner>-<repo>-pr-<N>`
+(an infra vhost, or a hashed/overflow label) is skipped, a per-PR GitHub read
+error keeps that preview, and a failure to mint the installation token aborts the
+whole pass so a total GitHub outage prunes **nothing**.
 
 ### GitHub reporting (optional)
 
diff --git a/src/config.rs b/src/config.rs
index 38a98ab..c1d7b12 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -279,15 +279,15 @@ pub struct Config {
     /// 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.
+    /// interval it asks **GitHub** whether each on-disk preview's PR is still
+    /// open, and prunes the ones whose PR has merged or closed. It needs no KV
+    /// (the preview nodes keep ePHPm's RESP listener off) and no ePHPm-side
+    /// change — only the switchboard App credentials it already reports with.
     ///
-    /// 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.
+    /// Requires `--app-id`/`--app-key`; the daemon refuses to start with a
+    /// non-zero interval and no App credentials (the pass would query nothing).
+    /// 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,
 
@@ -299,47 +299,21 @@ pub struct Config {
     #[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.
+    /// `site-a,site-b,preview.ephpm.dev`). The API's own site key is always
+    /// protected in addition to this list. Belt-and-braces: a non-`<owner>-<repo>-pr-<N>`
+    /// directory never parses as a preview and so is never a prune candidate
+    /// regardless.
     #[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>,
+    /// The GitHub owner/org the previews belong to — the leading segment of a
+    /// preview site key `<owner>-<repo>-pr-<N>`, and the owner the reconcile
+    /// queries PR state under. Defaults to `ephpm` (the fleet's org); set it if
+    /// previews are built for a different org.
+    #[arg(long, default_value = "ephpm", env = "SWITCHBOARD_RECONCILE_OWNER")]
+    pub reconcile_owner: String,
 
     // ── pre-serve static-analysis gate ─────────────────────────────────
     /// Path to the operator-controlled `ephpm analyze` policy file (YAML), used
@@ -521,35 +495,24 @@ impl Config {
         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.
+    /// The switchboard-api vhost's canonical site key, derived from
+    /// `--drain-host` with this node's suffix rule — added to the reconcile
+    /// keep-list so the API's own vhost is never a prune candidate.
     ///
-    /// # 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);
-        };
+    /// `None` when no drain host is configured (single-node mode) or it does not
+    /// normalize to a valid site key; the keep-list simply omits it then (and the
+    /// `<owner>-<repo>-pr-<N>` parse rule already excludes an infra vhost anyway).
+    #[must_use]
+    pub fn drain_host_site_key(&self) -> Option<String> {
+        let host = self.drain_host.as_ref()?;
         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:?}")
-            })
+        crate::site_key::site_key(host, suffix.as_deref()).ok()
     }
 
     /// The reconcile keep-list — vhost names never pruned — from
-    /// `--reconcile-keep-sites`, with the API site key always added.
+    /// `--reconcile-keep-sites`, with the API site key added when known.
     #[must_use]
-    pub fn reconcile_keep_set(&self, api_site: &str) -> std::collections::HashSet<String> {
+    pub fn reconcile_keep_set(&self, api_site: Option<&str>) -> std::collections::HashSet<String> {
         let mut set: std::collections::HashSet<String> = self
             .reconcile_keep_sites
             .split(',')
@@ -557,7 +520,9 @@ impl Config {
             .filter(|s| !s.is_empty())
             .map(str::to_string)
             .collect();
-        set.insert(api_site.to_string());
+        if let Some(api_site) = api_site {
+            set.insert(api_site.to_string());
+        }
         set
     }
 
@@ -599,22 +564,20 @@ 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.
+        // The reconcile decides prune-vs-keep by asking GitHub for each preview's
+        // PR state, so it needs the App credentials. Without them every pass would
+        // abort (mint nothing, prune nothing) — 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)"
+                self.github_reporting_enabled(),
+                "--reconcile-interval-secs is non-zero but --app-id/--app-key are unset — \
+                 the reconcile queries GitHub for each preview's PR state and cannot without \
+                 the switchboard App credentials (set the interval to 0 to disable it)"
             );
             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"
+                !self.reconcile_owner.trim().is_empty(),
+                "--reconcile-owner must not be empty when the reconcile is enabled"
             );
         }
         // A daemon that cannot remove a tenant's database when its PR closes is
diff --git a/src/kv.rs b/src/kv.rs
index afbc485..50c283d 100644
--- a/src/kv.rs
+++ b/src/kv.rs
@@ -305,177 +305,6 @@ 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> {
@@ -791,98 +620,6 @@ 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 a9daca3..ddd9d6c 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -242,15 +242,14 @@ async fn main() -> anyhow::Result<()> {
     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.
+    // into the shared state. `validate()` has already guaranteed App credentials
+    // whenever the interval is non-zero — the pass queries GitHub for PR state.
     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);
+        // The API vhost's own site key is added to the keep-list so it is never a
+        // prune candidate (the parse rule already excludes it, but belt-and-braces).
+        let keep = config.reconcile_keep_set(config.drain_host_site_key().as_deref());
         let interval = Duration::from_secs(config.reconcile_interval_secs.max(1));
-        Some((api_site, keep, interval))
+        Some((keep, interval))
     } else {
         info!("level-triggered reconcile disabled (--reconcile-interval-secs 0)");
         None
@@ -266,37 +265,21 @@ 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 {
+    if let Some((keep, interval)) = reconcile_setup {
+        info!(
+            interval_s = interval.as_secs(),
+            prune = state.config.reconcile_prune,
+            owner = %state.config.reconcile_owner,
+            keep = keep.len(),
+            "level-triggered reconcile enabled (authority: GitHub PR state)"
+        );
+        if !state.config.reconcile_prune {
             tracing::warn!(
-                "reconcile is enabled but no KV secret resolved — reconcile pass skipped"
+                "reconcile is in dry-run: orphans (merged/closed PRs) will be logged as \
+                 WOULD-prune but NOT removed. Set --reconcile-prune once the logs look right"
             );
         }
+        tokio::spawn(reconcile_loop(Arc::clone(&state), keep, interval));
     }
 
     if webhook_server_enabled {
@@ -329,21 +312,19 @@ 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
+/// Reconcile this node's on-disk previews to **GitHub PR state** on a timer,
+/// independent of the drain/queue path's `switchboard:gen` cursor. Every failure
+/// mode is transient (GitHub 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");
+    // Guarded at the spawn site (validate() requires App creds when enabled);
+    // kept here so a future refactor cannot start the loop without them.
+    let (Some(app_id), Some(app_key)) = (state.config.app_id, state.config.app_key.clone()) else {
+        tracing::warn!("reconcile loop started without App credentials — not running");
         return;
     };
 
@@ -361,22 +342,21 @@ async fn reconcile_loop(
             vhost_temp_base: cfg.vhost_temp_base.as_deref(),
             state_dir: &cfg.state_dir,
             allow_incomplete: cfg.allow_incomplete_teardown,
+            kv_secret: state.kv_secret.as_deref(),
             kv_addr: &cfg.kv_addr,
-            kv_secret: &kv_secret,
-            api_site: &api_site,
+            app_id,
+            app_key: &app_key,
+            owner: &cfg.reconcile_owner,
             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"
+                "reconcile pass failed (GitHub unreachable?) — nothing pruned; will retry"
             );
         }
     }
@@ -767,7 +747,8 @@ async fn fetch_installation_token(state: &AppState, req: &PreviewRequest) -> Opt
     }
 }
 
-/// Get a short-lived installation access token from GitHub.
+/// Get a short-lived installation access token from GitHub for a known
+/// installation id.
 ///
 /// GitHub Apps authenticate by:
 /// 1. Creating a JWT signed with the app's private key
@@ -777,6 +758,35 @@ async fn get_installation_token(
     app_key: &Path,
     installation_id: u64,
 ) -> anyhow::Result<String> {
+    let jwt = app_jwt(app_id, app_key).await?;
+    exchange_jwt_for_token(&jwt, installation_id).await
+}
+
+/// Get an installation token for an **owner/org** without a pre-known
+/// installation id — the path the reconcile uses, which has only on-disk site
+/// keys and no webhook payload to read an id from.
+///
+/// Mints the App JWT once and uses it for both the installation-lookup and the
+/// token exchange (a JWT is valid for both and for ~10 minutes).
+///
+/// # Errors
+///
+/// Returns an error if the JWT cannot be minted, the org has no installation of
+/// this App, or the token exchange fails. The reconcile treats any error as a
+/// total GitHub-auth failure and prunes nothing that pass.
+async fn installation_token_for_owner(
+    app_id: u64,
+    app_key: &Path,
+    owner: &str,
+) -> anyhow::Result<String> {
+    let jwt = app_jwt(app_id, app_key).await?;
+    let installation_id = org_installation_id(&jwt, owner).await?;
+    exchange_jwt_for_token(&jwt, installation_id).await
+}
+
+/// Build and sign the App-level JWT (RS256), shelling out to `openssl` for the
+/// signature.
+async fn app_jwt(app_id: u64, app_key: &Path) -> anyhow::Result<String> {
     let now = std::time::SystemTime::now()
         .duration_since(std::time::UNIX_EPOCH)?
         .as_secs();
@@ -812,9 +822,34 @@ async fn get_installation_token(
     anyhow::ensure!(output.status.success(), "openssl signing failed");
 
     let signature = base64_url_encode(&output.stdout);
-    let jwt = format!("{signing_input}.{signature}");
+    Ok(format!("{signing_input}.{signature}"))
+}
+
+/// Discover the id of this App's installation on `owner` via the App JWT.
+async fn org_installation_id(jwt: &str, owner: &str) -> anyhow::Result<u64> {
+    let url = format!("https://api.github.com/orgs/{owner}/installation");
+    let resp = reqwest::Client::new()
+        .get(&url)
+        .header("Authorization", format!("Bearer {jwt}"))
+        .header("User-Agent", "switchboard")
+        .header("Accept", "application/vnd.github+json")
+        .send()
+        .await?;
+
+    anyhow::ensure!(
+        resp.status().is_success(),
+        "failed to resolve the App installation on {owner}: {}",
+        resp.status()
+    );
+
+    let body: serde_json::Value = resp.json().await?;
+    body["id"]
+        .as_u64()
+        .ok_or_else(|| anyhow::anyhow!("installation response for {owner} missing 'id'"))
+}
 
-    // Exchange JWT for installation token.
+/// Exchange an App JWT for a short-lived installation access token.
+async fn exchange_jwt_for_token(jwt: &str, installation_id: u64) -> anyhow::Result<String> {
     let url = format!("https://api.github.com/app/installations/{installation_id}/access_tokens");
     let resp = reqwest::Client::new()
         .post(&url)
diff --git a/src/queue.rs b/src/queue.rs
index 9bfb7e2..d0046aa 100644
--- a/src/queue.rs
+++ b/src/queue.rs
@@ -248,92 +248,6 @@ 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
@@ -373,35 +287,6 @@ 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
@@ -674,72 +559,6 @@ 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
index a81311d..07cafdb 100644
--- a/src/reconcile.rs
+++ b/src/reconcile.rs
@@ -1,79 +1,63 @@
-//! Level-triggered convergence of this node's on-disk previews to the cluster's
-//! desired state.
+//! Level-triggered convergence of this node's on-disk previews to **GitHub PR
+//! state** — the authoritative source of whether a preview is still wanted.
 //!
-//! # The bug this closes
+//! # Why not the KV index / the drain cursor
 //!
-//! 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:
+//! Preview fan-out is edge-triggered on the `switchboard:gen` counter: a node
+//! only re-walks desired state when the counter advances, so a lost or
+//! not-yet-replicated increment strands a teardown (the preview stays served,
+//! its database and `<key>.toml` override on disk, every health check green —
+//! switchboard#24). The first cut of this reconcile (v0.2.0) read desired state
+//! from the KV index over RESP. But the preview nodes keep ePHPm's KV RESP
+//! listener **off** (there is no `[kv.redis_compat]` listener and no `[kv]`
+//! secret), so that reader was inert there.
 //!
-//! * 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.
+//! This version removes the KV dependency entirely and uses the one authority
+//! that needs no ePHPm-side change and is *more* authoritative than the KV index
+//! (which has an acknowledged lost-update window): **the pull request's state on
+//! GitHub**. The daemon already holds the switchboard App credentials it mints
+//! installation tokens with for reporting; this reuses that path.
 //!
-//! 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.
+//! # What a pass does (prune-only)
 //!
-//! # The fix: reconcile the level, not the edge
+//! For every preview directory under `sites_dir`:
 //!
-//! 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*:
+//! * parse its canonical site key `<owner>-<repo>-pr-<N>` back to `(repo, N)`
+//!   ([`parse_preview_site_key`]);
+//! * ask GitHub what PR `owner/repo#N` is **now**;
+//! * **open (or an unrecognised state) ⇒ keep; merged or closed ⇒ prune** with
+//!   the same KEEP-guarded, exact-path [`crate::teardown::teardown_preview`] the
+//!   webhook teardown uses (site dir, `<key>.db*`, `<key>.toml`,
+//!   `/tmp/ephpm-vhosts/<key>-<hash>`, preserving `*.single-db-bak`).
 //!
-//! * **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.
+//! Re-*deploying* a missing but still-open preview is deliberately **not** done
+//! here — that stays on the existing webhook path. This module only removes what
+//! GitHub says is gone.
 //!
-//! # Why this is safe under concurrent drains
+//! # Fail-safe, in every direction
 //!
-//! 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.
+//! Uncertainty always resolves to **keep**:
 //!
-//! # Fail-safe posture
+//! * a site key that does not parse as `<owner>-<repo>-pr-<N>` (a hashed/overflow
+//!   label, or an infra vhost) is skipped — never a prune candidate;
+//! * a per-PR GitHub error (rate limit, transient 5xx, a repo/PR that 404s) skips
+//!   *that* entry;
+//! * a failure to mint the installation token aborts the **whole** pass, so a
+//!   total GitHub outage prunes nothing;
+//! * an infra/non-preview keep-list still applies on top (the API vhost is added
+//!   automatically), though the parse rule already excludes those dirs.
 //!
-//! * 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};
+//! Pruning is opt-in (`--reconcile-prune`); until it is set the pass logs each
+//! orphan it *would* remove at WARN and removes nothing.
+
+use std::collections::HashMap;
 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};
+use crate::validate::{PullRequestState, pr_state_verdict};
 
-/// Everything one reconcile pass needs. Mirrors the teardown/deploy context the
-/// job path already assembles from [`crate::config::Config`], plus the reconcile
-/// policy knobs.
+/// Everything one reconcile pass needs.
 pub struct ReconcileContext<'a> {
     // ── where the artifacts live (same as `TeardownContext`) ────────────
     pub sites_dir: &'a Path,
@@ -82,58 +66,51 @@ pub struct ReconcileContext<'a> {
     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`).
+    /// ePHPm's `[kv] secret`, only for teardown's best-effort share-link
+    /// revocation. `None` (the preview cluster's case — no `[kv]` secret) skips
+    /// it cleanly; the reconcile itself never reads KV.
+    pub kv_secret: Option<&'a str>,
     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) ────────
+
+    // ── GitHub PR-state authority ───────────────────────────────────────
+    /// switchboard App id (`--app-id`).
+    pub app_id: u64,
+    /// switchboard App private key (`--app-key`).
+    pub app_key: &'a Path,
+    /// The GitHub owner/org previews belong to (the leading label segment). All
+    /// preview site keys are `<owner>-<repo>-pr-<N>`.
+    pub owner: &'a str,
+
+    // ── host → label 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.
+    /// Vhost directory names never pruned (infra/test sites). The parse rule
+    /// already excludes non-`<owner>-<repo>-pr-<N>` dirs; this is belt-and-braces.
+    pub keep: &'a std::collections::HashSet<String>,
+    /// Actually remove orphans. When `false` the pass logs WOULD-prune 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.
+/// What one reconcile pass observed and did.
 #[derive(Debug, Default, PartialEq, Eq)]
 pub struct ReconcileReport {
-    /// Preview directories found under `sites_dir` (excluding keep-list and
-    /// staging `.tmp` dirs).
+    /// Preview directories found (excluding keep-list and `.tmp` staging dirs).
     pub on_disk: usize,
-    /// Directories whose own preview key says `deploy` — live, kept.
-    pub live: usize,
+    /// Directories whose PR is still open (or an unrecognised state) — kept.
+    pub kept_open: 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,
+    /// Directories whose site key did not parse as `<owner>-<repo>-pr-<N>` — kept.
+    pub skipped_unparseable: usize,
+    /// Directories whose PR state could not be read from GitHub — kept.
+    pub skipped_api_error: usize,
 }
 
 /// A single orphan to remove: its site key (names every ePHPm artifact) and its
@@ -144,96 +121,170 @@ struct PrunePlan {
     label: String,
 }
 
-/// The decisions a pass reached, separated from executing them so the logic is
-/// unit-testable without a filesystem or a KV server.
+/// A per-preview GitHub lookup outcome. `Error` is a read that failed (rate
+/// limit, transient 5xx, a 404) — treated as "keep", never a prune.
+#[derive(Debug, Clone)]
+enum PrLookup {
+    State(PullRequestState),
+    Error,
+}
+
+/// The decisions a pass reached, plus the counts for the report — separated from
+/// executing them so the logic is unit-testable without a filesystem or GitHub.
 #[derive(Debug, Default, PartialEq, Eq)]
-struct Plan {
+struct PlanResult {
     prune: Vec<PrunePlan>,
-    deploy: Vec<String>,
+    kept_open: usize,
+    skipped_unparseable: usize,
+    skipped_api_error: usize,
 }
 
 /// 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.
+/// Returns an error only if the installation token could not be minted (a total
+/// GitHub-auth failure) — the caller logs it and waits for the next interval, and
+/// nothing is pruned. Per-PR read failures and per-site prune failures are
+/// counted in the [`ReconcileReport`], not surfaced as an error.
 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);
-        }
+    if on_disk.is_empty() {
+        return Ok(ReconcileReport::default());
     }
 
-    // (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?;
+    // Ask GitHub about every parseable preview. A token-mint failure aborts here
+    // (Err) so a total outage prunes nothing; a per-PR failure becomes `Error`.
+    let lookups = fetch_lookups(ctx, &on_disk).await?;
 
-    // (4) Decide.
     let plan = build_plan(
         &on_disk,
-        &snapshot.index,
-        &snapshot.docs,
-        suffix,
-        domain,
-        ctx.deploy_missing,
+        ctx.owner,
+        ctx.sites_domain_suffix,
+        ctx.preview_domain,
+        &lookups,
     );
 
-    // (5) Execute.
     let mut report = ReconcileReport {
         on_disk: on_disk.len(),
-        live: on_disk.len().saturating_sub(plan.prune.len()),
+        kept_open: plan.kept_open,
+        skipped_unparseable: plan.skipped_unparseable,
+        skipped_api_error: plan.skipped_api_error,
         ..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,
+        kept_open = report.kept_open,
         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,
+        skipped_unparseable = report.skipped_unparseable,
+        skipped_api_error = report.skipped_api_error,
         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;
+/// Ask GitHub for the PR state of every parseable preview under one installation
+/// token.
+///
+/// Returns a map from site key to its lookup outcome. Only parseable sites are
+/// queried; unparseable ones are handled by [`build_plan`]. A failure to mint the
+/// token is an error (aborts the pass); a per-PR read failure is recorded as
+/// [`PrLookup::Error`] (kept).
+async fn fetch_lookups(
+    ctx: &ReconcileContext<'_>,
+    on_disk: &[String],
+) -> anyhow::Result<HashMap<String, PrLookup>> {
+    use anyhow::Context as _;
+
+    let parsed: Vec<(String, String, u64)> = on_disk
+        .iter()
+        .filter_map(|site| {
+            parse_preview_site_key(site, ctx.owner).map(|(repo, pr)| (site.clone(), repo, pr))
+        })
+        .collect();
+
+    if parsed.is_empty() {
+        // Nothing to ask GitHub about — don't mint a token for no reason.
+        return Ok(HashMap::new());
+    }
+
+    // One installation token for the whole pass (rate-limit-friendly: a handful
+    // of reads well under any limit).
+    let token = crate::installation_token_for_owner(ctx.app_id, ctx.app_key, ctx.owner)
+        .await
+        .context("reconcile: could not mint a GitHub installation token — pruning nothing")?;
+    let client = crate::github::GitHubClient::new(token);
+
+    let mut map = HashMap::with_capacity(parsed.len());
+    for (site, repo, pr) in parsed {
+        let lookup = match client.pull_request_state(ctx.owner, &repo, pr).await {
+            Ok(state) => PrLookup::State(state),
+            Err(e) => {
+                tracing::warn!(
+                    site_key = %site,
+                    repo = %format!("{}/{repo}", ctx.owner),
+                    pr,
+                    error = %format!("{e:#}"),
+                    "reconcile: could not read PR state — keeping this preview (fail-safe)"
+                );
+                PrLookup::Error
+            }
+        };
+        map.insert(site, lookup);
+    }
+    Ok(map)
+}
+
+/// Decide what to prune. Pure — no IO — so it can be tested against hand-built
+/// lookup maps.
+fn build_plan(
+    on_disk: &[String],
+    owner: &str,
+    suffix: Option<&str>,
+    preview_domain: &str,
+    lookups: &HashMap<String, PrLookup>,
+) -> PlanResult {
+    let mut plan = PlanResult::default();
+
+    for site in on_disk {
+        let Some((_repo, _pr)) = parse_preview_site_key(site, owner) else {
+            // A hashed/overflow label or an infra vhost — never a prune candidate.
+            plan.skipped_unparseable += 1;
             continue;
+        };
+        match lookups.get(site) {
+            Some(PrLookup::State(state)) if pr_state_verdict(state).is_discard() => {
+                // Merged or closed ⇒ the preview is gone.
+                plan.prune.push(PrunePlan {
+                    site_key: site.clone(),
+                    label: label_for_site_key(site, suffix, preview_domain),
+                });
+            }
+            Some(PrLookup::State(_)) => plan.kept_open += 1, // open / unrecognised ⇒ keep
+            Some(PrLookup::Error) | None => plan.skipped_api_error += 1, // unreadable ⇒ keep
         }
+    }
 
+    plan
+}
+
+/// Remove (or, in dry-run, report) each planned orphan.
+async fn execute_prune(
+    ctx: &ReconcileContext<'_>,
+    plan: &PlanResult,
+    report: &mut ReconcileReport,
+) {
+    for target in &plan.prune {
         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 — \
+                "reconcile: orphan preview on disk whose PR is merged/closed — \
                  WOULD prune (set --reconcile-prune to remove it)"
             );
             continue;
@@ -246,7 +297,7 @@ async fn execute_prune(ctx: &ReconcileContext<'_>, plan: &Plan, report: &mut Rec
             vhost_temp_base: ctx.vhost_temp_base,
             state_dir: ctx.state_dir,
             allow_incomplete: ctx.allow_incomplete,
-            kv_secret: Some(ctx.kv_secret),
+            kv_secret: ctx.kv_secret,
             kv_addr: ctx.kv_addr,
         };
         let preview = Preview {
@@ -258,7 +309,7 @@ async fn execute_prune(ctx: &ReconcileContext<'_>, plan: &Plan, report: &mut Rec
                 report.pruned += 1;
                 tracing::info!(
                     site_key = %target.site_key,
-                    "reconcile: pruned an orphan preview (no live desired state)"
+                    "reconcile: pruned an orphan preview (PR merged/closed)"
                 );
             }
             Err(e) => {
@@ -273,153 +324,41 @@ async fn execute_prune(ctx: &ReconcileContext<'_>, plan: &Plan, report: &mut Rec
     }
 }
 
-/// 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`]).
+/// Parse a canonical preview site key `<owner>-<repo>-pr-<N>` into `(repo, N)`.
 ///
-/// # Errors
+/// Returns `None` for anything that is not a clean preview label for `owner`: an
+/// infra vhost (no `<owner>-` prefix), or a hashed/overflow label
+/// (`<base>-<6hex>`, whose tail after the last `-pr-` is not all digits). Both
+/// are fail-safe — the caller keeps them.
 ///
-/// 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)
+/// `<repo>` may itself contain hyphens (`php-sdk`, `switchboard-api`), so the
+/// split is on the **last** `-pr-<digits>`, matching how the label is built
+/// (`<owner>-<repo>-pr-<N>`); a repo that itself contains `-pr-` still resolves
+/// because the trailing number anchors the last one.
+fn parse_preview_site_key(key: &str, owner: &str) -> Option<(String, u64)> {
+    const SEP: &str = "-pr-";
+    let prefix = format!("{owner}-");
+    let rest = key.strip_prefix(&prefix)?;
+    let idx = rest.rfind(SEP)?;
+    let repo = &rest[..idx];
+    if repo.is_empty() {
+        return None;
+    }
+    let num = &rest[idx + SEP.len()..];
+    if num.is_empty() || !num.bytes().all(|b| b.is_ascii_digit()) {
+        return None;
+    }
+    let pr: u64 = num.parse().ok()?;
+    Some((repo.to_string(), pr))
 }
 
-/// Reconstruct the preview *label* from an on-disk site key — the inverse of
-/// [`site_key_for_label`].
+/// Reconstruct the preview *label* from an on-disk site key — the inverse of the
+/// deploy's `site_key(<label>.<preview_domain>)`.
 ///
-/// 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).
+/// With a suffix configured the key *is* the label; 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, which fails safe if the
+/// reconstruction is off (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();
@@ -440,7 +379,10 @@ fn label_for_site_key(site_key: &str, suffix: Option<&str>, preview_domain: &str
 /// # 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>> {
+fn list_preview_dirs(
+    sites_dir: &Path,
+    keep: &std::collections::HashSet<String>,
+) -> anyhow::Result<Vec<String>> {
     use anyhow::Context as _;
 
     let entries = match std::fs::read_dir(sites_dir) {
@@ -453,17 +395,14 @@ fn list_preview_dirs(sites_dir: &Path, keep: &HashSet<String>) -> anyhow::Result
 
     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.
+        // Follow symlinks: 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;
+            continue; // a deploy-in-flight staging dir, not a preview
         }
         if !site_key::is_valid_site_key(&name) {
             continue;
@@ -480,150 +419,155 @@ fn list_preview_dirs(sites_dir: &Path, keep: &HashSet<String>) -> anyhow::Result
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::job::sample_json;
+    use crate::queue::Queue;
+    use std::collections::HashSet;
 
     const DOMAIN: &str = "preview.ephpm.dev";
     const SUFFIX: Option<&str> = Some(".preview.ephpm.dev");
+    const OWNER: &str = "ephpm";
 
-    /// 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>> {
+    fn lookups(pairs: &[(&str, PrLookup)]) -> HashMap<String, PrLookup> {
         pairs
             .iter()
             .map(|(k, v)| ((*k).to_string(), v.clone()))
             .collect()
     }
 
+    // ── the site-key parser ─────────────────────────────────────────────
+
     #[test]
-    fn classify_reads_the_intent() {
-        assert_eq!(classify(Some(&deploy_doc("a"))), Desired::Live);
+    fn parses_clean_preview_keys_including_hyphenated_repos() {
+        assert_eq!(
+            parse_preview_site_key("ephpm-lab-pr-8", OWNER),
+            Some(("lab".to_string(), 8))
+        );
+        // Repos with hyphens: the split is the LAST `-pr-<digits>`.
+        assert_eq!(
+            parse_preview_site_key("ephpm-php-sdk-pr-67", OWNER),
+            Some(("php-sdk".to_string(), 67))
+        );
+        assert_eq!(
+            parse_preview_site_key("ephpm-switchboard-api-pr-5", OWNER),
+            Some(("switchboard-api".to_string(), 5))
+        );
+        // The org repo doubles the owner segment.
         assert_eq!(
-            classify(Some(&Some(sample_json("a", "teardown")))),
-            Desired::Gone
+            parse_preview_site_key("ephpm-ephpm-pr-488", OWNER),
+            Some(("ephpm".to_string(), 488))
         );
-        // 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.
+        // A repo that itself contains `-pr-` still resolves on the trailing number.
         assert_eq!(
-            classify(Some(&Some("{not json".to_string()))),
-            Desired::Gone
+            parse_preview_site_key("ephpm-my-pr-tool-pr-5", OWNER),
+            Some(("my-pr-tool".to_string(), 5))
         );
     }
 
     #[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");
+    fn refuses_non_preview_and_hashed_keys() {
+        // Infra vhosts have no `<owner>-` prefix.
+        assert_eq!(parse_preview_site_key("switchboard", OWNER), None);
+        assert_eq!(parse_preview_site_key("site-a", OWNER), None);
+        assert_eq!(parse_preview_site_key("preview.ephpm.dev", OWNER), None);
+        // A hashed/overflow label ends in `-<6hex>`, not `-pr-<digits>`.
+        assert_eq!(
+            parse_preview_site_key("ephpm-somelongrepo-pr-12-a1b2c3", OWNER),
+            None
+        );
+        // No number, empty repo, or non-digit tail.
+        assert_eq!(parse_preview_site_key("ephpm-lab-pr-", OWNER), None);
+        assert_eq!(parse_preview_site_key("ephpm--pr-5", OWNER), None);
+        assert_eq!(parse_preview_site_key("ephpm-lab-pr-x", OWNER), None);
+        // A different owner is not ours.
+        assert_eq!(parse_preview_site_key("other-lab-pr-8", OWNER), None);
     }
 
-    #[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");
-    }
+    // ── the prune decision ──────────────────────────────────────────────
 
     #[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),
+    fn merged_and_closed_prune_open_and_unknown_keep() {
+        let on_disk = vec![
+            "ephpm-lab-pr-1".to_string(), // open  -> keep
+            "ephpm-lab-pr-2".to_string(), // merged -> prune
+            "ephpm-lab-pr-3".to_string(), // closed -> prune
+            "ephpm-lab-pr-4".to_string(), // unknown -> keep
+        ];
+        let l = lookups(&[
+            ("ephpm-lab-pr-1", PrLookup::State(PullRequestState::Open)),
+            ("ephpm-lab-pr-2", PrLookup::State(PullRequestState::Merged)),
+            ("ephpm-lab-pr-3", PrLookup::State(PullRequestState::Closed)),
+            (
+                "ephpm-lab-pr-4",
+                PrLookup::State(PullRequestState::Unknown("locked".into())),
+            ),
         ]);
-        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());
+        let plan = build_plan(&on_disk, OWNER, SUFFIX, DOMAIN, &l);
+        let pruned: Vec<&str> = plan.prune.iter().map(|p| p.site_key.as_str()).collect();
+        assert_eq!(pruned, vec!["ephpm-lab-pr-2", "ephpm-lab-pr-3"]);
+        assert_eq!(plan.kept_open, 2, "open + unknown are kept");
+        assert_eq!(plan.skipped_api_error, 0);
+        assert_eq!(plan.skipped_unparseable, 0);
+        // The label is reconstructed for the marker (suffix configured => == key).
+        assert_eq!(plan.prune[0].label, "ephpm-lab-pr-2");
     }
 
     #[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());
+    fn an_api_error_keeps_the_preview() {
+        let on_disk = vec!["ephpm-lab-pr-9".to_string()];
+        let l = lookups(&[("ephpm-lab-pr-9", PrLookup::Error)]);
+        let plan = build_plan(&on_disk, OWNER, SUFFIX, DOMAIN, &l);
+        assert!(plan.prune.is_empty(), "a read failure must never prune");
+        assert_eq!(plan.skipped_api_error, 1);
     }
 
     #[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()]);
+    fn a_missing_lookup_keeps_the_preview() {
+        // Parseable but no lookup entry (e.g. it was added between listing and
+        // fetching): fail-safe to keep.
+        let on_disk = vec!["ephpm-lab-pr-9".to_string()];
+        let plan = build_plan(&on_disk, OWNER, SUFFIX, DOMAIN, &HashMap::new());
         assert!(plan.prune.is_empty());
+        assert_eq!(plan.skipped_api_error, 1);
     }
 
     #[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");
+    fn an_unparseable_key_is_never_a_candidate_even_if_a_lookup_exists() {
+        // A hashed label is kept regardless of any (spurious) lookup.
+        let on_disk = vec!["ephpm-x-pr-1-a1b2c3".to_string()];
+        let l = lookups(&[(
+            "ephpm-x-pr-1-a1b2c3",
+            PrLookup::State(PullRequestState::Merged),
+        )]);
+        let plan = build_plan(&on_disk, OWNER, SUFFIX, DOMAIN, &l);
+        assert!(plan.prune.is_empty(), "a hashed label is never pruned");
+        assert_eq!(plan.skipped_unparseable, 1);
     }
 
     #[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());
+    fn label_reconstruction_without_a_suffix_strips_the_domain() {
+        assert_eq!(
+            label_for_site_key("ephpm-lab-pr-8.preview.ephpm.dev", None, DOMAIN),
+            "ephpm-lab-pr-8"
+        );
+        assert_eq!(
+            label_for_site_key("ephpm-lab-pr-8", SUFFIX, DOMAIN),
+            "ephpm-lab-pr-8"
+        );
     }
 
+    // ── directory listing ───────────────────────────────────────────────
+
     #[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"] {
+        for dir in [
+            "ephpm-lab-pr-8",
+            "ephpm-switchboard-pr-36",
+            "switchboard",
+            "ephpm-lab-pr-8.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();
@@ -632,7 +576,10 @@ mod tests {
         let found = list_preview_dirs(sites, &keep).unwrap();
         assert_eq!(
             found,
-            vec!["live-pr-1".to_string(), "orphan-pr-2".to_string()],
+            vec![
+                "ephpm-lab-pr-8".to_string(),
+                "ephpm-switchboard-pr-36".to_string()
+            ],
             "keep-list, .tmp staging and plain files are all excluded"
         );
     }
@@ -644,41 +591,52 @@ mod tests {
         assert!(found.is_empty());
     }
 
+    // ── execute_prune (dry-run vs real, keep-guarded) ───────────────────
+
+    /// A context whose teardown roots point at one tempdir; GitHub/KV fields are
+    /// unused by `execute_prune` (it only runs the plan).
+    fn exec_ctx<'a>(
+        sites: &'a Path,
+        state: &'a Path,
+        keep: &'a HashSet<String>,
+        prune: bool,
+    ) -> ReconcileContext<'a> {
+        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 in the test
+            kv_secret: None,
+            kv_addr: "127.0.0.1:6379",
+            app_id: 1,
+            app_key: Path::new("/dev/null"),
+            owner: OWNER,
+            preview_domain: DOMAIN,
+            sites_domain_suffix: SUFFIX,
+            keep,
+            prune,
+        }
+    }
+
     #[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(sites.join("ephpm-lab-pr-2")).unwrap();
         std::fs::create_dir_all(&state).unwrap();
-        let queue = Queue::new(&state);
-        queue.ensure_dirs().unwrap();
+        Queue::new(&state).ensure_dirs().unwrap();
         let keep = HashSet::new();
+        let ctx = exec_ctx(&sites, &state, &keep, false);
 
-        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 {
+        let plan = PlanResult {
             prune: vec![PrunePlan {
-                site_key: "orphan-pr-2".to_string(),
-                label: "orphan-pr-2".to_string(),
+                site_key: "ephpm-lab-pr-2".to_string(),
+                label: "ephpm-lab-pr-2".to_string(),
             }],
-            deploy: vec![],
+            ..PlanResult::default()
         };
         let mut report = ReconcileReport::default();
         execute_prune(&ctx, &plan, &mut report).await;
@@ -686,7 +644,7 @@ mod tests {
         assert_eq!(report.would_prune, 1);
         assert_eq!(report.pruned, 0);
         assert!(
-            sites.join("orphan-pr-2").exists(),
+            sites.join("ephpm-lab-pr-2").exists(),
             "dry-run must not remove the directory"
         );
     }
@@ -696,92 +654,39 @@ mod tests {
         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(sites.join("ephpm-lab-pr-2")).unwrap();
         std::fs::create_dir_all(&state).unwrap();
-        let queue = Queue::new(&state);
-        queue.ensure_dirs().unwrap();
+        Queue::new(&state).ensure_dirs().unwrap();
         let keep = HashSet::new();
+        let ctx = exec_ctx(&sites, &state, &keep, true);
 
-        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 {
+        let plan = PlanResult {
             prune: vec![PrunePlan {
-                site_key: "orphan-pr-2".to_string(),
-                label: "orphan-pr-2".to_string(),
+                site_key: "ephpm-lab-pr-2".to_string(),
+                label: "ephpm-lab-pr-2".to_string(),
             }],
-            deploy: vec![],
+            ..PlanResult::default()
         };
         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());
+        assert_eq!(report.pruned, 1);
+        assert!(!sites.join("ephpm-lab-pr-2").exists());
     }
 
-    #[tokio::test]
-    async fn the_per_cycle_cap_defers_the_rest() {
+    /// The keep-list is honoured end-to-end: a keep-listed dir never reaches the
+    /// plan because `list_preview_dirs` excludes it.
+    #[test]
+    fn keep_list_excludes_a_dir_from_candidacy() {
         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;
+        let sites = root.path();
+        std::fs::create_dir_all(sites.join("site-a")).unwrap();
+        std::fs::create_dir_all(sites.join("ephpm-lab-pr-2")).unwrap();
+        let mut keep = HashSet::new();
+        keep.insert("site-a".to_string());
 
-        assert_eq!(report.pruned, 2, "cap honoured");
-        assert_eq!(
-            report.prune_deferred, 1,
-            "the third waits for the next pass"
-        );
+        let found = list_preview_dirs(sites, &keep).unwrap();
+        assert!(!found.contains(&"site-a".to_string()));
+        assert!(found.contains(&"ephpm-lab-pr-2".to_string()));
     }
 }