omni-dev 0.46.0

//! The cross-window worktree registry engine.
//!
//! Maintains the live, authoritative set of repos/worktrees open across *every*
//! VS Code window, fed by a first-party companion extension that reports from
//! each window over the daemon's control socket. The resident daemon is the
//! rendezvous point the per-window extension sandbox cannot replace: each window
//! can see only its own `workspace.workspaceFolders`, so a single process
//! aggregating those registrations is the only cross-window source of truth.
//! See ADR-0040.
//!
//! This is the standalone engine, analogous to [`crate::browser`] and
//! [`crate::snowflake`]; the daemon adapter lives in
//! [`crate::daemon::services::worktrees`].
//!
//! Like the Snowflake engine this is cheap and in-memory — no async setup, no
//! secret persisted. The registry lives behind a [`std::sync::Mutex`] that is
//! **never held across an `.await`** (the Snowflake rule); every op is pure CPU
//! under the lock, so liveness reaping happens inline on each read rather than
//! from a background task. The liveness TTL is measured on an `AwakeClock`, so
//! a system sleep ages nothing (#2126).

use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, MutexGuard, PoisonError};
use std::time::Duration;

use chrono::{DateTime, Duration as ChronoDuration, Utc};
use serde::{Deserialize, Serialize};
use tokio::sync::watch;

use crate::utils::awake_clock::AwakeClock;

/// How long a window may go silent before it ages out of the registry. Three
/// missed ~10s heartbeats; a window that crashed without firing `unregister`
/// disappears on the next read. The resident process is what makes this
/// liveness correct — a flat shared file could not reap stale entries.
///
/// Measured in *awake* time (#2126): see [`AwakeClock`]. A window cannot
/// heartbeat while the machine sleeps, so a system sleep of any length ages no
/// window, and the first read after a wake-up no longer reaps every one.
const DEFAULT_TTL: Duration = Duration::from_secs(30);

/// How long a per-repository PR-poll lease lasts before it auto-expires (#1376).
/// Enabling polling for a repo is deliberately **temporary** — 15 minutes — so an
/// idle repo stops costing GitHub budget without the user remembering to disable
/// it; re-enabling refreshes the lease.
const DEFAULT_POLL_LEASE: Duration = Duration::from_secs(15 * 60);

/// Ceiling on live registry entries, so a misbehaving companion flooding
/// `register` with distinct keys cannot grow daemon memory faster than the TTL
/// reaps it (#1140). Far above any real window count; when a new key would
/// exceed it, the longest-silent entry is evicted instead of rejecting the
/// request — an evicted live window self-heals via the `heartbeat` →
/// `{known: false}` → re-register path, so `register` stays infallible for the
/// companion.
const MAX_WINDOWS: usize = 256;

/// A `register` request from a companion extension.
///
/// The companion owns its `key` (a per-`activate()` UUID) so the registry never
/// has to reason about whether `vscode.env.sessionId` is unique per window;
/// everything else is best-effort metadata.
#[derive(Debug, Clone, Deserialize)]
pub struct RegisterRequest {
    /// Stable per-window identity, generated by the companion on activation.
    pub key: String,
    /// Absolute paths of the window's workspace folders.
    #[serde(default)]
    pub folders: Vec<PathBuf>,
    /// Repository root or name, when the window has one.
    #[serde(default)]
    pub repo: Option<String>,
    /// The window title, for display.
    #[serde(default)]
    pub title: Option<String>,
    /// The reporting extension-host process id.
    #[serde(default)]
    pub pid: Option<u32>,
}

/// One open window's live registration. Serialized verbatim into `list` /
/// `status` payloads; consumers compute age from `last_seen` (RFC 3339).
#[derive(Debug, Clone, Serialize)]
pub struct WindowEntry {
    /// The companion-owned per-window key.
    pub key: String,
    /// Absolute workspace-folder paths.
    pub folders: Vec<PathBuf>,
    /// Repository root or name, if reported.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub repo: Option<String>,
    /// Window title, if reported.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub title: Option<String>,
    /// Reporting extension-host pid, if reported.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub pid: Option<u32>,
    /// When the registry last heard from this window (register or heartbeat), by
    /// the wall clock. For display only: liveness is measured on `last_active`.
    pub last_seen: DateTime<Utc>,
    /// The registry's [`AwakeClock`] reading when this window was last heard
    /// from. [`reap`] and [`evict_oldest`] compare it, so a system sleep ages no
    /// window (#2126). Never serialized: a reading means nothing outside the
    /// registry that took it.
    #[serde(skip)]
    pub(crate) last_active: Duration,
}

/// The cross-window worktree registry: the in-memory, TTL-reaped set of open
/// windows. Hosted by
/// [`WorktreesService`](crate::daemon::services::worktrees::WorktreesService).
pub struct WorktreesRegistry {
    /// Open windows keyed by their companion-owned `key`.
    windows: Mutex<HashMap<String, WindowEntry>>,
    /// How long an entry survives without a heartbeat, in awake time.
    ttl: Duration,
    /// The clock the liveness TTL is measured on; see [`AwakeClock`].
    clock: AwakeClock,
    /// A monotonically-bumped version counter, incremented whenever the visible
    /// set of windows changes (a `register`, a removing `unregister`, or a
    /// mutation-driven reap that drops a stale entry). A push-subscription
    /// consumer holds a [`watch::Receiver`] from
    /// [`subscribe_changes`](Self::subscribe_changes) and wakes on each bump to
    /// re-snapshot (#1267). The counter's *value* is immaterial —
    /// only that it changed — so a burst coalesces into one wake and the
    /// subscriber diffs the resulting snapshot to suppress duplicate frames.
    ///
    /// `watch` needs no runtime and never blocks, so it fits this engine's
    /// no-async-setup posture; every [`bump`](Self::bump) happens *after* the map
    /// guard is dropped, so the `std::Mutex`-never-across-`.await` rule is intact
    /// (and the watch's own internal lock is never nested under the map lock).
    changes: watch::Sender<u64>,
    /// Window keys with a pending "close yourself" directive, set by the
    /// `close` op (#1277) when a cross-window close must reach a window the
    /// daemon can only *reply* to, never call. Each key is surfaced — and
    /// cleared — on that window's next `heartbeat` (the `known:false →
    /// re-register` precedent, riding the same reply). In-memory only, like the
    /// window map: a daemon restart drops any pending directive (the close op
    /// aborts and the user retries — an accepted failure mode). Behind its own
    /// `Mutex`, taken independently of the window map's, so neither nests.
    ///
    /// The op that sets a key also withdraws it
    /// ([`cancel_close_pending`](Self::cancel_close_pending)) when its wait
    /// *fails*, so a close the caller was told had failed has no later side
    /// effect (#2139). A successful close leaves its keys alone, and a reap clears
    /// nothing: a window counted closed only because it was reaped is handed the
    /// directive if it returns, and closes rather than staying open on a worktree
    /// the op removed.
    close_pending: Mutex<HashSet<String>>,
    /// Window keys with a pending "reload yourself" directive, set by the
    /// `reload` op (#1417). The second directive of the
    /// [`close_pending`](Self::close_pending) shape, and for the same reason: a
    /// window can only be reached on the `heartbeat` it initiates, so a
    /// cross-window reload rides that reply and is taken-and-cleared to fire
    /// exactly once. Unlike a close, nothing waits for it — a reload has no
    /// completion the daemon can observe (the window re-registers under the same
    /// key), so the op reports what it *signalled*, never what reloaded.
    /// In-memory only: a daemon restart drops any pending directive, and the
    /// user simply reloads again. Behind its **own** `Mutex`, taken
    /// independently of the window map's and of `close_pending`'s, so none nest.
    ///
    /// Only `unregister` clears a key; a reap deliberately does not (#2139). A
    /// window silent past the TTL that returns re-registers off its `known:
    /// false` heartbeat and is handed the directive there, which is the reload
    /// the user was told was signalled; clearing on reap would drop it. The cost
    /// is a UUID string left behind by a window that was signalled and then died
    /// without unregistering — accepted, since it takes a crash inside the
    /// signal-to-heartbeat gap and is not a growth path.
    reload_pending: Mutex<HashSet<String>>,
    /// Worktree paths the daemon is **currently rebasing** (#1415) — the
    /// transient half of the tree view's rebase cue.
    ///
    /// Two orthogonal facts drive that cue and neither substitutes for the other.
    /// The *durable* one is the worktree's own `repo.state()`, read fresh off disk
    /// into each snapshot: it survives a daemon restart and keeps showing a
    /// left-in-place conflict until the user resolves it. This is the *transient*
    /// one: a rebase that is running right now has not yet written a
    /// `.git/rebase-merge` state the snapshot can see for most of its life, and a
    /// clean rebase never leaves one at all — so without this a multi-second
    /// rebase would render as nothing happening.
    ///
    /// Unlike the two directives above this is **consumer-visible state**, not a
    /// message to one window: it rides the `tree` snapshot, so marking and
    /// clearing it bumps the change-notify (a directive never does).
    ///
    /// Keyed by **canonicalized worktree path**, not window key: a rebase targets
    /// worktrees, and a worktree need not have a window open on it. Behind its own
    /// `Mutex`, taken independently of the window map's, `close_pending`'s and
    /// `reload_pending`'s, so none of the four ever nest. In-memory only — a
    /// daemon restart clears it, which is correct: nothing is rebasing any more,
    /// and the durable `operation` field still shows any conflict left behind.
    rebasing: Mutex<HashSet<PathBuf>>,
    /// Worktree paths the daemon is **currently pushing** (#1443) — the twin of
    /// [`rebasing`](Self::rebasing), with the same keying, locking and
    /// bump-only-on-a-real-change contract.
    ///
    /// Unlike a rebase this cue has **no durable half**: a push writes no on-disk
    /// state for a later snapshot to rediscover, so there is no `operation`
    /// equivalent and this set is the whole of it. What makes a *completed* push
    /// visible instead is `upstream_sha` moving on the snapshot — which is exactly
    /// why that field is carried (#1344).
    pushing: Mutex<HashSet<PathBuf>>,
    /// The daemon-backed **show/hide-closed** toggle (#1301): whether the
    /// companion's tree view shows worktrees with no open window. A single
    /// cross-window value carried in every `tree`/`subscribe` snapshot so all
    /// windows read (and live-sync) the same state — `context.globalState` could
    /// not, being read-once with no cross-window change event. Defaults to `true`
    /// (show all, the original behavior). A lock-free [`AtomicBool`] rather than a
    /// `Mutex`, so it is never a `.await`-holding-a-lock hazard; a flip
    /// [`bump`](Self::bump)s the change-notify so subscribers re-push. In-memory
    /// like the window map: a daemon restart resets it to the default, which the
    /// next snapshot propagates to every window.
    show_closed: AtomicBool,
    /// The **per-repository PR-poll** enable set (#1376): the GitHub repos
    /// (`"owner/name"`) whose PR badges the daemon polls. Polling defaults
    /// **off** — a repo not in this set issues zero `gh` — so the user enables
    /// only the handful of repos they are actively working on, rather than the
    /// daemon polling all 29 open repos and exhausting the GitHub budget. A
    /// cross-window value like [`show_closed`](Self::show_closed): the daemon
    /// stamps each repo's state onto the `tree` snapshot (`polling_enabled`) and
    /// a [`set_polling`](Self::set_polling) flip [`bump`](Self::bump)s the
    /// change-notify so every window recolors and drops/keeps badges in sync.
    ///
    /// Unlike `show_closed` this survives a daemon restart: the adapter seeds it
    /// from a `0600` file on startup ([`seed_polling`](Self::seed_polling)) and
    /// persists it on each change — otherwise a restart would silently re-disable
    /// every repo. Behind its **own** `Mutex`, taken independently of the window
    /// map's (neither nests) and never held across an `.await`.
    ///
    /// Each enable is a **time-boxed lease**, not a permanent flag (#1376): the
    /// value is the wall-clock instant the lease **expires** ([`poll_ttl`] after
    /// it was enabled), so an idle repo auto-disables and stops costing `gh`
    /// without the user remembering to turn it back off. Expired entries are
    /// reaped on read — the window-TTL precedent — so the icon greys, badges
    /// drop, and the poller stops within one snapshot tick of expiry.
    ///
    /// [`poll_ttl`]: Self::poll_ttl
    polling_enabled: Mutex<HashMap<String, DateTime<Utc>>>,
    /// How long a repo's PR-poll lease lasts before it auto-expires (#1376).
    /// [`DEFAULT_POLL_LEASE`] in production; tests inject a short value via the
    /// `#[cfg(test)]` `with_poll_ttl` constructor (not linked — it does not exist
    /// in a non-test doc build).
    poll_ttl: Duration,
}

impl WorktreesRegistry {
    /// Creates the registry with the default liveness TTL. Cheap — no I/O.
    #[must_use]
    pub fn new() -> Self {
        Self {
            windows: Mutex::new(HashMap::new()),
            ttl: DEFAULT_TTL,
            clock: AwakeClock::new(),
            changes: watch::channel(0).0,
            close_pending: Mutex::new(HashSet::new()),
            reload_pending: Mutex::new(HashSet::new()),
            rebasing: Mutex::new(HashSet::new()),
            pushing: Mutex::new(HashSet::new()),
            show_closed: AtomicBool::new(true),
            polling_enabled: Mutex::new(HashMap::new()),
            poll_ttl: DEFAULT_POLL_LEASE,
        }
    }

    /// Creates a registry with a custom PR-poll lease duration, for tests that
    /// exercise auto-expiry without waiting the full 15 minutes.
    #[cfg(test)]
    #[must_use]
    pub fn with_poll_ttl(poll_ttl: Duration) -> Self {
        Self {
            poll_ttl,
            ..Self::new()
        }
    }

    /// A change-notification receiver for the push subscription: it observes a
    /// new value each time the visible window set changes (see [`changes`] and
    /// [`bump`]). Created with the current version already marked seen, so the
    /// first [`watch::Receiver::changed`] resolves on the *next* change — the
    /// subscriber sends its own initial snapshot up front and then waits for
    /// deltas (#1267).
    ///
    /// [`changes`]: Self::changes
    /// [`bump`]: Self::bump
    #[must_use]
    pub fn subscribe_changes(&self) -> watch::Receiver<u64> {
        self.changes.subscribe()
    }

    /// Signals subscribers that the visible state changed. Non-blocking and
    /// runtime-free; called only *after* the map guard is released so the two
    /// locks never nest. A send never fails here (the sender is owned by the
    /// registry, which outlives every receiver, and `send_modify` bumps even
    /// with no receivers).
    ///
    /// Visible outside the registry so the daemon's PR badge poller can signal a
    /// changed CI verdict (#1337) — the tree snapshot carries more than the window
    /// set. Callers must bump **only on a real change**: an unconditional bump
    /// defeats the server's snapshot diff and re-pushes to every window on every
    /// tick.
    pub(crate) fn bump(&self) {
        self.changes.send_modify(|v| *v = v.wrapping_add(1));
    }

    /// The current change-notify generation — the counter [`bump`](Self::bump)
    /// increments on every visible-set change. Read (never subscribed) so a
    /// coalescing consumer — the service's shared tree-snapshot cache (#1303) —
    /// can tell whether the registry has changed since it last computed and
    /// rebuild only then. The value itself is immaterial; only whether it
    /// differs between two reads matters, so `wrapping_add` overflow is benign.
    #[must_use]
    pub fn change_generation(&self) -> u64 {
        *self.changes.borrow()
    }

    /// Locks the registry, recovering from a poisoned mutex (a panic in a prior
    /// critical section must not wedge the whole registry).
    fn lock(&self) -> MutexGuard<'_, HashMap<String, WindowEntry>> {
        self.windows.lock().unwrap_or_else(PoisonError::into_inner)
    }

    /// Records (upserts) a window registration. Reaps stale entries first, then
    /// — only when a genuinely new key would grow the map past [`MAX_WINDOWS`] —
    /// evicts the longest-silent entry. Infallible: an upsert never evicts, and
    /// callers validate the `key` before reaching here.
    pub fn register(&self, req: RegisterRequest) {
        let now = Utc::now();
        let awake = self.clock.now();
        {
            let mut windows = self.lock();
            reap(&mut windows, self.ttl, awake);
            // Upserts never evict; only a genuinely new key can grow the map, and
            // never past MAX_WINDOWS.
            if !windows.contains_key(&req.key) && windows.len() >= MAX_WINDOWS {
                evict_oldest(&mut windows);
            }
            windows.insert(
                req.key.clone(),
                WindowEntry {
                    key: req.key,
                    folders: req.folders,
                    repo: req.repo,
                    title: req.title,
                    pid: req.pid,
                    last_seen: now,
                    last_active: awake,
                },
            );
        }
        // Always bump: a register is infrequent (once per companion `activate()`,
        // not per heartbeat) and may add or alter a window's folders/repo. A
        // no-op re-register with identical data is harmless — the subscriber
        // diffs the snapshot and suppresses the duplicate frame.
        self.bump();
    }

    /// Pauses the liveness clock for `outage`: advances every live entry's
    /// `last_active` (and its display `last_seen`) by that long, never past now.
    ///
    /// For a stretch when the daemon could not hear its windows — its `accept`
    /// loop was failing (#2111) — so their silence says nothing about whether they
    /// are alive. Without this the first heartbeat after an outage reaps every
    /// *other* window that has been quiet more than the TTL, before it has had a
    /// chance to send its own, and whole repos vanish from the tree for a few
    /// seconds. Only entries still in the map are credited: an entry that was
    /// already reaped stays gone, and one that was already stale before the outage
    /// began is credited no more than its neighbours. Not a visible change, so it
    /// does not bump the change-notify.
    ///
    /// `outage` is measured on a monotonic [`Instant`](std::time::Instant), the
    /// same awake time as the `AwakeClock` the entries are stamped on, so the
    /// two compose: the credit shifts a stamp by time the daemon was awake but
    /// deaf, and a system sleep still ages nothing (#2126).
    pub fn credit_outage(&self, outage: Duration) {
        let credit = ChronoDuration::from_std(outage).ok();
        let now = Utc::now();
        let mut windows = self.lock();
        // Read under the lock, not before it: every stamp already in the map was
        // taken before its writer locked, so a reading taken now is at least as
        // new as any of them and the clamp below can never pull one backwards.
        let awake = self.clock.now();
        for entry in windows.values_mut() {
            // Saturate rather than overflow on an absurd credit.
            entry.last_active = entry
                .last_active
                .checked_add(outage)
                .map_or(awake, |credited| credited.min(awake));
            entry.last_seen = credit
                .and_then(|credit| entry.last_seen.checked_add_signed(credit))
                .map_or(now, |credited| credited.min(now));
        }
    }

    /// Refreshes a window's liveness. Returns whether the key was known: a
    /// `false` tells a window that started before the daemon — or survived a
    /// daemon restart — to re-`register`, since the registry is in-memory and
    /// has no record of it.
    pub fn heartbeat(&self, key: &str) -> bool {
        let now = Utc::now();
        let awake = self.clock.now();
        let (known, reaped) = {
            let mut windows = self.lock();
            let reaped = reap(&mut windows, self.ttl, awake);
            let known = match windows.get_mut(key) {
                Some(entry) => {
                    entry.last_seen = now;
                    entry.last_active = awake;
                    true
                }
                None => false,
            };
            (known, reaped)
        };
        // A heartbeat is frequent (~every 10 s per window); a pure liveness
        // refresh does not change the visible set, so bump *only* when this
        // heartbeat's inline reap actually aged a stale sibling out.
        if reaped > 0 {
            self.bump();
        }
        known
    }

    /// Drops a window's registration. Returns whether an entry was present.
    pub fn unregister(&self, key: &str) -> bool {
        let awake = self.clock.now();
        let (removed, reaped) = {
            let mut windows = self.lock();
            let removed = windows.remove(key).is_some();
            let reaped = reap(&mut windows, self.ttl, awake);
            (removed, reaped)
        };
        // The window is gone; any directive for it is fulfilled or moot.
        // (Keys are per-`activate()` UUIDs, never reused, so a stale directive
        // would only ever leak a little memory — but clearing keeps it tidy.)
        // Both takes are outside the map's critical section, so no lock nests.
        self.take_close_pending(key);
        self.take_reload_pending(key);
        if removed || reaped > 0 {
            self.bump();
        }
        removed
    }

    /// Records a pending "close yourself" directive for `key`, to be surfaced on
    /// that window's next `heartbeat`. Set by the `close` op when signalling a
    /// window it can only reply to. Idempotent; infallible.
    pub fn mark_close_pending(&self, key: &str) {
        self.close_pending
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .insert(key.to_string());
    }

    /// Takes (returns and clears) `key`'s pending close directive. Called on
    /// each `heartbeat` so the directive fires exactly once; a `false` means no
    /// close is pending.
    pub fn take_close_pending(&self, key: &str) -> bool {
        self.close_pending
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .remove(key)
    }

    /// Withdraws `key`'s pending close directive without delivering it (#2139):
    /// the `close` op's counterpart to [`mark_close_pending`](Self::mark_close_pending),
    /// called when its wait fails so a directive it set cannot outlive an op the
    /// caller was told had failed. Without it a timed-out close reports failure
    /// while the window closes itself on its next heartbeat.
    ///
    /// Returns whether the directive was still pending. A `false` means the
    /// window already took it (it is closing, only slowly — a delivered
    /// directive cannot be recalled) or it was cleared by an `unregister`.
    ///
    /// The set holds a key once, not a count, so two concurrent closes whose
    /// targets share a multi-root window share one entry: whichever gives up
    /// first withdraws it for both. The ops are fanned out together, so the other
    /// loses at most the skew between their starts, and only if the window comes
    /// back inside it after being unresponsive for the whole of the first's wait.
    /// That is not worth a per-op ticket.
    pub fn cancel_close_pending(&self, key: &str) -> bool {
        self.take_close_pending(key)
    }

    /// Records a pending "reload yourself" directive for `key`, to be surfaced
    /// on that window's next `heartbeat` (#1417). Set by the `reload` op for
    /// every target window, which — unlike a close — includes no waiting: the
    /// caller learns only that the directive was marked. Idempotent; infallible.
    pub fn mark_reload_pending(&self, key: &str) {
        self.reload_pending
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .insert(key.to_string());
    }

    /// Takes (returns and clears) `key`'s pending reload directive. Called on
    /// each `heartbeat` so the directive fires exactly once; a `false` means no
    /// reload is pending.
    pub fn take_reload_pending(&self, key: &str) -> bool {
        self.reload_pending
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .remove(key)
    }

    /// Marks `paths` as being rebased right now (#1415), returning whether the
    /// set actually changed.
    ///
    /// A real change [`bump`](Self::bump)s the change-notify so every subscribed
    /// window re-pushes a snapshot carrying the spinner — the same cross-window
    /// sync `set_show_closed` / `set_polling` rely on. Bumping **only** on a real
    /// change is load-bearing rather than an optimization: an unconditional bump
    /// defeats the server's snapshot diff and re-pushes to every window on every
    /// tick.
    ///
    /// Callers pass **already-canonicalized** paths, so these match the tree
    /// snapshot's own keys. Canonicalizing is disk I/O, which belongs in the
    /// adapter (where `canonical()` lives), not in this engine — the same split
    /// that keeps the git enrichment out of here.
    pub fn mark_rebasing(&self, paths: &[PathBuf]) -> bool {
        self.mutate_in_flight(&self.rebasing, paths, true)
    }

    /// Clears the rebasing mark on `paths`, returning whether the set changed.
    /// Called on **every** exit from a phase-2 execute, so a panicking or failing
    /// rebase can never leave a permanent spinner on a row.
    pub fn clear_rebasing(&self, paths: &[PathBuf]) -> bool {
        self.mutate_in_flight(&self.rebasing, paths, false)
    }

    /// Marks `paths` as being pushed right now (#1443), returning whether the set
    /// actually changed. The [`mark_rebasing`](Self::mark_rebasing) twin, with the
    /// same canonicalized-paths-in and bump-only-on-a-real-change contract.
    pub fn mark_pushing(&self, paths: &[PathBuf]) -> bool {
        self.mutate_in_flight(&self.pushing, paths, true)
    }

    /// Clears the pushing mark on `paths`, returning whether the set changed.
    /// Called on **every** exit from a phase-2 execute — and it is the *only* thing
    /// that clears this cue, since a push leaves no on-disk state a later snapshot
    /// could correct a stale mark from.
    pub fn clear_pushing(&self, paths: &[PathBuf]) -> bool {
        self.mutate_in_flight(&self.pushing, paths, false)
    }

    /// The shared body of the in-flight-set mutators: mutate under that set's own
    /// lock, drop the guard, then bump if anything moved (never bump while holding
    /// a lock, per the engine's `std::Mutex`-never-across-`.await` discipline).
    ///
    /// Bumping **only** on a real change is load-bearing rather than an
    /// optimization: an unconditional bump defeats the server's snapshot diff and
    /// re-pushes to every window on every tick.
    fn mutate_in_flight(
        &self,
        set: &Mutex<HashSet<PathBuf>>,
        paths: &[PathBuf],
        insert: bool,
    ) -> bool {
        let changed = {
            let mut set = set.lock().unwrap_or_else(PoisonError::into_inner);
            paths.iter().fold(false, |changed, path| {
                let moved = if insert {
                    set.insert(path.clone())
                } else {
                    set.remove(path)
                };
                changed || moved
            })
        };
        if changed {
            self.bump();
        }
        changed
    }

    /// The worktree paths currently being rebased, canonicalized. Read into each
    /// `tree`/`subscribe` snapshot; cheap (a set clone of at most a batch's worth
    /// of paths) and never held across an `.await`.
    #[must_use]
    pub fn rebasing_paths(&self) -> HashSet<PathBuf> {
        Self::snapshot_paths(&self.rebasing)
    }

    /// The worktree paths currently being pushed, canonicalized (#1443). The
    /// [`rebasing_paths`](Self::rebasing_paths) twin, read into the same snapshot.
    #[must_use]
    pub fn pushing_paths(&self) -> HashSet<PathBuf> {
        Self::snapshot_paths(&self.pushing)
    }

    /// A clone of one in-flight set, for folding into a tree snapshot.
    fn snapshot_paths(set: &Mutex<HashSet<PathBuf>>) -> HashSet<PathBuf> {
        set.lock().unwrap_or_else(PoisonError::into_inner).clone()
    }

    /// The current show/hide-closed toggle: whether the tree view shows
    /// worktrees with no open window (#1301). Read into every `tree`/`subscribe`
    /// snapshot so every window renders the same, live-synced state.
    #[must_use]
    pub fn show_closed(&self) -> bool {
        self.show_closed.load(Ordering::Relaxed)
    }

    /// Sets the show/hide-closed toggle, returning whether the value actually
    /// changed. A real change [`bump`](Self::bump)s the change-notify so every
    /// subscriber re-pushes a snapshot carrying the new value — the reliable
    /// cross-window sync `context.globalState` could not provide. A no-op set
    /// (same value) neither bumps nor wakes anyone.
    pub fn set_show_closed(&self, show_closed: bool) -> bool {
        let changed = self.show_closed.swap(show_closed, Ordering::Relaxed) != show_closed;
        if changed {
            self.bump();
        }
        changed
    }

    /// Locks the per-repo PR-poll lease map (`"owner/name"` → lease-expiry
    /// instant), recovering from a poisoned mutex (a panic in a prior critical
    /// section must not wedge polling for the whole daemon).
    fn polling_lock(&self) -> MutexGuard<'_, HashMap<String, DateTime<Utc>>> {
        self.polling_enabled
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
    }

    /// Whether PR polling is currently leased for the GitHub repo `owner/name`
    /// (#1376) — the entry exists **and** its lease has not expired. Defaults
    /// **false** (a never-toggled repo does not poll), so only repos the user has
    /// explicitly enabled, within the last [`poll_ttl`](Self::poll_ttl), poll.
    #[must_use]
    pub fn is_polling_enabled(&self, owner: &str, name: &str) -> bool {
        let now = Utc::now();
        self.polling_lock()
            .get(&polling_key(owner, name))
            .is_some_and(|expiry| *expiry > now)
    }

    /// The repos with a **live** (unexpired) lease, as a set of `"owner/name"`
    /// keys — what stamps `polling_enabled` onto the `tree` snapshot. Reaps
    /// expired leases first (the window-TTL reap-on-read precedent), so an idle
    /// repo drops out on the next snapshot build without a background timer.
    /// Cloned out so the lock is never held across the (blocking-thread) tree
    /// build that reads it.
    #[must_use]
    pub fn enabled_polling_repos(&self) -> HashSet<String> {
        let now = Utc::now();
        let mut map = self.polling_lock();
        map.retain(|_, expiry| *expiry > now);
        map.keys().cloned().collect()
    }

    /// The live leases as `(repo, expiry)` pairs sorted by repo, for
    /// deterministic persistence to the `0600` prefs file (#1376) — the expiry is
    /// stored so a daemon restart within the lease window keeps the *remaining*
    /// time rather than resetting the clock. Reaps expired leases first, so a
    /// stale entry is never written back.
    #[must_use]
    pub fn polling_snapshot(&self) -> Vec<(String, DateTime<Utc>)> {
        let now = Utc::now();
        let mut map = self.polling_lock();
        map.retain(|_, expiry| *expiry > now);
        let mut entries: Vec<(String, DateTime<Utc>)> =
            map.iter().map(|(k, v)| (k.clone(), *v)).collect();
        entries.sort_by(|a, b| a.0.cmp(&b.0));
        entries
    }

    /// Enables (leases for [`poll_ttl`](Self::poll_ttl)) or disables PR polling
    /// for `owner/name`, returning whether the stored map changed — which the
    /// adapter uses to decide whether to persist. Enabling an already-leased repo
    /// **refreshes** the lease (a new expiry), which is a change worth persisting.
    ///
    /// [`bump`](Self::bump)s the change-notify only when the repo's **effective**
    /// enabled state flips (off→on or on→off), so every subscribed window
    /// re-pushes a snapshot that recolours the icon and drops/keeps badges — the
    /// [`set_show_closed`](Self::set_show_closed) precedent. A lease *refresh*
    /// (already on, still on) changes the expiry but not the visible state, so it
    /// persists without waking anyone.
    pub fn set_polling(&self, owner: &str, name: &str, enabled: bool) -> bool {
        let key = polling_key(owner, name);
        let now = Utc::now();
        let (changed, flipped) = {
            let mut map = self.polling_lock();
            let was_enabled = map.get(&key).is_some_and(|expiry| *expiry > now);
            if enabled {
                let expiry = now
                    + ChronoDuration::from_std(self.poll_ttl).unwrap_or_else(|_| {
                        ChronoDuration::seconds(DEFAULT_POLL_LEASE.as_secs() as i64)
                    });
                let changed = map.insert(key, expiry) != Some(expiry);
                (changed, !was_enabled)
            } else {
                let removed = map.remove(&key).is_some();
                (removed, was_enabled)
            }
        };
        if flipped {
            self.bump();
        }
        changed
    }

    /// Replaces the lease map wholesale from the persisted `0600` prefs file
    /// (#1376), dropping any lease that already expired while the daemon was down.
    /// Does **not** [`bump`](Self::bump): it runs before any window subscribes, so
    /// there is no one to notify, and each window's first snapshot already
    /// reflects the seeded leases.
    pub fn seed_polling(&self, leases: impl IntoIterator<Item = (String, DateTime<Utc>)>) {
        let now = Utc::now();
        *self.polling_lock() = leases
            .into_iter()
            .filter(|(_, expiry)| *expiry > now)
            .collect();
    }

    /// Test-only: advances the awake clock by `by`, so a test outside this module
    /// can age a window past the TTL without sleeping for it.
    #[cfg(test)]
    pub fn advance_clock(&self, by: Duration) {
        self.clock.advance(by);
    }

    /// Test-only: forces `owner/name`'s lease to `expiry`, so a test can simulate
    /// an elapsed lease deterministically without sleeping for [`poll_ttl`].
    #[cfg(test)]
    pub fn set_polling_expiry(&self, owner: &str, name: &str, expiry: DateTime<Utc>) {
        self.polling_lock().insert(polling_key(owner, name), expiry);
    }

    /// Reaps stale entries, then returns the live set sorted for deterministic
    /// output. Holds the lock only for pure-CPU work.
    ///
    /// Like the other reads ([`open_folders`](Self::open_folders),
    /// [`first_folder`](Self::first_folder)) this reaps but never
    /// [`bump`](Self::bump)s: the only observer of a read-path reap is the push
    /// subscription's own re-snapshot (or `status`/`menu`), and the
    /// subscription's periodic tick already re-samples read-only staleness — so
    /// bumping here would only make the subscription wake itself (#1267).
    pub fn list(&self) -> Vec<WindowEntry> {
        let awake = self.clock.now();
        let mut windows = self.lock();
        reap(&mut windows, self.ttl, awake);
        sorted_entries(&windows)
    }

    /// The first workspace folder of a still-live window, if it has one. Used by
    /// the tray "focus" action to resolve a key to a folder to open. Does not
    /// reap — a menu action races the reaper either way, and the caller handles
    /// a `None` (the window may have closed).
    pub fn first_folder(&self, key: &str) -> Option<PathBuf> {
        let windows = self.lock();
        windows.get(key).and_then(|e| e.folders.first().cloned())
    }

    /// Snapshots the distinct workspace folders across all live windows — the
    /// seed set the adapter resolves to repositories (each folder → its git
    /// common dir → repo root) to enumerate every worktree per repo (#1265).
    ///
    /// Reaps stale entries first, then returns the folders sorted and
    /// deduplicated. Like [`list`](Self::list) it is pure CPU under the lock:
    /// the git resolution the "distinct repos" derivation needs is disk I/O and
    /// stays in the adapter, off the registry lock, honouring the
    /// `Mutex`-never-across-`.await` invariant.
    pub fn open_folders(&self) -> Vec<PathBuf> {
        let awake = self.clock.now();
        let mut windows = self.lock();
        reap(&mut windows, self.ttl, awake);
        let mut folders: Vec<PathBuf> = windows
            .values()
            .flat_map(|e| e.folders.iter().cloned())
            .collect();
        folders.sort();
        folders.dedup();
        folders
    }
}

impl Default for WorktreesRegistry {
    fn default() -> Self {
        Self::new()
    }
}

/// Removes entries last heard from longer than `ttl` ago on the awake clock,
/// returning how many were dropped. `awake` is the registry's [`AwakeClock`]
/// reading, so time spent asleep never counts (#2126). Pure CPU; the caller holds
/// the registry lock but never `.await`s while holding it. The count lets a
/// *mutation* path ([`register`](WorktreesRegistry::register) et al.) decide
/// whether to [`bump`](WorktreesRegistry::bump) the change-notify; read paths
/// ignore it (see [`list`](WorktreesRegistry::list)).
fn reap(windows: &mut HashMap<String, WindowEntry>, ttl: Duration, awake: Duration) -> usize {
    let before = windows.len();
    windows.retain(|_, e| awake.saturating_sub(e.last_active) <= ttl);
    before - windows.len()
}

/// Removes the entry with the oldest `last_active` (ties broken by lowest key
/// for determinism). Called when a `register` of a new key would grow the
/// registry past [`MAX_WINDOWS`]. Pure CPU under the registry lock, like
/// [`reap`].
fn evict_oldest(windows: &mut HashMap<String, WindowEntry>) {
    let oldest = windows
        .values()
        .min_by(|a, b| {
            a.last_active
                .cmp(&b.last_active)
                .then_with(|| a.key.cmp(&b.key))
        })
        .map(|e| e.key.clone());
    if let Some(key) = oldest {
        windows.remove(&key);
    }
}

/// The canonical key for a GitHub repo in the per-repo PR-poll set: `owner/name`
/// (#1376). One place so the registry's set, the snapshot stamp, and the poller
/// filter all agree on the exact string.
fn polling_key(owner: &str, name: &str) -> String {
    format!("{owner}/{name}")
}

/// Snapshots the registry into a stably-ordered vector (by repo, then key) so
/// `list`/`status`/`menu` output is deterministic despite `HashMap` ordering.
fn sorted_entries(windows: &HashMap<String, WindowEntry>) -> Vec<WindowEntry> {
    let mut entries: Vec<WindowEntry> = windows.values().cloned().collect();
    entries.sort_by(|a, b| a.repo.cmp(&b.repo).then_with(|| a.key.cmp(&b.key)));
    entries
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
    use super::*;

    fn register_request(key: &str, repo: Option<&str>, folder: &str) -> RegisterRequest {
        RegisterRequest {
            key: key.to_string(),
            folders: vec![PathBuf::from(folder)],
            repo: repo.map(str::to_string),
            title: Some(format!("{key}-title")),
            pid: Some(1234),
        }
    }

    #[test]
    fn list_is_empty_initially() {
        let reg = WorktreesRegistry::new();
        assert!(reg.list().is_empty());
    }

    #[test]
    fn register_then_list_round_trips() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", Some("repo-a"), "/tmp/a"));
        let windows = reg.list();
        assert_eq!(windows.len(), 1);
        assert_eq!(windows[0].key, "w1");
        assert_eq!(windows[0].repo.as_deref(), Some("repo-a"));
    }

    #[test]
    fn register_is_idempotent_upsert() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", Some("repo-a"), "/tmp/a"));
        // Re-registering the same key updates rather than duplicates.
        reg.register(register_request("w1", Some("repo-b"), "/tmp/b"));
        let windows = reg.list();
        assert_eq!(windows.len(), 1);
        assert_eq!(windows[0].repo.as_deref(), Some("repo-b"));
    }

    #[test]
    fn heartbeat_reports_known_and_unknown() {
        let reg = WorktreesRegistry::new();
        // Unknown before registration: the window must re-register.
        assert!(!reg.heartbeat("w1"));
        reg.register(register_request("w1", None, "/tmp/a"));
        assert!(reg.heartbeat("w1"));
    }

    #[test]
    fn unregister_removes() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        assert!(reg.unregister("w1"));
        // Removing again is a no-op.
        assert!(!reg.unregister("w1"));
    }

    #[test]
    fn first_folder_returns_first_folder_or_none() {
        let reg = WorktreesRegistry::new();
        // No such key.
        assert!(reg.first_folder("missing").is_none());
        reg.register(register_request("w1", None, "/tmp/a"));
        assert_eq!(reg.first_folder("w1"), Some(PathBuf::from("/tmp/a")));
        // A folderless window resolves to None rather than a folder.
        reg.register(RegisterRequest {
            key: "w2".to_string(),
            folders: vec![],
            repo: None,
            title: None,
            pid: None,
        });
        assert!(reg.first_folder("w2").is_none());
    }

    #[test]
    fn open_folders_dedups_and_sorts_across_windows() {
        let reg = WorktreesRegistry::new();
        assert!(reg.open_folders().is_empty());
        // Two windows sharing a folder, plus a multi-folder window: the shared
        // path collapses and the result is sorted.
        reg.register(register_request("w1", Some("repo-a"), "/tmp/shared"));
        reg.register(RegisterRequest {
            key: "w2".to_string(),
            folders: vec![PathBuf::from("/tmp/shared"), PathBuf::from("/tmp/b")],
            repo: Some("repo-a".to_string()),
            title: None,
            pid: None,
        });
        reg.register(register_request("w3", Some("repo-b"), "/tmp/a"));
        assert_eq!(
            reg.open_folders(),
            vec![
                PathBuf::from("/tmp/a"),
                PathBuf::from("/tmp/b"),
                PathBuf::from("/tmp/shared"),
            ]
        );
    }

    #[test]
    fn open_folders_reaps_stale_windows() {
        let reg = registry_with_headroom();
        insert_silent(&reg, "fresh", Duration::ZERO);
        insert_silent(&reg, "stale", Duration::from_secs(120));
        // The stale window's folder is reaped out of the snapshot.
        reg.lock().get_mut("fresh").unwrap().folders = vec![PathBuf::from("/tmp/fresh")];
        reg.lock().get_mut("stale").unwrap().folders = vec![PathBuf::from("/tmp/stale")];
        assert_eq!(reg.open_folders(), vec![PathBuf::from("/tmp/fresh")]);
    }

    #[test]
    fn reap_evicts_only_stale_entries() {
        let awake = Duration::from_secs(1000);
        let mut windows = HashMap::new();
        windows.insert(
            "fresh".to_string(),
            entry_at("fresh", awake.saturating_sub(Duration::from_secs(5))),
        );
        windows.insert(
            "stale".to_string(),
            entry_at("stale", awake.saturating_sub(Duration::from_secs(120))),
        );
        assert_eq!(reap(&mut windows, DEFAULT_TTL, awake), 1);
        assert!(windows.contains_key("fresh"));
        assert!(!windows.contains_key("stale"));
    }

    #[test]
    fn reap_keeps_an_entry_exactly_at_the_ttl() {
        let awake = Duration::from_secs(1000);
        let mut windows = HashMap::new();
        windows.insert(
            "edge".to_string(),
            entry_at("edge", awake.saturating_sub(DEFAULT_TTL)),
        );
        // A stamp ahead of the reading (a credit never produces one, but a
        // saturating subtraction must not wrap if it ever did) is simply fresh.
        windows.insert(
            "ahead".to_string(),
            entry_at("ahead", awake + Duration::from_secs(5)),
        );
        assert_eq!(reap(&mut windows, DEFAULT_TTL, awake), 0);
        assert_eq!(
            reap(&mut windows, DEFAULT_TTL, awake + Duration::from_millis(1)),
            1,
            "a hair past the TTL reaps the edge entry only"
        );
        assert!(windows.contains_key("ahead"));
    }

    // --- Outage credit (#2111) ----------------------------------------------

    /// A registry whose awake clock has room to back-date a stamp by minutes (a
    /// fresh clock reads ~0, and a stamp cannot be set before it started).
    fn registry_with_headroom() -> WorktreesRegistry {
        let reg = WorktreesRegistry::new();
        reg.clock.advance(Duration::from_secs(1000));
        reg
    }

    /// Inserts `key` straight into the map, last heard from `silent_for` ago on
    /// the registry's awake clock.
    fn insert_silent(reg: &WorktreesRegistry, key: &str, silent_for: Duration) {
        let last_active = reg
            .clock
            .now()
            .checked_sub(silent_for)
            .expect("a registry built with `registry_with_headroom` has room");
        reg.lock()
            .insert(key.to_string(), entry_at(key, last_active));
    }

    #[test]
    fn credit_outage_keeps_a_window_the_outage_alone_would_have_expired() {
        let reg = registry_with_headroom();
        // 40 s silent is past the 30 s TTL, but 25 s of it was the daemon's own
        // outage, so it has really been silent for 15 s.
        insert_silent(&reg, "w", Duration::from_secs(40));
        reg.credit_outage(Duration::from_secs(25));
        assert_eq!(reg.list().len(), 1, "an outage must not age a window out");
    }

    #[test]
    fn without_a_credit_the_same_window_is_reaped() {
        let reg = registry_with_headroom();
        insert_silent(&reg, "w", Duration::from_secs(40));
        assert!(reg.list().is_empty(), "the control for the test above");
    }

    #[test]
    fn credit_outage_does_not_resurrect_a_window_that_was_already_gone() {
        let reg = registry_with_headroom();
        insert_silent(&reg, "dead", Duration::from_secs(120));
        // Reaped before the credit lands (any read does this).
        assert!(reg.list().is_empty());
        reg.credit_outage(Duration::from_secs(300));
        assert!(reg.list().is_empty());
    }

    #[test]
    fn credit_outage_never_moves_a_stamp_into_the_future() {
        let reg = registry_with_headroom();
        insert_silent(&reg, "w", Duration::from_secs(1));
        reg.credit_outage(Duration::from_secs(3600));
        assert!(reg.lock()["w"].last_active <= reg.clock.now());
        assert!(reg.lock()["w"].last_seen <= Utc::now());
        // Even a duration neither clock can add is saturated, not a panic and
        // not skipped: the awake stamp still moves up to now.
        insert_silent(&reg, "x", Duration::from_secs(20));
        reg.credit_outage(Duration::MAX);
        assert!(reg.lock()["x"].last_active <= reg.clock.now());
        assert!(
            reg.lock()["x"].last_active > reg.clock.now().saturating_sub(Duration::from_secs(1))
        );
        assert!(reg.lock()["x"].last_seen <= Utc::now());
    }

    #[test]
    fn credit_outage_advances_the_display_stamp_with_the_liveness_stamp() {
        let reg = registry_with_headroom();
        insert_silent(&reg, "w", Duration::from_secs(40));
        let active = reg.lock()["w"].last_active;
        reg.lock().get_mut("w").unwrap().last_seen = Utc::now() - ChronoDuration::seconds(40);

        reg.credit_outage(Duration::from_secs(25));

        let guard = reg.lock();
        assert_eq!(guard["w"].last_active, active + Duration::from_secs(25));
        let shown_age = (Utc::now() - guard["w"].last_seen).num_seconds();
        assert!(
            (14..=16).contains(&shown_age),
            "40 s silent less the 25 s credit reads ~15 s, got {shown_age}"
        );
    }

    #[test]
    fn credit_outage_is_not_a_visible_change() {
        let reg = registry_with_headroom();
        insert_silent(&reg, "w", Duration::from_secs(10));
        let rx = reg.subscribe_changes();
        reg.credit_outage(Duration::from_secs(5));
        assert!(
            !rx.has_changed().unwrap(),
            "crediting must not wake every subscriber"
        );
    }

    /// A minimal entry for reap/cap/eviction tests; only `key` and `last_active`
    /// participate in liveness and eviction order.
    fn entry_at(key: &str, last_active: Duration) -> WindowEntry {
        WindowEntry {
            key: key.to_string(),
            folders: vec![],
            repo: None,
            title: None,
            pid: None,
            last_seen: Utc::now(),
            last_active,
        }
    }

    #[test]
    fn evict_oldest_removes_oldest_with_key_tiebreak() {
        let now = Duration::from_secs(1000);
        let mut windows = HashMap::new();
        windows.insert("young".to_string(), entry_at("young", now));
        windows.insert(
            "old-b".to_string(),
            entry_at("old-b", now.saturating_sub(Duration::from_secs(10))),
        );
        windows.insert(
            "old-a".to_string(),
            entry_at("old-a", now.saturating_sub(Duration::from_secs(10))),
        );
        // Oldest `last_active` is shared by two entries; the lowest key loses.
        evict_oldest(&mut windows);
        assert!(!windows.contains_key("old-a"));
        assert!(windows.contains_key("old-b"));
        assert!(windows.contains_key("young"));
        // Empty map is a no-op rather than a panic.
        let mut empty: HashMap<String, WindowEntry> = HashMap::new();
        evict_oldest(&mut empty);
        assert!(empty.is_empty());
    }

    #[test]
    fn register_at_cap_evicts_only_the_oldest() {
        let reg = WorktreesRegistry::new();
        // Seed a full registry directly (registering 256 times would work too,
        // but sub-millisecond stamps may tie; explicit stamps make the
        // highest-numbered key unambiguously the oldest).
        {
            let mut windows = reg.lock();
            let base = Duration::from_secs(1000);
            for i in 0..MAX_WINDOWS {
                let key = format!("w{i:03}");
                windows.insert(
                    key.clone(),
                    entry_at(&key, base.saturating_sub(Duration::from_millis(i as u64))),
                );
            }
        }
        // A new key at the cap displaces exactly the longest-silent entry.
        reg.register(register_request("fresh", None, "/tmp/f"));
        let windows = reg.lock();
        assert_eq!(windows.len(), MAX_WINDOWS);
        assert!(windows.contains_key("fresh"));
        assert!(!windows.contains_key(&format!("w{:03}", MAX_WINDOWS - 1)));
        assert!(windows.contains_key("w000"));
    }

    #[test]
    fn register_upsert_at_cap_does_not_evict() {
        let reg = WorktreesRegistry::new();
        {
            let mut windows = reg.lock();
            let base = Duration::from_secs(1000);
            for i in 0..MAX_WINDOWS {
                let key = format!("w{i:03}");
                windows.insert(
                    key.clone(),
                    entry_at(&key, base.saturating_sub(Duration::from_millis(i as u64))),
                );
            }
        }
        // Re-registering an existing key is an upsert: nothing is displaced,
        // not even the oldest entry.
        let oldest = format!("w{:03}", MAX_WINDOWS - 1);
        reg.register(register_request(&oldest, Some("r"), "/tmp/a"));
        let windows = reg.lock();
        assert_eq!(windows.len(), MAX_WINDOWS);
        assert!(windows.contains_key(&oldest));
        assert!(windows.contains_key("w000"));
    }

    #[test]
    fn sorted_entries_orders_by_repo_then_key() {
        let mut windows = HashMap::new();
        for (key, repo) in [("z", "repo-a"), ("a", "repo-b"), ("m", "repo-a")] {
            windows.insert(
                key.to_string(),
                WindowEntry {
                    repo: Some(repo.to_string()),
                    ..entry_at(key, Duration::ZERO)
                },
            );
        }
        let entries = sorted_entries(&windows);
        let ordered: Vec<(&str, &str)> = entries
            .iter()
            .map(|e| (e.key.as_str(), e.repo.as_deref().unwrap()))
            .collect();
        assert_eq!(
            ordered,
            vec![("m", "repo-a"), ("z", "repo-a"), ("a", "repo-b")]
        );
    }

    #[test]
    fn default_constructs_an_empty_registry() {
        let reg = WorktreesRegistry::default();
        assert!(reg.lock().is_empty());
    }

    // --- Change-notify for the push subscription (#1267) --------------------

    #[test]
    fn subscribe_changes_starts_seen_and_register_bumps() {
        let reg = WorktreesRegistry::new();
        let mut rx = reg.subscribe_changes();
        // A fresh receiver has the current version already marked seen.
        assert!(!rx.has_changed().unwrap());
        // A register changes the visible set → the receiver observes a new value.
        reg.register(register_request("w1", None, "/tmp/a"));
        assert!(rx.has_changed().unwrap(), "register should bump");
        // Marking it seen clears the pending change.
        rx.borrow_and_update();
        assert!(!rx.has_changed().unwrap());
    }

    #[test]
    fn unregister_bumps_only_when_it_removes() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        // Subscribe *after* the register so its bump is already seen.
        let rx = reg.subscribe_changes();
        // Removing a missing key changes nothing (and reaps nothing) → no bump.
        assert!(!reg.unregister("ghost"));
        assert!(
            !rx.has_changed().unwrap(),
            "a no-op unregister must not bump"
        );
        // Removing a present key bumps.
        assert!(reg.unregister("w1"));
        assert!(
            rx.has_changed().unwrap(),
            "a removing unregister should bump"
        );
    }

    #[test]
    fn change_generation_advances_only_on_a_visible_change() {
        let reg = WorktreesRegistry::new();
        let g0 = reg.change_generation();
        // A no-op (heartbeat of an unknown key) leaves the generation untouched.
        assert!(!reg.heartbeat("ghost"));
        assert_eq!(
            reg.change_generation(),
            g0,
            "a no-op must not advance the generation"
        );
        // A register changes the visible set → the generation advances, so a
        // cache keyed on it rebuilds.
        reg.register(register_request("w1", None, "/tmp/a"));
        assert_ne!(
            reg.change_generation(),
            g0,
            "a register should advance the generation"
        );
    }

    #[test]
    fn heartbeat_bumps_only_when_it_reaps() {
        let reg = registry_with_headroom();
        reg.register(register_request("w1", None, "/tmp/a"));
        let rx = reg.subscribe_changes();
        // A plain heartbeat refreshes liveness but changes no visible state.
        assert!(reg.heartbeat("w1"));
        assert!(!rx.has_changed().unwrap(), "a pure heartbeat must not bump");
        // Seed a stale sibling directly; a heartbeat that reaps it *does* bump.
        insert_silent(&reg, "stale", Duration::from_secs(120));
        assert!(reg.heartbeat("w1"));
        assert!(
            rx.has_changed().unwrap(),
            "a heartbeat that reaps a stale sibling should bump"
        );
    }

    // --- Sleep-safe TTL (#2126) --------------------------------------------

    /// A hair under the window TTL, still inside it.
    fn just_inside_the_ttl() -> Duration {
        DEFAULT_TTL.checked_sub(Duration::from_secs(1)).unwrap()
    }

    /// Moves every wall-clock stamp back by `by`, as a laptop sleeping that long
    /// would, without any awake time passing.
    fn jump_wall_clock_forward(reg: &WorktreesRegistry, by: ChronoDuration) {
        for entry in reg.lock().values_mut() {
            entry.last_seen -= by;
        }
    }

    #[test]
    fn a_wall_clock_jump_with_no_awake_time_reaps_nothing() {
        // The #2108 regression, for windows: the first read after a sleep longer
        // than the TTL used to reap every window, emptying the tree and the tray
        // until each one re-registered off its next heartbeat.
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", Some("repo-a"), "/tmp/a"));
        reg.register(register_request("w2", Some("repo-b"), "/tmp/b"));
        reg.register(register_request("w3", None, "/tmp/c"));

        jump_wall_clock_forward(&reg, ChronoDuration::days(3));

        // Every reading path, and the mutating one that reaps first.
        assert_eq!(reg.list().len(), 3, "list");
        assert_eq!(reg.open_folders().len(), 3, "open_folders");
        assert!(
            reg.heartbeat("w1"),
            "a heartbeat after the wake-up is known"
        );
        assert!(reg.unregister("w3"));
        assert_eq!(reg.list().len(), 2, "unregister reaped nothing else");
        // Display still reads the wall clock: the stale `last_seen` is shown.
        assert!(reg.list()[1].last_seen < Utc::now() - ChronoDuration::days(2));
    }

    #[test]
    fn a_wall_clock_jump_does_not_bump_the_change_notify() {
        // Nothing was reaped, so nothing visible changed.
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        let rx = reg.subscribe_changes();
        jump_wall_clock_forward(&reg, ChronoDuration::hours(8));
        assert!(reg.heartbeat("w1"));
        assert!(!rx.has_changed().unwrap());
    }

    #[test]
    fn awake_time_past_the_ttl_still_reaps() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.clock.advance(just_inside_the_ttl());
        assert_eq!(reg.list().len(), 1, "inside the TTL it survives");
        reg.clock.advance(Duration::from_secs(2));
        assert!(
            reg.list().is_empty(),
            "past the TTL of awake silence it reaps"
        );
        // And the reaped window learns it must re-register.
        assert!(!reg.heartbeat("w1"));
    }

    #[test]
    fn a_heartbeat_restarts_the_awake_ttl() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.clock.advance(just_inside_the_ttl());
        assert!(reg.heartbeat("w1"));
        reg.clock.advance(just_inside_the_ttl());
        assert_eq!(reg.list().len(), 1, "the heartbeat restarted the clock");
    }

    #[test]
    fn activity_after_a_sleep_restarts_the_awake_ttl() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.clock.advance(just_inside_the_ttl());
        jump_wall_clock_forward(&reg, ChronoDuration::hours(8));
        // A heartbeat after the wake-up refreshes the awake stamp as well.
        assert!(reg.heartbeat("w1"));
        reg.clock.advance(just_inside_the_ttl());
        assert_eq!(reg.list().len(), 1);
    }

    #[test]
    fn re_registering_restarts_the_awake_ttl() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.clock.advance(just_inside_the_ttl());
        reg.register(register_request("w1", Some("repo"), "/tmp/a"));
        reg.clock.advance(just_inside_the_ttl());
        assert_eq!(reg.list().len(), 1);
    }

    #[test]
    fn an_outage_credit_and_a_sleep_compose() {
        // 35 s of awake time pass across a sleep (which adds none of its own),
        // and the daemon could not accept for 15 s of it. The window has really
        // been silent for 20 s, inside the TTL, so it lives; the control shows
        // the same 35 s with no credit is past it.
        let control = WorktreesRegistry::new();
        control.register(register_request("w", None, "/tmp/a"));
        control.clock.advance(Duration::from_secs(35));
        jump_wall_clock_forward(&control, ChronoDuration::hours(8));
        assert!(control.list().is_empty(), "the control for the test");

        let reg = WorktreesRegistry::new();
        reg.register(register_request("w", None, "/tmp/a"));
        reg.clock.advance(Duration::from_secs(35));
        jump_wall_clock_forward(&reg, ChronoDuration::hours(8));
        reg.credit_outage(Duration::from_secs(15));
        assert_eq!(reg.list().len(), 1);
    }

    #[test]
    fn eviction_at_the_cap_goes_by_awake_silence_not_the_wall_clock() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("quiet", None, "/tmp/q"));
        reg.clock.advance(Duration::from_secs(5));
        for i in 0..MAX_WINDOWS - 1 {
            reg.register(register_request(&format!("w{i:03}"), None, "/tmp/w"));
        }
        // Make the *newest* registrations look oldest by the wall clock; only the
        // awake stamp may decide who is the longest-silent.
        for entry in reg.lock().values_mut() {
            if entry.key != "quiet" {
                entry.last_seen -= ChronoDuration::days(1);
            }
        }
        reg.register(register_request("fresh", None, "/tmp/f"));
        let windows = reg.lock();
        assert_eq!(windows.len(), MAX_WINDOWS);
        assert!(!windows.contains_key("quiet"), "longest awake silence goes");
        assert!(windows.contains_key("fresh"));
    }

    #[test]
    fn a_pending_directive_survives_a_sleep_and_is_delivered() {
        // Neither directive set is touched by a reap, and a sleep no longer
        // reaps: the window is still known on its next heartbeat, so a close or
        // reload signalled just before the lid shut is still taken, exactly once.
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.mark_close_pending("w1");
        reg.mark_reload_pending("w1");

        jump_wall_clock_forward(&reg, ChronoDuration::hours(8));

        assert!(reg.heartbeat("w1"), "the window is known after the wake-up");
        assert!(reg.take_close_pending("w1"));
        assert!(reg.take_reload_pending("w1"));
        assert!(!reg.take_close_pending("w1"), "fires exactly once");
    }

    // --- Close-pending directive (#1277) -----------------------------------

    #[test]
    fn close_pending_is_taken_once_then_cleared() {
        let reg = WorktreesRegistry::new();
        // No directive by default.
        assert!(!reg.take_close_pending("w1"));
        // Marked → the first take observes it, the next does not (fires once).
        reg.mark_close_pending("w1");
        assert!(reg.take_close_pending("w1"));
        assert!(!reg.take_close_pending("w1"));
    }

    #[test]
    fn unregister_clears_a_pending_close_directive() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.mark_close_pending("w1");
        // Unregistering the window drops any pending directive with it.
        assert!(reg.unregister("w1"));
        assert!(!reg.take_close_pending("w1"));
    }

    #[test]
    fn cancel_close_pending_withdraws_an_undelivered_directive() {
        let reg = WorktreesRegistry::new();
        reg.mark_close_pending("w1");
        assert!(reg.cancel_close_pending("w1"), "it was still pending");
        assert!(
            !reg.take_close_pending("w1"),
            "a withdrawn directive is never delivered"
        );
        assert!(!reg.cancel_close_pending("w1"), "nothing left to withdraw");
    }

    #[test]
    fn cancel_close_pending_cannot_recall_a_directive_the_window_took() {
        let reg = WorktreesRegistry::new();
        reg.mark_close_pending("w1");
        // The window heartbeated first, so it already holds `close: true`.
        assert!(reg.take_close_pending("w1"));
        assert!(!reg.cancel_close_pending("w1"), "already delivered");
    }

    #[test]
    fn cancel_close_pending_touches_only_its_own_key_and_set() {
        let reg = WorktreesRegistry::new();
        reg.mark_close_pending("w1");
        reg.mark_close_pending("w2");
        reg.mark_reload_pending("w1");

        assert!(reg.cancel_close_pending("w1"));

        assert!(reg.take_close_pending("w2"), "another key is untouched");
        assert!(reg.take_reload_pending("w1"), "the reload set is untouched");
    }

    #[test]
    fn a_reap_leaves_pending_directives_for_a_window_that_returns() {
        // Pins the accepted half of #2139: a reap clears neither directive, so a
        // window that goes quiet past the TTL and comes back is handed what it
        // was signalled — on the heartbeat that answers `known: false`.
        let reg = registry_with_headroom();
        insert_silent(&reg, "w", Duration::from_secs(40));
        reg.mark_close_pending("w");
        reg.mark_reload_pending("w");

        assert!(reg.list().is_empty(), "the window was reaped");
        assert!(!reg.heartbeat("w"), "it returns unknown");

        assert!(reg.take_close_pending("w"));
        assert!(reg.take_reload_pending("w"));
    }

    // --- Reload-pending directive (#1417) ----------------------------------

    #[test]
    fn reload_pending_is_taken_once_then_cleared() {
        let reg = WorktreesRegistry::new();
        // No directive by default.
        assert!(!reg.take_reload_pending("w1"));
        // Marked → the first take observes it, the next does not (fires once).
        reg.mark_reload_pending("w1");
        assert!(reg.take_reload_pending("w1"));
        assert!(!reg.take_reload_pending("w1"));
    }

    #[test]
    fn unregister_clears_a_pending_reload_directive() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        reg.mark_reload_pending("w1");
        // Unregistering the window drops any pending directive with it.
        assert!(reg.unregister("w1"));
        assert!(!reg.take_reload_pending("w1"));
    }

    #[test]
    fn close_and_reload_directives_are_independent() {
        let reg = WorktreesRegistry::new();
        // Separate sets: marking one must not set or consume the other, so the
        // heartbeat can surface both fields and the companion pick a winner.
        reg.mark_reload_pending("w1");
        assert!(!reg.take_close_pending("w1"));
        reg.mark_close_pending("w1");
        assert!(reg.take_close_pending("w1"));
        assert!(reg.take_reload_pending("w1"));
        // Each key is tracked on its own.
        reg.mark_reload_pending("w1");
        assert!(!reg.take_reload_pending("w2"));
        assert!(reg.take_reload_pending("w1"));
    }

    #[test]
    fn marking_a_reload_does_not_bump() {
        let reg = WorktreesRegistry::new();
        reg.register(register_request("w1", None, "/tmp/a"));
        let rx = reg.subscribe_changes();
        // A directive is not consumer-visible state — it never reaches a tree
        // snapshot — so marking one must not push a redundant frame to every
        // subscriber. Same rule as the close directive.
        reg.mark_reload_pending("w1");
        assert!(!rx.has_changed().unwrap(), "marking a reload must not bump");
        assert!(reg.take_reload_pending("w1"));
        assert!(!rx.has_changed().unwrap(), "taking a reload must not bump");
    }

    // --- Rebasing set (#1415) ----------------------------------------------

    #[test]
    fn rebasing_marks_and_clears_the_named_paths() {
        let reg = WorktreesRegistry::new();
        let a = PathBuf::from("/tmp/a");
        let b = PathBuf::from("/tmp/b");
        assert!(
            reg.rebasing_paths().is_empty(),
            "nothing rebases by default"
        );

        assert!(reg.mark_rebasing(&[a.clone(), b.clone()]));
        assert_eq!(reg.rebasing_paths(), [a.clone(), b.clone()].into());

        // Clearing one leaves the other — a batch reports per worktree.
        assert!(reg.clear_rebasing(std::slice::from_ref(&a)));
        assert_eq!(reg.rebasing_paths(), [b.clone()].into());
        assert!(reg.clear_rebasing(&[b]));
        assert!(reg.rebasing_paths().is_empty());
    }

    #[test]
    fn rebasing_bumps_only_on_a_real_change() {
        // Load-bearing: an unconditional bump would defeat the server's snapshot
        // diff and re-push a `tree` frame to every window on every tick.
        let reg = WorktreesRegistry::new();
        let path = PathBuf::from("/tmp/a");

        let mut rx = reg.subscribe_changes();
        assert!(reg.mark_rebasing(std::slice::from_ref(&path)));
        assert!(rx.has_changed().unwrap(), "the first mark bumps");
        let _ = rx.borrow_and_update();

        assert!(
            !reg.mark_rebasing(std::slice::from_ref(&path)),
            "re-marking an already-rebasing path is not a change"
        );
        assert!(!rx.has_changed().unwrap(), "a no-op mark must not bump");

        assert!(
            !reg.clear_rebasing(&[PathBuf::from("/tmp/never-marked")]),
            "clearing an unmarked path is not a change"
        );
        assert!(!rx.has_changed().unwrap(), "a no-op clear must not bump");

        assert!(reg.clear_rebasing(&[path]));
        assert!(rx.has_changed().unwrap(), "a real clear bumps");
    }

    // --- Pushing set (#1443) -----------------------------------------------

    #[test]
    fn pushing_marks_and_clears_the_named_paths() {
        let reg = WorktreesRegistry::new();
        let a = PathBuf::from("/tmp/a");
        let b = PathBuf::from("/tmp/b");
        assert!(reg.pushing_paths().is_empty(), "nothing pushes by default");

        assert!(reg.mark_pushing(&[a.clone(), b.clone()]));
        assert_eq!(reg.pushing_paths(), [a.clone(), b.clone()].into());

        assert!(reg.clear_pushing(std::slice::from_ref(&a)));
        assert_eq!(reg.pushing_paths(), [b.clone()].into());
        assert!(reg.clear_pushing(&[b]));
        assert!(reg.pushing_paths().is_empty());
    }

    #[test]
    fn pushing_bumps_only_on_a_real_change() {
        let reg = WorktreesRegistry::new();
        let path = PathBuf::from("/tmp/a");

        let mut rx = reg.subscribe_changes();
        assert!(reg.mark_pushing(std::slice::from_ref(&path)));
        assert!(rx.has_changed().unwrap(), "the first mark bumps");
        let _ = rx.borrow_and_update();

        assert!(
            !reg.mark_pushing(std::slice::from_ref(&path)),
            "re-marking an already-pushing path is not a change"
        );
        assert!(!rx.has_changed().unwrap(), "a no-op mark must not bump");

        assert!(reg.clear_pushing(&[path]));
        assert!(rx.has_changed().unwrap(), "a real clear bumps");
    }

    #[test]
    fn pushing_and_rebasing_are_independent_sets() {
        // The two cues are orthogonal — the same worktree can be marked by one op
        // without the other's cue appearing, and clearing one must not clear the
        // other.
        let reg = WorktreesRegistry::new();
        let path = PathBuf::from("/tmp/a");

        assert!(reg.mark_rebasing(std::slice::from_ref(&path)));
        assert!(
            reg.pushing_paths().is_empty(),
            "a rebase mark must not show as a push"
        );

        assert!(reg.mark_pushing(std::slice::from_ref(&path)));
        assert!(reg.clear_rebasing(std::slice::from_ref(&path)));
        assert_eq!(
            reg.pushing_paths(),
            [path].into(),
            "clearing the rebase mark must leave the push mark alone"
        );
    }

    // --- Show/hide-closed toggle (#1301) -----------------------------------

    #[test]
    fn show_closed_defaults_to_true() {
        let reg = WorktreesRegistry::new();
        assert!(reg.show_closed(), "default is show all");
    }

    #[test]
    fn set_show_closed_reports_change_and_is_idempotent() {
        let reg = WorktreesRegistry::new();
        // Flipping to a new value reports a change and is observable.
        assert!(reg.set_show_closed(false));
        assert!(!reg.show_closed());
        // Setting the same value again is a no-op (no change reported).
        assert!(!reg.set_show_closed(false));
        // Flipping back reports a change again.
        assert!(reg.set_show_closed(true));
        assert!(reg.show_closed());
    }

    #[test]
    fn set_show_closed_bumps_only_on_change() {
        let reg = WorktreesRegistry::new();
        let rx = reg.subscribe_changes();
        // A no-op set (already the default) does not wake subscribers.
        assert!(!reg.set_show_closed(true));
        assert!(
            !rx.has_changed().unwrap(),
            "a no-op toggle must not bump the change-notify"
        );
        // A real flip bumps so subscribers re-push a snapshot with the new value.
        assert!(reg.set_show_closed(false));
        assert!(
            rx.has_changed().unwrap(),
            "flipping the toggle should bump the change-notify"
        );
    }

    #[test]
    fn polling_defaults_off_for_an_untoggled_repo() {
        let reg = WorktreesRegistry::new();
        // #1376: the whole point — a repo the user has never enabled is not polled.
        assert!(!reg.is_polling_enabled("rust-works", "omni-dev"));
        assert!(reg.enabled_polling_repos().is_empty());
        assert!(reg.polling_snapshot().is_empty());
    }

    #[test]
    fn set_polling_leases_and_disables() {
        let reg = WorktreesRegistry::new();
        // Enabling a fresh repo reports a change and leases it live.
        assert!(reg.set_polling("rust-works", "omni-dev", true));
        assert!(reg.is_polling_enabled("rust-works", "omni-dev"));
        // Re-enabling refreshes the lease — the map changes (new expiry).
        assert!(reg.set_polling("rust-works", "omni-dev", true));
        assert!(reg.is_polling_enabled("rust-works", "omni-dev"));
        // Disabling reports a change and clears it.
        assert!(reg.set_polling("rust-works", "omni-dev", false));
        assert!(!reg.is_polling_enabled("rust-works", "omni-dev"));
        // Disabling an already-disabled repo is a no-op.
        assert!(!reg.set_polling("rust-works", "omni-dev", false));
    }

    #[test]
    fn polling_lease_auto_expires() {
        // The 15-minute lease (#1376): an enabled repo drops out once its lease
        // elapses, reaped on the next read — no background timer.
        let reg = WorktreesRegistry::new();
        reg.set_polling("rust-works", "omni-dev", true);
        assert!(reg.is_polling_enabled("rust-works", "omni-dev"));
        // Force the lease into the past (as if 15 min elapsed).
        reg.set_polling_expiry(
            "rust-works",
            "omni-dev",
            Utc::now() - ChronoDuration::seconds(1),
        );
        assert!(
            !reg.is_polling_enabled("rust-works", "omni-dev"),
            "an expired lease reads as disabled"
        );
        // The read paths reap it, so it is gone from the snapshot entirely.
        assert!(reg.enabled_polling_repos().is_empty());
        assert!(reg.polling_snapshot().is_empty());
        // A tiny real TTL expires on its own after a short sleep.
        let short = WorktreesRegistry::with_poll_ttl(Duration::from_millis(30));
        short.set_polling("o", "n", true);
        assert!(short.is_polling_enabled("o", "n"));
        std::thread::sleep(Duration::from_millis(60));
        assert!(!short.is_polling_enabled("o", "n"));
    }

    #[test]
    fn set_polling_bumps_only_on_an_effective_flip() {
        let reg = WorktreesRegistry::new();
        let rx = reg.subscribe_changes();
        // A no-op (disabling an already-off repo) does not wake subscribers.
        assert!(!reg.set_polling("o", "n", false));
        assert!(
            !rx.has_changed().unwrap(),
            "a no-op poll toggle must not bump the change-notify"
        );
        // A real enable flips off→on and bumps so every window recolors.
        assert!(reg.set_polling("o", "n", true));
        assert!(
            rx.has_changed().unwrap(),
            "enabling a repo should bump the change-notify"
        );
    }

    #[test]
    fn refreshing_a_live_lease_does_not_bump() {
        // A lease refresh (already on, still on) persists a new expiry but does
        // not change the visible state, so it must not wake every window.
        let reg = WorktreesRegistry::new();
        reg.set_polling("o", "n", true);
        let rx = reg.subscribe_changes();
        assert!(
            reg.set_polling("o", "n", true),
            "re-enabling refreshes the lease (map changed → persist)"
        );
        assert!(
            !rx.has_changed().unwrap(),
            "refreshing a live lease must not bump — the visible state is unchanged"
        );
    }

    #[test]
    fn seed_polling_loads_leases_and_drops_expired() {
        let reg = WorktreesRegistry::new();
        reg.set_polling("a", "z", true);
        let future = Utc::now() + ChronoDuration::minutes(10);
        let past = Utc::now() - ChronoDuration::minutes(1);
        // Seeding (the startup load) replaces wholesale, dropping the prior entry
        // and any already-expired lease from the file.
        reg.seed_polling([
            ("rust-works/omni-dev".to_string(), future),
            ("acme/widgets".to_string(), future),
            ("stale/repo".to_string(), past),
        ]);
        assert!(!reg.is_polling_enabled("a", "z"));
        assert!(reg.is_polling_enabled("rust-works", "omni-dev"));
        assert!(
            !reg.is_polling_enabled("stale", "repo"),
            "expired lease dropped"
        );
        // The persisted form is deterministic (sorted) and carries the expiries.
        let snap = reg.polling_snapshot();
        assert_eq!(
            snap.iter().map(|(k, _)| k.clone()).collect::<Vec<_>>(),
            vec![
                "acme/widgets".to_string(),
                "rust-works/omni-dev".to_string()
            ]
        );
        assert!(snap.iter().all(|(_, expiry)| *expiry == future));
    }
}