forge_workspace/

gotify.rs

//! The Gotify seam on [`Workspace`]: the forge.toml `[gotify]` config
//! accessor, subscription CRUD (durable rows in the redb store,
//! ephemeral ones in memory), the subsystem start/stop lifecycle, and
//! the [`GotifyHost`] port impl the forge-connectors pump drives.
//!
//! Everything here stays on `Workspace` as a second `impl` block, so
//! every caller (the boot path in [`crate::workspace`], the
//! `mcp::gotify` facade, `spawn::deliver_gotify_message`, the
//! Inspector GOTIFY snapshot) keeps its path. The `gotify_*` fields
//! these methods own are `pub(crate)` for the same reason `db` is: so
//! this sibling module can reach them without a wrapper. Stream,
//! matching and the pump live in `forge_connectors::gotify`; store IO
//! in [`crate::store::gotify`]; the MCP tool surface in
//! [`crate::mcp::gotify`]; delivery in [`crate::spawn`].

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use forge_connectors::gotify::GotifyHost;
use forge_primitives::GotifySubscription;

use crate::protocol::{Command, SessionUpdate};
use crate::target::ProjectKey;
use crate::workspace::Workspace;

impl Workspace {
    /// The `[gotify]` server connection from forge.toml, or `None`
    /// when the section is absent. `None` keeps the Gotify subsystem
    /// dormant and makes `gotify__subscribe` error.
    pub fn gotify_config(&self) -> Option<forge_primitives::GotifyConfig> {
        self.config_arc().gotify.clone()
    }

    /// Register a Gotify subscription in the active set. Durable ones
    /// (lead / team-worker) also persist to the redb store; ephemeral
    /// ad-hoc-worker ones stay in memory only and drop on restart.
    pub(crate) fn add_gotify_subscription(
        &self,
        sub: forge_primitives::GotifySubscription,
        durable: bool,
    ) {
        let project = sub.project.clone();
        if durable
            && let Some(db) = self.db.lock().as_ref()
            && let Err(error) = crate::store::gotify::insert(db, &sub)
        {
            tracing::warn!(
                target: "forge_workspace::gotify",
                %error,
                "persisting a Gotify subscription failed",
            );
        }
        self.gotify_subs.lock().push(sub);
        self.announce_connector_subscriptions_changed(&project);
    }

    /// Remove the subscription `id` in `project` only when its owner
    /// matches `owner` (`None` = a lead subscription, `Some(label)` =
    /// that worker's), from both the active set and (when present) the
    /// redb store. Returns the row as it stood (`None` when nothing
    /// matched). Backs the owner-scoped `gotify__unsubscribe` so a caller
    /// removes only what it subscribed, mirroring
    /// [`Self::remove_cron_owned_by`]. Worker teardown uses
    /// [`Self::remove_gotify_subscriptions_for_worker`] instead and is
    /// deliberately not owner-gated.
    pub(crate) fn remove_gotify_subscription_owned_by(
        &self,
        project: &str,
        id: uuid::Uuid,
        owner: Option<&str>,
    ) -> Option<GotifySubscription> {
        let removed = {
            let mut subs = self.gotify_subs.lock();
            let pos = subs.iter().position(|s| {
                s.id == id && s.project == project && s.team_role.as_deref() == owner
            });
            pos.map(|pos| subs.remove(pos))
        };
        if removed.is_some()
            && let Some(db) = self.db.lock().as_ref()
            && let Err(error) = crate::store::gotify::remove(db, id)
        {
            tracing::warn!(
                target: "forge_workspace::gotify",
                %error,
                "removing a persisted Gotify subscription failed",
            );
        }
        if removed.is_some() {
            self.announce_connector_subscriptions_changed(project);
        }
        removed
    }

    /// Remove the worker `label`'s Gotify subscriptions in `project_key`
    /// from both the active set and the redb store, scoped to
    /// `(project name, team_role == Some(label))`. Backs
    /// `spawn::teardown_worker` so a despawned durable dynamic worker can't
    /// orphan a persisted sub. Subscriptions are project-scoped by NAME, so
    /// the key is resolved to a name via the same view lookup
    /// `resolve_identity` uses when subscribing.
    pub(crate) fn remove_gotify_subscriptions_for_worker(
        &self,
        project_key: &ProjectKey,
        label: &str,
    ) {
        let Some(project_name) =
            self.list_projects().into_iter().find(|v| v.key == *project_key).map(|v| v.name)
        else {
            tracing::warn!(
                target: "forge_workspace::gotify",
                project = %project_key.as_str(),
                label = %label,
                "could not resolve a project name at worker teardown; its Gotify subs may be stranded",
            );
            return;
        };
        let removed_ids: Vec<uuid::Uuid> = {
            let mut subs = self.gotify_subs.lock();
            let mut removed = Vec::new();
            subs.retain(|s| {
                let owned = s.project == project_name && s.team_role.as_deref() == Some(label);
                if owned {
                    removed.push(s.id);
                }
                !owned
            });
            removed
        };
        if removed_ids.is_empty() {
            return;
        }
        if let Some(db) = self.db.lock().as_ref() {
            for id in removed_ids {
                if let Err(error) = crate::store::gotify::remove(db, id) {
                    // Same zombie-durable-state class as a stranded
                    // dynamic-worker row: match its error! severity.
                    tracing::error!(
                        target: "forge_workspace::gotify",
                        %error,
                        id = %id,
                        project = %project_name,
                        label = %label,
                        "deleting a persisted Gotify subscription failed; it reloads into the active set on restart and a future same-label worker inherits it",
                    );
                }
            }
        }
        self.announce_connector_subscriptions_changed(&project_name);
    }

    /// The active subscriptions for `project`, whatever their owner. The
    /// scoping consumers narrow it - `gotify__list` to the caller's own
    /// subscriptions, the Inspector GOTIFY snapshot to the active tab's
    /// stamped project name - while the refused-unsubscribe determination
    /// reads it whole, because telling "not in this project" from "not
    /// yours" needs the other owners' rows.
    pub fn gotify_subscriptions_for_project(
        &self,
        project: &str,
    ) -> Vec<forge_primitives::GotifySubscription> {
        self.gotify_subs.lock().iter().filter(|s| s.project == project).cloned().collect()
    }

    /// Tell every view the project's connector subscriptions moved, as the
    /// two sets the core holds, on the project's lead seat. One update
    /// carries both connectors: they are one section of the board's project
    /// row, and a reader that patches it needs the pair. A name no project
    /// carries has no seat to route on and announces nothing.
    ///
    /// Lives here rather than on either connector: the section is the pair,
    /// so both seams announce through the one site.
    pub(crate) fn announce_connector_subscriptions_changed(&self, project_name: &str) {
        let Some(key) = self.lead_slot_for_project(project_name) else {
            tracing::debug!(
                target: "forge_workspace::gotify",
                event_name = "connector_subscriptions_changed_unroutable",
                project = %project_name,
                "a subscription write named a project no project carries; nothing was announced",
            );
            return;
        };
        let gotify = self.gotify_subscriptions_for_project(project_name);
        let slack = self.slack_subscriptions_for_project(project_name);
        let _ = self.update_tx.send(SessionUpdate::ConnectorSubscriptionsChanged {
            key,
            gotify,
            slack,
        });
    }

    /// Whether the Gotify stream is currently connected. Backs the
    /// Inspector GOTIFY status line.
    pub fn gotify_connected(&self) -> bool {
        *self.gotify_connected.lock()
    }

    /// Start the Gotify subsystem when it's configured, has at least one
    /// active subscription, and isn't already running. Idempotent. Spawns
    /// the forge-connectors pump, which runs the reconnecting stream task
    /// and translates its events into the `gotify_*` state below and
    /// `GotifyHost::deliver` calls. Called at boot and after a subscribe
    /// grows the active set.
    ///
    /// The pump's exit guard clears the handle when the task ends, so a
    /// pump that dies no longer wedges the subsystem: the next call here
    /// starts a new one.
    pub fn start_gotify_subsystem(self: &Arc<Self>) {
        let Some(cfg) = self.gotify_config() else {
            tracing::debug!(
                target: "forge_workspace::gotify",
                event_name = "gotify_subsystem_declined",
                reason = "no_config",
                "the Gotify subsystem did not start",
            );
            return;
        };
        let subscriptions = self.gotify_subs.lock().len();
        if subscriptions == 0 {
            tracing::debug!(
                target: "forge_workspace::gotify",
                event_name = "gotify_subsystem_declined",
                reason = "no_subscriptions",
                "the Gotify subsystem did not start",
            );
            return;
        }
        let mut guard = self.gotify_subsystem.lock();
        if guard.is_some() {
            drop(guard);
            tracing::debug!(
                target: "forge_workspace::gotify",
                event_name = "gotify_subsystem_declined",
                reason = "already_running",
                "the Gotify subsystem did not start",
            );
            return;
        }
        let token = Arc::new(());
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel();
        *guard = Some(GotifySubsystem { token: Arc::clone(&token), shutdown: shutdown_tx });
        drop(guard);
        tracing::info!(
            target: "forge_workspace::gotify",
            event_name = "gotify_subsystem_started",
            host = url_host(&cfg.url),
            subscriptions,
            "the Gotify subsystem started",
        );
        let host: Arc<dyn GotifyHost> = Arc::new(SubsystemHost::new(self));
        // Detached: dropping the handle lets the task run on its own, and
        // the exit guard is what observes its end.
        let _pump = self.spawn_gotify_subsystem_task(
            token,
            forge_connectors::gotify::run_subsystem(host, cfg, shutdown_rx),
        );
    }

    /// Spawn the Gotify pump behind the guard that clears this run's
    /// handle and liveness however the task ends: a return, a cancellation,
    /// or an unwind. Returns the task handle so a test can await the end.
    fn spawn_gotify_subsystem_task<F>(
        self: &Arc<Self>,
        token: Arc<()>,
        pump: F,
    ) -> tokio::task::JoinHandle<()>
    where
        F: std::future::Future<Output = ()> + Send + 'static,
    {
        let workspace = Arc::downgrade(self);
        tokio::spawn(async move {
            let _exit = GotifyExitGuard { workspace, token };
            pump.await;
        })
    }

    /// Stop the Gotify subsystem once no subscriptions remain: signal the
    /// stream task to exit and mark disconnected. No-op while any
    /// subscription is active or the subsystem isn't running.
    pub fn stop_gotify_subsystem_if_idle(&self) {
        if !self.gotify_subs.lock().is_empty() {
            return;
        }
        self.stop_gotify_subsystem();
    }

    /// Signal the running pump to stop and mark the connector
    /// disconnected, whatever subscriptions remain. The workspace's
    /// shutdown calls this so a clean stop reaches the pump's own exit
    /// rather than leaving the handle armed for the task's drop.
    pub(crate) fn stop_gotify_subsystem(&self) {
        // The handle lock is held across the flag write, as in the exit
        // guard: a subscribe can arm a new run the moment the handle is
        // taken, and a late write would blank that run's liveness.
        let subsystem = {
            let mut slot = self.gotify_subsystem.lock();
            let subsystem = slot.take();
            if subsystem.is_some() {
                *self.gotify_connected.lock() = false;
            }
            subsystem
        };
        if let Some(subsystem) = subsystem {
            let _ = subsystem.shutdown.send(());
        }
    }
}

/// One running Gotify pump as the workspace stores it: the shutdown signal
/// plus the token that tells this run from a later one.
pub(crate) struct GotifySubsystem {
    token: Arc<()>,
    shutdown: tokio::sync::oneshot::Sender<()>,
}

/// Clears a finished pump's handle and liveness when its task ends.
///
/// A guard rather than a trailing statement: an unwind reaches it too,
/// which is the defect this exists for - a panicking stream task left the
/// handle armed and no start could revive the subsystem.
struct GotifyExitGuard {
    workspace: std::sync::Weak<Workspace>,
    token: Arc<()>,
}

impl Drop for GotifyExitGuard {
    fn drop(&mut self) {
        let Some(workspace) = self.workspace.upgrade() else { return };
        let cleared = {
            // The handle lock is held across the flag write: a stop-then-
            // resubscribe can arm a new run the moment this one's handle is
            // taken, and a late write would blank that run's liveness.
            let mut slot = workspace.gotify_subsystem.lock();
            let ours = slot.as_ref().is_some_and(|run| Arc::ptr_eq(&run.token, &self.token));
            if ours {
                slot.take();
                *workspace.gotify_connected.lock() = false;
            }
            ours
        };
        if cleared {
            // A handle still armed here means nobody asked this run to
            // stop: the stream task died under it, or the pump unwound.
            let reason = if std::thread::panicking() { "panic" } else { "stream_ended" };
            tracing::warn!(
                target: "forge_workspace::gotify",
                event_name = "gotify_subsystem_exited",
                reason,
                "the Gotify subsystem exited without a shutdown",
            );
        } else {
            tracing::info!(
                target: "forge_workspace::gotify",
                event_name = "gotify_subsystem_exited",
                reason = "shutdown",
                "the Gotify subsystem stopped",
            );
        }
    }
}

/// The host a configured Gotify url names, for a log line: the scheme's
/// authority and nothing after it, so the client token never rides along.
fn url_host(url: &str) -> &str {
    let authority = url.split_once("://").map_or(url, |(_, rest)| rest);
    authority.split('/').next().unwrap_or_default()
}

/// The [`GotifyHost`] the forge-connectors pump drives: a
/// `Weak<Workspace>` behind the port. The pump holds the host strongly,
/// so the wrapper stays the weak boundary - the workspace can drop while
/// the subsystem runs, and every port call degrades to a no-op then,
/// exactly like the per-event upgrade the pump did when it lived here.
pub(crate) struct SubsystemHost(std::sync::Weak<Workspace>);

impl SubsystemHost {
    pub(crate) fn new(workspace: &Arc<Workspace>) -> Self {
        Self(Arc::downgrade(workspace))
    }
}

impl GotifyHost for SubsystemHost {
    fn http_client(&self, timeout: Duration) -> Result<reqwest::Client, String> {
        forge_agent::http_trust::with_extra_roots(reqwest::Client::builder().timeout(timeout))
            .build()
            .map_err(|error| error.to_string())
    }

    fn subscriptions(&self) -> Vec<forge_primitives::GotifySubscription> {
        self.0.upgrade().map(|ws| ws.gotify_subs.lock().clone()).unwrap_or_default()
    }

    /// The application NAME for `appid` via the reverse of the app index,
    /// or `None` when the id isn't known (index not yet fetched, or a new
    /// app the server added after the last refresh).
    fn app_name(&self, appid: u64) -> Option<String> {
        let ws = self.0.upgrade()?;
        ws.gotify_app_index
            .lock()
            .iter()
            .find(|&(_, &id)| id == appid)
            .map(|(name, _)| name.clone())
    }

    fn store_app_index(&self, index: HashMap<String, u64>) {
        if let Some(ws) = self.0.upgrade() {
            *ws.gotify_app_index.lock() = index;
        }
    }

    fn set_connected(&self, connected: bool) {
        if let Some(ws) = self.0.upgrade() {
            *ws.gotify_connected.lock() = connected;
        }
    }

    /// Wrap the matched message into the notification wire shape and
    /// dispatch it to the subscriber. This is the only place a
    /// forge-connectors match becomes a `GotifyNotification`, so the
    /// prose shape stays workspace-owned
    /// (`forge_server::envelope::detect_inbound` keys on it).
    fn deliver(
        &self,
        subscription: &forge_primitives::GotifySubscription,
        app: &str,
        message: &forge_primitives::GotifyMessage,
    ) {
        let Some(ws) = self.0.upgrade() else { return };
        let notification = crate::mcp::gotify::types::GotifyNotification {
            app: app.to_owned(),
            title: message.title.clone(),
            message: message.message.clone(),
            priority: message.priority,
        };
        if let Err(err) = ws.dispatch(Command::DeliverGotifyMessage {
            project: subscription.project.clone(),
            team_role: subscription.team_role.clone(),
            notification,
        }) {
            tracing::warn!(
                target: "forge_workspace::gotify",
                project = %subscription.project,
                error = ?err,
                "gotify DeliverGotifyMessage dispatch failed",
            );
        }
    }
}

#[cfg(any(test, feature = "testing"))]
impl Workspace {
    /// Register a Gotify subscription directly, bypassing the MCP
    /// subscribe path. Cross-crate test access so a view can exercise
    /// the Inspector's `refresh_gotify` resolution, mirroring
    /// [`Self::seed_test_cron`].
    ///
    /// The active-set write without the announcement: a fixture seeds state,
    /// it does not perform the write a view is owed a frame for.
    pub fn seed_test_gotify_subscription(&self, sub: forge_primitives::GotifySubscription) {
        self.gotify_subs.lock().push(sub);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::collections::HashMap;

    use tempfile::tempdir;

    use crate::target::SessionSlot;

    /// The Gotify notifications parked for `(project, label)`, under the
    /// org the project is declared in.
    fn parked_gotify(
        ws: &crate::Workspace,
        project: &str,
        label: Option<&str>,
    ) -> Vec<crate::mcp::gotify::types::GotifyNotification> {
        let org = ws.list_projects().into_iter().find(|v| v.name == project).expect("project").org;
        ws.parked_by_slot
            .lock()
            .get(&crate::SessionSlot::for_label(&org, project, label))
            .map(|parked| parked.gotify.clone())
            .unwrap_or_default()
    }

    /// Persist a worker row for `label` in `project_key`, the registry
    /// entry the durability checks read, failing loudly when it did not
    /// land. The key must resolve to a configured project: the row lands
    /// in the `(org, project name, label)` keyed `sessions` table.
    fn seed_worker_row(ws: &crate::Workspace, project_key: &ProjectKey, label: &str) {
        ws.seed_test_worker_row(project_key, label);
        assert!(
            ws.stored_worker_row(project_key, label).expect("read the seeded row").is_some(),
            "the worker row for {label} did not land; does {project_key:?} resolve to a project?",
        );
    }

    fn live_worker_entry(project: &str, label: &str) -> crate::mcp::workers::types::WorkerEntry {
        crate::mcp::workers::types::WorkerEntry {
            label: label.to_owned(),
            charter: "c".to_owned(),
            slot: SessionSlot::worker("TestOrg", project, label),
            session_id: None,
            status: forge_primitives::WorkerLiveness::Running,
            spawned_at: std::time::SystemTime::UNIX_EPOCH,
            spawned_by: SessionSlot::lead("TestOrg", project),
            needs_tag: false,
            is_git_repo_at_spawn: false,
            cwd: None,
            diagnostic: None,
            kick: None,
        }
    }

    fn gotify_sub(
        project: &str,
        applications: &[&str],
        min_priority: Option<u8>,
    ) -> forge_primitives::GotifySubscription {
        forge_primitives::GotifySubscription {
            id: uuid::Uuid::new_v4(),
            project: project.to_owned(),
            team_role: None,
            applications: applications.iter().map(|s| (*s).to_owned()).collect(),
            min_priority,
            created_at: std::time::SystemTime::UNIX_EPOCH,
        }
    }

    fn gotify_msg(appid: u64, priority: u8) -> forge_primitives::GotifyMessage {
        forge_primitives::GotifyMessage {
            id: 1,
            appid,
            title: "Alert".to_owned(),
            message: "body".to_owned(),
            priority,
            date: "2026-07-03T00:00:00Z".to_owned(),
        }
    }

    fn gotify_notif(
        app: &str,
        title: &str,
        message: &str,
        priority: u8,
    ) -> crate::GotifyNotification {
        crate::GotifyNotification {
            app: app.to_owned(),
            title: title.to_owned(),
            message: message.to_owned(),
            priority,
        }
    }

    /// Drain every currently-queued `SessionUpdate` from the test rx.
    fn drain_updates(
        rx: &mut tokio::sync::mpsc::UnboundedReceiver<crate::SeatUpdate>,
    ) -> Vec<crate::protocol::SessionUpdate> {
        let mut out = Vec::new();
        while let Ok(frame) = rx.try_recv() {
            out.push(std::sync::Arc::unwrap_or_clone(frame.update));
        }
        out
    }

    #[test]
    fn durable_gotify_subscription_persists_ephemeral_stays_in_memory() {
        use forge_primitives::GotifySubscription;

        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        let db = crate::store::Db::open(&dir.path().join("db.redb")).expect("open db");
        ws.install_db_for_test(db);

        let sub = |team_role: Option<&str>| GotifySubscription {
            id: uuid::Uuid::new_v4(),
            project: "p".to_owned(),
            team_role: team_role.map(str::to_owned),
            applications: vec![],
            min_priority: None,
            created_at: std::time::SystemTime::UNIX_EPOCH,
        };
        let durable = sub(None);
        let ephemeral = sub(Some("scratch"));
        ws.add_gotify_subscription(durable.clone(), true);
        ws.add_gotify_subscription(ephemeral, false);

        assert_eq!(
            ws.gotify_subscriptions_for_project("p").len(),
            2,
            "both live in the active in-memory set",
        );
        let persisted = || {
            crate::store::gotify::list(ws.db.lock().as_ref().expect("db installed")).expect("list")
        };
        assert_eq!(persisted().len(), 1, "only the durable subscription hit redb");
        assert_eq!(persisted()[0].id, durable.id);

        assert_eq!(
            ws.remove_gotify_subscription_owned_by("p", durable.id, None).map(|s| s.id),
            Some(durable.id),
            "the lead's own durable id removes, and the row comes back",
        );
        assert!(persisted().is_empty(), "removal cleared the persisted record");
    }

    /// `remove_gotify_subscriptions_for_worker` drops only the target
    /// worker's subs (matched by project name + label) from both the
    /// in-memory set and redb; the lead's sub and a sibling worker's sub
    /// survive, proving the removal is scoped, not a blanket project wipe.
    #[test]
    fn remove_gotify_subscriptions_for_worker_is_scoped_to_project_and_label() {
        let (ws, _rx) = Workspace::testing_stub();
        let dir = tempdir().expect("tempdir");
        ws.install_db_for_test(
            crate::store::Db::open(&dir.path().join("db.redb")).expect("open db"),
        );
        ws.seed_test_project("forge", "/tmp/gotify-durability");
        let view_key = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .map(|v| v.key)
            .expect("seeded project view");

        let mut scratch_sub = gotify_sub("forge", &[], None);
        scratch_sub.team_role = Some("scratch".to_owned());
        let lead_sub = gotify_sub("forge", &[], None);
        let mut sibling_sub = gotify_sub("forge", &[], None);
        sibling_sub.team_role = Some("other".to_owned());
        ws.add_gotify_subscription(scratch_sub.clone(), true);
        ws.add_gotify_subscription(lead_sub.clone(), true);
        ws.add_gotify_subscription(sibling_sub.clone(), true);

        ws.remove_gotify_subscriptions_for_worker(&view_key, "scratch");

        let in_mem = ws.gotify_subscriptions_for_project("forge");
        assert!(
            in_mem.iter().all(|s| s.id != scratch_sub.id),
            "the scratch worker's sub is gone from memory",
        );
        assert!(
            in_mem.iter().any(|s| s.id == lead_sub.id)
                && in_mem.iter().any(|s| s.id == sibling_sub.id),
            "the lead sub and the sibling worker's sub survive in memory",
        );

        let persisted = {
            let guard = ws.db.lock();
            crate::store::gotify::list(guard.as_ref().expect("db installed")).expect("list")
        };
        assert!(
            persisted.iter().all(|s| s.id != scratch_sub.id),
            "the scratch worker's sub is gone from redb",
        );
        assert!(
            persisted.iter().any(|s| s.id == lead_sub.id)
                && persisted.iter().any(|s| s.id == sibling_sub.id),
            "the survivors are still persisted in redb",
        );
    }

    /// Closing a worker drops its durable Gotify subs alongside its
    /// dynamic-worker row; the lead's sub survives.
    #[tokio::test]
    async fn closing_a_worker_drops_the_workers_durable_gotify_subs() {
        let (ws, _rx) = Workspace::testing_stub();
        let dir = tempdir().expect("tempdir");
        ws.install_db_for_test(
            crate::store::Db::open(&dir.path().join("db.redb")).expect("open db"),
        );
        ws.seed_test_project("forge", "/tmp/gotify-durability");
        let view_key = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .map(|v| v.key)
            .expect("seeded project view");
        ws.insert_live_worker(&view_key, live_worker_entry("forge", "scratch"));

        let mut scratch_sub = gotify_sub("forge", &[], None);
        scratch_sub.team_role = Some("scratch".to_owned());
        let lead_sub = gotify_sub("forge", &[], None);
        ws.add_gotify_subscription(scratch_sub.clone(), true);
        ws.add_gotify_subscription(lead_sub.clone(), true);

        crate::spawn::handle_close_worker(&ws, &view_key, "scratch");

        let in_mem = ws.gotify_subscriptions_for_project("forge");
        assert!(
            in_mem.iter().all(|s| s.id != scratch_sub.id),
            "teardown removed the worker's sub from memory",
        );
        assert!(in_mem.iter().any(|s| s.id == lead_sub.id), "the lead sub survives teardown");

        let persisted = {
            let guard = ws.db.lock();
            crate::store::gotify::list(guard.as_ref().expect("db installed")).expect("list")
        };
        assert!(
            persisted.iter().all(|s| s.id != scratch_sub.id),
            "teardown removed the worker's sub from redb",
        );
        assert!(persisted.iter().any(|s| s.id == lead_sub.id), "the lead sub is still persisted");
    }

    /// The seam the bug lived in: `resolve_identity` gathers the caller's
    /// dynamic-worker labels from the table and marks a table-backed
    /// worker durable, so its sub persists through the subscribe path even
    /// though the worker is not the lead.
    #[test]
    fn resolve_identity_persists_a_table_backed_dynamic_workers_sub() {
        let (ws, _rx) = Workspace::testing_stub();
        let dir = tempdir().expect("tempdir");
        ws.install_db_for_test(
            crate::store::Db::open(&dir.path().join("db.redb")).expect("open db"),
        );
        ws.seed_test_project("forge", "/tmp/gotify-durability-seam");
        let view_key = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .map(|v| v.key)
            .expect("seeded project view");

        // "scratch" is not the lead: durability
        // must come solely from its persisted worker row.
        seed_worker_row(&ws, &view_key, "scratch");
        let caller = SessionSlot::worker("TestOrg", "forge", "scratch");
        ws.insert_live_worker(&view_key, live_worker_entry("forge", "scratch"));

        let (name, team_role, durable) = crate::mcp::gotify::facade::resolve_identity(&ws, &caller)
            .expect("the worker caller resolves to its project");
        assert_eq!(
            (name.as_str(), team_role.as_deref(), durable),
            ("forge", Some("scratch"), true),
            "a table-backed dynamic worker resolves as a durable subscriber",
        );

        // The durable identity persists through the subscribe path's write.
        let sub = forge_primitives::GotifySubscription {
            id: uuid::Uuid::new_v4(),
            project: name,
            team_role,
            applications: vec![],
            min_priority: None,
            created_at: std::time::SystemTime::UNIX_EPOCH,
        };
        let sub_id = sub.id;
        ws.add_gotify_subscription(sub, durable);
        let persisted = {
            let guard = ws.db.lock();
            crate::store::gotify::list(guard.as_ref().expect("db installed")).expect("list")
        };
        assert!(
            persisted.iter().any(|s| s.id == sub_id),
            "the table-backed worker's sub is persisted to redb",
        );
    }

    /// A `ProjectKey` that matches no project resolves to no name, so the
    /// removal is a safe no-op that leaves every sub intact.
    #[test]
    fn remove_gotify_subscriptions_for_worker_unknown_project_is_a_no_op() {
        let (ws, _rx) = Workspace::testing_stub();
        let dir = tempdir().expect("tempdir");
        ws.install_db_for_test(
            crate::store::Db::open(&dir.path().join("db.redb")).expect("open db"),
        );
        ws.seed_test_project("forge", "/tmp/gotify-durability-noop");

        let mut scratch_sub = gotify_sub("forge", &[], None);
        scratch_sub.team_role = Some("scratch".to_owned());
        ws.add_gotify_subscription(scratch_sub.clone(), true);

        ws.remove_gotify_subscriptions_for_worker(&ProjectKey::new("ghost"), "scratch");

        assert!(
            ws.gotify_subscriptions_for_project("forge").iter().any(|s| s.id == scratch_sub.id),
            "an unknown project key leaves the in-memory subs intact",
        );
        let persisted = {
            let guard = ws.db.lock();
            crate::store::gotify::list(guard.as_ref().expect("db installed")).expect("list")
        };
        assert!(
            persisted.iter().any(|s| s.id == scratch_sub.id),
            "and leaves the persisted subs intact",
        );
    }

    /// With no store installed the removal still scrubs the in-memory set;
    /// the redb delete is simply skipped (mirrors
    /// `record_worker_row_errors_when_store_unavailable`).
    #[test]
    fn remove_gotify_subscriptions_for_worker_scrubs_memory_without_a_db() {
        let (ws, _rx) = Workspace::testing_stub();
        // No install_db_for_test: the store is closed for this session.
        ws.seed_test_project("forge", "/tmp/gotify-durability-nodb");
        let view_key = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .map(|v| v.key)
            .expect("seeded project view");

        let mut scratch_sub = gotify_sub("forge", &[], None);
        scratch_sub.team_role = Some("scratch".to_owned());
        ws.add_gotify_subscription(scratch_sub.clone(), true);

        ws.remove_gotify_subscriptions_for_worker(&view_key, "scratch");

        assert!(
            ws.gotify_subscriptions_for_project("forge").iter().all(|s| s.id != scratch_sub.id),
            "the in-memory sub is scrubbed even with no store installed",
        );
    }

    /// The GotifyHost impl the forge-connectors pump drives: the port
    /// reads the `gotify_*` state, and `deliver` wraps the matched
    /// message into the notification wire shape addressed to the
    /// subscriber's project + team role.
    #[test]
    fn gotify_host_impl_reads_state_and_delivers_to_the_subscriber() {
        use forge_connectors::gotify::GotifyHost as _;

        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        let mut worker_sub = gotify_sub("p1", &[], None);
        worker_sub.team_role = Some("reviewer".to_owned());
        ws.add_gotify_subscription(worker_sub.clone(), false);
        ws.add_gotify_subscription(gotify_sub("p2", &[], None), false);

        let host = SubsystemHost::new(&ws);
        host.store_app_index(HashMap::from([("alerts".to_owned(), 3u64)]));
        assert_eq!(ws.gotify_app_index.lock().len(), 1, "the index lands on the workspace");
        assert_eq!(host.app_name(3).as_deref(), Some("alerts"), "appid resolves via the index");
        assert_eq!(host.app_name(9), None, "an unknown appid resolves to nothing");

        host.set_connected(true);
        assert!(ws.gotify_connected(), "liveness reaches the Inspector's flag");

        assert_eq!(host.subscriptions().len(), 2, "the pump matches against the active set");

        ws.enable_test_dispatch_intercept();
        host.deliver(&worker_sub, "alerts", &gotify_msg(3, 5));
        let dispatched = ws.drain_test_dispatch_buffer();
        let Some(crate::protocol::Command::DeliverGotifyMessage {
            project,
            team_role,
            notification,
        }) = dispatched.first()
        else {
            panic!("deliver dispatches exactly one DeliverGotifyMessage: {dispatched:?}");
        };
        assert_eq!(project, "p1");
        assert_eq!(team_role.as_deref(), Some("reviewer"), "the team role rides along");
        assert_eq!(
            notification.app, "alerts",
            "the connector-resolved display name becomes the notification's app",
        );
        assert_eq!(notification.title, "Alert", "the message title rides into the wrap");
        assert_eq!(notification.message, "body", "the message body rides into the wrap");
        assert_eq!(notification.priority, 5, "the priority rides into the wrap");
    }

    #[test]
    fn deliver_gotify_message_spawns_asleep_project_and_buffers() {
        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/gotify-forge");

        let notif = gotify_notif("forge", "Heads up", "hello envelope", 5);
        ws.enable_test_dispatch_intercept();
        crate::spawn::deliver_gotify_message(&ws, "forge", None, notif.clone());
        let dispatched = ws.drain_test_dispatch_buffer();

        let spawns = dispatched
            .iter()
            .filter(|c| {
                matches!(c, crate::protocol::Command::SpawnProject { project_name, .. }
                    if project_name == "forge")
            })
            .count();
        assert_eq!(spawns, 1, "the asleep project got a spawn");

        assert_eq!(
            parked_gotify(&ws, "forge", None),
            vec![notif],
            "the notification is parked for the project's lead",
        );
    }

    #[test]
    fn deliver_gotify_message_skips_gone_project() {
        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        // No project seeded: "ghost" is not in forge.toml.
        ws.enable_test_dispatch_intercept();
        crate::spawn::deliver_gotify_message(
            &ws,
            "ghost",
            None,
            gotify_notif("ghost", "t", "x", 1),
        );
        let dispatched = ws.drain_test_dispatch_buffer();
        assert!(
            dispatched.iter().all(|c| !matches!(c, crate::protocol::Command::SpawnProject { .. })),
            "a gone target is skipped without a spawn or panic",
        );
    }

    #[test]
    fn deliver_gotify_message_delivers_to_running_team_worker() {
        let dir = tempdir().expect("tempdir");
        let (ws, mut rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/gotify-team");
        let view_key = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .map(|v| v.key)
            .expect("seeded project view");
        let worker_key = SessionSlot::worker("TestOrg", "forge", "reviewer");
        ws.insert_live_worker(
            &view_key,
            crate::mcp::workers::types::WorkerEntry {
                label: "reviewer".to_owned(),
                charter: "review".to_owned(),
                slot: worker_key.clone(),
                session_id: None,
                status: forge_primitives::WorkerLiveness::Running,
                spawned_at: std::time::SystemTime::UNIX_EPOCH,
                spawned_by: SessionSlot::lead("TestOrg", "forge"),
                needs_tag: false,
                is_git_repo_at_spawn: false,
                cwd: None,
                diagnostic: None,
                kick: None,
            },
        );

        ws.mark_session_connected_for_test(&worker_key, "worker-reviewer");
        let notif = gotify_notif("Backups", "Nightly backup", "done", 5);
        ws.enable_test_dispatch_intercept();
        crate::spawn::deliver_gotify_message(&ws, "forge", Some("reviewer"), notif.clone());
        let dispatched = ws.drain_test_dispatch_buffer();

        assert!(
            dispatched.iter().any(|c| matches!(
                c,
                crate::protocol::Command::Prompt { key, .. }
                    | crate::protocol::Command::PromptUnder { key, .. }
                    if key == &worker_key
            )),
            "a running team worker receives the notification directly",
        );
        assert!(
            dispatched.iter().all(|c| !matches!(c, crate::protocol::Command::SpawnProject { .. })),
            "no project spawn when the target worker is already running",
        );

        // The delivery ALSO echoes the notification block into the worker's
        // chat so the user sees what arrived (mirrors the peer echo).
        let echoed = drain_updates(&mut rx).into_iter().any(|update| {
            matches!(
                update,
                crate::protocol::SessionUpdate::GotifyNotificationAppended {
                    key,
                    notification,
                    ..
                } if key == worker_key && notification == notif
            )
        });
        assert!(echoed, "a running-target delivery emits a GotifyNotificationAppended echo");
    }

    #[test]
    fn deliver_gotify_message_team_worker_asleep_falls_through_to_lead() {
        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/gotify-team");

        // No live worker of that role: the subscription falls through to
        // lead delivery, which spawns the asleep project (the lead brings up
        // its team) and buffers the envelope for the Connected drain.
        let notif = gotify_notif("Backups", "backup", "env", 5);
        ws.enable_test_dispatch_intercept();
        crate::spawn::deliver_gotify_message(&ws, "forge", Some("reviewer"), notif.clone());
        let dispatched = ws.drain_test_dispatch_buffer();

        let spawns = dispatched
            .iter()
            .filter(|c| {
                matches!(c, crate::protocol::Command::SpawnProject { project_name, .. }
                    if project_name == "forge")
            })
            .count();
        assert_eq!(spawns, 1, "an asleep team-worker subscription spawns the project lead");

        assert_eq!(
            parked_gotify(&ws, "forge", None),
            vec![notif],
            "the notification is parked for the project's lead",
        );
    }

    /// A `Failed` entry is kept by design but is NOT live, so it is no more
    /// of a delivery target than a label with no entry at all: routing to it
    /// would park the notification in a bucket nothing drains. It falls
    /// through to the lead exactly like the gone-worker case beside this one.
    #[test]
    fn deliver_gotify_message_failed_team_worker_falls_through_to_lead() {
        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/gotify-team");
        let view_key = ws.project_key_for_name("forge").expect("seeded project");
        let mut failed = live_worker_entry("forge", "reviewer");
        failed.status = forge_primitives::WorkerLiveness::Failed;
        failed.diagnostic = Some("connection closed".to_owned());
        ws.insert_live_worker(&view_key, failed);

        let notif = gotify_notif("Backups", "backup", "env", 5);
        ws.enable_test_dispatch_intercept();
        crate::spawn::deliver_gotify_message(&ws, "forge", Some("reviewer"), notif.clone());
        let dispatched = ws.drain_test_dispatch_buffer();

        assert_eq!(
            parked_gotify(&ws, "forge", Some("reviewer")),
            Vec::new(),
            "nothing parks for a failed worker's slot",
        );
        let spawns = dispatched
            .iter()
            .filter(|c| {
                matches!(c, crate::protocol::Command::SpawnProject { project_name, .. }
                    if project_name == "forge")
            })
            .count();
        assert_eq!(spawns, 1, "the fall-through spawns the asleep project");
        assert_eq!(
            parked_gotify(&ws, "forge", None),
            vec![notif],
            "and the notification waits for the project's lead",
        );
    }

    /// A team-worker subscription whose worker is a live entry but still
    /// Spawning (session_id None) must buffer on the worker's own
    /// DomainSession for its Connected drain - not dispatch a bare
    /// Command::Prompt (dropped) and not fall through to the lead.
    #[test]
    fn deliver_gotify_message_to_spawning_team_worker_buffers_on_its_domain() {
        let dir = tempdir().expect("tempdir");
        let (ws, _rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/gotify-spawning");
        let view_key = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .map(|v| v.key)
            .expect("seeded project view");
        let worker_key = SessionSlot::worker("TestOrg", "forge", "reviewer");
        ws.insert_live_worker(&view_key, live_worker_entry("forge", "reviewer"));
        // Register the domain WITHOUT stamping session_id: still Spawning.
        ws.register_domain_session(worker_key.clone(), None);

        let notif = gotify_notif("Backups", "backup", "env", 5);
        ws.enable_test_dispatch_intercept();
        crate::spawn::deliver_gotify_message(&ws, "forge", Some("reviewer"), notif.clone());
        let dispatched = ws.drain_test_dispatch_buffer();

        assert!(
            dispatched.is_empty(),
            "no bare Prompt (dropped) and no lead fallback for a spawning worker",
        );
        assert_eq!(
            parked_gotify(&ws, "forge", Some("reviewer")),
            vec![notif],
            "parked for the worker's own label, for its Connected drain",
        );
    }

    /// The one `ConnectorSubscriptionsChanged` on a test's update stream, as
    /// its key and the two sets it announces.
    fn next_connectors_changed(
        rx: &mut tokio::sync::mpsc::UnboundedReceiver<crate::SeatUpdate>,
    ) -> (
        SessionSlot,
        Vec<forge_primitives::GotifySubscription>,
        Vec<forge_primitives::slack::SlackSubscription>,
    ) {
        match rx.try_recv().map(|frame| std::sync::Arc::unwrap_or_clone(frame.update)) {
            Ok(crate::protocol::SessionUpdate::ConnectorSubscriptionsChanged {
                key,
                gotify,
                slack,
            }) => (key, gotify, slack),
            other => {
                panic!("expected a ConnectorSubscriptionsChanged on the stream, got {other:?}")
            }
        }
    }

    /// A gotify subscribe, unsubscribe and worker teardown each announce the
    /// project's connector sets on its lead seat - BOTH connectors, because
    /// the section a client patches is one - so the inspector's GOTIFY
    /// section pops on the frame that moved it.
    ///
    /// Mutants: drop any one emission (the drain is empty); announce only the
    /// connector that moved (a patch would leave the sibling stale); announce
    /// a set taken before the write.
    #[test]
    fn every_gotify_write_announces_the_projects_subscriptions() {
        let dir = tempdir().expect("tempdir");
        let (ws, mut rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/tp-gotify");
        let view_key = ws.project_key_for_name("forge").expect("seeded project");
        let lead = SessionSlot::lead("TestOrg", "forge");

        // The sibling connector carries a set of its own, so an announcer
        // that sent a hardcoded empty beside the moved set cannot pass: the
        // frame carries the pair, and a later "send only what moved" would
        // clear the client's slack rows on every gotify write.
        ws.add_slack_subscription(
            forge_primitives::slack::SlackSubscription {
                id: uuid::Uuid::new_v4(),
                workspace: "acme".to_owned(),
                project: "forge".to_owned(),
                team_role: None,
                target: forge_primitives::slack::SlackSubscriptionTarget::DirectMessages,
                created_at: std::time::SystemTime::UNIX_EPOCH,
            },
            true,
        );
        let (_, _, slack) = next_connectors_changed(&mut rx);
        assert_eq!(slack.len(), 1, "precondition: the sibling set is seeded and announced");

        let subscription = gotify_sub("forge", &["alerts"], Some(5));
        ws.add_gotify_subscription(subscription.clone(), true);
        let (key, gotify, slack) = next_connectors_changed(&mut rx);
        assert_eq!(key, lead, "a subscribe routes on the project's lead seat");
        assert_eq!(
            gotify.iter().map(|sub| sub.id).collect::<Vec<_>>(),
            [subscription.id],
            "and carries the set the core now holds",
        );
        assert_eq!(slack.len(), 1, "beside the sibling connector's own set: {slack:?}");

        assert!(
            ws.remove_gotify_subscription_owned_by("forge", subscription.id, None).is_some(),
            "precondition: the lead removes its own subscription",
        );
        let (_, gotify, slack) = next_connectors_changed(&mut rx);
        assert!(gotify.is_empty(), "an unsubscribe announces the set it left behind");
        assert_eq!(slack.len(), 1, "and still carries the sibling's set: {slack:?}");

        // A worker's teardown announces the survivors.
        let mut worker_subscription = gotify_sub("forge", &[], None);
        worker_subscription.team_role = Some("reviewer".to_owned());
        ws.add_gotify_subscription(worker_subscription, true);
        let (_, gotify, _) = next_connectors_changed(&mut rx);
        assert_eq!(gotify.len(), 1, "precondition: the worker's subscribe announced it");
        ws.remove_gotify_subscriptions_for_worker(&view_key, "reviewer");
        let (announced_key, gotify, _) = next_connectors_changed(&mut rx);
        assert_eq!(announced_key, lead, "a teardown routes the same way");
        assert!(gotify.is_empty(), "and announces what the teardown left");
    }

    /// A workspace with a real `[gotify]` config and one subscription, so
    /// `start_gotify_subsystem` takes its start path rather than declining.
    /// The url is a refused local port, so a pump the real start spawns
    /// fails its dial without reaching anything.
    fn subsystem_fixture() -> (Arc<Workspace>, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("tempdir");
        let mut config = crate::config::LoadedConfig::empty_for_test();
        config.gotify = Some(forge_primitives::GotifyConfig {
            url: "http://127.0.0.1:1".to_owned(),
            client_token: "Ctest".to_owned(),
        });
        let (ws, _rx) =
            Workspace::testing_stub_with_config(dir.path().to_owned(), config).expect("stub");
        ws.seed_test_gotify_subscription(gotify_sub("forge", &[], None));
        (ws, dir)
    }

    /// The defect: a pump whose stream task dies leaves its handle armed,
    /// and every later start no-ops against it. A dead pump clears the
    /// handle and liveness, so the next start runs.
    ///
    /// Mutant: clear only on the shutdown path - a trailing statement,
    /// which is what left the subsystem wedged.
    #[tokio::test]
    async fn a_dead_pump_clears_its_handle_and_the_next_start_runs() {
        let (ws, _dir) = subsystem_fixture();
        let token = Arc::new(());
        let (shutdown, _shutdown_rx) = tokio::sync::oneshot::channel();
        *ws.gotify_subsystem.lock() = Some(GotifySubsystem { token: Arc::clone(&token), shutdown });
        ws.seed_test_gotify_connected(true);

        // The pump body returning is the stream task ending under it.
        ws.spawn_gotify_subsystem_task(token, async {}).await.expect("the pump task returns");

        assert!(ws.gotify_subsystem.lock().is_none(), "a dead pump's handle is cleared");
        assert!(!ws.gotify_connected(), "and liveness goes false with it");

        ws.start_gotify_subsystem();
        assert!(ws.gotify_subsystem.lock().is_some(), "a second start runs");
    }

    /// An unwind inside the pump - a panic out of `route` or a host call -
    /// clears the handle too. A panic in the inner stream task is the other
    /// shape: it ends that task, the pump returns normally, and the exit
    /// guard reports `stream_ended`.
    ///
    /// Mutant: a trailing statement instead of a guard, where the unwind
    /// skips it.
    #[tokio::test]
    async fn a_panicking_pump_clears_its_handle() {
        let (ws, _dir) = subsystem_fixture();
        let token = Arc::new(());
        let (shutdown, _shutdown_rx) = tokio::sync::oneshot::channel();
        *ws.gotify_subsystem.lock() = Some(GotifySubsystem { token: Arc::clone(&token), shutdown });
        ws.seed_test_gotify_connected(true);

        let task = ws.spawn_gotify_subsystem_task(token, async {
            panic!("the pump unwound before it could clear its handle");
        });
        let error = task.await.expect_err("the pump task unwinds");
        assert!(error.is_panic(), "the task ended by unwinding, not cancellation");

        assert!(ws.gotify_subsystem.lock().is_none(), "an unwind clears the handle too");
        assert!(!ws.gotify_connected(), "and liveness goes false with it");
    }

    /// A guard that fires after its run was stopped must leave a run armed
    /// since alone: stop-then-resubscribe can start the next pump before
    /// the old task unwinds.
    ///
    /// Mutant: take the slot unconditionally, where the stale guard
    /// disarms a live pump and clears its liveness.
    #[tokio::test]
    async fn a_stale_guard_leaves_a_newer_runs_handle_armed() {
        let (ws, _dir) = subsystem_fixture();
        let stale = Arc::new(());
        let (stale_tx, _stale_rx) = tokio::sync::oneshot::channel();
        *ws.gotify_subsystem.lock() =
            Some(GotifySubsystem { token: Arc::clone(&stale), shutdown: stale_tx });
        let task = ws.spawn_gotify_subsystem_task(stale, async {});

        // The stale run's task has not run yet: its stop took the handle
        // and a new subscription armed a new run.
        let newer = Arc::new(());
        let (newer_tx, _newer_rx) = tokio::sync::oneshot::channel();
        *ws.gotify_subsystem.lock() =
            Some(GotifySubsystem { token: Arc::clone(&newer), shutdown: newer_tx });
        ws.seed_test_gotify_connected(true);

        task.await.expect("the stale pump task returns");

        let slot = ws.gotify_subsystem.lock();
        let held = slot.as_ref().expect("the newer run's handle stays armed");
        assert!(Arc::ptr_eq(&held.token, &newer), "the stale guard took nothing");
        drop(slot);
        assert!(ws.gotify_connected(), "and left the newer run's liveness alone");
    }

    /// A clean stop signals the pump, so its exit guard takes the shutdown
    /// branch instead of reporting a stream end at every process stop.
    ///
    /// Mutant: leave the handle to the task's drop, where every clean stop
    /// logs a stream end.
    #[tokio::test]
    async fn a_workspace_shutdown_stops_the_gotify_pump() {
        use tracing_subscriber::layer::SubscriberExt;

        let (ws, _dir) = subsystem_fixture();
        let token = Arc::new(());
        let (shutdown, shutdown_rx) = tokio::sync::oneshot::channel();
        *ws.gotify_subsystem.lock() = Some(GotifySubsystem { token: Arc::clone(&token), shutdown });
        ws.seed_test_gotify_connected(true);
        // The pump body ends on the shutdown signal, as the real pump does.
        let pump = ws.spawn_gotify_subsystem_task(token, async {
            let _ = shutdown_rx.await;
        });

        let caught = Caught::default();
        let subscriber = tracing_subscriber::Registry::default().with(caught.clone());
        {
            let _default = tracing::subscriber::set_default(subscriber);
            ws.shutdown_within(std::time::Duration::from_secs(5)).await;
            pump.await.expect("the pump task returns");
        }

        assert!(ws.gotify_subsystem.lock().is_none(), "shutdown takes the handle");
        assert!(!ws.gotify_connected(), "and marks the connector disconnected");
        let records = caught.0.lock().expect("capture").clone();
        assert!(
            records.iter().any(|(level, _, fields)| {
                *level == tracing::Level::INFO
                    && fields.contains("gotify_subsystem_exited")
                    && fields.contains("reason=\"shutdown\"")
            }),
            "the guard takes the shutdown branch, not the stream-ended one: {records:?}",
        );
    }

    /// One captured tracing record: its level, its target and its fields.
    #[derive(Clone, Default)]
    struct Caught(std::sync::Arc<std::sync::Mutex<Vec<(tracing::Level, String, String)>>>);

    #[derive(Default)]
    struct CaughtFields(String);

    impl tracing::field::Visit for CaughtFields {
        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
            if !self.0.is_empty() {
                self.0.push(' ');
            }
            let _ = std::fmt::write(&mut self.0, format_args!("{}={value:?}", field.name()));
        }
    }

    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Caught {
        fn on_event(
            &self,
            event: &tracing::Event<'_>,
            _ctx: tracing_subscriber::layer::Context<'_, S>,
        ) {
            let mut fields = CaughtFields::default();
            event.record(&mut fields);
            self.0.lock().expect("capture").push((
                *event.metadata().level(),
                event.metadata().target().to_owned(),
                fields.0,
            ));
        }
    }

    /// A decline is a normal state, so it is recorded at debug and never
    /// warns. `forge-server`'s logging test pins the directive that keeps
    /// the record readable.
    ///
    /// Mutant: put any decline back at warn, where every boot without
    /// Gotify reads as a problem.
    #[test]
    fn a_declined_start_is_recorded_at_debug() {
        use tracing_subscriber::layer::SubscriberExt;

        let (ws, _rx) = Workspace::testing_stub();
        let caught = Caught::default();
        let subscriber = tracing_subscriber::Registry::default().with(caught.clone());
        tracing::subscriber::with_default(subscriber, || ws.start_gotify_subsystem());

        let records = caught.0.lock().expect("capture").clone();
        assert!(
            records.iter().any(|(level, target, fields)| {
                *level == tracing::Level::DEBUG
                    && target == "forge_workspace::gotify"
                    && fields.contains("gotify_subsystem_declined")
                    && fields.contains("reason=\"no_config\"")
            }),
            "a start with no Gotify config declines at debug: {records:?}",
        );
        assert!(
            records.iter().all(|(level, _, _)| *level != tracing::Level::WARN),
            "no decline is a warning: {records:?}",
        );
    }

    /// A write that changed nothing announces nothing: an unsubscribe for an
    /// id nothing holds and a teardown matching no worker subscription are
    /// both no-ops.
    ///
    /// Mutant: emit unconditionally, where each of these is news the section
    /// never moved for.
    #[test]
    fn a_gotify_write_that_changed_nothing_announces_nothing() {
        let dir = tempdir().expect("tempdir");
        let (ws, mut rx) = Workspace::testing_stub_with_config_dir(dir.path().to_owned());
        ws.seed_test_project("forge", "/tmp/tp-gotify-noop");
        let view_key = ws.project_key_for_name("forge").expect("seeded project");

        assert!(
            ws.remove_gotify_subscription_owned_by("forge", uuid::Uuid::new_v4(), None).is_none(),
            "precondition: no subscription carries the id",
        );
        ws.remove_gotify_subscriptions_for_worker(&view_key, "nobody");

        let announced = rx.try_recv();
        assert!(
            announced.is_err(),
            "no write here moved a subscription, so nothing is announced, and this was: \
             {announced:?}",
        );
    }
}