exfiltrate 0.4.0

// SPDX-License-Identifier: MIT OR Apache-2.0

//! Live-state providers, and the `snapshot` command that drives them.
//!
//! # Why this is not the log path
//!
//! A recorder answers *what just happened*. A provider answers *what exists
//! now*: which tasks are live, how deep a queue is, what a routing table
//! contains. Reconstructing the second from the first fails in exactly the
//! cases where you need it — after records were dropped, or when the
//! interesting state never emitted anything because it never changed.
//!
//! # What a subsystem has to do
//!
//! Implement [`Provider`] and call [`add_provider`]. Nothing else. In
//! particular a subsystem does **not** register a command, so twelve
//! subsystems do not produce twelve slightly different command shapes; they
//! all appear under `snapshot --subsystem <name>`.
//!
//! This layer has no dependency on `logwise` — neither the facade nor the
//! runtime — and must not grow one. A logwise-aware provider belongs in a crate
//! above both, such as `logwise_agent_exfiltrate`.
//!
//! # The contract a provider is held to
//!
//! * **Bounded.** [`SnapshotRequest::limit`] is the most rows the caller wants.
//!   A provider that ignores it is truncated by the registry and the result is
//!   reported `Partial` rather than silently shortened.
//! * **Cancellable.** [`SnapshotRequest::should_stop`] goes true when the
//!   client disconnected or the deadline passed. Cancellation is cooperative:
//!   nothing can preempt a provider mid-scan, so one that never checks is
//!   reported `TimedOut` after the fact rather than killed.
//! * **Non-blocking.** A provider that cannot take its lock should return
//!   [`ProviderResult::Busy`] instead of waiting. Blocking here blocks the
//!   debug server, which is the one thing that still has to work when the
//!   application is wedged.
//!
//! # Privacy and authorization
//!
//! Authorization happens on the connection, before any command runs: a peer
//! that did not satisfy the configured token (see [`crate::Config`]) never reaches
//! this code. What this layer adds is that the **default projection is
//! support-safe**, so a deployment that got its transport wrong leaks less.
//! `--view local` is the opt-in for local-only fields, and there is no way to
//! mark a field secret and have it shipped anyway — see
//! [`exfiltrate_internal::snapshot::FieldPrivacy`].

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use exfiltrate_internal::args::{ArgKind, ArgSpec, ParsedArgs};
use exfiltrate_internal::command::{Command, CommandContext, Response};
use exfiltrate_internal::snapshot::{
    FieldPrivacy, SnapshotField, SnapshotOutcome, SnapshotResponse, SnapshotRow, SnapshotSet,
    SnapshotValue,
};
use wasm_lite_std::Mutex;
use wasm_lite_std::time::{Duration, Instant};

/// What the caller asked for.
#[derive(Debug)]
pub struct SnapshotRequest<'a> {
    selector: Option<&'a str>,
    limit: usize,
    deadline: Instant,
    cancelled: &'a AtomicBool,
}

impl<'a> SnapshotRequest<'a> {
    /// The `--id` the caller passed, if any. Its meaning is the subsystem's.
    pub fn selector(&self) -> Option<&'a str> {
        self.selector
    }

    /// The most rows worth producing. Producing more is not an error, but the
    /// extra will be discarded and the answer marked `Partial`.
    pub fn limit(&self) -> usize {
        self.limit
    }

    /// True once the caller has gone or the deadline has passed.
    ///
    /// Check it in any loop that could run long. Nothing can interrupt a
    /// provider from outside.
    pub fn should_stop(&self) -> bool {
        self.cancelled.load(Ordering::Relaxed) || Instant::now() >= self.deadline
    }
}

/// One row, built with an explicit privacy level per field.
///
/// The level is not a hint. It decides whether the value survives projection,
/// and the default view drops local-only fields.
#[derive(Clone, Debug, Default)]
pub struct Row(Vec<SnapshotField>);

impl Row {
    /// An empty row. Add fields with [`support`](Row::support) or
    /// [`local`](Row::local), which is where the privacy level is chosen.
    pub fn new() -> Row {
        Row(Vec::new())
    }

    /// A value safe for any peer that reached this process — a count, a state
    /// name, an id this crate minted.
    pub fn support(mut self, name: impl Into<String>, value: impl Into<SnapshotValue>) -> Row {
        self.0.push(SnapshotField {
            name: name.into(),
            privacy: FieldPrivacy::SupportSafe,
            value: Some(value.into()),
        });
        self
    }

    /// A value derived from the application's own data — a label someone typed,
    /// a path, a URL. Withheld unless the caller asked for a local view.
    pub fn local(mut self, name: impl Into<String>, value: impl Into<SnapshotValue>) -> Row {
        self.0.push(SnapshotField {
            name: name.into(),
            privacy: FieldPrivacy::LocalOnly,
            value: Some(value.into()),
        });
        self
    }

    /// How many fields the row carries, before any view drops any of them.
    pub fn len(&self) -> usize {
        self.0.len()
    }

    /// Whether the row carries no fields at all.
    pub fn is_empty(&self) -> bool {
        self.0.is_empty()
    }
}

/// Not `#[non_exhaustive]`: the three outcomes are the contract between a
/// provider and the CLI, and a caller that wildcards them would report a
/// partial answer as a complete one.
/// What a provider produced.
#[derive(Debug)]
pub enum ProviderResult {
    /// The rows matching the selector.
    Rows(Vec<Row>),
    /// Everything it could get before it had to stop, and why.
    Partial(Vec<Row>, String),
    /// The subsystem is here but cannot answer now.
    Unavailable(String),
    /// Compiled out of this build.
    NotCompiled,
    /// Its state was locked. Return this rather than blocking.
    Busy,
}

/// A subsystem that can describe its live state.
pub trait Provider: Send + Sync + 'static {
    /// The name this appears under, as `snapshot --subsystem <name>`.
    ///
    /// A short lowercase noun: `tasks`, `executor`, `context`.
    fn subsystem(&self) -> &'static str;

    /// One line for `snapshot --list`.
    fn description(&self) -> &'static str;

    /// Produce the rows. See the module docs for the contract.
    fn snapshot(&self, request: &SnapshotRequest<'_>) -> ProviderResult;
}

static PROVIDERS: Mutex<Vec<Arc<dyn Provider>>> = Mutex::new(Vec::new());

/// Registers a provider. A second registration under the same name replaces the
/// first, so a subsystem that re-initialises does not accumulate duplicates.
pub fn add_provider<P: Provider>(provider: P) {
    let provider: Arc<dyn Provider> = Arc::new(provider);
    PROVIDERS.with_mut_sync(|providers| {
        providers.retain(|existing| existing.subsystem() != provider.subsystem());
        providers.push(provider);
    });
}

/// The subsystems currently registered.
pub fn registered() -> Vec<(&'static str, &'static str)> {
    PROVIDERS.with_sync(|providers| {
        providers
            .iter()
            .map(|provider| (provider.subsystem(), provider.description()))
            .collect()
    })
}

#[cfg(test)]
pub(crate) fn clear_providers() {
    PROVIDERS.with_mut_sync(|providers| providers.clear());
}

/// Runs one provider, enforcing everything the provider is not trusted to do
/// itself: the row limit, the deadline, panic isolation, and projection.
fn run(
    provider: &Arc<dyn Provider>,
    selector: Option<&str>,
    limit: usize,
    timeout: Duration,
    cancelled: &AtomicBool,
    local_view: bool,
) -> SnapshotResponse {
    let started = Instant::now();
    let deadline = started + timeout;
    let request = SnapshotRequest {
        selector,
        limit,
        deadline,
        cancelled,
    };

    // A provider is someone else's code. On native a panic is caught and
    // reported, because a crashing provider is a bug in that subsystem and
    // reporting it as merely "no rows" hides it. On wasm32 the profile builds
    // std with panic_abort, so there is nothing to catch -- stated rather than
    // silently different.
    #[cfg(not(target_arch = "wasm32"))]
    let produced =
        std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| provider.snapshot(&request)));
    #[cfg(target_arch = "wasm32")]
    let produced = Ok::<_, ()>(provider.snapshot(&request));

    let elapsed_ms = started.elapsed().as_millis() as u64;
    let overran = Instant::now() >= deadline;

    let (mut rows, mut outcome, mut reason) = match produced {
        Err(_) => (
            Vec::new(),
            SnapshotOutcome::Panicked,
            Some("the provider panicked; this is a bug in that subsystem".to_string()),
        ),
        Ok(ProviderResult::Rows(rows)) => (rows, SnapshotOutcome::Ready, None),
        Ok(ProviderResult::Partial(rows, why)) => (rows, SnapshotOutcome::Partial, Some(why)),
        Ok(ProviderResult::Unavailable(why)) => {
            (Vec::new(), SnapshotOutcome::Unavailable, Some(why))
        }
        Ok(ProviderResult::NotCompiled) => (
            Vec::new(),
            SnapshotOutcome::NotCompiled,
            Some("the subsystem is not compiled into this build".to_string()),
        ),
        Ok(ProviderResult::Busy) => (
            Vec::new(),
            SnapshotOutcome::Busy,
            Some("the subsystem's state was locked; retry".to_string()),
        ),
    };

    // A provider that ignored the deadline is reported, not trusted.
    if overran && outcome == SnapshotOutcome::Ready {
        outcome = SnapshotOutcome::TimedOut;
        reason = Some(format!(
            "the provider ran past its {}ms deadline; its result may be stale",
            timeout.as_millis()
        ));
    }

    // ...and one that ignored the limit is truncated here rather than
    // returning more than was asked for.
    if rows.len() > limit {
        rows.truncate(limit);
        if outcome == SnapshotOutcome::Ready {
            outcome = SnapshotOutcome::Partial;
            reason = Some(format!("truncated to the requested limit of {limit}"));
        }
    }

    let rows: Vec<SnapshotRow> = rows
        .into_iter()
        .map(|row| SnapshotRow {
            fields: row
                .0
                .into_iter()
                .map(|mut field| {
                    if field.privacy == FieldPrivacy::LocalOnly && !local_view {
                        // The slot and the policy stay; only the value goes.
                        field.value = None;
                    }
                    field
                })
                .collect(),
        })
        .collect();

    SnapshotResponse {
        subsystem: provider.subsystem().to_string(),
        outcome,
        reason,
        returned: rows.len(),
        rows,
        selector: selector.map(str::to_string),
        view: if local_view { "local" } else { "remote" }.to_string(),
        elapsed_ms,
    }
}

static ARGS: &[ArgSpec] = &[
    ArgSpec::flag(
        "subsystem",
        "which subsystem to ask; comma-separated for several, omit for all",
        ArgKind::String,
    ),
    ArgSpec::flag(
        "id",
        "a selector whose meaning is the subsystem's, e.g. a task or context id",
        ArgKind::String,
    ),
    ArgSpec::flag(
        "limit",
        "the most rows to return per subsystem",
        ArgKind::Integer,
    ),
    ArgSpec::flag(
        "timeout-ms",
        "how long a single provider may take before its result is reported stale",
        ArgKind::Integer,
    ),
    ArgSpec::flag(
        "view",
        "remote is support-safe only; local additionally includes local-only fields",
        ArgKind::Enum(&["remote", "local"]),
    ),
    ArgSpec::flag(
        "list",
        "list the registered subsystems and return",
        ArgKind::Bool,
    ),
];

const DEFAULT_LIMIT: usize = 256;
const DEFAULT_TIMEOUT_MS: u64 = 250;

/// The `snapshot` command.
///
/// Public so an embedder that builds its own command surface, or a test, can
/// invoke it without going through the global registry.
#[derive(Debug)]
pub struct Snapshot;

impl Command for Snapshot {
    fn name(&self) -> &'static str {
        "snapshot"
    }

    fn short_description(&self) -> &'static str {
        "Shows live state — what exists now, rather than what just happened."
    }

    fn full_description(&self) -> &'static str {
        "Asks registered subsystems to describe their current state.\n\n\
This is the counterpart to reading logs, not a replacement for it. A log says what \
happened; this says what is there — live tasks, queue depths, routing tables. Rebuilding \
that from an event history is unreliable exactly when you need it: after records were \
dropped, or when the state you care about never emitted anything because it never changed.\n\n\
`--list` shows what is registered. With no `--subsystem`, every subsystem is asked.\n\n\
Each answer carries an outcome, and they are kept apart because they call for different \
things:\n\
  ready         everything the selector matched is here\n\
  partial       some of it; `reason` says what stopped it\n\
  unavailable   the subsystem is here but cannot answer now\n\
  not-compiled  it was compiled out of this build\n\
  busy          its state was locked; retry\n\
  panicked      the provider crashed — a bug in that subsystem\n\
  timed-out     it ran past its deadline; the result may be stale\n\n\
`--limit` and `--timeout-ms` are enforced by this command, not left to the provider: one \
that ignores them is truncated or reported rather than trusted.\n\n\
The default view is `remote`, which is support-safe only. `--view local` additionally \
shows local-only fields — values derived from the application's own data. A withheld \
field keeps its slot, so you can tell 'no such field' from 'not shown to you'."
    }

    fn args(&self) -> &'static [ArgSpec] {
        ARGS
    }

    fn execute(&self, args: Vec<String>) -> Result<Response, Response> {
        self.execute_with(args, &CommandContext::detached())
    }

    fn execute_with(
        &self,
        args: Vec<String>,
        context: &CommandContext,
    ) -> Result<Response, Response> {
        let parsed = ParsedArgs::parse(self.args(), args).map_err(Response::String)?;

        if parsed.boolean("list") {
            let listed: Vec<_> = registered()
                .into_iter()
                .map(|(name, description)| SnapshotRow {
                    fields: vec![
                        SnapshotField {
                            name: "subsystem".to_string(),
                            privacy: FieldPrivacy::SupportSafe,
                            value: Some(SnapshotValue::String(name.to_string())),
                        },
                        SnapshotField {
                            name: "description".to_string(),
                            privacy: FieldPrivacy::SupportSafe,
                            value: Some(SnapshotValue::String(description.to_string())),
                        },
                    ],
                })
                .collect();
            return Response::from_serialize(&SnapshotSet {
                snapshots: vec![SnapshotResponse {
                    subsystem: "registry".to_string(),
                    outcome: SnapshotOutcome::Ready,
                    reason: None,
                    returned: listed.len(),
                    rows: listed,
                    selector: None,
                    view: "remote".to_string(),
                    elapsed_ms: 0,
                }],
                unknown: Vec::new(),
            });
        }

        // Comma-separated rather than a repeated flag: `ParsedArgs` rejects a
        // non-variadic flag given twice, and variadic is only meaningful for a
        // positional parameter.
        let wanted: Vec<String> = parsed
            .get("subsystem")
            .map(|value| {
                value
                    .split(',')
                    .map(str::trim)
                    .filter(|name| !name.is_empty())
                    .map(str::to_string)
                    .collect()
            })
            .unwrap_or_default();
        let selector = parsed.get("id");
        let limit = parsed
            .integer("limit")
            .filter(|limit| *limit > 0)
            .map_or(DEFAULT_LIMIT, |limit| limit as usize);
        let timeout = Duration::from_millis(
            parsed
                .integer("timeout-ms")
                .filter(|ms| *ms > 0)
                .map_or(DEFAULT_TIMEOUT_MS, |ms| ms as u64),
        );
        let local_view = parsed.get("view") == Some("local");

        let providers = PROVIDERS.with_sync(|providers| providers.clone());
        let selected: Vec<_> = if wanted.is_empty() {
            providers
        } else {
            providers
                .into_iter()
                .filter(|provider| wanted.iter().any(|name| name == provider.subsystem()))
                .collect()
        };
        let unknown: Vec<String> = wanted
            .iter()
            .filter(|name| {
                !selected
                    .iter()
                    .any(|provider| provider.subsystem() == name.as_str())
            })
            .cloned()
            .collect();

        let cancelled = context.cancel_flag();
        let mut snapshots = Vec::with_capacity(selected.len());
        for provider in &selected {
            // A cancelled request stops between providers; within one, stopping
            // is the provider's own job via `should_stop`.
            context.check_cancelled()?;
            snapshots.push(run(
                provider, selector, limit, timeout, &cancelled, local_view,
            ));
        }

        Response::from_serialize(&SnapshotSet { snapshots, unknown })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use exfiltrate_internal::snapshot::SnapshotSet;
    use std::sync::atomic::AtomicU64;

    /// The registry is process-global, so cases take turns and each starts from
    /// an empty one.
    fn session() -> std::sync::MutexGuard<'static, ()> {
        static SERIALIZE: std::sync::Mutex<()> = std::sync::Mutex::new(());
        let guard = SERIALIZE
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        clear_providers();
        guard
    }

    fn ask(args: &[&str]) -> SnapshotSet {
        let response = Snapshot
            .execute(args.iter().map(|arg| arg.to_string()).collect())
            .expect("snapshot should not fail");
        match response {
            Response::Bytes(bytes) => rmp_serde::from_slice(&bytes).expect("decode"),
            other => panic!("expected a structured response, got {other:?}"),
        }
    }

    fn only(set: &SnapshotSet) -> &SnapshotResponse {
        assert_eq!(set.snapshots.len(), 1, "{set:?}");
        &set.snapshots[0]
    }

    struct Fixed {
        name: &'static str,
        result: fn(&SnapshotRequest<'_>) -> ProviderResult,
    }

    impl Provider for Fixed {
        fn subsystem(&self) -> &'static str {
            self.name
        }
        fn description(&self) -> &'static str {
            "a test provider"
        }
        fn snapshot(&self, request: &SnapshotRequest<'_>) -> ProviderResult {
            (self.result)(request)
        }
    }

    /// The five non-`Ready` outcomes stay distinct, because they call for
    /// different responses from the caller.
    #[wasm_lite::wasm_lite_test]
    fn the_outcomes_are_distinct() {
        let _session = session();
        add_provider(Fixed {
            name: "unavailable",
            result: |_| ProviderResult::Unavailable("no window yet".to_string()),
        });
        add_provider(Fixed {
            name: "notcompiled",
            result: |_| ProviderResult::NotCompiled,
        });
        add_provider(Fixed {
            name: "busy",
            result: |_| ProviderResult::Busy,
        });

        for (subsystem, expected) in [
            ("unavailable", SnapshotOutcome::Unavailable),
            ("notcompiled", SnapshotOutcome::NotCompiled),
            ("busy", SnapshotOutcome::Busy),
        ] {
            let set = ask(&["--subsystem", subsystem]);
            let answer = only(&set);
            assert_eq!(answer.outcome, expected, "{answer:?}");
            assert!(
                answer.reason.is_some(),
                "a non-ready outcome should say why: {answer:?}"
            );
        }
    }

    /// A provider that returns more than was asked for is truncated here, and
    /// the answer says so rather than looking complete.
    #[wasm_lite::wasm_lite_test]
    fn ignoring_the_limit_is_truncated_and_reported_partial() {
        let _session = session();
        add_provider(Fixed {
            name: "greedy",
            result: |_| {
                ProviderResult::Rows((0..50).map(|n| Row::new().support("n", n as u64)).collect())
            },
        });

        let set = ask(&["--subsystem", "greedy", "--limit", "5"]);
        let answer = only(&set);
        assert_eq!(answer.returned, 5);
        assert_eq!(answer.outcome, SnapshotOutcome::Partial);
        assert!(answer.reason.as_deref().unwrap().contains("limit"));
    }

    /// A provider that reports its own truncation keeps its reason.
    #[wasm_lite::wasm_lite_test]
    fn a_self_reported_partial_keeps_its_reason() {
        let _session = session();
        add_provider(Fixed {
            name: "partial",
            result: |_| {
                ProviderResult::Partial(
                    vec![Row::new().support("n", 1u64)],
                    "one shard was locked".to_string(),
                )
            },
        });
        let set = ask(&["--subsystem", "partial"]);
        let answer = only(&set);
        assert_eq!(answer.outcome, SnapshotOutcome::Partial);
        assert_eq!(answer.reason.as_deref(), Some("one shard was locked"));
    }

    /// A provider that ignores the deadline is reported, not trusted. Nothing
    /// can preempt it, so the check is after the fact.
    #[wasm_lite::wasm_lite_test]
    fn ignoring_the_deadline_is_reported_timed_out() {
        let _session = session();
        add_provider(Fixed {
            name: "slow",
            result: |_request| {
                let start = Instant::now();
                while start.elapsed() < Duration::from_millis(30) {
                    std::hint::spin_loop();
                }
                ProviderResult::Rows(vec![Row::new().support("n", 1u64)])
            },
        });

        let set = ask(&["--subsystem", "slow", "--timeout-ms", "5"]);
        let answer = only(&set);
        assert_eq!(answer.outcome, SnapshotOutcome::TimedOut, "{answer:?}");
        assert!(answer.reason.as_deref().unwrap().contains("deadline"));
    }

    /// A well-behaved provider sees `should_stop` go true and stops itself.
    #[wasm_lite::wasm_lite_test]
    fn a_cooperative_provider_observes_the_deadline() {
        let _session = session();
        add_provider(Fixed {
            name: "cooperative",
            result: |request| {
                let mut rows = Vec::new();
                loop {
                    if request.should_stop() {
                        return ProviderResult::Partial(rows, "deadline".to_string());
                    }
                    rows.push(Row::new().support("n", rows.len() as u64));
                    // A safety net only: 5ms cannot reach this, so reaching it
                    // would mean `should_stop` is broken.
                    if rows.len() > 5_000_000 {
                        return ProviderResult::Rows(rows);
                    }
                }
            },
        });

        let set = ask(&[
            "--subsystem",
            "cooperative",
            "--timeout-ms",
            "5",
            "--limit",
            "100000",
        ]);
        let answer = only(&set);
        assert_eq!(answer.outcome, SnapshotOutcome::Partial, "{answer:?}");
        assert_eq!(answer.reason.as_deref(), Some("deadline"));
    }

    /// The whole privacy contract: a remote view is withheld the local-only
    /// value, but the field keeps its slot and its policy so the caller can
    /// tell "no such field" from "not shown to you".
    #[wasm_lite::wasm_lite_test]
    fn local_only_fields_are_withheld_from_a_remote_view() {
        let _session = session();
        add_provider(Fixed {
            name: "mixed",
            result: |_| {
                ProviderResult::Rows(vec![
                    Row::new()
                        .support("depth", 3u64)
                        .local("label", "user typed this"),
                ])
            },
        });

        let remote = ask(&["--subsystem", "mixed"]);
        let row = &only(&remote).rows[0];
        assert_eq!(only(&remote).view, "remote");
        assert_eq!(row.get("depth").unwrap().value, Some(SnapshotValue::U64(3)));
        let label = row.get("label").expect("the field keeps its slot");
        assert_eq!(label.privacy, FieldPrivacy::LocalOnly);
        assert_eq!(label.value, None, "withheld from a remote view");

        let local = ask(&["--subsystem", "mixed", "--view", "local"]);
        let row = &only(&local).rows[0];
        assert_eq!(
            row.get("label").unwrap().value,
            Some(SnapshotValue::String("user typed this".to_string()))
        );
    }

    /// A subsystem nobody registered is named back, rather than silently
    /// producing an empty answer that looks like "nothing there".
    #[wasm_lite::wasm_lite_test]
    fn an_unknown_subsystem_is_reported() {
        let _session = session();
        add_provider(Fixed {
            name: "known",
            result: |_| ProviderResult::Rows(Vec::new()),
        });
        let set = ask(&["--subsystem", "known,nope"]);
        assert_eq!(set.unknown, vec!["nope".to_string()]);
        assert_eq!(set.snapshots.len(), 1);
    }

    /// Re-registering a subsystem replaces it, so a component that
    /// re-initialises does not accumulate duplicates.
    #[wasm_lite::wasm_lite_test]
    fn registering_twice_replaces_rather_than_duplicates() {
        let _session = session();
        add_provider(Fixed {
            name: "dup",
            result: |_| ProviderResult::Rows(vec![Row::new().support("v", 1u64)]),
        });
        add_provider(Fixed {
            name: "dup",
            result: |_| ProviderResult::Rows(vec![Row::new().support("v", 2u64)]),
        });
        assert_eq!(registered().len(), 1);
        let set = ask(&["--subsystem", "dup"]);
        assert_eq!(
            only(&set).rows[0].get("v").unwrap().value,
            Some(SnapshotValue::U64(2))
        );
    }

    /// The selector is passed through to the provider unchanged; its meaning is
    /// the subsystem's.
    #[wasm_lite::wasm_lite_test]
    fn the_selector_reaches_the_provider_and_is_echoed() {
        let _session = session();
        add_provider(Fixed {
            name: "select",
            result: |request| {
                ProviderResult::Rows(vec![
                    Row::new().support("saw", request.selector().unwrap_or("<none>")),
                ])
            },
        });
        let set = ask(&["--subsystem", "select", "--id", "task-7"]);
        let answer = only(&set);
        assert_eq!(answer.selector.as_deref(), Some("task-7"));
        assert_eq!(
            answer.rows[0].get("saw").unwrap().value,
            Some(SnapshotValue::String("task-7".to_string()))
        );
    }

    /// State changing underneath a provider yields a coherent bounded answer
    /// rather than a torn one: the provider reads a consistent value per row and
    /// the registry bounds the result.
    ///
    /// `(worker)` because it joins the writer, and a blocking join is
    /// `Atomics.wait`, which traps on the browser main thread.
    #[wasm_lite::wasm_lite_test(worker)]
    fn concurrent_mutation_still_yields_a_coherent_answer() {
        static COUNTER: AtomicU64 = AtomicU64::new(0);
        let _session = session();
        COUNTER.store(0, Ordering::SeqCst);

        add_provider(Fixed {
            name: "mutating",
            result: |request| {
                let mut rows = Vec::new();
                for _ in 0..64 {
                    if request.should_stop() {
                        return ProviderResult::Partial(rows, "stopped".to_string());
                    }
                    // Each row observes one atomic read; the value may differ
                    // between rows, which is what a live snapshot means.
                    rows.push(Row::new().support("seen", COUNTER.load(Ordering::SeqCst)));
                }
                ProviderResult::Rows(rows)
            },
        });

        // `std::thread::spawn` is unsupported on wasm32; threads go through
        // wasm_lite_std there.
        #[cfg(not(target_arch = "wasm32"))]
        let writer = std::thread::spawn(|| {
            for _ in 0..10_000 {
                COUNTER.fetch_add(1, Ordering::SeqCst);
            }
        });
        #[cfg(target_arch = "wasm32")]
        let writer = wasm_lite_std::spawn(|| {
            for _ in 0..10_000 {
                COUNTER.fetch_add(1, Ordering::SeqCst);
            }
        });
        let set = ask(&["--subsystem", "mutating", "--limit", "64"]);
        writer.join().expect("writer");

        let answer = only(&set);
        assert!(answer.returned <= 64);
        for row in &answer.rows {
            // Every row is well-formed: the field is present and is a number.
            assert!(matches!(
                row.get("seen").unwrap().value,
                Some(SnapshotValue::U64(_))
            ));
        }
    }

    /// `--list` names what is registered without running anything.
    #[wasm_lite::wasm_lite_test]
    fn list_names_the_registered_subsystems() {
        let _session = session();
        add_provider(Fixed {
            name: "listed",
            result: |_| panic!("--list must not run providers"),
        });
        let set = ask(&["--list"]);
        let answer = only(&set);
        assert_eq!(answer.subsystem, "registry");
        assert!(
            answer
                .rows
                .iter()
                .any(|row| row.get("subsystem").unwrap().value
                    == Some(SnapshotValue::String("listed".to_string()))),
            "{answer:?}"
        );
    }

    /// A cancelled request stops instead of asking every provider.
    #[wasm_lite::wasm_lite_test]
    fn a_cancelled_request_stops() {
        let _session = session();
        add_provider(Fixed {
            name: "never",
            result: |_| panic!("a cancelled request must not reach a provider"),
        });
        let context = CommandContext::new(Arc::new(AtomicBool::new(true)), Arc::new(|_| Ok(())));
        assert!(Snapshot.execute_with(Vec::new(), &context).is_err());
    }

    /// A provider that panics is reported as a bug in that subsystem, not as an
    /// empty result.
    ///
    /// Native only: the wasm32 profile builds std with `panic_abort`, so there
    /// is nothing to catch. That difference is stated in `run` rather than
    /// silently tolerated.
    #[cfg(not(target_arch = "wasm32"))]
    #[test]
    fn a_panicking_provider_is_reported_not_swallowed() {
        let _session = session();
        add_provider(Fixed {
            name: "boom",
            result: |_| panic!("deliberate provider panic"),
        });
        // The default hook would print the panic; quiet it for this one case.
        let previous = std::panic::take_hook();
        std::panic::set_hook(Box::new(|_| {}));
        let set = ask(&["--subsystem", "boom"]);
        std::panic::set_hook(previous);

        let answer = only(&set);
        assert_eq!(answer.outcome, SnapshotOutcome::Panicked);
        assert!(answer.reason.as_deref().unwrap().contains("bug"));
        assert!(answer.rows.is_empty());
    }
}