faculties 0.21.0

//! Reusable Orient observations and persistent or one-shot waiting.
//!
//! Facts, proof evidence and selected event IDs remain frozen while
//! exact payloads are acquired. Application evaluation time is explicit.
//! Condition scripts run only after acquisition,
//! once per evaluation. Presentation follows successful output acceptance.

#[path = "health.rs"]
mod health;
use health::HealthSources;
#[path = "receipt_import.rs"]
mod receipt_import;

#[derive(Clone, Debug)]
pub struct Orient {
    storage: Storage,
    health_max_age: Duration,
}

#[derive(Clone, Debug)]
pub struct ShowOptions {
    pub message_limit: usize,
    pub doing_limit: usize,
    pub todo_limit: usize,
    pub evaluate_habits: bool,
}
impl Default for ShowOptions {
    fn default() -> Self {
        Self {
            message_limit: 10,
            doing_limit: 5,
            todo_limit: 5,
            evaluate_habits: true,
        }
    }
}
#[derive(Clone, Debug)]
pub struct WakeOptions {
    pub chars: usize,
    pub doing_limit: usize,
    pub todo_limit: usize,
}
impl Default for WakeOptions {
    fn default() -> Self {
        Self {
            chars: 800_000,
            doing_limit: 5,
            todo_limit: 5,
        }
    }
}
#[derive(Clone, Debug)]
pub struct WaitOptions {
    pub timeout: Option<Duration>,
    pub poll_interval: Duration,
}
impl Default for WaitOptions {
    fn default() -> Self {
        Self {
            timeout: None,
            poll_interval: Duration::from_secs(1),
        }
    }
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct BaselineReceipt {
    pub persona: Id,
    pub events: usize,
}

impl Orient {
    pub fn new(pile: PathBuf, key: Option<PathBuf>) -> Self {
        Self::with_storage(Storage::new(pile, key))
    }
    pub fn with_storage(storage: Storage) -> Self {
        Self {
            storage,
            health_max_age: crate::schemas::swarm_health::DEFAULT_MAX_AGE,
        }
    }
    /// Choose how long this reader treats the latest health observation as current.
    /// This does not change the report facts or the reporting daemon.
    pub fn with_health_max_age(mut self, max_age: Duration) -> Self {
        self.health_max_age = max_age;
        self
    }
    /// Situational overview. Evaluates stored Habit conditions: trusted local execution.
    pub fn show(
        &self,
        persona: Option<&str>,
        options: &ShowOptions,
        out: &mut Out<'_>,
    ) -> Result<()> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(cmd_show(
                pile,
                signer,
                self.storage.path(),
                persona,
                options.message_limit,
                options.doing_limit,
                options.todo_limit,
                options.evaluate_habits,
                self.health_max_age,
                out,
            ))
        })
    }
    pub fn wake(
        &self,
        persona: Option<&str>,
        options: &WakeOptions,
        out: &mut Out<'_>,
    ) -> Result<()> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(cmd_wake(
                pile,
                signer,
                persona,
                options.chars,
                options.doing_limit,
                options.todo_limit,
                out,
            ))
        })
    }
    pub fn poll(&self, persona: &str, peek: bool, out: &mut Out<'_>) -> Result<()> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(cmd_poll(
                pile,
                signer,
                Some(persona),
                peek,
                self.health_max_age,
                out,
            ))
        })
    }
    pub fn baseline(&self, persona: &str) -> Result<BaselineReceipt> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(cmd_baseline(
                pile,
                signer,
                Some(persona),
                self.health_max_age,
            ))
        })
    }
    /// Explicit additive import of one legacy persona's resident receipts into
    /// this signing zooid's private source. It never baselines unseen events.
    pub fn import_receipts(&self, legacy_persona: &str) -> Result<usize> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(receipt_import::import(pile, signer, legacy_persona))
        })
    }
    /// One-shot wait. It returns after the first complete news report or timeout.
    /// No permanent process or competing observer is started by constructing Orient.
    pub fn wait(&self, persona: &str, options: &WaitOptions, out: &mut Out<'_>) -> Result<()> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(cmd_wait(
                pile,
                signer,
                self.storage.path(),
                Some(persona),
                options,
                self.health_max_age,
                out,
            ))
        })
    }

    /// Keep one store, input selection and Habit transition baseline alive
    /// across accepted reports. The output sink is the sole delivery boundary:
    /// an error ends the daemon before that report's receipts are committed.
    /// SIGINT/SIGTERM or the optional monotonic timeout ends observation and
    /// returns through the ordinary checked storage-close path. A synchronous
    /// output sink must bound its own execution time.
    pub fn daemon(&self, persona: &str, options: &WaitOptions, out: &mut Out<'_>) -> Result<()> {
        self.storage.with_store(|pile, signer, runtime| {
            runtime.block_on(async {
                #[cfg(unix)]
                let mut terminate =
                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
                let stop = async {
                    #[cfg(unix)]
                    tokio::select! {
                        result = tokio::signal::ctrl_c() => result?,
                        _ = terminate.recv() => {},
                    }
                    #[cfg(not(unix))]
                    tokio::signal::ctrl_c().await?;
                    Ok::<(), anyhow::Error>(())
                };
                tokio::select! {
                    result = stop => result,
                    result = cmd_observe(
                        pile, signer, self.storage.path(), Some(persona), options,
                        self.health_max_age, true, out,
                    ) => result,
                }
            })
        })
    }
}

use crate::collection_names::{configured_handle, open_configured, open_exact_in};
use crate::memory_cover::{render_cover_report, CoverOpts};
use crate::out::Out;
use crate::schemas::archive::archive;
use crate::schemas::compass::DEFAULT_SCOPE_ID as COMPASS_SCOPE_ID;
use crate::schemas::compass::{board, KIND_GOAL_ID, KIND_NOTE_ID, KIND_STATUS_ID};
use crate::schemas::discord::{discord, DEFAULT_SCOPE_ID as DISCORD_SCOPE_ID};
use crate::schemas::files::file;
use crate::schemas::habit::DEFAULT_SCOPE_ID as HABIT_SCOPE_ID;
use crate::schemas::habit::{
    attrs as habit_attrs, KIND_DONE_ID as KIND_HABIT_DONE_ID, KIND_HABIT_ID,
    KIND_STATE_ID as KIND_HABIT_STATE_ID, STATE_ACTIVE, STATE_PAUSED,
};
use crate::schemas::mail::DEFAULT_SCOPE_ID as MAIL_SCOPE_ID;
use crate::schemas::mail::{
    imported as imported_mail, observation as mail_observation, projection as mail_projection,
    read as mail_read, IMPORT_RECEIVED, KIND_IMPORTED_OBSERVATION, KIND_PARSED_PROJECTION,
    KIND_POP_OBSERVATION, KIND_READ_OBSERVATION, RECIPE_RFC5322_V1,
};
use crate::schemas::memory::DEFAULT_SCOPE_ID as MEMORY_SCOPE_ID;
use crate::schemas::message::{
    local as local_message, DEFAULT_SCOPE_ID as MESSAGE_SCOPE_ID, KIND_MESSAGE_ID, KIND_READ_ID,
};
use crate::schemas::orient::{presentation, KIND_PRESENTED};
use crate::schemas::relations::{
    group as relation_group, identity as relation_identity, lifecycle as relation_lifecycle,
    profile as relation_profile, DEFAULT_SCOPE_ID as RELATIONS_SCOPE_ID, KIND_GROUP,
    KIND_GROUP_SNAPSHOT, KIND_IDENTITY_VERDICT, KIND_PERSON_ID, KIND_PERSON_LIFECYCLE,
    KIND_PERSON_PROFILE,
};
use crate::schemas::status::DEFAULT_SCOPE_ID as STATUS_SCOPE_ID;
use crate::schemas::status::{status as window_status, KIND_STATUS_UPDATE};
use crate::schemas::teams::{teams, DEFAULT_SCOPE_ID as TEAMS_SCOPE_ID};
use crate::schemas::wiki::DEFAULT_SCOPE_ID as WIKI_SCOPE_ID;
use crate::storage::FacultySnapshot;
#[cfg(test)]
use crate::storage::{open_store, open_store_as, runtime};
use crate::storage::{read, FactArchive, FacultyStore, Storage};
use crate::storage::FactView;
use crate::{
    clock, compass, discord as discord_model, habits, mail as mail_model, message,
    orient as orient_model, relations, status, teams as teams_model, wiki as wiki_model,
};
use anybytes::{Bytes, View};
use anyhow::{anyhow, bail, Context, Result};
use ed25519_dalek::SigningKey;
use hifitime::Epoch;
use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use triblespace::core::blob::encodings::simplearchive::SimpleArchive;
use triblespace::core::blob::encodings::succinctarchive::{
    Rank9AcceleratedSuccinctArchiveBlob, SuccinctArchiveBlob,
};
use triblespace::core::collection::lww_register::{LwwIndex, LwwQuery, LwwRegisterBlob};
use triblespace::core::collection::observed_store::{DependencyTracker, ObservedStore};
#[cfg(test)]
use triblespace::core::collection::Support;
use triblespace::core::collection::{
    AttachedSnapshot, Collection, CollectionEncoding, CollectionRealizationError,
    CollectionSnapshotExt, CollectionStoreExt, Cover,
};
use triblespace::core::metadata;
use triblespace::core::query::TriblePattern;
use triblespace::core::repo::{
    BlobStoreGet, BlobStoreList, CapabilityProofRead, MissingBlob, StoreChanges, StoreSnapshot,
};
use triblespace::macros::{find, pattern};
use triblespace::prelude::*;

type IntervalValue = Inline<inlineencodings::NsTAIInterval>;

// Temporary, opt-in diagnostics. These retain only existing cover roots and
// report to stderr; they never choose a view or change a retry decision.
fn trace_refresh_enabled() -> bool {
    static ENABLED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
    *ENABLED.get_or_init(|| std::env::var_os("ORIENT_TRACE_REFRESH").is_some_and(|v| v == "1"))
}

fn trace_refresh_line(fields: std::fmt::Arguments<'_>) {
    use std::io::Write;
    // A closed diagnostic pipe must not turn a successful observation into a
    // failed command (or interrupt receipt publication).
    let _ = writeln!(std::io::stderr().lock(), "ORIENT_TRACE_REFRESH {fields}");
}

fn trace_refresh_call<T, E>(
    target: &'static str,
    stage: &'static str,
    operation: impl FnOnce() -> std::result::Result<T, E>,
) -> std::result::Result<T, E> {
    if !trace_refresh_enabled() {
        return operation();
    }
    let started = Instant::now();
    let result = operation();
    // Attachment/view/query stages are synchronous. In one `wait` command
    // they occur in order inside the enclosing refresh begin/end pair, before
    // ordinary payload acquisition can yield to another health poll.
    trace_refresh_line(format_args!(
        "event=stage target={target:?} stage={stage} elapsed_us={} ok={}",
        started.elapsed().as_micros(),
        result.is_ok(),
    ));
    result
}

struct RefreshProbe {
    id: u64,
    scope: &'static str,
    started: Instant,
}

impl RefreshProbe {
    fn begin(
        scope: &'static str,
        sampled: &FacultySnapshot,
        previous: Option<&FacultySnapshot>,
        pending: Option<&PendingWaitFrame>,
    ) -> Option<Self> {
        if !trace_refresh_enabled() {
            return None;
        }
        static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
        let probe = Self {
            id: NEXT.fetch_add(1, std::sync::atomic::Ordering::Relaxed),
            scope,
            started: Instant::now(),
        };
        let changes = previous.map(|previous| sampled.changes_since(previous));
        trace_refresh_line(format_args!(
            "event=begin refresh={} scope={} initial={} blobs={:?} records={:?} proofs={:?} wants={:?} pending={:?} exact_missing={}",
            probe.id,
            scope,
            previous.is_none(),
            changes.map(|c| c.contains(StoreChanges::BLOBS)),
            changes.map(|c| c.contains(StoreChanges::COLLECTION_RECORDS)),
            changes.map(|c| c.contains(StoreChanges::CAPABILITY_PROOFS)),
            changes.map(|c| c.contains(StoreChanges::WANTS)),
            pending.map(|p| p.reason),
            pending.is_some_and(|p| p.missing.is_some()),
        ));
        Some(probe)
    }

    fn finish(&self, outcome: &'static str) {
        trace_refresh_line(format_args!(
            "event=end refresh={} scope={} outcome={} elapsed_us={}",
            self.id,
            self.scope,
            outcome,
            self.started.elapsed().as_micros(),
        ));
    }

    fn cover<E: CollectionEncoding>(
        &self,
        target: &'static str,
        previous: Option<&Cover<E>>,
        cover: &Cover<E>,
    ) {
        trace_refresh_line(format_args!(
            "event=target refresh={} scope={} target={target:?} cover_equal={:?} cover_before={:?} cover_after={}",
            self.id,
            self.scope,
            previous.map(|old| old == cover),
            previous.map(|old| old.len()),
            cover.len(),
        ));
    }

    fn fact(&self, target: &'static str, previous: Option<&OrientFact>, current: &OrientFact) {
        self.cover(
            target,
            previous.map(|old| old.collection.cover()),
            current.collection.cover(),
        );
    }

    fn ordinary(&self, previous: Option<&OrientObservation>, current: &OrientObservation) {
        for (target, old, new) in [
            (
                "Message",
                previous.map(|p| &p.facts.messages),
                &current.facts.messages,
            ),
            ("Mail", previous.map(|p| &p.facts.mail), &current.facts.mail),
            (
                "Teams",
                previous.map(|p| &p.facts.teams),
                &current.facts.teams,
            ),
            (
                "Discord",
                previous.map(|p| &p.facts.discord),
                &current.facts.discord,
            ),
            (
                "Compass",
                previous.map(|p| &p.facts.compass),
                &current.facts.compass,
            ),
            (
                "Relations",
                previous.map(|p| &p.facts.relations),
                &current.facts.relations,
            ),
            (
                "Status",
                previous.map(|p| &p.facts.status),
                &current.facts.status,
            ),
        ] {
            self.fact(target, old, new);
        }
        if let Some(habits) = &current.facts.habits {
            self.fact(
                "Habit",
                previous.and_then(|p| p.facts.habits.as_ref()),
                habits,
            );
        }
        self.cover(
            "Orient receipts",
            previous.map(|p| p.facts.presentations.collection.cover()),
            current.facts.presentations.collection.cover(),
        );
        self.cover(
            "Compass status",
            previous.map(|p| p.compass_status_collection.cover()),
            current.compass_status_collection.cover(),
        );
    }

    fn frame(&self, previous: Option<&OrientObservation>, result: &Result<Option<WaitFrameLoad>>) {
        // Initial pending attempts have no comparison baseline. Later pending
        // frames compare to the last ready observation, not to one another;
        // never clone their prepared queries just for this diagnostic.
        trace_refresh_line(format_args!(
            "event=baseline refresh={} scope={} baseline={}",
            self.id,
            self.scope,
            if previous.is_some() {
                "last_ready"
            } else {
                "absent"
            },
        ));
        let outcome = match result {
            Ok(Some(WaitFrameLoad::Retained(_))) => "retained",
            Ok(Some(WaitFrameLoad::Ready(frame))) => {
                self.ordinary(previous, &frame.observation);
                "ready"
            }
            Ok(Some(WaitFrameLoad::Pending(pending))) => {
                if let Some(observation) = &pending.observation {
                    self.ordinary(previous, observation);
                }
                trace_refresh_line(format_args!(
                    "event=pending refresh={} scope={} reason={:?} observation={} exact_missing={}",
                    self.id,
                    self.scope,
                    pending.reason,
                    pending.observation.is_some(),
                    pending.missing.is_some(),
                ));
                "pending"
            }
            Ok(None) => "health_deadline",
            Err(_) => "error",
        };
        self.finish(outcome);
    }
}

fn interval_key(interval: IntervalValue) -> i128 {
    let (lower, _): (i128, i128) = interval.try_from_inline().unwrap();
    lower
}

fn epoch_seconds(epoch: Epoch) -> i64 {
    (epoch.to_tai_duration().total_nanoseconds() / 1_000_000_000) as i64
}

/// The earlier of two optional deadlines.
fn earliest(left: Option<Epoch>, right: Option<Epoch>) -> Option<Epoch> {
    match (left, right) {
        (Some(left), Some(right)) => Some(if left <= right { left } else { right }),
        (left, right) => left.or(right),
    }
}

/// The wall-clock second at which a cooling habit in `seen` can next fall due.
fn habit_deadline(seen: &Option<HabitObservation>) -> Option<Epoch> {
    seen.as_ref()
        .and_then(|seen| seen.next_cooldown_at)
        .map(|secs| Epoch::from_tai_seconds(secs as f64))
}

// Presentation history survives a temporarily unreadable initial frame in a
// daemon. Its clock does not: only currently readable Habit inputs may impose
// a deadline on a payload retry. One-shot waiting keeps its original policy.
fn pending_habit_deadline(
    continuous: bool,
    readable: Option<&HabitObservation>,
    seen: &Option<HabitObservation>,
) -> Option<Epoch> {
    if continuous {
        readable
            .and_then(|habits| habits.next_cooldown_at)
            .map(|secs| Epoch::from_tai_seconds(secs as f64))
    } else {
        habit_deadline(seen)
    }
}

fn invalidate_pending_habit_context(continuous: bool, seen: &mut Option<HabitObservation>) {
    if !continuous {
        *seen = None;
    }
}

fn format_age(now_key: i128, past_key: i128) -> String {
    let delta_ns = now_key.saturating_sub(past_key);
    let delta_s = (delta_ns / 1_000_000_000).max(0) as i64;
    if delta_s < 60 {
        format!("{delta_s}s")
    } else if delta_s < 60 * 60 {
        format!("{}m", delta_s / 60)
    } else if delta_s < 60 * 60 * 24 {
        format!("{}h", delta_s / (60 * 60))
    } else {
        format!("{}d", delta_s / (60 * 60 * 24))
    }
}

fn fmt_id(id: Id) -> String {
    format!("{id:x}")
}

fn entity_tags<P: TriblePattern>(space: &P, entity_id: Id) -> Vec<String> {
    let mut tags: Vec<String> =
        find!(tag: String, pattern!(space, [{ entity_id @ board::tag: ?tag }])).collect();
    tags.sort();
    tags.dedup();
    tags
}

fn visible_notes<P: TriblePattern>(
    space: &P,
    own_identities: &HashSet<Id>,
    attention_keys: &HashSet<String>,
    relevant_goals: &HashSet<Id>,
) -> BTreeMap<Id, Id> {
    let mut notes = BTreeMap::new();
    for (note_id, goal_id) in find!(
        (note_id: Id, goal_id: Id),
        pattern!(space, [
            {
                ?note_id @
                metadata::tag: &KIND_NOTE_ID,
                board::task: ?goal_id,
                board::note: _?body,
            },
            { ?goal_id @ metadata::tag: &KIND_GOAL_ID },
        ])
    ) {
        let own_note = exists!((by: Id), and!(
            own_identities.has(by),
            pattern!(space, [{ note_id @ board::by: ?by }])
        ));
        if own_note {
            continue;
        }
        let directly_addressed = entity_tags(space, note_id)
            .iter()
            .any(|tag| attention_keys.contains(&tag.to_ascii_lowercase()));
        if directly_addressed || relevant_goals.contains(&goal_id) {
            insert_note_goal(&mut notes, note_id, goal_id);
        }
    }
    notes
}

/// One Orient input's authored collection and explicit query projections.
#[derive(Clone, Copy)]
struct OrientSource {
    source: Collection<SimpleArchive>,
    succinct: Collection<SuccinctArchiveBlob>,
    rank9: Collection<Rank9AcceleratedSuccinctArchiveBlob>,
    label: &'static str,
}

impl OrientSource {
    async fn open(
        pile: &mut FacultyStore,
        signer: &SigningKey,
        scope: Id,
        label: &'static str,
    ) -> Result<Self> {
        let authority = signer.verifying_key();
        let source = if let Some(handle) = configured_handle(scope)? {
            let snapshot = pile.snapshot()?;
            read(pile, &snapshot, |reader| {
                open_exact_in(reader, scope, handle)
            })
            .await?
        } else {
            open_configured(pile, scope, authority)?
        };
        Self::register(pile, source, label)
    }

    fn register<S>(
        pile: &mut S,
        source: Collection<SimpleArchive>,
        label: &'static str,
    ) -> Result<Self>
    where
        S: CollectionStoreExt + SnapshotSource,
        S::Snapshot: BlobStoreGet + CapabilityProofRead,
    {
        let succinct = pile
            .attach::<SuccinctArchiveBlob>(source, ())
            .with_context(|| format!("register {label} Succinct collection"))?;
        let rank9 = pile
            .attach::<Rank9AcceleratedSuccinctArchiveBlob>(source, succinct)
            .with_context(|| format!("register {label} Rank9 collection"))?;
        Ok(Self {
            source,
            succinct,
            rank9,
            label,
        })
    }

    /// Carry the source and attach what the carry leaves, Succinct first,
    /// before selecting the immutable query views. Attached collections need
    /// no WRITE: the store is opened as this signer, whose MAPs are the ones
    /// believed. A foundation left unattached is read from its own bytes.
    async fn maintain(&self, pile: &mut FacultyStore, signer: &SigningKey) -> Result<()> {
        crate::storage::tolerate_own_lag(pile.maintain_attached(self.succinct, signer).await)
            .with_context(|| format!("maintain {} Succinct collection", self.label))?;
        crate::storage::tolerate_own_lag(pile.maintain_attached(self.rank9, signer).await)
            .with_context(|| format!("maintain {} Rank9 collection", self.label))?;
        Ok(())
    }

    fn observe(&self, snapshot: &FacultySnapshot) -> Result<OrientFact> {
        let collection = trace_refresh_call(self.label, "attach", || snapshot.attached(self.rank9))
            .with_context(|| format!("observe resident {} Rank9 projection", self.label))?;
        let view = trace_refresh_call(self.label, "view", || {
            crate::storage::attached_facts(&collection)
        })
        .with_context(|| format!("read resident {} Rank9 projection", self.label))?;
        Ok(OrientFact { collection, view })
    }
}

/// Receipt facts, projected the same way every other Orient input is.
/// The signing key is the observer's identity; routing aliases are not authority.
///
/// This used to derive an `EntityIdSetBlob` over `presentation::event`, and a
/// set of ids can answer exactly one question: is THIS id present. Every
/// caller therefore had to COMPUTE an id in order to ask -- which is the
/// hash-join the substrate rules forbid, and it forced a presented occurrence
/// to have a derived identity whether or not it had any business having one.
/// A habit's due event has none: it is an intention and an instant. Projecting
/// receipts as an ordinary Rank9 fact archive lets each caller JOIN on whatever
/// actually identifies its occurrence, so nothing has to be named to be found.
struct ReceiptSource {
    source: Collection<SimpleArchive>,
    succinct: Collection<SuccinctArchiveBlob>,
    rank9: Collection<Rank9AcceleratedSuccinctArchiveBlob>,
}

impl ReceiptSource {
    fn register<S: CollectionStoreExt>(pile: &mut S, signer: &SigningKey) -> Result<Self> {
        let policy = crate::collection_names::private_policy(signer.verifying_key());
        let source = pile.collection(
            crate::schemas::orient::RECEIPT_COLLECTION_NAME,
            policy.clone(),
        )?;
        let succinct = pile.attach::<SuccinctArchiveBlob>(source, ())?;
        let rank9 = pile.attach::<Rank9AcceleratedSuccinctArchiveBlob>(source, succinct)?;
        Ok(Self {
            source,
            succinct,
            rank9,
        })
    }

    fn observe(&self, snapshot: &FacultySnapshot) -> Result<ReceiptObservation> {
        let collection = trace_refresh_call("Orient receipts", "attach", || {
            snapshot.attached(self.rank9)
        })?;
        let view = trace_refresh_call("Orient receipts", "view", || {
            crate::storage::attached_facts(&collection)
        })?;
        Ok(ReceiptObservation { collection, view })
    }

    /// Attach this run's own receipts into the queryable projection. A run that
    /// is about to report does this before observing, so a re-armed run reads
    /// its receipts through the projection rather than from their bytes. The
    /// background maintainer attaches the same projection; this call only
    /// closes the window between the commit and that maintainer's next pass.
    /// Both the ordinary path and the resident-only health path call this,
    /// each with its own store, so the two cannot disagree about what counts
    /// as lag.
    async fn maintain<S>(&self, pile: &mut S, signer: &SigningKey) -> Result<()>
    where
        S: triblespace::core::repo::Store
            + triblespace::core::repo::async_store::AsyncBlobStoreAcquire
            + Send,
    {
        // Both, in order: Rank9 is built from the Succinct attachment of the
        // same node, so the Succinct one comes first.
        crate::storage::tolerate_own_lag(pile.maintain_attached(self.succinct, signer).await)
            .context("maintain Orient receipt Succinct collection")?;
        crate::storage::tolerate_own_lag(pile.maintain_attached(self.rank9, signer).await)
            .context("maintain Orient receipt Rank9 collection")?;
        Ok(())
    }
}

/// Refresh the receipt projection before the observation that decides what
/// to report. Best effort by construction: the projection makes the read exact,
/// it is not a precondition of it, so a failure is reported and the run goes on
/// — see `observe_snapshot`, where projection lag may repeat an event and never
/// blocks one.
async fn refresh_receipts_before_observation(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    sources: &OrientSources,
    output: &mut Out<'_>,
) -> Result<()> {
    if let Err(error) = sources.presentations.maintain(pile, signer).await {
        output.line(format!(
            "note: Orient receipt membership not refreshed ({error:#}); a recent event may repeat"
        ))?;
    }
    Ok(())
}

struct ReceiptObservation {
    collection: AttachedSnapshot<FacultySnapshot, Rank9AcceleratedSuccinctArchiveBlob>,
    view: FactArchive,
}

impl ReceiptObservation {
    fn view(&self) -> &FactArchive {
        &self.view
    }

    #[cfg(test)]
    fn contains(&self, event: Id) -> bool {
        event_presented(&self.view, event)
    }

    /// Every event id this signer holds a receipt for.
    #[cfg(test)]
    fn presented_events(&self) -> BTreeSet<Id> {
        find!(
            event: Id,
            pattern!(&self.view, [{ _?receipt @ presentation::event: ?event }])
        )
        .collect()
    }

    #[cfg(test)]
    fn is_empty(&self) -> bool {
        !exists!(pattern!(&self.view, [{ _?receipt @
            metadata::tag: &crate::schemas::orient::KIND_PRESENTED,
        }]))
    }

    fn is_current(&self, snapshot: &FacultySnapshot) -> bool {
        self.collection.is_current(snapshot)
    }
}

struct OrientSources {
    messages: OrientSource,
    mail: OrientSource,
    teams: OrientSource,
    discord: OrientSource,
    compass: OrientSource,
    relations: OrientSource,
    status: OrientSource,
    habits: Option<OrientSource>,
    presentations: ReceiptSource,
    compass_status: Collection<LwwRegisterBlob>,
    #[cfg(test)]
    observations: std::sync::atomic::AtomicUsize,
}

impl OrientSources {
    async fn open(
        pile: &mut FacultyStore,
        signer: &SigningKey,
        include_habits: bool,
    ) -> Result<Self> {
        let authority = signer.verifying_key();
        let messages = OrientSource::open(pile, signer, MESSAGE_SCOPE_ID, "Message").await?;
        let mail = OrientSource::open(pile, signer, MAIL_SCOPE_ID, "Mail").await?;
        let teams = OrientSource::open(pile, signer, TEAMS_SCOPE_ID, "Teams").await?;
        let discord = OrientSource::open(pile, signer, DISCORD_SCOPE_ID, "Discord").await?;
        let compass = OrientSource::open(pile, signer, COMPASS_SCOPE_ID, "Compass").await?;
        let relations = OrientSource::open(pile, signer, RELATIONS_SCOPE_ID, "Relations").await?;
        let status = OrientSource::open(pile, signer, STATUS_SCOPE_ID, "Status").await?;
        let habits = if include_habits {
            Some(OrientSource::open(pile, signer, HABIT_SCOPE_ID, "Habit").await?)
        } else {
            None
        };
        let presentations = ReceiptSource::register(pile, signer)?;
        let compass_status = compass::status_register_collection(pile, authority)?;
        Ok(Self {
            messages,
            mail,
            teams,
            discord,
            compass,
            relations,
            status,
            habits,
            presentations,
            compass_status,
            #[cfg(test)]
            observations: std::sync::atomic::AtomicUsize::new(0),
        })
    }
}

struct OrientFact {
    collection: AttachedSnapshot<FacultySnapshot, Rank9AcceleratedSuccinctArchiveBlob>,
    view: FactArchive,
}

impl OrientFact {
    #[cfg(test)]
    fn support(&self) -> &Support<SimpleArchive> {
        self.collection.support()
    }

    fn is_current(&self, snapshot: &FacultySnapshot) -> bool {
        self.collection.is_current(snapshot)
    }

    fn view(&self) -> &FactArchive {
        &self.view
    }
}

struct OrientFacts {
    messages: OrientFact,
    mail: OrientFact,
    teams: OrientFact,
    discord: OrientFact,
    compass: OrientFact,
    relations: OrientFact,
    status: OrientFact,
    habits: Option<OrientFact>,
    presentations: ReceiptObservation,
}

/// One coherent semantic observation. Each source stays in its own resident
/// target collection and Rank9 query view; shared vocabulary never turns those
/// authority boundaries into an accidental global fact union.
struct OrientObservation {
    /// Resident payload reader for the selected observation.
    /// Exact acquisition may advance its blob residency without
    /// changing any selected fact view, support, or authorization boundary.
    snapshot: FacultySnapshot,
    /// Every payload this observation renders -- message bodies, titles,
    /// subjects, habit scripts -- consulted through a reader that records the
    /// handle, present or not. A payload that lands after the envelope is then
    /// an arrival this observation depends on, and the sweep refreshes it.
    /// Read through the raw snapshot instead and it never would: the fact
    /// views only ever depend on records in their own lineage.
    payloads: DependencyTracker,
    facts: OrientFacts,
    compass_status: LwwQuery,
    compass_status_collection: AttachedSnapshot<FacultySnapshot, LwwRegisterBlob>,
}

impl OrientObservation {
    fn is_current(&self, snapshot: &FacultySnapshot) -> bool {
        [
            &self.facts.messages,
            &self.facts.mail,
            &self.facts.teams,
            &self.facts.discord,
            &self.facts.compass,
            &self.facts.relations,
            &self.facts.status,
        ]
        .into_iter()
        .all(|fact| fact.is_current(snapshot))
            && self
                .facts
                .habits
                .as_ref()
                .is_none_or(|fact| fact.is_current(snapshot))
            && self.facts.presentations.is_current(snapshot)
            && self.compass_status_collection.is_current(snapshot)
            && StoreSnapshot::changes_for(
                snapshot,
                &self.snapshot,
                &self
                    .payloads
                    .lock()
                    .expect("payload read-set is not poisoned"),
            ) == StoreChanges::NONE
    }

    fn query<'a>(&'a self, snapshot: &'a FacultySnapshot) -> OrientQuery<'a> {
        OrientQuery {
            messages: self.facts.messages.view(),
            mail: self.facts.mail.view(),
            teams: self.facts.teams.view(),
            discord: self.facts.discord.view(),
            compass: self.facts.compass.view(),
            relations: self.facts.relations.view(),
            status: self.facts.status.view(),
            habits: self.facts.habits.as_ref().map(OrientFact::view),
            presentations: self.facts.presentations.view(),
            compass_status: &self.compass_status,
            payloads: ObservedStore::with_tracker(
                snapshot.clone(),
                std::sync::Arc::clone(&self.payloads),
            ),
        }
    }
}

/// Bring authorized source projections forward before one coherent observation.
/// Receipt upkeep has its own best-effort boundary so a failure to refresh
/// presentation membership cannot suppress otherwise readable attention.
async fn maintain_inputs(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    sources: &OrientSources,
) -> Result<()> {
    for source in [
        Some(&sources.messages),
        Some(&sources.mail),
        Some(&sources.teams),
        Some(&sources.discord),
        Some(&sources.compass),
        Some(&sources.relations),
        Some(&sources.status),
        sources.habits.as_ref(),
    ]
    .into_iter()
    .flatten()
    {
        source.maintain(pile, signer).await?;
    }
    crate::storage::tolerate_own_lag(pile.maintain_attached(sources.compass_status, signer).await)
        .context("maintain Compass status register")?;
    Ok(())
}

#[cfg(test)]
async fn maintain_sources(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    sources: &OrientSources,
) -> Result<()> {
    maintain_inputs(pile, signer, sources).await?;
    sources.presentations.maintain(pile, signer).await
}

/// Read every target collection as it actually exists at one immutable store
/// boundary. Application evaluation time is supplied separately; this function
/// performs no writes and reads no clock.
#[cfg(test)]
fn observe_sources(
    snapshot: FacultySnapshot,
    sources: &OrientSources,
) -> Result<OrientObservation> {
    let messages = sources.messages.observe(&snapshot)?;
    let mail = sources.mail.observe(&snapshot)?;
    let teams = sources.teams.observe(&snapshot)?;
    let discord = sources.discord.observe(&snapshot)?;
    let compass = sources.compass.observe(&snapshot)?;
    let relations = sources.relations.observe(&snapshot)?;
    let status = sources.status.observe(&snapshot)?;
    let habits = sources
        .habits
        .as_ref()
        .map(|source| source.observe(&snapshot))
        .transpose()?;
    let presentations = sources.presentations.observe(&snapshot)?;
    // Positive known-winner membership is an ordinary relation over the
    // same foundations the facts read: a status no attachment reaches yet is
    // built in memory, so the register never answers for less than the facts.
    let status_collection = trace_refresh_call("Compass status", "attach", || {
        snapshot.attached(sources.compass_status)
    })
    .map_err(|error| anyhow!("observe Compass status register: {error}"))?;
    let status_index = trace_refresh_call("Compass status", "view", || {
        status_collection
            .read::<LwwIndex>()
            .map(|read| read.into_value())
    })
    .map_err(|error| anyhow!("read Compass status register: {error}"))?;
    let compass_status = trace_refresh_call("Compass status", "query", || status_index.query())
        .map_err(|error| anyhow!("prepare Compass status register query: {error}"))?;
    Ok(OrientObservation {
        snapshot,
        payloads: DependencyTracker::default(),
        facts: OrientFacts {
            messages,
            mail,
            teams,
            discord,
            compass,
            relations,
            status,
            habits,
            presentations,
        },
        compass_status,
        compass_status_collection: status_collection,
    })
}

/// Observe the resident targets at the caller's exact immutable boundary.
/// Receipt projection lag may repeat an event; it never blocks an observation.
#[cfg(test)]
fn observe_snapshot(
    snapshot: FacultySnapshot,
    sources: &OrientSources,
) -> Result<OrientObservation> {
    #[cfg(test)]
    sources
        .observations
        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    observe_sources(snapshot, sources)
}

#[cfg(test)]
fn observe_current_sources(
    pile: &mut FacultyStore,
    sources: &OrientSources,
) -> Result<OrientObservation> {
    let snapshot = pile
        .snapshot()
        .map_err(|error| anyhow!("freeze shared Orient native store snapshot: {error}"))?;
    observe_snapshot(snapshot, sources)
}

/// Select foreground observations against frozen records and proof evidence,
/// acquiring exact authority bytes before the coverage fold. Only the reader
/// wrapper is stripped afterward: selected cover/support/dependencies survive
/// unchanged, and later arrival cannot introduce another record or proof.
async fn observe_snapshot_acquiring(
    snapshot: FacultySnapshot,
    sources: &OrientSources,
    receipts: bool,
) -> Result<OrientObservation> {
    #[cfg(test)]
    sources.observations.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
    let fact_sources = [
        sources.messages, sources.mail, sources.teams, sources.discord,
        sources.compass, sources.relations, sources.status,
    ];
    let habits = sources.habits;
    let receipts_target = sources.presentations.rank9;
    let status_target = sources.compass_status;
    let runtime = tokio::runtime::Handle::current();
    tokio::task::spawn_blocking(move || {
        let reader = crate::storage::AcquiringReader::with_handle(snapshot.clone(), runtime);
        let observe = |source: OrientSource| -> Result<OrientFact> {
            let collection = reader.attached_acquiring(source.rank9)?;
            let view = crate::storage::acquire_attached_facts(&collection)?;
            Ok(OrientFact { collection: collection.into_frozen(), view })
        };
        let [messages, mail, teams, discord, compass, relations, status] = fact_sources;
        let presentations = reader.attached(receipts_target)?.into_frozen();
        let presentation_view = if receipts {
            // Receipt history is advisory: absence can repeat an event but
            // cannot suppress readable attention behind a network wait. Read
            // these already-selected coordinates resident-only, as before.
            presentations.facts()?
        } else {
            // Wake presents an overview, not unseen-news membership. Do not
            // demand historical receipt payloads it will never consult.
            FactArchive::new(vec![
                triblespace::core::blob::encodings::succinctarchive::SuccinctArchive::from(
                    &TribleSet::new(),
                ),
            ])
        };
        let status_collection = reader.attached_acquiring(status_target)?;
        let status_index = crate::storage::require_complete_attached_read(
            status_collection.read_acquiring::<LwwIndex>()?,
        )?;
        Ok(OrientObservation {
            snapshot,
            payloads: DependencyTracker::default(),
            facts: OrientFacts {
                messages: observe(messages)?, mail: observe(mail)?, teams: observe(teams)?,
                discord: observe(discord)?, compass: observe(compass)?,
                relations: observe(relations)?, status: observe(status)?,
                habits: habits.map(observe).transpose()?,
                presentations: ReceiptObservation {
                    collection: presentations, view: presentation_view,
                },
            },
            compass_status: status_index.query()?,
            compass_status_collection: status_collection.into_frozen(),
        })
    }).await.context("join frozen Orient input selection")?
}

async fn observe_current_sources_acquiring(
    pile: &mut FacultyStore,
    sources: &OrientSources,
) -> Result<OrientObservation> {
    observe_snapshot_acquiring(pile.snapshot()?, sources, true).await
}

/// Borrowed inputs for one declarative Orient query.
///
/// Every source remains a separately admitted, maintained Succinct relation.
/// Cross-source joins are explicit at the query site.
struct OrientQuery<'a> {
    messages: &'a FactArchive,
    mail: &'a FactArchive,
    teams: &'a FactArchive,
    discord: &'a FactArchive,
    compass: &'a FactArchive,
    relations: &'a FactArchive,
    status: &'a FactArchive,
    habits: Option<&'a FactArchive>,
    presentations: &'a FactArchive,
    compass_status: &'a LwwQuery,
    /// The reader every payload is read through: the current snapshot, with
    /// each consultation charged to the observation's read-set.
    payloads: ObservedStore<FacultySnapshot>,
}

fn is_payload_pending(error: &anyhow::Error) -> bool {
    error
        .chain()
        .any(|source| source.downcast_ref::<MissingBlob>().is_some())
}

/// An unavailable observation may be retried; authority, malformed descriptors,
/// contradictory equations, and other storage failures are not availability.
/// A blob to wait for (`MissingDependency`, or a root that cannot yet stand
/// on its admitted commits) is pending. An own foundation left without a leaf
/// (`Unmappable`) is not: upkeep treats it as lag and never returns it
/// ([`crate::storage::tolerate_own_lag`]), and waiting would not change it.
fn is_preparation_pending(error: &anyhow::Error) -> bool {
    error.chain().any(|source| {
        source.downcast_ref::<MissingBlob>().is_some()
            // Typed incomplete support includes unavailable dependencies and
            // a mapping's bounded inability to represent a foundation. It is
            // pending evidence, not an invented exact missing-blob claim.
            || source.downcast_ref::<crate::storage::IncompleteAttachedRead>().is_some()
            || matches!(
                source.downcast_ref::<CollectionRealizationError>(),
                Some(
                    CollectionRealizationError::IncompleteCover { .. }
                        | CollectionRealizationError::MissingDependency { .. }
                )
            )
    })
}

fn read_utf8<R: BlobStoreList + BlobStoreGet>(
    reader: &R,
    handle: Inline<inlineencodings::Handle<blobencodings::UTF8String>>,
    label: &str,
) -> Result<String> {
    let unknown = Inline::<inlineencodings::Handle<blobencodings::UnknownBlob>>::new(handle.raw);
    if !reader
        .contains_blob(unknown)
        .map_err(|error| anyhow!("inspect {label} residency: {error}"))?
    {
        return Err(MissingBlob { handle: unknown }.into());
    }
    let value: View<str> = BlobStoreGet::get(reader, handle)
        .with_context(|| format!("read {label} payload {}", hex::encode(handle.raw)))?;
    Ok(value.to_string())
}

fn read_bytes<R: BlobStoreList + BlobStoreGet>(
    reader: &R,
    handle: Inline<inlineencodings::Handle<blobencodings::RawBytes>>,
    label: &str,
) -> Result<Vec<u8>> {
    let unknown = Inline::<inlineencodings::Handle<blobencodings::UnknownBlob>>::new(handle.raw);
    if !reader
        .contains_blob(unknown)
        .map_err(|error| anyhow!("inspect {label} residency: {error}"))?
    {
        return Err(MissingBlob { handle: unknown }.into());
    }
    let value: Bytes = BlobStoreGet::get(reader, handle)
        .with_context(|| format!("read {label} payload {}", hex::encode(handle.raw)))?;
    Ok(value.to_vec())
}

fn ids_of_kind<P: TriblePattern>(space: &P, kind: Id) -> BTreeSet<Id> {
    find!(id: Id, pattern!(space, [{ ?id @ metadata::tag: &kind }])).collect()
}

fn track_heads<P: TriblePattern>(
    space: &P,
    kind: Id,
    owner_attribute: &Attribute<inlineencodings::GenId>,
    owner: Id,
) -> Vec<Id> {
    let members: BTreeSet<Id> = find!(
        id: Id,
        pattern!(space, [{ ?id @
            metadata::tag: &kind,
            owner_attribute: &owner,
        }])
    )
    .collect();
    let superseded: BTreeSet<Id> = find!(
        old: Id,
        pattern!(space, [{ _?new @
            metadata::tag: &kind,
            owner_attribute: &owner,
            metadata::supersedes: ?old,
        }])
    )
    .collect();
    members.difference(&superseded).copied().collect()
}

fn person_anchors<P: TriblePattern>(space: &P) -> BTreeSet<Id> {
    ids_of_kind(space, KIND_PERSON_ID)
}

fn group_anchors<P: TriblePattern>(space: &P) -> BTreeSet<Id> {
    ids_of_kind(space, KIND_GROUP)
}

fn profile_heads<P: TriblePattern>(space: &P, person: Id) -> Vec<Id> {
    track_heads(space, KIND_PERSON_PROFILE, &relation_profile::of, person)
}

fn profile_lookup_handles<P: TriblePattern>(space: &P, person: Id) -> Vec<relations::TextHandle> {
    let mut handles = BTreeSet::new();
    for head in profile_heads(space, person) {
        handles.extend(find!(
            value: relations::TextHandle,
            pattern!(space, [{ head @ metadata::name: ?value }])
        ));
        handles.extend(find!(
            value: relations::TextHandle,
            pattern!(space, [{ head @ relation_profile::alias: ?value }])
        ));
    }
    handles.into_iter().collect()
}

#[derive(Default)]
struct IdentityIndex {
    parent: HashMap<Id, Id>,
    contradictory: BTreeSet<Id>,
    unsettled: BTreeSet<(Id, Id)>,
}

fn identity_root(parent: &HashMap<Id, Id>, mut id: Id) -> Id {
    while let Some(next) = parent.get(&id).copied() {
        if next == id {
            break;
        }
        id = next;
    }
    id
}

fn union_identity(parent: &mut HashMap<Id, Id>, left: Id, right: Id) {
    let left = identity_root(parent, left);
    let right = identity_root(parent, right);
    if left == right {
        return;
    }
    let (low, high) = if left < right {
        (left, right)
    } else {
        (right, left)
    };
    parent.insert(low, low);
    parent.insert(high, low);
}

impl IdentityIndex {
    fn from_relations<P: TriblePattern>(space: &P) -> Self {
        let mut index = Self::default();
        for person in person_anchors(space) {
            index.parent.insert(person, person);
        }

        let rows: Vec<(Id, Id, Id, bool)> = find!(
            (event: Id, low: Id, high: Id, same: bool),
            pattern!(space, [{ ?event @
                metadata::tag: &KIND_IDENTITY_VERDICT,
                relation_identity::low: ?low,
                relation_identity::high: ?high,
                relation_identity::same: ?same,
            }])
        )
        .collect();
        let superseded: BTreeSet<Id> = find!(
            old: Id,
            pattern!(space, [{ _?event @
                metadata::tag: &KIND_IDENTITY_VERDICT,
                metadata::supersedes: ?old,
            }])
        )
        .collect();
        let mut pairs = BTreeMap::<(Id, Id), BTreeSet<bool>>::new();
        for (event, left, right, same) in rows {
            let (low, high) = if left < right {
                (left, right)
            } else {
                (right, left)
            };
            index.parent.entry(low).or_insert(low);
            index.parent.entry(high).or_insert(high);
            if !superseded.contains(&event) {
                pairs.entry((low, high)).or_default().insert(same);
            }
        }

        for (&(low, high), values) in &pairs {
            if values.len() == 1 && values.contains(&true) {
                union_identity(&mut index.parent, low, high);
            } else if values.len() > 1 {
                index.unsettled.insert((low, high));
            }
        }
        for (&(low, high), values) in &pairs {
            if values.len() == 1 && values.contains(&false) {
                let low = identity_root(&index.parent, low);
                let high = identity_root(&index.parent, high);
                if low == high {
                    index.contradictory.insert(low);
                }
            }
        }
        index
    }

    fn equivalent(&self, left: Id, right: Id) -> Result<bool> {
        if left == right {
            return Ok(true);
        }
        let left = identity_root(&self.parent, left);
        let right = identity_root(&self.parent, right);
        if self.contradictory.contains(&left) || self.contradictory.contains(&right) {
            bail!("identity comparison touches a contradictory component");
        }
        let roots = if left < right {
            (left, right)
        } else {
            (right, left)
        };
        if self.unsettled.iter().any(|(low, high)| {
            let low = identity_root(&self.parent, *low);
            let high = identity_root(&self.parent, *high);
            (low.min(high), low.max(high)) == roots
        }) {
            bail!("identity comparison is unsettled by a forked verdict");
        }
        Ok(left == right)
    }

    fn component(&self, person: Id) -> Result<BTreeSet<Id>> {
        let root = identity_root(&self.parent, person);
        if self.contradictory.contains(&root) {
            bail!("identity component containing {person:x} is contradictory");
        }
        let mut component: BTreeSet<Id> = self
            .parent
            .keys()
            .copied()
            .filter(|candidate| identity_root(&self.parent, *candidate) == root)
            .collect();
        component.insert(person);
        Ok(component)
    }
}

fn lifecycle_retired<P: TriblePattern>(space: &P, person: Id) -> Result<Option<bool>> {
    let heads = track_heads(
        space,
        KIND_PERSON_LIFECYCLE,
        &relation_lifecycle::of,
        person,
    );
    if heads.is_empty() {
        return Ok(Some(false));
    }
    let values: BTreeSet<bool> = heads
        .into_iter()
        .flat_map(|head| {
            find!(
                value: bool,
                pattern!(space, [{ head @ relation_lifecycle::retired: ?value }])
            )
        })
        .collect();
    Ok((values.len() == 1).then(|| *values.first().expect("one lifecycle value")))
}

fn group_head_ids<P: TriblePattern>(space: &P, group: Id) -> Vec<Id> {
    track_heads(
        space,
        KIND_GROUP_SNAPSHOT,
        &relation_group::snapshot_of,
        group,
    )
}

fn group_members<P: TriblePattern>(space: &P, snapshot: Id) -> BTreeSet<Id> {
    find!(
        member: Id,
        pattern!(space, [{ snapshot @ relation_group::member: ?member }])
    )
    .collect()
}

#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct NativeMessage {
    id: Id,
    from: Id,
    to: Id,
    body: message::TextHandle,
    created_at: IntervalValue,
}

fn native_message_rows(query: &OrientQuery<'_>) -> Vec<NativeMessage> {
    find!(
        (id: Id, from: Id, to: Id, body: message::TextHandle, created_at: IntervalValue),
        pattern!(query.messages, [{ ?id @
            metadata::tag: &KIND_MESSAGE_ID,
            local_message::from: ?from,
            local_message::to: ?to,
            local_message::body: ?body,
            metadata::created_at: ?created_at,
        }])
    )
    .map(|(id, from, to, body, created_at)| NativeMessage {
        id,
        from,
        to,
        body,
        created_at,
    })
    .collect()
}

fn message_is_inbox(
    query: &OrientQuery<'_>,
    identities: &IdentityIndex,
    own_identities: &HashSet<Id>,
    row: &NativeMessage,
    persona: Id,
) -> Result<bool> {
    // A repeated sender field is another attribution witness, not another
    // envelope. An external witness cannot make our own envelope into news.
    if exists!((sender: Id), and!(
        own_identities.has(sender),
        pattern!(query.messages, [{ row.id @ local_message::from: ?sender }])
    )) {
        return Ok(false);
    }
    let snapshots: Vec<Id> = find!(
        snapshot: Id,
        pattern!(query.messages, [{ row.id @ local_message::group_snapshot: ?snapshot }])
    )
    .collect();
    if snapshots.is_empty() {
        return identities.equivalent(row.to, persona);
    }
    for snapshot in snapshots {
        for member in group_members(query.relations, snapshot) {
            if identities.equivalent(member, persona)? {
                return Ok(true);
            }
        }
    }
    Ok(false)
}

fn message_is_read(
    query: &OrientQuery<'_>,
    identities: &IdentityIndex,
    message: Id,
    persona: Id,
) -> Result<bool> {
    for reader in find!(
        reader: Id,
        pattern!(query.messages, [{ _?read @
            metadata::tag: &KIND_READ_ID,
            local_message::about_message: &message,
            local_message::reader: ?reader,
        }])
    ) {
        if identities.equivalent(reader, persona)? {
            return Ok(true);
        }
    }
    Ok(false)
}

fn unread_messages(query: &OrientQuery<'_>, persona: Id) -> Result<Vec<NativeMessage>> {
    if !person_anchors(query.relations).contains(&persona) {
        return Ok(Vec::new());
    }
    let identities = IdentityIndex::from_relations(query.relations);
    let own_identities = identities.component(persona)?.into_iter().collect();
    let mut rows = Vec::new();
    for row in native_message_rows(query) {
        if message_is_inbox(query, &identities, &own_identities, &row, persona)?
            && !message_is_read(query, &identities, row.id, persona)?
        {
            rows.push(row);
        }
    }
    rows.sort_by_key(|row| (std::cmp::Reverse(interval_key(row.created_at)), row.id));
    rows.dedup_by_key(|row| row.id);
    Ok(rows)
}

fn native_task_title(query: &OrientQuery<'_>, task: Id) -> Result<String> {
    find!(
        handle: compass::TextHandle,
        pattern!(query.compass, [{ task @ board::title: ?handle }])
    )
    .next()
    .map(|handle| read_utf8(&query.payloads, handle, "Compass title"))
    .transpose()
    .map(|title| title.unwrap_or_default())
}

fn render_native_messages(
    query: &OrientQuery<'_>,
    persona: Option<Id>,
    limit: usize,
) -> Result<(String, BTreeSet<Id>)> {
    use std::fmt::Write as _;

    let mut out = String::new();
    let Some(persona) = persona else {
        writeln!(out, "Local messages:").unwrap();
        writeln!(
            out,
            "- Unavailable: no persona (pass --persona <label-or-hex> or set $PERSONA)"
        )
        .unwrap();
        return Ok((out, BTreeSet::new()));
    };

    let unread = unread_messages(query, persona)?;

    writeln!(
        out,
        "Local messages (unread inbox for {}):",
        read_native_person_label(query, persona)?
    )
    .unwrap();
    if unread.is_empty() {
        writeln!(out, "- None").unwrap();
        return Ok((out, BTreeSet::new()));
    }
    let now = interval_key(clock::point_now()?);
    let mut shown = BTreeSet::new();
    for row in unread.into_iter().take(limit) {
        shown.insert(row.id);
        writeln!(
            out,
            "- [{}] {} {} -> {} (unread)",
            fmt_id(row.id),
            format_age(now, interval_key(row.created_at)),
            read_native_person_label(query, row.from)?,
            read_native_person_label(query, row.to)?,
        )
        .unwrap();
        let body = read_utf8(&query.payloads, row.body, "Message body")?;
        if body.is_empty() {
            writeln!(out, "    ").unwrap();
        } else {
            for line in body.lines() {
                writeln!(out, "    {}", line.trim_end_matches('\r')).unwrap();
            }
        }
    }
    Ok((out, shown))
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct MailSummary {
    claimed_at: Option<i128>,
    from: Option<mail_model::TextHandle>,
    subject: mail_model::TextHandle,
}

/// One attention item per unread, non-spam inbound wire message. Re-observing
/// the same wire through another source is idempotent when its presentation
/// agrees; conflicting parser projections are not silently arbitrated.
fn native_unread_mail(query: &OrientQuery<'_>, persona: Id) -> Result<BTreeMap<Id, MailSummary>> {
    // A raw exact anchor is a valid observer before its Relations profile
    // arrives. Until then it has no identity component and therefore no
    // Relations-dependent Mail inbox projection.
    if !person_anchors(query.relations).contains(&persona) {
        return Ok(BTreeMap::new());
    }
    let component = IdentityIndex::from_relations(query.relations).component(persona)?;
    let read_wires: BTreeSet<Id> = find!(
        (wire: Id, reader: Id),
        pattern!(query.mail, [{ _?read @
            metadata::tag: &KIND_READ_OBSERVATION,
            mail_read::wire: ?wire,
            mail_read::reader: ?reader,
        }])
    )
    .filter_map(|(wire, reader)| component.contains(&reader).then_some(wire))
    .collect();

    let mut sources: BTreeSet<(Id, Id)> = find!(
        (source: Id, wire: Id),
        pattern!(query.mail, [{ ?source @
            metadata::tag: &KIND_POP_OBSERVATION,
            mail_observation::wire: ?wire,
        }])
    )
    .collect();
    sources.extend(find!(
        (source: Id, wire: Id),
        pattern!(query.mail, [{ ?source @
            metadata::tag: &KIND_IMPORTED_OBSERVATION,
            imported_mail::direction: &IMPORT_RECEIVED,
            mail_observation::wire: ?wire,
        }])
    ));

    let mut by_wire = BTreeMap::new();
    for (source, wire) in sources {
        if read_wires.contains(&wire) {
            continue;
        }
        let projections: BTreeSet<Id> = find!(
            projection: Id,
            pattern!(query.mail, [{ ?projection @
                metadata::tag: &KIND_PARSED_PROJECTION,
                mail_projection::source: &source,
                mail_projection::recipe: &RECIPE_RFC5322_V1,
            }])
        )
        .collect();
        for projection in projections {
            for (subject, spam) in find!(
                (subject: mail_model::TextHandle, spam: bool),
                pattern!(query.mail, [{ projection @
                    mail_projection::subject: ?subject,
                    mail_projection::spam: ?spam,
                }])
            ) {
                if spam {
                    continue;
                }
                let from = find!(
                    value: mail_model::TextHandle,
                    pattern!(query.mail, [{ projection @ mail_projection::from: ?value }])
                )
                .min_by_key(|value| value.raw);
                let claimed_at = find!(
                    value: IntervalValue,
                    pattern!(query.mail, [{ projection @ mail_projection::claimed_date: ?value }])
                )
                .map(interval_key)
                .min();
                by_wire.entry(wire).or_insert(MailSummary {
                    claimed_at,
                    from,
                    subject,
                });
            }
        }
    }
    Ok(by_wire)
}

/// Logical Teams messages that are attention items for this pile.
///
/// Teams carries no per-reader read state, so the attention set is every
/// present (not deleted) logical message written by somebody other than us;
/// news is that set minus the persona's relational `Presented` ledger. There is
/// no persona gating: one tenant account serves every window sharing this
/// pile, so a colleague's message is addressed to the pile rather than to one
/// window — the same reading as a peer message sent to a group you are in.
///
/// This reads only what the pile already holds. `orient` never calls Graph:
/// `wait` re-arms after every turn, and a network round trip on that path
/// would both slow the common case and rate-limit the tenant. `teams read`
/// remains the only thing that pulls new messages into the pile.
fn native_teams_messages(query: &OrientQuery<'_>) -> Result<BTreeSet<Id>> {
    // Author entities for the account this pile posts as. An author's
    // `teams::user_id` and an auth profile's `teams::auth_user_id` are both
    // content-derived UTF8String handles, so equal Graph user ids are equal
    // handle values and the join needs no blob reads.
    let own_authors: BTreeSet<Id> = find!(
        author: Id,
        pattern!(query.teams, [
            {
                ?author @
                metadata::tag: archive::kind_author,
                teams::source: _?source,
                teams::user_id: _?user,
            },
            {
                _?profile @
                metadata::tag: teams::kind_auth_profile,
                teams::source: _?source,
                teams::auth_user_id: _?user,
            }
        ])
    )
    .collect();

    let present_state: Inline<inlineencodings::ShortString> = "present"
        .try_to_inline()
        .expect("Teams present state fits ShortString");
    let deleted_state: Inline<inlineencodings::ShortString> = "deleted"
        .try_to_inline()
        .expect("Teams deleted state fits ShortString");
    let mut present: BTreeSet<Id> = find!(
        message: Id,
        pattern!(query.teams, [{
            _?observation @
            metadata::tag: teams::kind_message_observation,
            teams::message: ?message,
            teams::message_state: &present_state,
        }])
    )
    .collect();
    let mut deleted: BTreeSet<Id> = find!(
        message: Id,
        pattern!(query.teams, [{
            _?observation @
            metadata::tag: teams::kind_message_observation,
            teams::message: ?message,
            teams::message_state: &deleted_state,
        }])
    )
    .collect();
    for message in find!(
        message: Id,
        pattern!(query.teams, [{
            _?tombstone @
            metadata::tag: teams::kind_message_tombstone,
            teams::message: ?message,
        }])
    ) {
        deleted.insert(message);
    }

    // Our own sends come back through the next delta pull. They must not wake
    // anybody, exactly as a persona's own peer sends and goal edits do not
    // wake its watcher. Attribution is also what separates a message from
    // Graph's authorless chat events (`<systemEventMessage/>` for a member
    // added, a chat renamed, ...): news is somebody writing to us, so an
    // unattributed observation is never news, and can never be mistaken for a
    // colleague when we cannot even check it against our own account.
    let mut own = BTreeSet::new();
    let mut from_others = BTreeSet::new();
    for (message, author) in find!(
        (message: Id, author: Id),
        pattern!(query.teams, [{
            _?observation @
            metadata::tag: teams::kind_message_observation,
            teams::message: ?message,
            archive::author: ?author,
            teams::message_state: &present_state,
        }])
    ) {
        if own_authors.contains(&author) {
            own.insert(message);
        } else {
            from_others.insert(message);
        }
    }

    present.retain(|message| {
        from_others.contains(message) && !own.contains(message) && !deleted.contains(message)
    });
    Ok(present)
}

/// Newest observation of one logical Teams message, with the display name and
/// body worth printing when it turns up as news.
#[derive(Clone, Copy)]
struct TeamsMessageDetailHandles {
    author: Option<teams_model::TextHandle>,
    content: Option<teams_model::TextHandle>,
}

fn teams_message_detail_handles(
    query: &OrientQuery<'_>,
    message: Id,
) -> Result<TeamsMessageDetailHandles> {
    let present_state: Inline<inlineencodings::ShortString> = "present"
        .try_to_inline()
        .expect("Teams present state fits ShortString");
    let newest = find!(
        (modified: IntervalValue, observation: Id),
        pattern!(query.teams, [{
            ?observation @
            metadata::tag: teams::kind_message_observation,
            teams::message: message,
            teams::modified_at: ?modified,
            teams::message_state: &present_state,
        }])
    )
    .map(|(modified, observation)| (interval_key(modified), observation))
    .max();
    let Some((_, observation)) = newest else {
        return Ok(TeamsMessageDetailHandles {
            author: None,
            content: None,
        });
    };
    let author = find!(
        handle: teams_model::TextHandle,
        pattern!(query.teams, [{ observation @ teams::author_name: ?handle }])
    )
    .next();
    let content = find!(
        handle: teams_model::TextHandle,
        pattern!(query.teams, [{ observation @ archive::content: ?handle }])
    )
    .next();
    Ok(TeamsMessageDetailHandles { author, content })
}

fn teams_message_detail(query: &OrientQuery<'_>, message: Id) -> Result<(String, String)> {
    let handles = teams_message_detail_handles(query, message)?;
    let author = handles
        .author
        .map(|handle| read_utf8(&query.payloads, handle, "Teams author display name"))
        .transpose()?
        .unwrap_or_else(|| "(unknown)".to_owned());
    let content = handles
        .content
        .map(|handle| read_utf8(&query.payloads, handle, "Teams message content"))
        .transpose()?
        .unwrap_or_else(|| "(no content)".to_owned());
    Ok((author, content))
}

/// Discord messages that are attention items for this pile.
///
/// Scoped exactly like Teams: Discord carries no per-reader read state, so
/// the attention set is every message somebody other than this pile's bot
/// wrote, and news is that set minus the persona's `Presented` ledger. There
/// is no persona gating: one bot serves every window sharing this pile. The
/// bot's messages are recognised by the account facts its intake (from READY)
/// and `discord send` record; a message edited or observed again is the same
/// anchor, so it is news once. As with Teams, news is somebody writing: a
/// system notice (a pin, a member joining) is not, though Discord gives it an
/// author.
///
/// And news is what was heard: a message sent in a channel at or after the
/// moment `discord live`'s intake hears the channel from, which intake
/// records in the collection with what it stores. Who stored the message does
/// not matter: one sent while intake was down is news when `discord read`
/// brings it in. A channel's history is never news, although `discord read`
/// stores it just as intake stores what it hears, and neither is a channel
/// intake never heard: reading history wakes no window, then or when a
/// window first reads Discord.
fn native_discord_messages(query: &OrientQuery<'_>) -> BTreeSet<Id> {
    let own: BTreeSet<Id> = find!(
        user: Id,
        pattern!(query.discord, [{
            _?account @
            metadata::tag: discord::kind_bot_account,
            discord::user: ?user,
        }])
    )
    .collect();
    let notices: BTreeSet<Id> = find!(
        anchor: Id,
        pattern!(query.discord, [{
            _?notice @
            metadata::tag: discord::kind_system_notice,
            discord::message: ?anchor,
        }])
    )
    .collect();
    find!(
        (anchor: Id, author: Id, created: IntervalValue, heard: IntervalValue),
        pattern!(query.discord, [
            {
                _?observation @
                metadata::tag: archive::kind_message,
                discord::message: ?anchor,
                discord::channel: _?channel,
                archive::author: ?author,
                metadata::created_at: ?created,
            },
            {
                _?intake @
                metadata::tag: discord::kind_intake,
                discord::channel: _?channel,
                discord::heard_from: ?heard,
            },
        ])
    )
    .filter(|(anchor, author, created, heard)| {
        interval_key(*created) >= interval_key(*heard)
            && !own.contains(author)
            && !notices.contains(anchor)
    })
    .map(|(anchor, ..)| anchor)
    .collect()
}

/// The newest observation of one Discord message, as worth printing when it
/// turns up as news: every distinct version at the newest time Discord gave
/// it, since the message is news once and a version left out of that one
/// report would never be presented. Ordinarily there is one.
///
/// Versions are told apart as `discord read` tells them apart (its
/// `select_messages`): by the whole semantic state of an observation, whose
/// attachments are attachment entities, Discord's attachment id and declared
/// size part of their identity; never by what is printed of them.
struct DiscordMessageDetail {
    author: String,
    channel: String,
    versions: Vec<DiscordVersion>,
    /// The distinct attachment entities of those versions.
    attachments: usize,
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct DiscordVersion {
    content: String,
    attachments: Vec<DiscordAttachment>,
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct DiscordAttachment {
    name: String,
    /// Its stored content, unless it was too large to store.
    stored: Option<crate::files::ContentHandle>,
    /// The size Discord declared for one too large to store.
    size: Option<u64>,
}

/// What `discord read` tells two observations of one message apart by: its
/// content, author, channel, times, reply and attachment entities.
#[derive(Clone, Debug, Default, Eq, Ord, PartialEq, PartialOrd)]
struct DiscordState {
    content: BTreeSet<discord_model::TextHandle>,
    author: BTreeSet<Id>,
    channel: BTreeSet<Id>,
    created: BTreeSet<IntervalValue>,
    edited: BTreeSet<IntervalValue>,
    reply_to: BTreeSet<Id>,
    attachments: BTreeSet<Id>,
}

fn discord_message_detail(query: &OrientQuery<'_>, anchor: Id) -> Result<DiscordMessageDetail> {
    // Newest by Discord's own version time: the edit, else the creation.
    let observations: Vec<(i128, Id)> = find!(
        (observation: Id, created: IntervalValue),
        pattern!(query.discord, [{
            ?observation @
            metadata::tag: archive::kind_message,
            discord::message: anchor,
            metadata::created_at: ?created,
        }])
    )
    .map(|(observation, created)| {
        let edited = find!(
            edited: IntervalValue,
            pattern!(query.discord, [{ observation @ archive::edited_at: ?edited }])
        )
        .map(interval_key)
        .max();
        (edited.unwrap_or(interval_key(created)), observation)
    })
    .collect();
    let latest = observations
        .iter()
        .map(|(time, _)| *time)
        .max()
        .ok_or_else(|| anyhow!("Discord message [{}] has no observation", fmt_id(anchor)))?;
    let newest: BTreeSet<Id> = observations
        .into_iter()
        .filter(|(time, _)| *time == latest)
        .map(|(_, observation)| observation)
        .collect();
    let first = *newest.first().expect("the latest time has an observation");

    let author = find!(
        author: Id,
        pattern!(query.discord, [{ first @ archive::author: ?author }])
    )
    .next();
    let author = match author {
        Some(author) => {
            let names = find!(
                handle: discord_model::TextHandle,
                pattern!(query.discord, [{
                    _?profile @
                    metadata::tag: discord::kind_user_profile,
                    discord::user: author,
                    archive::author_name: ?handle,
                }])
            )
            .chain(find!(
                handle: discord_model::TextHandle,
                pattern!(query.discord, [{ author @ discord::user_id: ?handle }])
            ))
            .next();
            names
                .map(|handle| read_utf8(&query.payloads, handle, "Discord author name"))
                .transpose()?
        }
        None => None,
    }
    .unwrap_or_else(|| "(unknown)".to_owned());
    let channel = find!(
        handle: discord_model::TextHandle,
        pattern!(query.discord, [
            { first @ discord::channel: _?channel },
            { _?channel @ discord::channel_id: ?handle },
        ])
    )
    .next()
    .map(|handle| read_utf8(&query.payloads, handle, "Discord channel id"))
    .transpose()?
    .unwrap_or_else(|| "(unknown)".to_owned());
    let mut versions: BTreeMap<DiscordState, DiscordVersion> = BTreeMap::new();
    for observation in newest {
        let state = DiscordState {
            content: find!(
                handle: discord_model::TextHandle,
                pattern!(query.discord, [{ observation @ archive::content: ?handle }])
            )
            .collect(),
            author: find!(
                author: Id,
                pattern!(query.discord, [{ observation @ archive::author: ?author }])
            )
            .collect(),
            channel: find!(
                channel: Id,
                pattern!(query.discord, [{ observation @ discord::channel: ?channel }])
            )
            .collect(),
            created: find!(
                created: IntervalValue,
                pattern!(query.discord, [{ observation @ metadata::created_at: ?created }])
            )
            .collect(),
            edited: find!(
                edited: IntervalValue,
                pattern!(query.discord, [{ observation @ archive::edited_at: ?edited }])
            )
            .collect(),
            reply_to: find!(
                reply: Id,
                pattern!(query.discord, [{ observation @ archive::reply_to: ?reply }])
            )
            .collect(),
            attachments: find!(
                attachment: Id,
                pattern!(query.discord, [{ observation @ archive::attachment: ?attachment }])
            )
            .collect(),
        };
        if versions.contains_key(&state) {
            continue;
        }
        let content = state
            .content
            .first()
            .map(|handle| read_utf8(&query.payloads, *handle, "Discord message content"))
            .transpose()?
            .unwrap_or_default();
        let mut attachments = Vec::new();
        for &attachment in &state.attachments {
            let name = find!(
                name: discord_model::TextHandle,
                pattern!(query.discord, [{ attachment @ archive::attachment_name: ?name }])
            )
            .next()
            .map(|name| read_utf8(&query.payloads, name, "Discord attachment name"))
            .transpose()?
            .unwrap_or_else(|| "(unnamed)".to_owned());
            let stored = find!(
                stored: crate::files::ContentHandle,
                pattern!(query.discord, [
                    { attachment @ archive::attachment_file: _?file },
                    { _?file @ file::content: ?stored },
                ])
            )
            .next();
            let size = find!(
                size: Inline<inlineencodings::U256BE>,
                pattern!(query.discord, [{ attachment @ archive::attachment_size_bytes: ?size }])
            )
            .find_map(|size| u64::try_from_inline(&size).ok());
            attachments.push(DiscordAttachment { name, stored, size });
        }
        attachments.sort();
        versions.insert(
            state,
            DiscordVersion {
                content,
                attachments,
            },
        );
    }
    let attachments = versions
        .keys()
        .flat_map(|state| &state.attachments)
        .collect::<BTreeSet<_>>()
        .len();
    let mut versions: Vec<DiscordVersion> = versions.into_values().collect();
    versions.sort();
    Ok(DiscordMessageDetail {
        author,
        channel,
        versions,
        attachments,
    })
}

/// What `orient` prints of one version of a Discord message: its author,
/// channel and whole text, which of how many versions at the newest time it
/// is when there are more, and per attachment the `files get` that writes its
/// stored bytes out, or the size Discord declared for one too large to store.
fn discord_version_text(detail: &DiscordMessageDetail, version: &DiscordVersion) -> String {
    let count = detail.versions.len();
    let of = if count > 1 {
        format!(" (one of {count} versions with the same time)")
    } else {
        String::new()
    };
    let mut text = format!(
        "- {} in channel {}{of}: {}\n",
        detail.author, detail.channel, version.content
    );
    // The bytes are in the pile, stored with the message; `files get` with
    // `@-` writes them out by content hash.
    for attachment in &version.attachments {
        let name = &attachment.name;
        match (attachment.stored, attachment.size) {
            (Some(content), _) => text.push_str(&format!(
                "  attachment {name}: files get files:{} @-\n",
                crate::files::content_hash_hex(content)
            )),
            (None, Some(size)) => text.push_str(&format!(
                "  attachment {name}: {size} bytes, too large to store\n"
            )),
            (None, None) => text.push_str(&format!("  attachment {name}: not stored\n")),
        }
    }
    text
}

/// `Discord message from <author> in channel <id>: <first line>`, and how
/// many attachments come with it (and how many versions, when Discord gave
/// more than one the same time). Best effort, like every News line: a body
/// not yet here only shortens the line; the detail below waits for it.
fn discord_news_line(query: &OrientQuery<'_>, anchor: Id) -> Option<String> {
    let detail = discord_message_detail(query, anchor).ok()?;
    let mut line = format!(
        "Discord message from {} in channel {}",
        detail.author, detail.channel
    );
    let first = detail.versions.first()?;
    if let Some(preview) = clip_line(&first.content, NEWS_PREVIEW_CHARS) {
        line.push_str(": ");
        line.push_str(&preview);
    }
    match detail.attachments {
        0 => {}
        1 => line.push_str(" (1 attachment)"),
        n => line.push_str(&format!(" ({n} attachments)")),
    }
    if detail.versions.len() > 1 {
        line.push_str(&format!(
            " ({} versions with the same time)",
            detail.versions.len()
        ));
    }
    Some(line)
}

/// Render the same unread native Mail projection that drives `orient wait`.
fn render_native_mail(
    query: &OrientQuery<'_>,
    persona: Option<Id>,
    limit: usize,
) -> Result<(String, BTreeSet<Id>)> {
    use std::fmt::Write as _;

    let mut out = String::new();
    let Some(persona) = persona else {
        writeln!(out, "Mail:").unwrap();
        writeln!(
            out,
            "- Unavailable: no persona (pass --persona <label-or-hex> or set $PERSONA)"
        )
        .unwrap();
        return Ok((out, BTreeSet::new()));
    };

    let mut rows = native_unread_mail(query, persona)?
        .into_iter()
        .collect::<Vec<_>>();
    rows.sort_by_key(|(wire, summary)| {
        (
            std::cmp::Reverse(summary.claimed_at),
            std::cmp::Reverse(*wire),
        )
    });

    writeln!(
        out,
        "Mail (unread for {}):",
        read_native_person_label(query, persona)?
    )
    .unwrap();
    if rows.is_empty() {
        writeln!(out, "- None").unwrap();
        return Ok((out, BTreeSet::new()));
    }
    let now = interval_key(clock::point_now()?);
    let mut shown = BTreeSet::new();
    for (wire, summary) in rows.into_iter().take(limit) {
        shown.insert(wire);
        let age = summary
            .claimed_at
            .map(|at| format_age(now, at))
            .unwrap_or_else(|| "?".to_owned());
        let from = summary
            .from
            .map(|handle| read_utf8(&query.payloads, handle, "Mail From"))
            .transpose()?
            .unwrap_or_else(|| "(no From)".to_owned());
        let subject = read_utf8(&query.payloads, summary.subject, "Mail subject")?;
        writeln!(out, "- [{}] {} {} — {}", fmt_id(wire), age, from, subject,).unwrap();
    }
    Ok((out, shown))
}

fn latest_goal_status(query: &OrientQuery<'_>, goal: Id) -> Option<(Id, String, IntervalValue)> {
    find!(
        (event: Id, status: String, at: IntervalValue),
        and!(
            pattern!(query.compass, [{ ?event @
                metadata::tag: &KIND_STATUS_ID,
                board::status_of: &goal,
                board::status: ?status,
                metadata::created_at: ?at,
            }]),
            query.compass_status.has(event),
        )
    )
    .next()
}

fn goal_priority_edges(query: &OrientQuery<'_>, goals: &BTreeSet<Id>) -> BTreeSet<(Id, Id)> {
    let mut latest = BTreeMap::<(Id, Id), ((i128, Id), bool)>::new();
    let mut absorb = |event: Id, higher: Id, lower: Id, at: IntervalValue, active: bool| {
        let order = (interval_key(at), event);
        let entry = latest.entry((higher, lower)).or_insert((order, active));
        if order > entry.0 {
            *entry = (order, active);
        }
    };
    for (event, higher, lower, at) in find!(
        (event: Id, higher: Id, lower: Id, at: IntervalValue),
        pattern!(query.compass, [{ ?event @
            metadata::tag: &crate::schemas::compass::KIND_PRIORITIZE_ID,
            board::higher: ?higher,
            board::lower: ?lower,
            metadata::created_at: ?at,
        }])
    ) {
        absorb(event, higher, lower, at, true);
    }
    for (event, higher, lower, at) in find!(
        (event: Id, higher: Id, lower: Id, at: IntervalValue),
        pattern!(query.compass, [{ ?event @
            metadata::tag: &crate::schemas::compass::KIND_DEPRIORITIZE_ID,
            board::higher: ?higher,
            board::lower: ?lower,
            metadata::created_at: ?at,
        }])
    ) {
        absorb(event, higher, lower, at, false);
    }
    let mut edges: BTreeSet<(Id, Id)> = latest
        .into_iter()
        .filter_map(|(edge, (_, active))| active.then_some(edge))
        .collect();
    for (child, parent) in find!(
        (child: Id, parent: Id),
        pattern!(query.compass, [{ ?child @
            metadata::tag: &KIND_GOAL_ID,
            board::parent: ?parent,
        }])
    ) {
        if goals.contains(&parent) {
            edges.insert((child, parent));
        }
    }
    edges
}

fn render_native_compass_goals(
    query: &OrientQuery<'_>,
    doing_limit: usize,
    todo_limit: usize,
) -> Result<(String, BTreeSet<Id>)> {
    use std::fmt::Write as _;

    let goals = ids_of_kind(query.compass, KIND_GOAL_ID);
    let ranks = compass::priority_ranks(goals.iter().copied(), &goal_priority_edges(query, &goals));
    let mut doing = Vec::<(usize, i128, Id, Option<Id>)>::new();
    let mut todo = Vec::<(usize, i128, Id, Option<Id>)>::new();
    for task in goals {
        let (status_event, status, status_at) = latest_goal_status(query, task)
            .map(|(event, value, at)| {
                (
                    Some(event),
                    value.to_ascii_lowercase(),
                    Some(interval_key(at)),
                )
            })
            .unwrap_or_else(|| (None, "todo".to_owned(), None));
        let created = find!(
            at: IntervalValue,
            pattern!(query.compass, [{ task @ metadata::created_at: ?at }])
        )
        .map(interval_key)
        .min()
        .unwrap_or(0);
        let key = status_at.unwrap_or(created);
        let rank = ranks.get(&task).copied().unwrap_or(usize::MAX);
        match status.as_str() {
            "doing" => doing.push((rank, key, task, status_event)),
            "todo" => todo.push((rank, key, task, status_event)),
            _ => {}
        }
    }
    let compare = |left: &(usize, i128, Id, Option<Id>), right: &(usize, i128, Id, Option<Id>)| {
        left.0
            .cmp(&right.0)
            .then_with(|| right.1.cmp(&left.1))
            .then_with(|| left.2.cmp(&right.2))
    };
    doing.sort_by(compare);
    todo.sort_by(compare);

    let mut out = String::new();
    let mut shown = BTreeSet::new();
    writeln!(out, "Compass:").unwrap();
    if doing.is_empty() && todo.is_empty() {
        writeln!(out, "- No goals.").unwrap();
        return Ok((out, shown));
    }
    writeln!(out, "Doing:").unwrap();
    if doing.is_empty() {
        writeln!(out, "- None").unwrap();
    } else {
        for (_, _, task, status_event) in doing.into_iter().take(doing_limit) {
            shown.insert(task);
            shown.extend(status_event);
            writeln!(
                out,
                "- [{}] {}{}",
                fmt_id(task),
                native_task_title(query, task)?,
                render_tags(&entity_tags(query.compass, task)),
            )
            .unwrap();
        }
    }
    writeln!(out, "Todo:").unwrap();
    if todo.is_empty() {
        writeln!(out, "- None").unwrap();
    } else {
        for (_, _, task, status_event) in todo.into_iter().take(todo_limit) {
            shown.insert(task);
            shown.extend(status_event);
            writeln!(
                out,
                "- [{}] {}{}",
                fmt_id(task),
                native_task_title(query, task)?,
                render_tags(&entity_tags(query.compass, task)),
            )
            .unwrap();
        }
    }
    Ok((out, shown))
}

fn render_window_status(query: &OrientQuery<'_>) -> Result<(String, BTreeSet<Id>)> {
    use std::fmt::Write as _;

    let mut latest = BTreeMap::<Id, ((i128, Id), status::TextHandle)>::new();
    for (event, window, text, at) in find!(
        (event: Id, window: Id, text: status::TextHandle, at: IntervalValue),
        pattern!(query.status, [{ ?event @
            metadata::tag: &KIND_STATUS_UPDATE,
            window_status::window: ?window,
            window_status::text: ?text,
            metadata::created_at: ?at,
        }])
    ) {
        let key = (interval_key(at), event);
        let entry = latest.entry(window).or_insert((key, text));
        if key > entry.0 {
            *entry = (key, text);
        }
    }
    let mut rows = Vec::new();
    for (person, (_, handle)) in &latest {
        let text = Some(read_utf8(&query.payloads, *handle, "Status text")?);
        rows.push((read_native_person_label(query, *person)?, text));
    }
    rows.sort_by(|left, right| left.0.cmp(&right.0));

    let mut out = String::new();
    let shown = latest.keys().copied().collect();
    writeln!(out, "Window status:").unwrap();
    if rows.is_empty() {
        writeln!(out, "- (none)").unwrap();
    }
    for (label, text) in rows {
        writeln!(out, "- {label}: {}", text.unwrap_or_else(|| "—".to_owned())).unwrap();
    }
    Ok((out, shown))
}

#[derive(Clone, Debug, Eq, PartialEq)]
struct DueHabit {
    label: String,
    nudge: String,
    /// The second this due began: the last completion plus the cooldown, or 0
    /// for an intention never completed. It is the identity of one due event,
    /// so a fresh completion and the due that follows it are a new event.
    since: i64,
    /// Addressed to the observing persona.
    targeted: bool,
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
struct HabitObservation {
    due: BTreeMap<Id, DueHabit>,
    attention: BTreeMap<Id, String>,
    /// Intentions addressed to the observing persona.
    targeted: BTreeSet<Id>,
    /// Earliest completion-relative deadline that can change a cooling row.
    next_cooldown_at: Option<i64>,
}

/// Prepare only the selected Habit evaluation inputs. No scripts run here:
/// a missing payload can safely retry this read against the same frozen facts.
fn prepare_habits(
    snapshot: &FacultySnapshot,
    facts: &FactArchive,
    persona: Option<Id>,
) -> Result<(Vec<habits::HabitRow>, HabitObservation)> {
    let mut rows = Vec::new();
    let mut observation = HabitObservation::default();
    let definitions: Vec<(Id, String, habits::TextHandle, habits::TextHandle)> = find!(
        (habit: Id, label: String, condition: habits::TextHandle, nudge: habits::TextHandle),
        pattern!(facts, [{ ?habit @
            metadata::tag: &KIND_HABIT_ID,
            habit_attrs::label: ?label,
            habit_attrs::condition: ?condition,
            habit_attrs::nudge: ?nudge,
        }])
    )
    .collect();
    let superseded: BTreeSet<Id> = find!(
        old: Id,
        pattern!(facts, [{ _?new @
            metadata::tag: &KIND_HABIT_ID,
            metadata::supersedes: ?old,
        }])
    )
    .collect();

    for (habit, label, condition_handle, nudge_handle) in definitions {
        if superseded.contains(&habit) {
            continue;
        }
        // Attention routing is a fact on the intention, not its author. Check
        // before fetching payloads or executing any predicate. Without an
        // observer, only global intentions apply.
        let targeted = exists!(pattern!(facts, [{ habit @ habit_attrs::persona: _?target }]));
        if targeted
            && !persona.is_some_and(|persona| {
                exists!(pattern!(facts, [{ habit @ habit_attrs::persona: &persona }]))
            })
        {
            continue;
        }
        if targeted {
            observation.targeted.insert(habit);
        }
        let condition = read_utf8(snapshot, condition_handle, "Habit condition")?;
        let nudge = read_utf8(snapshot, nudge_handle, "Habit nudge")?;
        let script_handle = find!(
            handle: habits::ScriptHandle,
            pattern!(facts, [{ habit @ habit_attrs::script: ?handle }])
        )
        .min_by_key(|handle| handle.raw);
        let script = match script_handle {
            Some(handle) => Some(habits::Script {
                handle,
                bytes: read_bytes(snapshot, handle, "Habit script")?,
            }),
            None => None,
        };

        let mut completed_at: Vec<i64> = find!(
            at: IntervalValue,
            pattern!(facts, [{ _?done @
                metadata::tag: &KIND_HABIT_DONE_ID,
                habit_attrs::of: &habit,
                metadata::created_at: ?at,
            }])
        )
        .map(|at| (interval_key(at) / 1_000_000_000) as i64)
        .collect();
        completed_at.sort_unstable();
        completed_at.dedup();

        let state_rows: Vec<(Id, String, IntervalValue)> = find!(
            (id: Id, state: String, at: IntervalValue),
            pattern!(facts, [{ ?id @
                metadata::tag: &KIND_HABIT_STATE_ID,
                habit_attrs::of: &habit,
                habit_attrs::state: ?state,
                metadata::created_at: ?at,
            }])
        )
        .collect();
        let state_superseded: BTreeSet<Id> = find!(
            old: Id,
            pattern!(facts, [{ _?new @
                metadata::tag: &KIND_HABIT_STATE_ID,
                habit_attrs::of: &habit,
                metadata::supersedes: ?old,
            }])
        )
        .collect();
        let mut heads = Vec::new();
        for (id, state, asserted_at) in state_rows {
            if state_superseded.contains(&id) {
                continue;
            }
            let state = match state.as_str() {
                STATE_ACTIVE => habits::DeclaredState::Active,
                STATE_PAUSED => habits::DeclaredState::Paused,
                other => {
                    observation.attention.insert(
                        habit,
                        format!("{label} [{habit:x}] has unknown state {other:?}"),
                    );
                    continue;
                }
            };
            heads.push(habits::StateAssertion {
                id,
                habit,
                state,
                predecessors: find!(
                    predecessor: Id,
                    pattern!(facts, [{ id @ metadata::supersedes: ?predecessor }])
                )
                .collect(),
                asserted_at,
            });
        }
        let activation = if heads.is_empty()
            || heads
                .iter()
                .all(|head| head.state == habits::DeclaredState::Active)
        {
            habits::Activation::Active(heads)
        } else if heads
            .iter()
            .all(|head| head.state == habits::DeclaredState::Paused)
        {
            habits::Activation::Paused(heads)
        } else {
            habits::Activation::Forked(heads)
        };
        rows.push(habits::HabitRow {
            id: habit,
            label,
            condition,
            nudge,
            script,
            activation,
            completed_at,
        });
    }
    Ok((rows, observation))
}

/// Evaluate once after payload preparation succeeds. The wall clock and
/// directory stay explicit so a wait can observe temporal edges without an
/// append, and an acquisition retry cannot repeat a condition script.
fn observe_habits(
    (rows, mut observation): (Vec<habits::HabitRow>, HabitObservation),
    pile: &Path,
    now_secs: i64,
) -> Result<HabitObservation> {
    let at = habits::evaluation_dir(pile);
    for row in rows {
        let state = habits::evaluate(&row, now_secs, &at);
        match &state {
            habits::State::Due => {
                let since = row
                    .next_cooldown_at()
                    .map_err(anyhow::Error::msg)?
                    .unwrap_or(0);
                observation.due.insert(
                    row.id,
                    DueHabit {
                        label: row.label.clone(),
                        nudge: row.nudge.clone(),
                        since,
                        targeted: observation.targeted.contains(&row.id),
                    },
                );
            }
            habits::State::Cooling => {
                if let Some(deadline) = row
                    .next_cooldown_at()
                    .map_err(anyhow::Error::msg)?
                    .filter(|deadline| *deadline > now_secs)
                {
                    observation.next_cooldown_at = Some(
                        observation
                            .next_cooldown_at
                            .map_or(deadline, |seen| seen.min(deadline)),
                    );
                }
            }
            habits::State::Forked(heads) => {
                observation.attention.insert(
                    row.id,
                    format!(
                        "{} [{:x}] has conflicting state heads: {}",
                        row.label,
                        row.id,
                        heads
                            .iter()
                            .map(|(id, state)| format!("{id:x}={}", state.as_str()))
                            .collect::<Vec<_>>()
                            .join(", ")
                    ),
                );
            }
            habits::State::Unparseable(error) | habits::State::Failed(error) => {
                observation.attention.insert(
                    row.id,
                    format!("{} [{:x}] {}: {error}", row.label, row.id, state.word()),
                );
            }
            habits::State::Waiting | habits::State::Paused => {}
        }
    }
    Ok(observation)
}

fn render_passive_habits(facts: &FactArchive, persona: Option<Id>) -> String {
    use std::fmt::Write as _;
    let superseded: BTreeSet<Id> = find!(old: Id, pattern!(facts, [{ _?new @ metadata::tag: &KIND_HABIT_ID, metadata::supersedes: ?old }])).collect();
    let mut text = String::from("Habits (passive; conditions not evaluated):\n");
    let mut count = 0;
    for (habit, label) in find!((habit: Id, label: String), pattern!(facts, [{ ?habit @ metadata::tag: &KIND_HABIT_ID, habit_attrs::label: ?label }]))
    {
        let targeted = exists!(pattern!(facts, [{ habit @ habit_attrs::persona: _?target }]));
        if targeted
            && !persona.is_some_and(|persona| {
                exists!(pattern!(facts, [{ habit @ habit_attrs::persona: &persona }]))
            })
        {
            continue;
        }
        if !superseded.contains(&habit) {
            writeln!(text, "- [{}] {label} (not evaluated)", fmt_id(habit)).unwrap();
            count += 1;
        }
    }
    if count == 0 {
        text.push_str("- None\n");
    }
    text
}

fn render_native_habits(observation: &HabitObservation) -> String {
    use std::fmt::Write as _;

    let mut out = String::new();
    writeln!(out, "Habits due:").unwrap();
    if observation.due.is_empty() {
        writeln!(out, "- None").unwrap();
    } else {
        for (id, habit) in &observation.due {
            writeln!(out, "- [{}] {}: {}", fmt_id(*id), habit.label, habit.nudge).unwrap();
        }
    }
    if !observation.attention.is_empty() {
        writeln!(out, "Habit attention:").unwrap();
        for warning in observation.attention.values() {
            writeln!(out, "- {warning}").unwrap();
        }
    }
    out
}

fn newly_due(previous: &HabitObservation, current: &HabitObservation) -> Vec<(Id, DueHabit)> {
    current
        .due
        .iter()
        .filter(|(id, habit)| {
            previous
                .due
                .get(*id)
                .is_none_or(|seen| seen.since != habit.since)
        })
        .map(|(id, habit)| (*id, habit.clone()))
        .collect()
}

fn newly_needing_attention(
    previous: &HabitObservation,
    current: &HabitObservation,
) -> Vec<(Id, String)> {
    current
        .attention
        .iter()
        .filter(|(id, warning)| previous.attention.get(*id) != Some(*warning))
        .map(|(id, warning)| (*id, warning.clone()))
        .collect()
}

fn push_due_news(out: &mut String, due: &[(Id, DueHabit)]) {
    use std::fmt::Write as _;
    // The label is the habit's name and `push_due_detail` spells out the nudge
    // directly below, so the hex id only cost the reader tokens.
    for (_, habit) in due {
        writeln!(out, "News: habit became due: {}", habit.label).unwrap();
    }
}

fn push_due_detail(out: &mut String, due: &[(Id, DueHabit)]) {
    use std::fmt::Write as _;
    if due.is_empty() {
        return;
    }
    writeln!(out, "\nHabits newly due:").unwrap();
    for (_, habit) in due {
        writeln!(out, "- {}: {}", habit.label, habit.nudge).unwrap();
    }
}

/// What this frame has to say about intentions: every due event not yet
/// receipted, plus warnings for intentions that newly need attention.
///
/// Due-ness is decided by RECEIPT, not by comparing against the previous
/// observation. A transition comparison answers "did this change since I last
/// looked", which is the wrong question -- an intention that fell due while
/// nothing was looking never transitions in any observation anyone holds, and
/// so was silently lost. Attention warnings keep the transition form because
/// they describe a change of condition rather than an occurrence to present.
fn render_habit_transitions(
    previous: &HabitObservation,
    current: &HabitObservation,
    presented: &FactArchive,
) -> Option<(String, Vec<(Id, i64)>)> {
    use std::fmt::Write as _;
    let (due_text, events) = render_due_habits_unreceipted(current, presented)
        .unwrap_or_else(|| (String::new(), Vec::new()));
    let attention = newly_needing_attention(previous, current);
    if due_text.is_empty() && attention.is_empty() {
        return None;
    }
    let mut out = String::new();
    // The warning already names the habit and its id, so it *is* the reason;
    // the separate attention block below it only repeated the same string.
    for (_, warning) in &attention {
        writeln!(out, "News: habit needs attention: {warning}").unwrap();
    }
    out.push_str(&due_text);
    Some((out, events))
}

/// Every due intention this persona has not already been shown, and the due
/// events to receipt for having shown them.
///
/// This replaces two mechanisms that both existed only because the answer to
/// "has this persona seen this already?" was being inferred from process-local
/// state. A fresh `orient wait` has no such state, so an arm could either stay
/// quiet (losing an intention whose due instant fell between one wait's exit
/// and the next arm) or repeat everything unsatisfied. The old split reported
/// only intentions addressed to the observing persona at arm and left shared
/// ones to whoever saw the transition -- which resolves to NOBODY exactly when
/// the transition happened in the gap. Observed 2026-09-18:
/// `work-ledger-grooming` (`every 7d`, addressed to everyone) went due
/// unobserved and would have sat due for a week.
///
/// A receipt answers it as a fact instead, the way it already does for
/// messages: each due event has a stable identity derived from
/// `(habit, since)`, so presenting it is recorded and a rearmed watcher simply
/// does not present it again. Completion ends the due event; the next
/// recurrence has a later `since`, hence a different identity, and is
/// presented once more.
fn render_due_habits_unreceipted(
    current: &HabitObservation,
    presented: &FactArchive,
) -> Option<(String, Vec<(Id, i64)>)> {
    let due: Vec<(Id, DueHabit)> = current
        .due
        .iter()
        .filter(|(id, habit)| !habit_due_presented(presented, **id, habit.since))
        .map(|(id, habit)| (*id, habit.clone()))
        .collect();
    if due.is_empty() {
        return None;
    }
    let presented_now = due.iter().map(|(id, habit)| (*id, habit.since)).collect();
    let mut out = String::new();
    push_due_news(&mut out, &due);
    push_due_detail(&mut out, &due);
    Some((out, presented_now))
}

/// Has this observer already been shown this intention's due occurrence?
///
/// An ordinary join on what identifies the occurrence -- the intention and the
/// instant its due began. The receipt collection is private to this signing
/// key, so the observer is the descriptor's authority and needs no clause.
fn event_presented(presented: &FactArchive, event: Id) -> bool {
    exists!(pattern!(presented, [{ _?receipt @ presentation::event: &event }]))
}

fn habit_due_presented(presented: &FactArchive, habit: Id, since: i64) -> bool {
    let Ok(due_at) = clock::point(Epoch::from_tai_seconds(since as f64)) else {
        return false;
    };
    exists!(pattern!(presented, [{ _?receipt @
        presentation::habit: &habit,
        presentation::due_at: &due_at,
    }]))
}

#[derive(Debug)]
struct PersonaNotFound(String);

impl std::fmt::Display for PersonaNotFound {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(formatter, "no person matches '{}'", self.0)
    }
}

impl std::error::Error for PersonaNotFound {}

fn is_persona_not_found(error: &anyhow::Error) -> bool {
    error.downcast_ref::<PersonaNotFound>().is_some()
}

fn resolve_native_persona(query: &OrientQuery<'_>, input: &str) -> Result<Id> {
    resolve_resident_persona(query.relations, &query.payloads, input)
}

fn resolve_resident_persona<R: BlobStoreList + BlobStoreGet>(
    facts: &FactArchive,
    reader: &R,
    input: &str,
) -> Result<Id> {
    let input = input.trim();
    if let Some(id) = Id::from_hex(input) {
        // Exact anchors remain useful before a profile has arrived.
        return Ok(id);
    }
    if input.is_empty() {
        bail!("empty person selector");
    }
    let wanted = relations::lookup_key(input);
    let mut settled = Vec::new();
    let mut forked = Vec::new();
    for person in person_anchors(facts) {
        let heads = profile_heads(facts, person);
        let mut matched = false;
        for handle in profile_lookup_handles(facts, person) {
            let value = read_utf8(reader, handle, "Relations profile selector")?;
            if relations::lookup_key(&value) == wanted {
                matched = true;
                break;
            }
        }
        if !matched || lifecycle_retired(facts, person)? == Some(true) {
            continue;
        }
        if heads.len() == 1 && lifecycle_retired(facts, person)?.is_some() {
            settled.push(person);
        } else {
            forked.push(person);
        }
    }
    if !forked.is_empty() {
        bail!(
            "cannot resolve person '{input}': unreconciled Relations state on {}",
            forked
                .iter()
                .map(|id| fmt_id(*id))
                .collect::<Vec<_>>()
                .join(", ")
        );
    }
    match settled.as_slice() {
        [person] => Ok(*person),
        [] => Err(PersonaNotFound(input.to_owned()).into()),
        _ => bail!(
            "multiple people match '{input}': {}",
            settled
                .iter()
                .map(|id| fmt_id(*id))
                .collect::<Vec<_>>()
                .join(", ")
        ),
    }
}

fn native_person_label_handle(
    query: &OrientQuery<'_>,
    person: Id,
) -> Result<Option<relations::TextHandle>> {
    let heads = profile_heads(query.relations, person);
    let [head] = heads.as_slice() else {
        return Ok(None);
    };
    let head = *head;
    Ok(find!(
        handle: relations::TextHandle,
        pattern!(query.relations, [{ head @ metadata::name: ?handle }])
    )
    .min_by_key(|handle| handle.raw))
}

fn read_native_person_label(query: &OrientQuery<'_>, person: Id) -> Result<String> {
    native_person_label_handle(query, person)?
        .map(|handle| read_utf8(&query.payloads, handle, "Relations person label"))
        .transpose()
        .map(|label| label.unwrap_or_else(|| fmt_id(person)))
}

fn persona_keys(query: &OrientQuery<'_>, persona: Id) -> Result<HashSet<String>> {
    profile_lookup_handles(query.relations, persona)
        .into_iter()
        .map(|handle| {
            read_utf8(&query.payloads, handle, "Relations persona selector")
                .map(|value| value.to_ascii_lowercase())
        })
        .collect()
}

/// Every textual group selector that may currently address `persona`.
///
/// Attention is a conservative read projection, not a mutation precondition:
/// a legitimate fork in one group must not disable every watcher. We therefore
/// inspect each maximal snapshot independently. A fork head names an
/// attention group only when that same snapshot contains the persona; names
/// from a sibling head cannot borrow another head's membership. Settled
/// same-person components still participate in membership, while an exact id
/// without a Relations person record simply belongs to no group yet.
fn group_attention_name_handles<P: TriblePattern>(
    facts: &P,
    persona: Id,
) -> Result<Vec<relations::TextHandle>> {
    if !person_anchors(facts).contains(&persona) {
        return Ok(Vec::new());
    }
    let equivalent = IdentityIndex::from_relations(facts).component(persona)?;
    let mut handles = BTreeSet::new();
    for group in group_anchors(facts) {
        for head in group_head_ids(facts, group) {
            if group_members(facts, head)
                .iter()
                .any(|member| equivalent.contains(member))
            {
                handles.extend(find!(
                    handle: relations::TextHandle,
                    pattern!(facts, [{ head @ metadata::name: ?handle }])
                ));
            }
        }
    }
    Ok(handles.into_iter().collect())
}

fn group_attention_keys<R: BlobStoreList + BlobStoreGet, P: TriblePattern>(
    reader: &R,
    facts: &P,
    persona: Id,
) -> Result<HashSet<String>> {
    group_attention_name_handles(facts, persona)?
        .into_iter()
        .map(|handle| {
            read_utf8(reader, handle, "Relations group name")
                .map(|value| relations::lookup_key(&value))
        })
        .collect()
}

fn attention_keys(query: &OrientQuery<'_>, persona: Id) -> Result<HashSet<String>> {
    let mut keys = persona_keys(query, persona)?;
    keys.extend(group_attention_keys(
        &query.payloads,
        query.relations,
        persona,
    )?);
    Ok(keys)
}

fn status_roster(query: &OrientQuery<'_>) -> Result<BTreeSet<Id>> {
    Ok(find!(
        window: Id,
        pattern!(query.status, [{ _?event @
            metadata::tag: &KIND_STATUS_UPDATE,
            window_status::window: ?window,
        }])
    )
    .collect())
}

fn insert_note_goal(notes: &mut BTreeMap<Id, Id>, note_id: Id, goal_id: Id) {
    notes
        .entry(note_id)
        .and_modify(|existing| {
            if goal_id < *existing {
                *existing = goal_id;
            }
        })
        .or_insert(goal_id);
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
enum CollectionSyncIssue {
    ComparisonUnavailable,
    DivergenceStalled,
    Recovered,
}

#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
struct CollectionSyncGroup {
    observer: String,
    peer_scope: String,
    issue: CollectionSyncIssue,
}

impl CollectionSyncGroup {
    fn reason(&self, collections: usize) -> String {
        let issue = match self.issue {
            CollectionSyncIssue::ComparisonUnavailable => {
                "collection comparisons unavailable beyond progress grace"
            }
            CollectionSyncIssue::DivergenceStalled => {
                "collections remain divergent without observed progress beyond grace"
            }
            CollectionSyncIssue::Recovered => {
                "collections recovered and converged at observed pairwise roots"
            }
        };
        format!(
            "swarm health: observer [{}] collection sync{}: {collections} {issue}",
            self.observer, self.peer_scope
        )
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
enum AttentionEvent {
    Message(Id),
    Mail(Id),
    Teams(Id),
    /// A Discord message, by its stable anchor.
    Discord(Id),
    Goal {
        event: Id,
        goal: Id,
        status: String,
    },
    Note {
        note: Id,
        goal: Id,
    },
    StatusWindow(Id),
    Health {
        event: Id,
        detail: String,
        collection_group: Option<CollectionSyncGroup>,
    },
}

impl AttentionEvent {
    fn id(&self) -> Id {
        match self {
            Self::Message(id)
            | Self::Mail(id)
            | Self::Teams(id)
            | Self::Discord(id)
            | Self::StatusWindow(id) => *id,
            Self::Goal { event, .. } => *event,
            Self::Note { note, .. } => *note,
            Self::Health { event, .. } => *event,
        }
    }

    fn reason(&self) -> String {
        match self {
            Self::Message(id) => format!("new message [{}]", fmt_id(*id)),
            Self::Mail(id) => format!("new mail [{}]", fmt_id(*id)),
            Self::Teams(id) => format!("new Teams message [{}]", fmt_id(*id)),
            Self::Discord(id) => format!("new Discord message [{}]", fmt_id(*id)),
            Self::Goal {
                event,
                goal,
                status,
            } if event == goal => format!("new goal [{}] ({status})", fmt_id(*goal)),
            Self::Goal { goal, status, .. } => {
                format!("goal [{}] is now {status}", fmt_id(*goal))
            }
            Self::Note { note, goal } => {
                format!("new note [{}] on goal [{}]", fmt_id(*note), fmt_id(*goal))
            }
            Self::StatusWindow(window) => {
                format!("new status window [{}]", fmt_id(*window))
            }
            Self::Health { detail, .. } => format!("swarm health: {detail}"),
        }
    }
}

#[derive(Clone, Debug, Default, Eq, PartialEq)]
struct AttentionView {
    events: BTreeMap<Id, AttentionEvent>,
}

impl AttentionView {
    fn insert(&mut self, event: AttentionEvent) {
        self.events.entry(event.id()).or_insert(event);
    }

    fn ids(&self) -> impl ExactSizeIterator<Item = Id> + '_ {
        self.events.keys().copied()
    }

    /// Everything this observer has not already been shown.
    ///
    /// An ordinary join against the receipt facts. It used to be a membership
    /// test against a set of ids, which answers only "is THIS id present" and
    /// so obliged every caller to have an id in hand before it could ask --
    /// fine for a message, which is a record with an identity, and the reason
    /// a habit's due occurrence had to be given a derived one it did not want.
    fn pending(&self, presented: &FactArchive) -> Self {
        Self {
            events: self
                .events
                .iter()
                .filter(|(event, _)| !event_presented(presented, **event))
                .map(|(event, detail)| (*event, detail.clone()))
                .collect(),
        }
    }

    fn is_empty(&self) -> bool {
        self.events.is_empty()
    }
}

/// Derive the exact attention-event set visible to one persona from the
/// current source collections. This is deliberately an ephemeral query view;
/// durable state consists only of receipt facts in the zooid's collection.
fn load_attention_view(query: &OrientQuery<'_>, persona_id: Id) -> Result<AttentionView> {
    let mut view = AttentionView::default();
    // Attribution is relational: settled same-person anchors are also self.
    // An absent actor remains unknown, never inferred from a shared signing key.
    let own_identities: HashSet<Id> = IdentityIndex::from_relations(query.relations)
        .component(persona_id)?
        .into_iter()
        .collect();
    for row in unread_messages(query, persona_id)? {
        view.insert(AttentionEvent::Message(row.id));
    }
    for wire in native_unread_mail(query, persona_id)?.into_keys() {
        view.insert(AttentionEvent::Mail(wire));
    }
    for message in native_teams_messages(query)? {
        view.insert(AttentionEvent::Teams(message));
    }
    for message in native_discord_messages(query) {
        view.insert(AttentionEvent::Discord(message));
    }

    let attention_keys = attention_keys(query, persona_id)?;

    let mut relevant_goals = HashSet::new();
    for id in find!(id: Id, pattern!(query.compass, [{ ?id @ metadata::tag: &KIND_GOAL_ID }])) {
        let authored_status = exists!((by: Id), and!(own_identities.has(by), pattern!(query.compass, [{
            _?evt @
            metadata::tag: &KIND_STATUS_ID,
            board::status_of: &id,
            board::by: ?by,
        }])));
        let authored_note = exists!((by: Id), and!(own_identities.has(by), pattern!(query.compass, [{
            _?evt @
            metadata::tag: &KIND_NOTE_ID,
            board::task: &id,
            board::note: _?body,
            board::by: ?by,
        }])));
        let involved = authored_status || authored_note;
        let tags = entity_tags(query.compass, id);
        let addressed = tags
            .iter()
            .any(|tag| attention_keys.contains(&tag.to_ascii_lowercase()));
        if involved || addressed {
            relevant_goals.insert(id);
            match latest_goal_status(query, id) {
                Some((event, status, _)) => {
                    let authored = exists!((by: Id), and!(
                        own_identities.has(by),
                        pattern!(query.compass, [{ event @ board::by: ?by }])
                    ));
                    if !authored {
                        view.insert(AttentionEvent::Goal {
                            event,
                            goal: id,
                            status,
                        });
                    }
                }
                None if addressed => view.insert(AttentionEvent::Goal {
                    event: id,
                    goal: id,
                    status: "todo".to_owned(),
                }),
                None => {}
            }
        }
    }

    for (note, goal) in visible_notes(
        query.compass,
        &own_identities,
        &attention_keys,
        &relevant_goals,
    ) {
        view.insert(AttentionEvent::Note { note, goal });
    }

    for window in status_roster(query)? {
        if !own_identities.contains(&window) {
            view.insert(AttentionEvent::StatusWindow(window));
        }
    }

    Ok(view)
}

fn presentation_collection_for_write(
    pile: &mut FacultyStore,
    signer: &SigningKey,
) -> Result<Collection<SimpleArchive>> {
    let collection = ReceiptSource::register(pile, signer)?.source;
    require_presentation_write(&pile.snapshot()?, collection, signer)?;
    Ok(collection)
}

fn require_presentation_write(
    snapshot: &FacultySnapshot,
    collection: Collection<SimpleArchive>,
    signer: &SigningKey,
) -> Result<()> {
    if !collection
        .writer_is_admitted(snapshot, signer.verifying_key())
        .context("check Orient presentation WRITE admission")?
    {
        bail!("Orient presentation collection requires WRITE to record Presented receipts");
    }
    Ok(())
}

fn save_presentations(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    events: impl IntoIterator<Item = Id>,
) -> Result<()> {
    let fragment = orient_model::receipt_fragment(events, clock::point_now()?);
    if fragment.facts().is_empty() {
        return Ok(());
    }
    let collection = presentation_collection_for_write(pile, signer)?;
    pile.commit(collection, signer, fragment)
        .map_err(|error| anyhow!("commit Orient presentation facts: {error}"))?;
    Ok(())
}

async fn cmd_baseline(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    persona: Option<&str>,
    health_max_age: Duration,
) -> Result<BaselineReceipt> {
    let Some(input) = persona else {
        bail!("baseline requires a persona (pass --persona <label-or-hex> or set $PERSONA)");
    };
    async {
        let health = HealthSources::open(pile, signer, health_max_age).await?.observe(pile, signer)?;
        let health_events = health.attention().attention;
        let sources = OrientSources::open(pile, signer, false).await?;
        maintain_inputs(pile, signer, &sources).await?;
        let observation = observe_current_sources_acquiring(pile, &sources).await?;
        let (persona, events) = read(pile, &observation.snapshot, |reader| {
            let query = observation.query(reader);
            let persona = resolve_native_persona(&query, input)?;
            let view = load_attention_view(&query, persona)?;
            Ok((
                persona,
                view.ids()
                    .chain(health_events.ids())
                    .collect::<BTreeSet<_>>()
                    .into_iter()
                    .collect::<Vec<_>>(),
            ))
        })
        .await?;
        save_presentations(pile, signer, events.iter().copied())?;
        Ok(BaselineReceipt {
            persona,
            events: events.len(),
        })
    }
    .await
}

async fn cmd_show(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    pile_path: &Path,
    persona: Option<&str>,
    message_limit: usize,
    doing_limit: usize,
    todo_limit: usize,
    evaluate_habits: bool,
    health_max_age: Duration,
    output: &mut Out<'_>,
) -> Result<()> {
    use std::fmt::Write as _;

    async {
        let health_sources = HealthSources::open(pile, signer, health_max_age).await?;
        let health = health_sources.observe_acquiring(pile, signer).await?;
        let health_report = health.report_acquiring().await?;
        write_complete_report(output, &health_report.text, "local swarm health overview")?;
        if let Some(input) = persona {
            match health.persona_acquiring(pile, input).await {
                Ok(_) => save_presentations(pile, signer, health_report.attention.ids())?,
                Err(error) if is_payload_pending(&error) || is_persona_not_found(&error) => {}
                Err(error) => return Err(error),
            }
        }
        let sources = OrientSources::open(pile, signer, true).await?;
        maintain_inputs(pile, signer, &sources).await?;
        refresh_receipts_before_observation(pile, signer, &sources, output).await?;
        let instant = clock::now()?;
        let observation = observe_current_sources_acquiring(pile, &sources).await?;
        let (persona_id, messages, mail, habits, goals, window_status, shown) =
            read(pile, &observation.snapshot, |reader| {
                let query = observation.query(reader);
                let persona_id = persona
                    .map(|input| resolve_native_persona(&query, input))
                    .transpose()?;
                let (messages, message_events) =
                    render_native_messages(&query, persona_id, message_limit)?;
                let (mail, mail_events) = render_native_mail(&query, persona_id, message_limit)?;
                let habits = if evaluate_habits {
                    Some(prepare_habits(
                        reader,
                        query
                            .habits
                            .expect("Show opens the Habit source collection"),
                        persona_id,
                    )?)
                } else {
                    None
                };
                let (goals, goal_events) =
                    render_native_compass_goals(&query, doing_limit, todo_limit)?;
                let (window_status, status_events) = render_window_status(&query)?;
                let shown = match persona_id {
                    Some(persona) => {
                        let candidates: BTreeSet<_> =
                            load_attention_view(&query, persona)?.ids().collect();
                        message_events
                            .into_iter()
                            .chain(mail_events)
                            .chain(goal_events)
                            .chain(status_events)
                            .filter(|event| candidates.contains(event))
                            .collect::<Vec<_>>()
                    }
                    None => Vec::new(),
                };
                Ok((
                    persona_id,
                    messages,
                    mail,
                    habits,
                    goals,
                    window_status,
                    shown,
                ))
            })
            .await?;
        let habits = match habits {
            Some(habits) => {
                render_native_habits(&observe_habits(habits, pile_path, epoch_seconds(instant))?)
            }
            None => render_passive_habits(
                observation
                    .facts
                    .habits
                    .as_ref()
                    .expect("Show opens Habits")
                    .view(),
                persona_id,
            ),
        };

        let mut report = String::new();
        writeln!(report, "Orient").unwrap();
        report.push_str(&messages);
        report.push_str(&mail);
        report.push('\n');
        report.push_str(&habits);
        report.push_str(&goals);
        report.push_str(&window_status);

        write_complete_report(output, &report, "Orient overview")?;

        if persona_id.is_some() {
            save_presentations(pile, signer, shown)?;
        }
        Ok(())
    }
    .await
}

/// Render only the *novel* content behind the news — new peer messages, Mail
/// and Teams messages, plus newly-arrived roster members — so a woken watcher gets what changed,
/// not a full re-dump of the snapshot. The `News:` reason lines are rendered by
/// the caller; this fills in the detail worth reading.
fn render_news_detail(
    query: &OrientQuery<'_>,
    pending: &AttentionView,
    persona_id: Id,
) -> Result<String> {
    use std::fmt::Write as _;

    let mut out = String::new();
    let new_msgs: Vec<Id> = pending
        .events
        .values()
        .filter_map(|event| match event {
            AttentionEvent::Message(id) => Some(*id),
            _ => None,
        })
        .collect();
    if !new_msgs.is_empty() {
        let rows = native_message_rows(query);
        writeln!(out, "\nNew messages:").unwrap();
        for id in &new_msgs {
            if let Some(row) = rows.iter().find(|r| r.id == *id) {
                let from = read_native_person_label(query, row.from)?;
                let body = read_utf8(&query.payloads, row.body, "Message body")?;
                writeln!(out, "- {from}: {body}").unwrap();
            }
        }
    }
    let new_mail: Vec<Id> = pending
        .events
        .values()
        .filter_map(|event| match event {
            AttentionEvent::Mail(id) => Some(*id),
            _ => None,
        })
        .collect();
    if !new_mail.is_empty() {
        let summaries = native_unread_mail(query, persona_id)?;
        writeln!(out, "\nNew mail:").unwrap();
        for wire in &new_mail {
            let summary = summaries.get(wire).ok_or_else(|| {
                anyhow!("new Mail wire {} vanished from current view", fmt_id(*wire))
            })?;
            let from = summary
                .from
                .map(|handle| read_utf8(&query.payloads, handle, "Mail From"))
                .transpose()?
                .unwrap_or_else(|| "(no From)".to_owned());
            let subject = read_utf8(&query.payloads, summary.subject, "Mail subject")?;
            writeln!(out, "- [{}] {} — {}", fmt_id(*wire), from, subject,).unwrap();
        }
    }
    let new_teams: Vec<Id> = pending
        .events
        .values()
        .filter_map(|event| match event {
            AttentionEvent::Teams(id) => Some(*id),
            _ => None,
        })
        .collect();
    if !new_teams.is_empty() {
        writeln!(out, "\nNew Teams messages:").unwrap();
        for message in &new_teams {
            let (author, content) = teams_message_detail(query, *message)?;
            writeln!(out, "- {author}: {content}").unwrap();
        }
    }
    let new_discord: Vec<Id> = pending
        .events
        .values()
        .filter_map(|event| match event {
            AttentionEvent::Discord(id) => Some(*id),
            _ => None,
        })
        .collect();
    if !new_discord.is_empty() {
        writeln!(out, "\nNew Discord messages:").unwrap();
        for message in &new_discord {
            let detail = discord_message_detail(query, *message)?;
            // Distinct versions that print alike (they differ in a Discord
            // attachment id, say) are counted, and printed once.
            let mut printed = BTreeSet::new();
            for version in &detail.versions {
                let text = discord_version_text(&detail, version);
                if printed.insert(text.clone()) {
                    out.push_str(&text);
                }
            }
        }
    }
    let new_people: Vec<Id> = pending
        .events
        .values()
        .filter_map(|event| match event {
            AttentionEvent::StatusWindow(id) if *id != persona_id => Some(*id),
            _ => None,
        })
        .collect();
    if !new_people.is_empty() {
        writeln!(out, "\nNew status window(s):").unwrap();
        for id in &new_people {
            writeln!(out, "- {}", read_native_person_label(query, *id)?).unwrap();
        }
    }
    Ok(out)
}

/// Longest preview one `News:` line may carry from a stored body.
///
/// News is an attention channel an agent reads a line at a time, so a bounded
/// first line is the whole point: the body stays in Compass instead of being
/// re-delivered on every wake.
const NEWS_PREVIEW_CHARS: usize = 96;

/// Longest goal title one `News:` line may carry.
const NEWS_TITLE_CHARS: usize = 72;

/// Short, still-actionable form of an id for a News line.
///
/// Compass resolves hex prefixes (`resolve_id_prefix`), so eight characters
/// remain a usable argument to `compass show`/`move`/`note` at a quarter of the
/// tokens a full id costs. The full id is printed wherever no human-readable
/// name accompanies it, because there an exact id is the only thing the reader
/// can act on.
fn fmt_short_id(id: Id) -> String {
    fmt_id(id).chars().take(8).collect()
}

/// The first non-empty line of `text`, clipped to `limit` characters.
///
/// `None` for a body with nothing to show, so callers drop the clause entirely
/// rather than printing an empty preview.
fn clip_line(text: &str, limit: usize) -> Option<String> {
    let mut lines = text.lines().map(str::trim).filter(|line| !line.is_empty());
    let first = lines.next()?;
    let more_lines = lines.next().is_some();
    let clipped: String = first.chars().take(limit).collect();
    let clipped_chars = clipped.chars().count();
    if clipped_chars < first.chars().count() || more_lines {
        Some(format!("{}…", clipped.trim_end()))
    } else {
        Some(clipped)
    }
}

/// Best-effort text behind a handle for a News line.
///
/// An absent or not-yet-acquired blob degrades to no preview. Propagating the
/// error instead would turn a decorative body into a `MissingBlob`, and
/// `prepare_news_once` treats that as a pending payload — withholding the whole
/// report over text the reader never had before.
fn news_text(query: &OrientQuery<'_>, handle: compass::TextHandle, label: &str) -> Option<String> {
    read_utf8(&query.payloads, handle, label).ok()
}

/// How a goal is named in a News line: `[short] "Title"` when a title reads,
/// the full id otherwise.
///
/// Open world: a goal with no `board::title`, or one whose title blob has not
/// arrived, is named by the id the reader can still act on. A placeholder word
/// would be strictly less useful than the identifier it replaced.
fn news_goal_name(query: &OrientQuery<'_>, goal: Id) -> String {
    let title = find!(
        handle: compass::TextHandle,
        pattern!(query.compass, [{ goal @ board::title: ?handle }])
    )
    .next()
    .and_then(|handle| news_text(query, handle, "Compass title"))
    .and_then(|title| clip_line(&title, NEWS_TITLE_CHARS));
    match title {
        Some(title) => format!("[{}] \"{title}\"", fmt_short_id(goal)),
        None => format!("[{}]", fmt_id(goal)),
    }
}

/// ` by <label>` when an event records an acting persona.
///
/// `board::by` is optional attribution with no workflow semantics, so an absent
/// author simply drops the clause.
fn news_actor(query: &OrientQuery<'_>, event: Id) -> String {
    find!(by: Id, pattern!(query.compass, [{ event @ board::by: ?by }]))
        .next()
        .and_then(|by| read_native_person_label(query, by).ok())
        .map(|label| format!(" by {label}"))
        .unwrap_or_default()
}

/// The lane a goal left, when the ledger records one before `current`.
///
/// One point-of-use query over the status events of a single goal — a handful
/// of rows — ordered the way the LWW register orders them, so the predecessor
/// named here is the state the winning event replaced. `None` for a goal whose
/// first status this is.
fn previous_goal_status(query: &OrientQuery<'_>, goal: Id, current: Id) -> Option<String> {
    let mut events: Vec<(i128, Id, String)> = find!(
        (event: Id, status: String, at: IntervalValue),
        pattern!(query.compass, [{ ?event @
            metadata::tag: &KIND_STATUS_ID,
            board::status_of: &goal,
            board::status: ?status,
            metadata::created_at: ?at,
        }])
    )
    .map(|(event, status, at)| (interval_key(at), event, status))
    .collect();
    events.sort();
    let position = events.iter().position(|(_, event, _)| *event == current)?;
    events[..position]
        .last()
        .map(|(_, _, status)| status.clone())
}

/// The `News:` line for one attention event, carrying the content the reader
/// would otherwise have to run another command to see.
///
/// Every lookup is a point-of-use query keyed on one entity, and this runs only
/// over *pending* events — the handful that are new since the last receipt, not
/// the candidate set. It therefore adds nothing to the per-goal scan
/// `load_attention_view` already performs on every call.
///
/// Mail and Teams keep their reason lines: `render_news_detail` already prints
/// sender, subject and body for those beneath the reasons.
fn news_line(query: &OrientQuery<'_>, event: &AttentionEvent) -> String {
    match event {
        AttentionEvent::Message(id) => {
            let id = *id;
            match find!(
                from: Id,
                pattern!(query.messages, [{ id @ local_message::from: ?from }])
            )
            .next()
            .and_then(|from| read_native_person_label(query, from).ok())
            {
                // The short id stays, unlike the goal lines where a title
                // carries the identity: a message is the one thing here you act
                // on BY id (`message ack <id> <persona>`), and without it the
                // only way back to that id is `message list`, which costs 43 s
                // on the live pile. A news line you cannot act on sends you to
                // a slow lookup to recover what it just threw away.
                Some(from) => format!("new message [{}] from {from}", fmt_short_id(id)),
                None => event.reason(),
            }
        }
        AttentionEvent::Goal {
            event: status_event,
            goal,
            status,
        } if status_event == goal => {
            format!("new goal {} ({status})", news_goal_name(query, *goal))
        }
        AttentionEvent::Goal {
            event: status_event,
            goal,
            status,
        } => {
            let name = news_goal_name(query, *goal);
            let actor = news_actor(query, *status_event);
            match previous_goal_status(query, *goal, *status_event) {
                Some(previous) if !previous.eq_ignore_ascii_case(status) => {
                    format!("goal {name}: {previous} -> {status}{actor}")
                }
                _ => format!("goal {name} is now {status}{actor}"),
            }
        }
        AttentionEvent::Note { note, goal } => {
            let name = news_goal_name(query, *goal);
            let actor = news_actor(query, *note);
            let note = *note;
            let preview = find!(
                handle: compass::TextHandle,
                pattern!(query.compass, [{ note @ board::note: ?handle }])
            )
            .next()
            .and_then(|handle| news_text(query, handle, "Compass note"))
            .and_then(|body| clip_line(&body, NEWS_PREVIEW_CHARS))
            .map(|body| format!(": {body}"))
            .unwrap_or_default();
            format!("note on {name}{actor}{preview}")
        }
        AttentionEvent::StatusWindow(window) => match read_native_person_label(query, *window) {
            Ok(label) => format!("new status window {label}"),
            Err(_) => event.reason(),
        },
        AttentionEvent::Discord(message) => match discord_news_line(query, *message) {
            Some(line) => line,
            None => event.reason(),
        },
        AttentionEvent::Mail(_) | AttentionEvent::Teams(_) | AttentionEvent::Health { .. } => {
            event.reason()
        }
    }
}

enum News {
    Quiet,
    Report { text: String, events: Vec<Id> },
}

fn write_complete_report(output: &mut Out<'_>, report: &str, description: &str) -> Result<()> {
    // An accepted complete native Part is the delivery boundary. The CLI sink
    // flushes it before returning. A collecting frontend accepts it into its
    // response; this is deliberately not a remote receipt or processing ack.
    output
        .text(report)
        .with_context(|| format!("deliver complete {description}"))
}

fn prepare_news_once(query: &OrientQuery<'_>, persona_id: Id) -> Result<News> {
    let candidates = load_attention_view(query, persona_id)?;
    let pending = candidates.pending(query.presentations);
    if pending.is_empty() {
        return Ok(News::Quiet);
    }
    use std::fmt::Write as _;

    let mut text = String::new();
    for event in pending.events.values() {
        writeln!(text, "News: {}", news_line(query, event)).unwrap();
    }
    text.push_str(&render_news_detail(query, &pending, persona_id)?);
    Ok(News::Report {
        text,
        events: pending.ids().collect(),
    })
}

/// Record that these due events were presented to this persona.
///
/// A habit report is not news and must never acknowledge a message body the
/// reader did not see -- that is why the two are receipted separately rather
/// than folded into one fragment. What it DOES acknowledge is itself: the due
/// events it just displayed, so a rearmed watcher does not display them again.
fn commit_habit_receipts(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    due: &[(Id, i64)],
) -> Result<()> {
    if due.is_empty() {
        return Ok(());
    }
    let mut pairs = Vec::with_capacity(due.len());
    for (habit, since) in due {
        pairs.push((
            *habit,
            clock::point(Epoch::from_tai_seconds(*since as f64))?,
        ));
    }
    let collection = presentation_collection_for_write(pile, signer)?;
    let fragment = orient_model::habit_receipt_fragment(pairs, clock::point_now()?);
    pile.commit(collection, signer, fragment)
        .map_err(|error| anyhow!("commit Orient habit presentation facts: {error}"))?;
    Ok(())
}

fn apply_prepared_news(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    peek: bool,
    prepared: &News,
    prefix: &str,
    output: &mut Out<'_>,
) -> Result<()> {
    match prepared {
        News::Report { text, events } => {
            // Do not deliver consuming news that this principal cannot record.
            // Target production authority is independent and may be remote.
            let receipt = if !peek && !events.is_empty() {
                Some((
                    presentation_collection_for_write(pile, signer)?,
                    orient_model::receipt_fragment(events.iter().copied(), clock::point_now()?),
                ))
            } else {
                None
            };
            let mut complete = String::with_capacity(prefix.len() + text.len());
            complete.push_str(prefix);
            complete.push_str(text);
            write_complete_report(output, &complete, "Orient news report")?;
            if let Some((collection, fragment)) = receipt {
                pile.commit(collection, signer, fragment)
                    .map_err(|error| anyhow!("commit Orient presentation facts: {error}"))?;
            }
        }
        News::Quiet => {
            if !prefix.is_empty() {
                write_complete_report(output, prefix, "Orient habit report")?;
            }
        }
    }
    Ok(())
}

/// One-shot `wait`: acquire selected payloads, report pending attention
/// tersely, and only then record presentation. No provider means quiet pending,
/// not acknowledgment of a body the recipient never saw.
async fn cmd_poll(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    persona: Option<&str>,
    peek: bool,
    health_max_age: Duration,
    output: &mut Out<'_>,
) -> Result<()> {
    let Some(input) = persona else {
        bail!("poll requires a persona (pass --persona <label-or-hex> or set $PERSONA)");
    };
    async {
        let mut health = HealthSources::open(pile, signer, health_max_age).await?;
        if health.poll(pile, signer, input, peek, output)?.0 {
            return Ok(());
        }
        let sources = OrientSources::open(pile, signer, false).await?;
        if let Err(error) = maintain_inputs(pile, signer, &sources).await {
            if is_preparation_pending(&error) {
                return Ok(());
            }
            return Err(error);
        }
        refresh_receipts_before_observation(pile, signer, &sources, output).await?;
        let observation = match observe_current_sources_acquiring(pile, &sources).await {
            Ok(observation) => observation,
            Err(error) if is_preparation_pending(&error) => return Ok(()),
            Err(error) => return Err(error),
        };
        let prepared = read(pile, &observation.snapshot, |reader| {
            let query = observation.query(reader);
            let persona = resolve_native_persona(&query, input)?;
            prepare_news_once(&query, persona)
        })
        .await;
        let news = match prepared {
            Ok(prepared) => prepared,
            Err(error) if is_payload_pending(&error) || is_persona_not_found(&error) => {
                return Ok(())
            }
            Err(error) => return Err(error),
        };
        apply_prepared_news(pile, signer, peek, &news, "", output)?;
        Ok(())
    }
    .await
}

struct WaitOutcome {
    news_printed: bool,
    view_pending: bool,
    had_ready_frame: bool,
}

struct WaitFrame {
    watermark: FacultySnapshot,
    observation: OrientObservation,
    persona: Id,
    habits: HabitObservation,
    news: News,
}

struct PendingWaitFrame {
    watermark: FacultySnapshot,
    reason: PendingWaitReason,
    /// An exact failed read can make progress when a provider appears even
    /// though the pile prefix is unchanged. Projection absence cannot.
    missing: Option<MissingBlob>,
    /// Keep the actual immutable targets selected before payload acquisition.
    /// Retrying a body or persona label must not replace these target views.
    observation: Option<OrientObservation>,
    /// The persona this observation resolved, once it did.
    persona: Option<Id>,
    /// Habit readiness depends on the Habit source alone: the observation
    /// evaluated at this watermark once persona and Habit payloads were
    /// readable, kept across pure body retries so no condition script reruns,
    /// and cleared whenever persona or preparation is pending again.
    habits: Option<HabitObservation>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum PendingWaitReason {
    /// A required descriptor or selected input cannot yet be read.
    Preparation,
    /// The persona selector cannot yet be resolved in this observation.
    PersonaSelection,
    /// The persona is settled, but another selected payload is unavailable.
    Payload,
}

impl PendingWaitFrame {
    fn awaiting_view(watermark: FacultySnapshot, reason: PendingWaitReason) -> Self {
        Self {
            watermark,
            reason,
            missing: None,
            observation: None,
            persona: None,
            habits: None,
        }
    }
}

fn wait_storage_changed(sampled: &FacultySnapshot, previous: &FacultySnapshot) -> bool {
    let changes = sampled.changes_since(previous);
    // Blob changes stay conservative: a new fallback occurrence can repair a
    // previously unusable member even when no new content hash was added.
    // Generic authority has no clock expiry, and WANT is not a query input.
    changes.contains(StoreChanges::BLOBS)
        || changes.contains(StoreChanges::COLLECTION_RECORDS)
        || changes.contains(StoreChanges::CAPABILITY_PROOFS)
}

fn pending_blob(error: &anyhow::Error) -> Option<MissingBlob> {
    error.chain().find_map(|source| {
        source.downcast_ref::<MissingBlob>().copied().or_else(|| {
            match source.downcast_ref::<CollectionRealizationError>() {
                Some(CollectionRealizationError::MissingDependency { member }) => {
                    Some(MissingBlob {
                        handle: Inline::new(member.raw),
                    })
                }
                _ => None,
            }
        })
    })
}

enum WaitFrameLoad {
    Pending(PendingWaitFrame),
    Ready(WaitFrame),
    /// Eager upkeep succeeded without changing any selected dependency.
    /// Keep the ready observation and its application-time Habit context.
    Retained(FacultySnapshot),
}

impl WaitFrameLoad {
    fn watermark_snapshot(&self) -> &FacultySnapshot {
        match self {
            Self::Pending(pending) => &pending.watermark,
            Self::Ready(frame) => &frame.watermark,
            Self::Retained(snapshot) => snapshot,
        }
    }
}

async fn load_wait_frame(
    pile: &mut FacultyStore,
    sources: &OrientSources,
    snapshot: FacultySnapshot,
    pending: &mut Option<PendingWaitFrame>,
    pile_path: &Path,
    persona_input: &str,
    evaluated_at: Epoch,
) -> Result<WaitFrameLoad> {
    // The observation and polling baseline are the same immutable prefix.
    // Later selected-payload acquisition cannot swallow a concurrent append.
    let observation = match observe_snapshot_acquiring(snapshot.clone(), sources, true).await {
        Ok(observation) => observation,
        Err(error) if is_preparation_pending(&error) => {
            let mut fresh =
                PendingWaitFrame::awaiting_view(snapshot, PendingWaitReason::Preparation);
            fresh.missing = pending_blob(&error);
            return Ok(WaitFrameLoad::Pending(fresh));
        }
        Err(error) => return Err(error),
    };
    // The fresh frame lives in the caller's slot while its payloads are read,
    // so a boundary that cancels the read keeps what the read had already
    // resolved: the persona and the habit observation.
    let mut fresh = PendingWaitFrame::awaiting_view(snapshot, PendingWaitReason::Payload);
    fresh.observation = Some(observation);
    let slot = pending.insert(fresh);
    match resume_wait_payloads(pile, slot, pile_path, persona_input, evaluated_at).await? {
        Some(frame) => {
            pending.take();
            Ok(WaitFrameLoad::Ready(frame))
        }
        None => Ok(WaitFrameLoad::Pending(
            pending.take().expect("pending payload frame"),
        )),
    }
}

/// Retry only the selected immutable inputs. This borrows the cached view so a
/// health deadline can cancel the network read without losing that selection.
async fn resume_wait_payloads<S>(
    pile: &mut S,
    pending: &mut PendingWaitFrame,
    pile_path: &Path,
    persona_input: &str,
    instant: Epoch,
) -> Result<Option<WaitFrame>>
where
    S: SnapshotSource<Snapshot = FacultySnapshot>
        + triblespace::core::repo::async_store::AsyncBlobStoreAcquire,
{
    let observation = pending
        .observation
        .as_mut()
        .expect("payload continuation retains its selected views");
    let (persona, reader) = match read(pile, &observation.snapshot, |reader| {
        let persona = resolve_native_persona(&observation.query(reader), persona_input)?;
        Ok((persona, reader.clone()))
    })
    .await
    {
        Ok(persona) => persona,
        Err(error) if is_payload_pending(&error) || is_persona_not_found(&error) => {
            pending.reason = PendingWaitReason::PersonaSelection;
            pending.missing = pending_blob(&error);
            // A different persona would select different intentions.
            pending.persona = None;
            pending.habits = None;
            return Ok(None);
        }
        Err(error) => return Err(error),
    };
    observation.snapshot = reader;
    pending.persona = Some(persona);
    // Habit readiness comes first and depends on the Habit source alone.
    // Evaluate it once at this watermark; a pure body retry reuses it and
    // reruns no condition script.
    if pending.habits.is_none() {
        let prepared = read(pile, &observation.snapshot, |reader| {
            let query = observation.query(reader);
            let habits = prepare_habits(
                reader,
                query
                    .habits
                    .expect("Wait opens the Habit source collection"),
                Some(persona),
            )?;
            Ok((reader.clone(), habits))
        })
        .await;
        let (reader, habits) = match prepared {
            Ok(prepared) => prepared,
            Err(error) if is_payload_pending(&error) => {
                pending.reason = PendingWaitReason::Payload;
                pending.missing = pending_blob(&error);
                return Ok(None);
            }
            Err(error) => return Err(error),
        };
        // Retain acquired bytes for later timer-driven Habit evaluation. The
        // selected target views and polling watermark remain untouched.
        observation.snapshot = reader;
        pending.habits = Some(observe_habits(habits, pile_path, epoch_seconds(instant))?);
    }
    // News readiness depends on the directed-news bodies. A missing body
    // keeps the frame pending without discarding the habit observation.
    let prepared = read(pile, &observation.snapshot, |reader| {
        let query = observation.query(reader);
        let news = prepare_news_once(&query, persona)?;
        Ok((reader.clone(), news))
    })
    .await;
    let (reader, news) = match prepared {
        Ok(prepared) => prepared,
        Err(error) if is_payload_pending(&error) => {
            pending.reason = PendingWaitReason::Payload;
            pending.missing = pending_blob(&error);
            return Ok(None);
        }
        Err(error) => return Err(error),
    };
    observation.snapshot = reader;
    let habits = pending
        .habits
        .take()
        .expect("habit readiness precedes news readiness");
    Ok(Some(WaitFrame {
        watermark: pending.watermark.clone(),
        observation: pending
            .observation
            .take()
            .expect("selected payloads are ready"),
        persona,
        habits,
        news,
    }))
}

fn observe_habits_in_observation(
    observation: &OrientObservation,
    pile_path: &Path,
    now_secs: i64,
    persona: Id,
) -> Result<HabitObservation> {
    let facts = observation
        .facts
        .habits
        .as_ref()
        .expect("a wait observation includes Habit");
    observe_habits(
        prepare_habits(&observation.snapshot, facts.view(), Some(persona))?,
        pile_path,
        now_secs,
    )
}

/// A timer sweep refreshes the same selected Habit context whose cached
/// evaluation it replaces. A newer pending frame may already have readable
/// habits even while its news body is missing; never copy an evaluation of
/// `current` into that frame. Pure payload retries do not call this function.
fn sweep_wait_habits(
    current: &OrientObservation,
    persona: Id,
    pending: &mut Option<PendingWaitFrame>,
    pile_path: &Path,
    now_secs: i64,
) -> Result<HabitObservation> {
    if let Some(pending) = pending.as_mut() {
        if let (Some(persona), Some(_), Some(observation)) = (
            pending.persona,
            pending.habits.as_ref(),
            pending.observation.as_ref(),
        ) {
            let habits = observe_habits_in_observation(observation, pile_path, now_secs, persona)?;
            pending.habits = Some(habits.clone());
            return Ok(habits);
        }
    }
    observe_habits_in_observation(current, pile_path, now_secs, persona)
}

/// Selected payload acquisition yields at a local health
/// validity boundary. Completed cache work remains reusable when the caller
/// re-observes health before resuming this attention view.
async fn load_wait_frame_before_health_deadline(
    pile: &mut FacultyStore,
    sources: &OrientSources,
    mut snapshot: FacultySnapshot,
    pending: &mut Option<PendingWaitFrame>,
    pile_path: &Path,
    persona_input: &str,
    next_health_change: Option<Epoch>,
    mut evaluated_at: Epoch,
) -> Result<Option<WaitFrameLoad>> {
    let unchanged = pending.as_ref().is_some_and(|pending| {
        pending.observation.as_ref().map_or_else(
            || !wait_storage_changed(&snapshot, &pending.watermark),
            |observation| observation.is_current(&snapshot),
        )
    });
    if unchanged {
        let cached = pending.as_mut().expect("unchanged pending frame exists");
        if cached.observation.is_some() {
            // A selected observation whose payload read failed or was cut at
            // a boundary is retried; the persona and habits it already
            // resolved are reused.
            let ready = tokio::select! {
                boundary = health::deadline(next_health_change) => {
                    boundary?;
                    return Ok(None);
                }
                frame = resume_wait_payloads(
                    pile, cached, pile_path, persona_input, evaluated_at,
                ) => frame?,
            };
            return Ok(Some(match ready {
                Some(frame) => WaitFrameLoad::Ready(frame),
                None => WaitFrameLoad::Pending(pending.take().expect("pending payload")),
            }));
        }
        if let Some(missing) = cached.missing {
            // An exact preparation dependency can also become available from
            // a new provider. Retry that handle, not the entire preparation.
            let acquired = tokio::select! {
                boundary = health::deadline(next_health_change) => {
                    boundary?;
                    return Ok(None);
                }
                bytes = pile.acquire(missing.handle) => bytes?,
            };
            if acquired.is_none() {
                return Ok(Some(WaitFrameLoad::Pending(
                    pending.take().expect("pending preparation dependency"),
                )));
            }
            // No observation was selected yet. Retry preparation at the
            // prefix containing the acquired dependency, not the old snapshot
            // which necessarily still lacks it. This starts a fresh application
            // evaluation as well; selected-payload retries above keep their time.
            evaluated_at = clock::now()?;
            snapshot = pile.snapshot()?;
        } else {
            // Nothing in this outcome performs I/O: absent projection records
            // or an absent persona require a different store observation.
            return Ok(Some(WaitFrameLoad::Pending(
                pending.take().expect("unchanged pending observation"),
            )));
        }
    }
    tokio::select! {
        boundary = health::deadline(next_health_change) => {
            boundary?;
            Ok(None)
        }
        frame = load_wait_frame(pile, sources, snapshot, pending, pile_path, persona_input, evaluated_at) => frame.map(Some),
    }
}

/// Eager upkeep belongs before view selection, never inside a selected payload
/// retry. Keep the pre-upkeep prefix as the attempted baseline: an append to an
/// earlier input during upkeep must still be noticed on the next poll. Our own
/// equations can consequently cause one extra, quiet upkeep pass.
async fn prepare_wait_frame_before_health_deadline(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    sources: &OrientSources,
    mut snapshot: FacultySnapshot,
    maintained_from: &mut Option<FacultySnapshot>,
    upkeep_interrupted: &mut bool,
    pending: &mut Option<PendingWaitFrame>,
    current: Option<&OrientObservation>,
    pile_path: &Path,
    persona_input: &str,
    next_health_change: Option<Epoch>,
    timeout_at: Option<tokio::time::Instant>,
    mut evaluated_at: Epoch,
    output: &mut Out<'_>,
) -> Result<Option<WaitFrameLoad>> {
    if maintained_from
        .as_ref()
        .is_none_or(|previous| wait_storage_changed(&snapshot, previous))
    {
        let before = snapshot.clone();
        let result = wait_upkeep_before_deadline(
            async {
                maintain_inputs(pile, signer, sources).await?;
                refresh_receipts_before_observation(pile, signer, sources, output).await
            },
            upkeep_interrupted,
            next_health_change,
            timeout_at,
        )
        .await;
        match result {
            Ok(false) => return Ok(None),
            Ok(true) => {}
            Err(error) => {
                if !is_preparation_pending(&error) {
                    return Err(error);
                }
                let mut fresh =
                    PendingWaitFrame::awaiting_view(before, PendingWaitReason::Preparation);
                fresh.missing = pending_blob(&error);
                return Ok(Some(WaitFrameLoad::Pending(fresh)));
            }
        }
        *maintained_from = Some(before);
        snapshot = pile.snapshot()?;
        evaluated_at = clock::now()?;
    }
    // A broad source change can require upkeep without changing the selected
    // projections. Reuse only a fully ready view, never a pending payload or
    // an interrupted upkeep attempt. Keep the pre-upkeep maintenance baseline
    // above so a concurrent append to an earlier input is still noticed.
    if pending.is_none() && current.is_some_and(|current| current.is_current(&snapshot)) {
        return Ok(Some(WaitFrameLoad::Retained(snapshot)));
    }
    tokio::select! {
        _ = wait_timeout_deadline(timeout_at) => Ok(None),
        frame = load_wait_frame_before_health_deadline(
            pile, sources, snapshot, pending, pile_path, persona_input,
            next_health_change, evaluated_at,
        ) => frame,
    }
}

/// Poll eager upkeep only until a clock or command boundary needs the caller.
/// A cancelled attempt cannot certify its prefix or replace a selected frame;
/// those actions remain in the caller's successful-completion branch.
async fn wait_upkeep_before_deadline<F>(
    upkeep: F,
    upkeep_interrupted: &mut bool,
    next_health_change: Option<Epoch>,
    timeout_at: Option<tokio::time::Instant>,
) -> Result<bool>
where
    F: std::future::Future<Output = Result<()>>,
{
    tokio::select! {
        _ = wait_timeout_deadline(timeout_at) => {
            *upkeep_interrupted = true;
            Ok(false)
        }
        boundary = health::deadline(next_health_change) => {
            boundary?;
            *upkeep_interrupted = true;
            Ok(false)
        }
        result = upkeep => {
            result?;
            *upkeep_interrupted = false;
            Ok(true)
        }
    }
}

/// The command budget is monotonic; wall-clock corrections only affect the
/// health/Habit deadlines, never how long a bounded wait may run.
async fn wait_timeout_deadline(timeout_at: Option<tokio::time::Instant>) {
    match timeout_at {
        Some(deadline) => tokio::time::sleep_until(deadline).await,
        None => std::future::pending::<()>().await,
    }
}

async fn cmd_wait(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    pile_path: &Path,
    persona: Option<&str>,
    options: &WaitOptions,
    health_max_age: Duration,
    output: &mut Out<'_>,
) -> Result<()> {
    cmd_observe(
        pile,
        signer,
        pile_path,
        persona,
        options,
        health_max_age,
        false,
        output,
    )
    .await
}

/// Both frontends use the same state machine. Continuous observation does not
/// re-open the pile, re-arm already-due habits, or discard pending reads after
/// a delivery. Successful output still precedes the existing receipt COMMIT.
async fn cmd_observe(
    pile: &mut FacultyStore,
    signer: &SigningKey,
    pile_path: &Path,
    persona: Option<&str>,
    options: &WaitOptions,
    health_max_age: Duration,
    continuous: bool,
    output: &mut Out<'_>,
) -> Result<()> {
    let Some(persona_input) = persona else {
        bail!("wait requires a persona (pass --persona <label-or-hex> or set $PERSONA)");
    };
    let timeout = options.timeout;
    let result: Result<WaitOutcome> = async {
        let mut health = HealthSources::open(pile, signer, health_max_age).await?;
        let poll = options.poll_interval.max(Duration::from_millis(1));
        let start = Instant::now();
        let timeout_at = timeout.map(|timeout| tokio::time::Instant::now() + timeout);
        let mut view_pending;
        let mut next_health_change;
        let mut pending_frame: Option<PendingWaitFrame> = None;
        let mut maintained_from: Option<FacultySnapshot> = None;
        let mut upkeep_interrupted = false;
        // While directed news is still pending, the persona's own clocks keep
        // running against the pending observation. Their baseline and sweep
        // cadence mirror the ready frame's.
        let mut pending_habits_seen: Option<HabitObservation> = None;
        let mut last_pending_sweep = Instant::now();

        let sources = loop {
            let (fired, deadline) = health.poll(pile, signer, persona_input, false, output)?;
            next_health_change = deadline;
            if fired && !continuous {
                return Ok(WaitOutcome {
                    news_printed: true,
                    view_pending: false,
                    had_ready_frame: true,
                });
            }
            tokio::select! {
                _ = wait_timeout_deadline(timeout_at) => return Ok(WaitOutcome {
                    news_printed: false,
                    view_pending: true,
                    had_ready_frame: false,
                }),
                boundary = health::deadline(next_health_change) => { boundary?; }
                sources = OrientSources::open(pile, signer, true) => break sources?,
            }
        };
        // Keep the prefix which selected these target views as the polling
        // watermark, including while their lazy payload reads are pending.
        let initial = loop {
            let (fired, deadline) = health.poll(pile, signer, persona_input, false, output)?;
            next_health_change = deadline;
            if fired && !continuous {
                return Ok(WaitOutcome {
                    news_printed: true,
                    view_pending: false,
                    had_ready_frame: true,
                });
            }
            let evaluated_at = clock::now()?;
            let sampled = pile.snapshot()?;
            let probe = RefreshProbe::begin(
                "ordinary",
                &sampled,
                pending_frame.as_ref().map(|pending| &pending.watermark),
                pending_frame.as_ref(),
            );
            // A pending body fetch yields at the next habit deadline as well
            // as at the health boundary, so the clock gets its turn. The very
            // first read, before any frame is retained, gets one poll interval,
            // so a stalled body cannot keep the persona's clocks unobserved. A
            // retry keeps the health boundary when its baseline has no deadline
            // (empty or script-only intentions) or no baseline exists yet, so a
            // body or preparation payload that answers within its budget lands.
            // Upkeep cut before it could select a frame also gets that retry:
            // resetting the same short cap would starve a slower acquisition.
            let boundary = match pending_habit_deadline(
                continuous,
                pending_frame
                    .as_ref()
                    .and_then(|pending| pending.habits.as_ref()),
                &pending_habits_seen,
            ) {
                Some(deadline) => earliest(next_health_change, Some(deadline)),
                None if pending_frame.is_none() && !upkeep_interrupted => {
                    let first_read = Epoch::from_tai_seconds(
                        clock::now()?.to_tai_seconds() + poll.as_secs_f64(),
                    );
                    earliest(next_health_change, Some(first_read))
                }
                None => next_health_change,
            };
            let attempt = prepare_wait_frame_before_health_deadline(
                pile,
                signer,
                &sources,
                sampled,
                &mut maintained_from,
                &mut upkeep_interrupted,
                &mut pending_frame,
                None,
                pile_path,
                persona_input,
                boundary,
                timeout_at,
                evaluated_at,
                output,
            )
            .await;
            if let Some(probe) = &probe {
                probe.frame(None, &attempt);
            }
            if let Some(attempt) = attempt? {
                match attempt {
                    WaitFrameLoad::Ready(frame) => {
                        pending_frame = None;
                        view_pending = false;
                        break frame;
                    }
                    WaitFrameLoad::Pending(pending) => {
                        if matches!(
                            pending.reason,
                            PendingWaitReason::PersonaSelection | PendingWaitReason::Preparation
                        ) {
                            // Stop this context's clock without forgetting a
                            // daemon's already-announced due occurrences.
                            invalidate_pending_habit_context(continuous, &mut pending_habits_seen);
                        }
                        pending_frame = Some(pending);
                    }
                    WaitFrameLoad::Retained(_) => {
                        unreachable!("initial preparation has no ready observation to retain")
                    }
                }
            }
            // A retained frame without a habit observation (its intentions'
            // own payloads are still missing) carries no clock: an older
            // baseline's deadline must not keep cutting its reads.
            if pending_frame
                .as_ref()
                .is_some_and(|pending| pending.habits.is_none())
            {
                invalidate_pending_habit_context(continuous, &mut pending_habits_seen);
            }
            // Whether the read returned pending or was cut at the boundary,
            // the retained frame is what the persona's clocks run against.
            let mut swept: Option<(HabitObservation, String, Vec<(Id, i64)>)> = None;
            if let Some(pending) = pending_frame.as_ref() {
                if let (Some(persona), Some(habits), Some(observation)) = (
                    pending.persona,
                    pending.habits.as_ref(),
                    pending.observation.as_ref(),
                ) {
                    match &pending_habits_seen {
                        None => {
                            // An intention already due when the watcher arms,
                            // shared or addressed, is presented at once, even
                            // while a news body is still pending -- and
                            // receipted, so a rearmed watcher does not present
                            // the same occurrence again.
                            let (due_report, due_events) = render_due_habits_unreceipted(
                                habits,
                                observation.facts.presentations.view(),
                            )
                            .unwrap_or_else(|| (String::new(), Vec::new()));
                            pending_habits_seen = Some(habits.clone());
                            last_pending_sweep = Instant::now();
                            if !due_report.is_empty() {
                                write_complete_report(output, &due_report, "Orient habit report")?;
                                commit_habit_receipts(pile, signer, &due_events)?;
                                if !continuous {
                                    return Ok(WaitOutcome {
                                        news_printed: true,
                                        view_pending: true,
                                        had_ready_frame: false,
                                    });
                                }
                            }
                        }
                        Some(seen) => {
                            let now_secs = epoch_seconds(clock::now()?);
                            let cooldown_elapsed = seen
                                .next_cooldown_at
                                .is_some_and(|deadline| now_secs >= deadline);
                            let periodic_condition_check =
                                last_pending_sweep.elapsed() >= Duration::from_secs(60);
                            if cooldown_elapsed || periodic_condition_check {
                                let current_habits = observe_habits_in_observation(
                                    observation,
                                    pile_path,
                                    now_secs,
                                    persona,
                                )?;
                                let (habit_report, due_events) = render_habit_transitions(
                                    seen,
                                    &current_habits,
                                    observation.facts.presentations.view(),
                                )
                                .unwrap_or_else(|| (String::new(), Vec::new()));
                                swept = Some((current_habits, habit_report, due_events));
                            } else if continuous {
                                // A newly readable pending frame may contain
                                // a completion or a new due occurrence even
                                // while its directed-news body is unavailable.
                                let (report, due_events) = render_habit_transitions(
                                    seen,
                                    habits,
                                    observation.facts.presentations.view(),
                                )
                                .unwrap_or_else(|| (String::new(), Vec::new()));
                                if seen != habits {
                                    swept = Some((habits.clone(), report, due_events));
                                }
                            }
                        }
                    }
                }
            }
            if let Some((current_habits, habit_report, swept_events)) = swept {
                // The sweep's observation is what the retained frame carries
                // into Ready, so Ready compares against what was last seen and
                // never against the observation its first preparation cached.
                if let Some(pending) = pending_frame.as_mut() {
                    pending.habits = Some(current_habits.clone());
                }
                pending_habits_seen = Some(current_habits);
                last_pending_sweep = Instant::now();
                if !habit_report.is_empty() {
                    // A habit-only report acknowledges no news.
                    write_complete_report(output, &habit_report, "Orient habit report")?;
                    commit_habit_receipts(pile, signer, &swept_events)?;
                    if !continuous {
                        return Ok(WaitOutcome {
                            news_printed: true,
                            view_pending: true,
                            had_ready_frame: false,
                        });
                    }
                }
            }
            view_pending = true;

            if timeout.is_some_and(|timeout| start.elapsed() >= timeout) {
                return Ok(WaitOutcome {
                    news_printed: false,
                    view_pending,
                    had_ready_frame: false,
                });
            }
            let boundary = earliest(
                next_health_change,
                pending_habit_deadline(
                    continuous,
                    pending_frame
                        .as_ref()
                        .and_then(|pending| pending.habits.as_ref()),
                    &pending_habits_seen,
                ),
            );
            let sleep =
                health::until(boundary, clock::now()?).map_or(poll, |delay| delay.min(poll));
            let sleep = timeout.map_or(sleep, |timeout| {
                sleep.min(timeout.saturating_sub(start.elapsed()))
            });
            tokio::time::sleep(sleep).await;
            // The next poll only retries an exact failed payload read unless
            // storage changed. Provider appearance does not require rebuilding
            // an otherwise identical collection observation.
        };

        let mut observed_snapshot = initial.watermark.clone();

        let WaitFrame {
            watermark: _,
            observation: mut current,
            persona: mut persona_id,
            habits: mut habit_seen,
            news,
        } = initial;
        // Due-ness is decided by receipt: whatever this arm presents is
        // receipted, so a rearmed one-shot watcher does not repeat it, and an
        // occurrence already presented is not repeated here.
        let mut last_habit_sweep = Instant::now();
        let mut current_habit_context_valid = true;

        let initial_report = matches!(news, News::Report { .. });
        // With a baseline taken while news was pending, what fell due since is
        // news at this frame and the fresh observation is kept; without one,
        // an owned intention already due is reported at arm.
        let presented = current.facts.presentations.view();
        let (arm_report, arm_due_events) = match pending_habits_seen.take() {
            Some(seen) => render_habit_transitions(&seen, &habit_seen, presented),
            None => render_due_habits_unreceipted(&habit_seen, presented),
        }
        .unwrap_or_else(|| (String::new(), Vec::new()));
        let arm_fired = !arm_report.is_empty();
        apply_prepared_news(pile, signer, false, &news, &arm_report, output)?;
        commit_habit_receipts(pile, signer, &arm_due_events)?;
        if (initial_report || arm_fired) && !continuous {
            return Ok(WaitOutcome {
                news_printed: true,
                view_pending: false,
                had_ready_frame: true,
            });
        }

        loop {
            if let Some(timeout) = timeout {
                if start.elapsed() >= timeout {
                    return Ok(WaitOutcome {
                        news_printed: false,
                        view_pending,
                        had_ready_frame: true,
                    });
                }
            }
            let sleep = health::until(next_health_change, clock::now()?)
                .map_or(poll, |delay| delay.min(poll));
            let sleep = timeout.map_or(sleep, |timeout| {
                sleep.min(timeout.saturating_sub(start.elapsed()))
            });
            tokio::time::sleep(sleep).await;
            let (fired, deadline) = health.poll(pile, signer, persona_input, false, output)?;
            next_health_change = deadline;
            if fired && !continuous {
                return Ok(WaitOutcome {
                    news_printed: true,
                    view_pending: false,
                    had_ready_frame: true,
                });
            }
            let now = clock::now()?;
            let sampled = pile
                .snapshot()
                .map_err(|error| anyhow!("refresh Orient wait snapshot: {error}"))?;
            let maintenance_changed = maintained_from
                .as_ref()
                .is_none_or(|previous| wait_storage_changed(&sampled, previous));
            let storage_changed = maintenance_changed || !current.is_current(&sampled);
            let probe = RefreshProbe::begin(
                "ordinary",
                &sampled,
                Some(&observed_snapshot),
                pending_frame.as_ref(),
            );
            let now_secs = epoch_seconds(now);
            let cooldown_elapsed = habit_seen
                .next_cooldown_at
                .is_some_and(|deadline| now_secs >= deadline);
            let periodic_condition_check = last_habit_sweep.elapsed() >= Duration::from_secs(60);
            if !storage_changed && !view_pending && !cooldown_elapsed && !periodic_condition_check {
                if let Some(probe) = &probe {
                    probe.finish("unchanged");
                }
                continue;
            }

            if storage_changed || view_pending {
                // Ready clocks continue to run while a changed source needs
                // upkeep. An interruption must reach the retained sweep below.
                let boundary = earliest(
                    next_health_change,
                    earliest(
                        habit_seen
                            .next_cooldown_at
                            .map(|secs| Epoch::from_tai_seconds(secs as f64)),
                        pending_frame
                            .as_ref()
                            .and_then(|pending| pending.habits.as_ref())
                            .and_then(|habits| habits.next_cooldown_at)
                            .map(|secs| Epoch::from_tai_seconds(secs as f64)),
                    ),
                );
                let attempt = prepare_wait_frame_before_health_deadline(
                    pile,
                    signer,
                    &sources,
                    sampled,
                    &mut maintained_from,
                    &mut upkeep_interrupted,
                    &mut pending_frame,
                    Some(&current),
                    pile_path,
                    persona_input,
                    boundary,
                    timeout_at,
                    now,
                    output,
                )
                .await;
                if let Some(probe) = &probe {
                    probe.frame(Some(&current), &attempt);
                }
                let attempt = attempt?;
                if let Some(attempt) = &attempt {
                    observed_snapshot = attempt.watermark_snapshot().clone();
                }
                match attempt {
                    Some(WaitFrameLoad::Retained(_)) => {
                        // Do not reset the Habit evaluation or its sweep
                        // clock. Any deadline crossed during upkeep is still
                        // handled below against this same selected view.
                        view_pending = false;
                    }
                    Some(WaitFrameLoad::Ready(candidate)) => {
                        pending_frame = None;
                        view_pending = false;
                        current_habit_context_valid = true;
                        let (habit_report, due_events) = render_habit_transitions(
                            &habit_seen,
                            &candidate.habits,
                            candidate.observation.facts.presentations.view(),
                        )
                        .unwrap_or_else(|| (String::new(), Vec::new()));
                        let habit_fired = !habit_report.is_empty();
                        let ordinary_fired = matches!(candidate.news, News::Report { .. });
                        apply_prepared_news(
                            pile,
                            signer,
                            false,
                            &candidate.news,
                            &habit_report,
                            output,
                        )?;
                        commit_habit_receipts(pile, signer, &due_events)?;
                        if (habit_fired || ordinary_fired) && !continuous {
                            return Ok(WaitOutcome {
                                news_printed: true,
                                view_pending: false,
                                had_ready_frame: true,
                            });
                        }
                        current = candidate.observation;
                        persona_id = candidate.persona;
                        habit_seen = candidate.habits;
                        last_habit_sweep = Instant::now();
                        continue;
                    }
                    Some(WaitFrameLoad::Pending(pending)) => {
                        view_pending = true;
                        if matches!(
                            pending.reason,
                            PendingWaitReason::PersonaSelection | PendingWaitReason::Preparation
                        ) {
                            // Once a formerly resolved selector is absent or
                            // undecidable, the old persona context cannot emit
                            // new time-driven reports.
                            current_habit_context_valid = false;
                        }
                        pending_frame = Some(pending);
                    }
                    None => view_pending = true,
                }
            } else if let Some(probe) = &probe {
                probe.finish("clock_only");
            }

            // A frame whose persona selection is unresolved remains only a
            // presentation baseline until a readable replacement arrives.
            if !current_habit_context_valid {
                last_habit_sweep = Instant::now();
                continue;
            }
            // Upkeep/payload acquisition may have crossed a clock boundary
            // after the pre-attempt check above.
            let now_secs = epoch_seconds(clock::now()?);
            let cooldown_elapsed = habit_seen
                .next_cooldown_at
                .is_some_and(|deadline| now_secs >= deadline)
                || pending_frame
                    .as_ref()
                    .and_then(|pending| pending.habits.as_ref())
                    .and_then(|habits| habits.next_cooldown_at)
                    .is_some_and(|deadline| now_secs >= deadline);
            let periodic_condition_check = last_habit_sweep.elapsed() >= Duration::from_secs(60);
            if !cooldown_elapsed && !periodic_condition_check {
                continue;
            }

            // Habit readiness is independent of a selected news body's
            // availability. Sweep the ready pending Habit context when there
            // is one, otherwise retain the last fully readable context.
            let current_habits = sweep_wait_habits(
                &current,
                persona_id,
                &mut pending_frame,
                pile_path,
                now_secs,
            )?;
            let (habit_report, due_events) = render_habit_transitions(
                &habit_seen,
                &current_habits,
                current.facts.presentations.view(),
            )
            .unwrap_or_else(|| (String::new(), Vec::new()));
            let habit_fired = !habit_report.is_empty();
            if habit_fired {
                write_complete_report(output, &habit_report, "Orient habit report")?;
                commit_habit_receipts(pile, signer, &due_events)?;
            }
            habit_seen = current_habits;
            last_habit_sweep = Instant::now();
            if habit_fired && !continuous {
                return Ok(WaitOutcome {
                    news_printed: true,
                    view_pending,
                    had_ready_frame: true,
                });
            }
        }
    }
    .await;
    let outcome = result?;
    if continuous {
        // Timeout/shutdown is not news and must never invoke the delivery sink.
        return Ok(());
    }
    if outcome.news_printed {
        // Terse path: the News: reasons and the novel detail were already
        // printed inside the wait loop — don't re-dump the full snapshot.
        return Ok(());
    }
    if outcome.view_pending {
        if outcome.had_ready_frame {
            output.line(format!(
                "The latest pile prefix does not yet provide a readable attention view; the watcher retained its last readable snapshot."
            ))?;
        } else {
            output.line(format!(
                "No fully readable attention view became available before wait ended."
            ))?;
        }
        return Ok(());
    }
    output.line(format!("No change detected since wait started."))?;
    Ok(())
}

fn render_tags(tags: &[String]) -> String {
    if tags.is_empty() {
        return String::new();
    }
    let mut sorted = tags.to_vec();
    sorted.sort();
    sorted.dedup();
    format!(
        " {}",
        sorted
            .iter()
            .map(|tag| {
                if tag.starts_with('#') {
                    tag.to_string()
                } else {
                    format!("#{}", tag)
                }
            })
            .collect::<Vec<_>>()
            .join(" ")
    )
}

/// `orient wake` — assemble the full wake bundle a fresh face reads to come
/// into itself: the memory cover (coarse → fine over ALL memories), then the
/// cover-tagged wiki beliefs (the ambient always-true set), then the compass
/// goals. A supplied persona records shown attention events after output
/// acceptance; previous presentation history does not filter this overview.
async fn cmd_wake(
    storage: &mut FacultyStore,
    signer: &SigningKey,
    persona: Option<&str>,
    chars: usize,
    doing_limit: usize,
    todo_limit: usize,
    output: &mut Out<'_>,
) -> Result<()> {
    use std::fmt::Write as _;

    async {
        // Register and maintain authorized inputs before freezing one query
        // boundary. Plain wake still never consults Embeddings.
        let sources = OrientSources::open(storage, signer, false).await?;
        let memory_collection =
            OrientSource::open(storage, signer, MEMORY_SCOPE_ID, "Memory").await?;
        let wiki_collection = OrientSource::open(storage, signer, WIKI_SCOPE_ID, "Wiki").await?;
        let wiki_latest = wiki_model::latest_collection(storage, signer.verifying_key())
            .context("register Wiki supersession index")?;
        maintain_inputs(storage, signer, &sources).await?;
        memory_collection.maintain(storage, signer).await?;
        wiki_collection.maintain(storage, signer).await?;
        // A revision the index cannot attach is its lag, exactly as it is for
        // the Wiki facts just above; wake reads what is here.
        crate::storage::tolerate_own_lag(storage.maintain_attached(wiki_latest, signer).await)
            .context("maintain Wiki supersession index")?;
        let snapshot = storage
            .snapshot()
            .map_err(|error| anyhow!("freeze shared wake observation: {error}"))?;
        // Wake renders the requested overview, not unseen news. Its output
        // does not depend on prior Presented facts; recording what this wake
        // shows afterward must not impose a historical receipt-read barrier.
        let observation = observe_snapshot_acquiring(snapshot, &sources, false).await?;
        let frozen = observation.snapshot.clone();
        let runtime = tokio::runtime::Handle::current();
        let (memory_facts, wiki_facts, wiki_order) = tokio::task::spawn_blocking(move || {
            let reader = crate::storage::AcquiringReader::with_handle(frozen, runtime);
            let memory_facts = crate::storage::acquire_facts(&reader, memory_collection.rank9)
                .context("read selected Memory collection")?;
            let wiki_facts = crate::storage::acquire_facts(&reader, wiki_collection.rank9)
                .context("read selected Wiki collection")?;
            let wiki_order = crate::storage::require_complete_attached_read(
                reader.attached_acquiring(wiki_latest)?
                    .read_acquiring::<triblespace::core::collection::latest::LatestIndex>()?,
            ).context("read selected Wiki supersession index")?;
            Ok::<_, anyhow::Error>((memory_facts, wiki_facts, wiki_order))
        }).await.context("join frozen wake input selection")??;
        let persona_id = read(storage, &observation.snapshot, |reader| {
            let query = observation.query(reader);
            persona
                .map(|input| resolve_native_persona(&query, input))
                .transpose()
        })
        .await?;
        // Prepare each independent section separately. A missing Wiki or
        // Compass attachment must not repeat memory-cover planning/diagnostics.
        // Plain wake never consults Embeddings; these facts stay shard-backed.
        let cover = read(storage, &observation.snapshot, |reader| {
            render_cover_report(
                &memory_facts,
                &TribleSet::new(),
                reader,
                &CoverOpts::plain(chars),
            )
        })
        .await?;
        let beliefs = read(storage, &observation.snapshot, |reader| {
            wiki_model::cover_fragments(reader, &wiki_facts, &wiki_order)
        })
        .await?;
        let (goals, shown) = read(storage, &observation.snapshot, |reader| {
            let query = observation.query(reader);
            let (goals, shown) = render_native_compass_goals(&query, doing_limit, todo_limit)?;
            let shown = match persona_id {
                Some(persona) => {
                    let candidates: BTreeSet<_> =
                        load_attention_view(&query, persona)?.ids().collect();
                    shown
                        .into_iter()
                        .filter(|event| candidates.contains(event))
                        .collect::<Vec<_>>()
                }
                None => Vec::new(),
            };

            Ok((goals, shown))
        })
        .await?;

        let mut report = cover.text;
        report.push('\n');
        writeln!(report, "Beliefs (cover):").unwrap();
        if beliefs.is_empty() {
            writeln!(report, "- None").unwrap();
        } else {
            for (title, content) in beliefs {
                writeln!(report, "- {title}").unwrap();
                for line in content.lines() {
                    writeln!(report, "    {line}").unwrap();
                }
            }
        }
        report.push('\n');
        report.push_str(&goals);

        write_complete_report(output, &report, "Orient wake report")?;
        if !cover.diagnostics.is_empty() {
            output.line(format!(
                "Memory diagnostics (outside the charged cover text):\n{}",
                cover.diagnostics.join("\n")
            ))?;
        }

        if persona_id.is_some() {
            save_presentations(storage, signer, shown)?;
        }
        Ok(())
    }
    .await
}

/// Foreground acquisition needs the same driven Tokio boundary as production.
/// Resident fixtures use this explicit driver rather than a passive executor.
#[cfg(test)]
fn test_block_on<F: std::future::Future>(future: F) -> F::Output {
    if let Ok(runtime) = tokio::runtime::Handle::try_current() {
        tokio::task::block_in_place(|| runtime.block_on(future))
    } else {
        tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .expect("Orient fixture runtime")
            .block_on(future)
    }
}

#[cfg(test)]
mod tests {
    /// Project receipt facts the way the live path does, so tests exercise the
    /// same join rather than a stand-in for it.
    fn receipts_archive(facts: &TribleSet) -> FactArchive {
        use triblespace::core::blob::encodings::succinctarchive::SuccinctArchive;
        FactArchive::new(vec![SuccinctArchive::from(facts)])
    }

    /// No receipts at all: every due occurrence is unpresented.
    fn nothing_presented() -> FactArchive {
        receipts_archive(&TribleSet::new())
    }

    fn presented_habit_due(due: impl IntoIterator<Item = (Id, i64)>) -> FactArchive {
        let pairs: Vec<_> = due
            .into_iter()
            .map(|(habit, since)| {
                (
                    habit,
                    clock::point(Epoch::from_tai_seconds(since as f64)).unwrap(),
                )
            })
            .collect();
        let fragment = orient_model::habit_receipt_fragment(pairs, clock::point_now().unwrap());
        receipts_archive(&TribleSet::from(fragment))
    }

    fn presented(events: impl IntoIterator<Item = Id>) -> FactArchive {
        let fragment = orient_model::receipt_fragment(events, clock::point_now().unwrap());
        receipts_archive(&TribleSet::from(fragment))
    }

    fn due_habit(label: &str, since: i64, targeted: bool) -> DueHabit {
        DueHabit {
            label: label.to_owned(),
            nudge: "do it".to_owned(),
            since,
            targeted,
        }
    }

    #[test]
    fn daemon_pending_recovery_keeps_seen_occurrence_but_not_an_unreadable_clock() {
        let habit = *fucid();
        let mut due = HabitObservation::default();
        due.due
            .insert(habit, due_habit("owned reminder", 100, true));
        due.next_cooldown_at = Some(200);
        assert!(render_due_habits_unreceipted(&due, &nothing_presented()).is_some());
        let mut seen = Some(due.clone());
        // All initial-pending invalidation sites use this boundary. Losing
        // inputs must stop their clock, not turn recovery into another arm.
        invalidate_pending_habit_context(true, &mut seen);
        assert_eq!(pending_habit_deadline(true, None, &seen), None);
        // Unchanged since the baseline, but still unreceipted, so still
        // reported: what decides is whether this observer has been SHOWN the
        // occurrence, not whether it changed since something it happens to
        // hold in memory.
        assert!(
            render_habit_transitions(seen.as_ref().unwrap(), &due, &nothing_presented()).is_some()
        );
        assert_eq!(
            pending_habit_deadline(true, Some(&due), &seen),
            Some(Epoch::from_tai_seconds(200.0))
        );
        let mut next = due.clone();
        next.due
            .insert(habit, due_habit("owned reminder", 300, true));
        assert!(
            render_habit_transitions(seen.as_ref().unwrap(), &next, &nothing_presented()).is_some()
        );
        invalidate_pending_habit_context(false, &mut seen);
        assert!(seen.is_none(), "one-shot rearm semantics remain unchanged");
    }

    #[test]
    fn a_fresh_completion_makes_the_next_due_a_new_event() {
        let habit = Id::new([7; 16]).unwrap();
        let mut previous = HabitObservation::default();
        previous.due.insert(habit, due_habit("tick", 100, true));
        let mut current = HabitObservation::default();
        current.due.insert(habit, due_habit("tick", 100, true));
        assert!(
            newly_due(&previous, &current).is_empty(),
            "the same due event is not news twice"
        );
        current.due.insert(habit, due_habit("tick", 1300, true));
        assert_eq!(
            newly_due(&previous, &current).len(),
            1,
            "the due after a fresh completion is a new event"
        );
    }

    #[test]
    fn a_due_occurrence_is_reported_until_its_receipt_exists_then_never_again() {
        let owned = Id::new([8; 16]).unwrap();
        let shared = Id::new([9; 16]).unwrap();
        let mut armed = HabitObservation::default();

        // A SHARED intention is reported. It used to stay a quiet baseline on
        // the reasoning that whoever saw its transition would take it -- but
        // when the due instant falls between one wait's exit and the next arm
        // NOBODY sees the transition, so that resolved to nobody, silently.
        // `work-ledger-grooming` (every 7d, addressed to everyone) was lost
        // exactly this way on 2026-09-18.
        armed
            .due
            .insert(shared, due_habit("work-ledger-grooming", 700, false));
        let (shared_report, shared_events) =
            render_due_habits_unreceipted(&armed, &nothing_presented())
                .expect("a shared due occurrence is reported");
        assert!(shared_report.contains("work-ledger-grooming"));
        assert_eq!(shared_events, vec![(shared, 700)]);

        armed.due.insert(owned, due_habit("cc-tick", 1300, true));
        let (both, events) = render_due_habits_unreceipted(&armed, &nothing_presented())
            .expect("both due occurrences are reported");
        assert!(both.contains("News: habit became due: cc-tick"));
        assert!(both.contains("work-ledger-grooming"));
        assert_eq!(events.len(), 2);

        // Receipting ONE occurrence silences exactly that one. Targeting has
        // nothing to do with it; being shown does.
        let seen_shared = presented_habit_due([(shared, 700)]);
        let (only_owned, owned_events) = render_due_habits_unreceipted(&armed, &seen_shared)
            .expect("the unreceipted occurrence is still reported");
        assert!(only_owned.contains("cc-tick"));
        assert!(!only_owned.contains("work-ledger-grooming"));
        assert_eq!(owned_events, vec![(owned, 1300)]);

        // Receipt both and the watcher is quiet, however often it rearms.
        let seen_both = presented_habit_due([(shared, 700), (owned, 1300)]);
        assert!(render_due_habits_unreceipted(&armed, &seen_both).is_none());
        assert!(render_due_habits_unreceipted(&armed, &seen_both).is_none());

        // A fresh due of the SAME habit is a different occurrence: later
        // `since`, so the old receipt does not cover it.
        let mut again = HabitObservation::default();
        again.due.insert(owned, due_habit("cc-tick", 1900, true));
        let (recurred, _) = render_due_habits_unreceipted(&again, &seen_both)
            .expect("a later due occurrence is presented again");
        assert!(recurred.contains("cc-tick"));
    }

    use super::super::cli::{parse_wait_target, WaitTarget};
    use super::*;
    use std::fs;
    use std::io::{self, Write};
    use std::sync::atomic::{AtomicU64, Ordering};
    use std::time::{SystemTime, UNIX_EPOCH};
    use triblespace::core::blob::encodings::succinctarchive::{
        OrderedUniverse, SuccinctArchive, UnionArchive,
    };
    use triblespace::core::blob::{Blob, IntoBlob};
    use triblespace::core::collection::{
        records::empty_metadata_handle, CollectionCommit, CollectionRecord, CollectionStore,
    };
    use triblespace::core::repo::{StorageClose, WantRead};

    fn write_report_to_writer(
        writer: &mut impl Write,
        report: &str,
        description: &str,
    ) -> Result<()> {
        let mut emit = |part| {
            let crate::out::Part::Text { text } = part else {
                bail!("expected report text")
            };
            writer.write_all(text.as_bytes())?;
            writer.flush()?;
            Ok(())
        };
        write_complete_report(&mut Out::new(&mut emit), report, description)
    }
    fn apply_news_to_writer(
        pile: &mut FacultyStore,
        signer: &SigningKey,
        _persona: Id,
        peek: bool,
        news: &News,
        prefix: &str,
        writer: &mut impl Write,
    ) -> Result<()> {
        let mut emit = |part| {
            let crate::out::Part::Text { text } = part else {
                bail!("expected report text")
            };
            writer.write_all(text.as_bytes())?;
            writer.flush()?;
            Ok(())
        };
        apply_prepared_news(pile, signer, peek, news, prefix, &mut Out::new(&mut emit))
    }

    static NEXT_TEST_PILE: AtomicU64 = AtomicU64::new(0);

    struct TestPile {
        dir: PathBuf,
        path: PathBuf,
        signer: SigningKey,
    }

    impl TestPile {
        fn new() -> Self {
            // Run isolated tests with collection override environment cleared by the runner.
            let nonce = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos();
            let sequence = NEXT_TEST_PILE.fetch_add(1, Ordering::Relaxed);
            let dir = std::env::temp_dir().join(format!(
                "faculties-orient-succinct-{}-{nonce}-{sequence}",
                std::process::id()
            ));
            fs::create_dir_all(&dir).unwrap();
            let path = dir.join("test.pile");
            fs::File::create(&path).unwrap();
            let signer = SigningKey::from_bytes(&[7; 32]);
            Self { dir, path, signer }
        }
    }

    impl Drop for TestPile {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.dir);
        }
    }

    fn id(byte: u8) -> Id {
        Id::new([byte; 16]).unwrap()
    }

    fn archive(facts: &TribleSet) -> FactArchive {
        UnionArchive::new(vec![SuccinctArchive::<OrderedUniverse>::from(facts)])
    }

    #[test]
    fn attention_excludes_every_self_attribution_but_keeps_unknown_and_other_actors() {
        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let persona = id(200);
            let alias = id(201);
            let other = id(1);
            let mut people = Fragment::empty();
            for (person, label) in [
                (persona, "observer"),
                (alias, "old-observer"),
                (other, "peer"),
            ] {
                people += relations::person_fragment(
                    person,
                    relations::ProfileInput {
                        label: label.to_owned(),
                        ..Default::default()
                    },
                )
                .unwrap()
                .0;
            }
            people += entity! {
                metadata::tag: &KIND_IDENTITY_VERDICT,
                relation_identity::low: &persona,
                relation_identity::high: &alias,
                relation_identity::same: true,
            };
            pile.commit(sources.relations.source, &fixture.signer, people)
                .unwrap();

            let now = clock::point_now().unwrap();
            let mut facts = Fragment::empty();
            let mut expected = BTreeSet::from([other]);
            // The mixed event puts the external actor first in key order:
            // choosing just one attribution cannot establish "not mine".
            for (label, actor, extra_self, visible) in [
                ("exact", Some(persona), false, false),
                ("alias", Some(alias), false, false),
                ("mixed", Some(other), true, false),
                ("external", Some(other), false, true),
                ("unknown", None, false, true),
            ] {
                let (goal, goal_id) =
                    compass::goal_fragment(label, vec!["observer".to_owned()], None, now).unwrap();
                facts += goal;
                let mut status = compass::status_fragment(goal_id, "doing", actor, now).unwrap();
                let status_id = status.root().unwrap();
                let (mut note, note_id) =
                    compass::note_fragment(goal_id, label, vec![], vec![], vec![], actor, now)
                        .unwrap();
                if extra_self {
                    status += entity! { ExclusiveId::force_ref(&status_id) @ board::by: &alias };
                    note += entity! { ExclusiveId::force_ref(&note_id) @ board::by: &alias };
                }
                facts += status;
                facts += note;
                if visible {
                    expected.extend([status_id, note_id]);
                }
            }
            // Involvement follows the same settled identity relation: another
            // person's later update on an untagged goal still deserves attention.
            let (goal, goal_id) = compass::goal_fragment("involved", vec![], None, now).unwrap();
            facts += goal;
            facts +=
                compass::note_fragment(goal_id, "mine", vec![], vec![], vec![], Some(alias), now)
                    .unwrap()
                    .0;
            let status = compass::status_fragment(goal_id, "done", Some(other), now).unwrap();
            expected.insert(status.root().unwrap());
            facts += status;
            // A goal without a status/actor is likewise unknown, not self.
            let (goal, goal_id) =
                compass::goal_fragment("bare", vec!["observer".to_owned()], None, now).unwrap();
            facts += goal;
            expected.insert(goal_id);
            pile.commit(sources.compass.source, &fixture.signer, facts)
                .unwrap();
            let mut windows = Fragment::empty();
            for window in [persona, alias, other] {
                windows +=
                    entity! { metadata::tag: &KIND_STATUS_UPDATE, window_status::window: &window };
            }
            pile.commit(sources.status.source, &fixture.signer, windows)
                .unwrap();
            let mut messages = Fragment::empty();
            for (label, sender, recipient, extra_self, visible) in [
                ("self", persona, persona, false, false),
                ("alias", alias, persona, false, false),
                ("mixed", other, persona, true, false),
                ("external", other, persona, false, true),
                // An opaque sender without a Relations profile is not self.
                ("unknown", id(199), persona, false, true),
                ("elsewhere", other, id(198), false, false),
            ] {
                let body = messages.put(label.to_owned());
                let mut envelope =
                    message::envelope_fragment(sender, recipient, body, now, None, None);
                let envelope_id = envelope.root().unwrap();
                if extra_self {
                    envelope += entity! { ExclusiveId::force_ref(&envelope_id) @ local_message::from: &alias };
                }
                messages += envelope;
                if visible {
                    expected.insert(envelope_id);
                }
            }
            // A senderless fragment still does not satisfy Message's typed
            // envelope query; do not change that pre-existing reader policy.
            let body: message::TextHandle = messages.put("no sender".to_owned());
            messages += entity! {
                metadata::tag: &KIND_MESSAGE_ID,
                local_message::to: &persona,
                local_message::body: body,
                metadata::created_at: now,
            };
            pile.commit(sources.messages.source, &fixture.signer, messages)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let view =
                load_attention_view(&observation.query(&observation.snapshot), persona).unwrap();
            assert_eq!(view.ids().collect::<BTreeSet<_>>(), expected);
            pile.close().unwrap();
        });
    }

    #[test]
    fn nonwriter_observes_lagging_external_input_and_retains_own_receipts() {
        test_block_on(async {
            let fixture = TestPile::new();
            let reader_key = SigningKey::from_bytes(&[73; 32]);
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            // External Relations are owned elsewhere; notification receipts
            // remain this local writer's collection.
            let mut sources = OrientSources::open(&mut pile, &reader_key, false)
                .await
                .unwrap();
            sources.relations =
                OrientSource::open(&mut pile, &fixture.signer, RELATIONS_SCOPE_ID, "Relations")
                    .await
                    .unwrap();
            let known = id(71);
            let pending = id(72);
            let person = |person, label: &str| {
                relations::person_fragment(
                    person,
                    crate::relations::ProfileInput {
                        label: label.to_owned(),
                        ..Default::default()
                    },
                )
                .unwrap()
                .0
            };
            pile.commit(
                sources.relations.source,
                &fixture.signer,
                person(known, "resident-reader"),
            )
            .unwrap();
            sources
                .relations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            pile.commit(
                sources.relations.source,
                &fixture.signer,
                person(pending, "pending-reader"),
            )
            .unwrap();
            let before = pile.snapshot().unwrap();
            let records: Vec<_> = before.records().unwrap().map(Result::unwrap).collect();
            // The unattached input is read from its own bytes.
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            assert_eq!(
                person_anchors(observation.facts.relations.view()),
                BTreeSet::from([known, pending]),
            );
            assert_eq!(
                pile.snapshot()
                    .unwrap()
                    .records()
                    .unwrap()
                    .map(Result::unwrap)
                    .collect::<Vec<_>>(),
                records,
                "a nonwriter read must not publish records for lagging external input",
            );

            let event = id(74);
            save_presentations(&mut pile, &reader_key, [event]).unwrap();
            sources
                .presentations
                .maintain(&mut pile, &reader_key)
                .await
                .unwrap();
            let next = observe_current_sources(&mut pile, &sources).unwrap();
            assert!(next.facts.presentations.contains(event));
            assert_eq!(
                person_anchors(next.facts.relations.view()),
                BTreeSet::from([known, pending])
            );

            // External upkeep attaches the new input; the read is unchanged.
            sources
                .relations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let caught_up = observe_current_sources(&mut pile, &sources).unwrap();
            assert_eq!(
                person_anchors(caught_up.facts.relations.view()),
                BTreeSet::from([known, pending]),
            );
            pile.close().unwrap();
        });
    }

    #[test]
    fn unprojected_receipts_do_not_block_passive_observation() {
        test_block_on(async {
            let fixture = TestPile::new();
            let reader_key = SigningKey::from_bytes(&[73; 32]);
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let mut sources = OrientSources::open(&mut pile, &reader_key, false)
                .await
                .unwrap();
            sources.presentations = ReceiptSource::register(&mut pile, &fixture.signer).unwrap();
            pile.commit(
                sources.presentations.source,
                &fixture.signer,
                orient_model::receipt_fragment([id(76)], clock::point_now().unwrap()),
            )
            .unwrap();
            let before = pile.snapshot().unwrap();
            let records: Vec<_> = before.records().unwrap().map(Result::unwrap).collect();
            // An unattached receipt is read from its own bytes, without
            // upkeep and without blocking.
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            assert!(observation.facts.presentations.contains(id(76)));
            assert_eq!(
                pile.snapshot()
                    .unwrap()
                    .records()
                    .unwrap()
                    .map(Result::unwrap)
                    .collect::<Vec<_>>(),
                records,
                "reading an unattached receipt never performs upkeep or blocks",
            );
            assert!(observation.snapshot.wants().unwrap().next().is_none());
            sources
                .presentations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let ready = observe_current_sources(&mut pile, &sources).unwrap();
            assert!(ready.facts.presentations.contains(id(76)));
            pile.close().unwrap();
        });
    }

    #[test]
    fn a_fresh_one_shot_sees_its_receipts_before_any_upkeep() {
        test_block_on(async {
            let fixture = TestPile::new();
            let reader = SigningKey::from_bytes(&[73; 32]);
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &reader, false)
                .await
                .unwrap();
            let persona = id(77);
            let event = id(78);
            let news = News::Report {
                text: "News: one delivery\n".to_owned(),
                events: vec![event],
            };
            let mut output = Vec::new();
            apply_news_to_writer(&mut pile, &reader, persona, false, &news, "", &mut output)
                .unwrap();
            assert_eq!(output, b"News: one delivery\n");

            // A fresh one-shot sees the same signer-owned receipt descriptors,
            // and reads the receipt from its bytes before any upkeep attaches
            // it: nothing repeats, and observing publishes nothing.
            let restarted = OrientSources::open(&mut pile, &reader, false)
                .await
                .unwrap();
            assert_eq!(restarted.presentations.source, sources.presentations.source);
            assert_eq!(restarted.presentations.rank9, sources.presentations.rank9);
            let before = pile.snapshot().unwrap();
            let records: Vec<_> = before.records().unwrap().map(Result::unwrap).collect();
            let unattached = observe_current_sources(&mut pile, &restarted).unwrap();
            let mut attention = AttentionView::default();
            attention.insert(AttentionEvent::Message(event));
            assert!(attention
                .pending(unattached.facts.presentations.view())
                .is_empty());
            assert_eq!(
                pile.snapshot()
                    .unwrap()
                    .records()
                    .unwrap()
                    .map(Result::unwrap)
                    .collect::<Vec<_>>(),
                records,
                "observation publishes nothing",
            );

            // Routing aliases do not split one zooid's receipt authority.
            save_presentations(&mut pile, &reader, [id(80)]).unwrap();
            let aliases = ReceiptSource::register(&mut pile, &reader).unwrap();
            assert_eq!(aliases.source, restarted.presentations.source);
            assert_eq!(aliases.rank9, restarted.presentations.rank9);
            let refreshed = observe_current_sources(&mut pile, &restarted).unwrap();
            assert_eq!(
                refreshed.facts.presentations.presented_events(),
                BTreeSet::from([event, id(80)]),
            );
            pile.close().unwrap();
        });
    }

    /// A run that is about to report refreshes the receipt projection itself,
    /// through the production helper, and says nothing when that succeeds;
    /// the event it presented does not repeat either before or after.
    #[test]
    fn a_reporting_run_refreshes_the_receipt_set_so_it_does_not_repeat_itself() {
        test_block_on(async {
            let fixture = TestPile::new();
            let reader = SigningKey::from_bytes(&[73; 32]);
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &reader, false)
                .await
                .unwrap();
            let persona = id(77);
            let event = id(78);
            let news = News::Report {
                text: "News: one delivery\n".to_owned(),
                events: vec![event],
            };
            let mut output = Vec::new();
            apply_news_to_writer(&mut pile, &reader, persona, false, &news, "", &mut output)
                .unwrap();

            let mut attention = AttentionView::default();
            attention.insert(AttentionEvent::Message(event));
            let unattached = observe_current_sources(&mut pile, &sources).unwrap();
            assert!(attention
                .pending(unattached.facts.presentations.view())
                .is_empty());

            let mut text = String::new();
            let mut emit = |part| {
                let crate::out::Part::Text { text: part } = part else {
                    bail!("expected text")
                };
                text.push_str(&part);
                Ok(())
            };
            refresh_receipts_before_observation(
                &mut pile,
                &reader,
                &sources,
                &mut Out::new(&mut emit),
            )
            .await
            .unwrap();
            assert!(text.is_empty(), "a successful refresh says nothing: {text}");

            let refreshed = observe_current_sources(&mut pile, &sources).unwrap();
            assert!(
                attention
                    .pending(refreshed.facts.presentations.view())
                    .is_empty(),
                "after the refresh the event is presented and must not repeat",
            );
            pile.close().unwrap();
        });
    }

    #[test]
    fn persona_wake_ignores_unrelated_historical_receipt_projection_lag() {
        runtime().unwrap().block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let persona = id(86);
            let prior_event = id(87);
            let (person, _, _) = relations::person_fragment(
                persona,
                relations::ProfileInput {
                    label: "wake-reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, person)
                .unwrap();
            let (goal, goal_id) = compass::goal_fragment(
                "a resident goal for this wake",
                vec!["wake-reader".to_owned()],
                None,
                clock::point_now().unwrap(),
            )
            .unwrap();
            pile.commit(sources.compass.source, &fixture.signer, goal)
                .unwrap();
            pile.commit(
                sources.presentations.source,
                &fixture.signer,
                orient_model::receipt_fragment([prior_event], clock::point_now().unwrap()),
            )
            .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();

            // An unrelated historical receipt record arrives without its
            // archive. Its absence cannot hide the existing projection.
            let cold = orient_model::receipt_fragment([id(89)], clock::point_now().unwrap());
            let cold = IntoBlob::<SimpleArchive>::to_blob(cold.facts().clone());
            let arriving = CollectionCommit::sign(
                &fixture.signer,
                sources.presentations.source.handle(),
                inlineencodings::Handle::<SimpleArchive>::to_hash(cold.get_handle()),
                empty_metadata_handle(),
            );
            pile.insert(CollectionRecord::Commit(arriving)).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let before = sources
                .presentations
                .observe(&pile.snapshot().unwrap())
                .unwrap();
            assert_eq!(before.presented_events(), BTreeSet::from([prior_event]),);
            assert!(!before.contains(id(89)));
            assert!(!pile
                .snapshot()
                .unwrap()
                .contains_blob(cold.get_handle())
                .unwrap());

            let mut report = String::new();
            cmd_wake(
                &mut pile,
                &fixture.signer,
                Some("wake-reader"),
                0,
                5,
                5,
                &mut Out::new(&mut |part| {
                    let crate::out::Part::Text { text } = part else {
                        bail!("expected wake text")
                    };
                    report.push_str(&text);
                    Ok(())
                }),
            )
            .await
            .unwrap();
            assert!(report.contains("Beliefs (cover):"));
            assert!(report.contains("a resident goal for this wake"));

            // The accepted overview still publishes its shown goal receipt,
            // and making it visible does not depend on the historical gap:
            // all ordinary receipt readers accept the resident set. Upkeep
            // attaches what is resident and fetches nothing, so the gap stays
            // residual, no WANT is recorded, and the network host is never
            // started.
            sources
                .presentations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let snapshot = pile.snapshot().unwrap();
            let after = sources.presentations.observe(&snapshot).unwrap();
            assert_eq!(
                after.presented_events(),
                BTreeSet::from([prior_event, goal_id]),
            );
            assert!(observe_current_sources(&mut pile, &sources)
                .unwrap()
                .facts
                .presentations
                .contains(goal_id));
            assert!(!snapshot.contains_blob(cold.get_handle()).unwrap());
            assert!(snapshot.wants().unwrap().next().is_none());
            assert!(pile.health().started_at.is_none());
            pile.close().unwrap();
        });
    }

    #[test]
    fn compact_receipt_targets_do_not_need_original_payloads_or_metadata() {
        test_block_on(async {
            let fixture = TestPile::new();
            let copy = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let mut omitted = BTreeSet::new();
            for event in [id(82), id(83)] {
                let commit = pile
                    .commit(
                        sources.presentations.source,
                        &fixture.signer,
                        orient_model::receipt_fragment([event], clock::point_now().unwrap()),
                    )
                    .unwrap();
                omitted.insert(commit.data().raw);
                omitted.insert(commit.metadata().raw);
            }
            sources
                .presentations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let before = pile.snapshot().unwrap();
            let mut copied = open_store_as(&copy.path, fixture.signer.verifying_key()).unwrap();
            for info in before.blobs() {
                let info = info.unwrap();
                if !omitted.contains(&info.handle.raw) {
                    let blob: Blob<blobencodings::UnknownBlob> =
                        BlobStoreGet::get(&before, info.handle).unwrap();
                    copied.put::<blobencodings::UnknownBlob, _>(blob).unwrap();
                }
            }
            for record in before.records().unwrap() {
                copied.insert(record.unwrap()).unwrap();
            }
            let snapshot = copied.snapshot().unwrap();
            for handle in omitted {
                assert!(!snapshot
                    .contains_blob(
                        Inline::<inlineencodings::Handle<blobencodings::UnknownBlob>>::new(handle)
                    )
                    .unwrap());
            }
            let records: Vec<_> = snapshot.records().unwrap().map(Result::unwrap).collect();
            let observation = observe_current_sources(&mut copied, &sources).unwrap();
            let observed = &observation.facts.presentations;
            assert_eq!(
                observed.presented_events(),
                BTreeSet::from([id(82), id(83)])
            );
            assert!(observation.snapshot.wants().unwrap().next().is_none());
            assert_eq!(
                observation
                    .snapshot
                    .records()
                    .unwrap()
                    .map(Result::unwrap)
                    .collect::<Vec<_>>(),
                records,
                "warm target reads do not reconstruct historical source records",
            );
            copied.close().unwrap();
            pile.close().unwrap();
        });
    }

    #[test]
    fn receipt_write_denial_is_not_pending_availability() {
        let fixture = TestPile::new();
        let stranger = SigningKey::from_bytes(&[73; 32]);
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let source = ReceiptSource::register(&mut pile, &fixture.signer)
            .unwrap()
            .source;
        let snapshot = pile.snapshot().unwrap();
        let error = require_presentation_write(&snapshot, source, &stranger).unwrap_err();
        assert!(format!("{error:#}").contains("requires WRITE"));
        assert!(!is_preparation_pending(&error));
        assert!(snapshot.records().unwrap().next().is_none());
        pile.close().unwrap();
    }

    #[test]
    fn cold_receipts_do_not_fetch_history_while_persona_is_pending() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let source = ReceiptSource::register(&mut pile, &fixture.signer)
            .unwrap()
            .source;
        let fragment = orient_model::receipt_fragment([id(85)], clock::point_now().unwrap());
        let blob = IntoBlob::<SimpleArchive>::to_blob(fragment.facts().clone());
        let commit = CollectionCommit::sign(
            &fixture.signer,
            source.handle(),
            inlineencodings::Handle::<SimpleArchive>::to_hash(blob.get_handle()),
            empty_metadata_handle(),
        );
        pile.insert(CollectionRecord::Commit(commit)).unwrap();
        let options = WaitOptions {
            timeout: Some(Duration::from_millis(25)),
            poll_interval: Duration::from_secs(1),
        };
        let mut text = String::new();
        let mut emit = |part| {
            let crate::out::Part::Text { text: part } = part else {
                bail!("expected text")
            };
            text.push_str(&part);
            Ok(())
        };
        let started = Instant::now();
        runtime().unwrap().block_on(async {
            tokio::time::timeout(
                Duration::from_secs(2),
                cmd_wait(
                    &mut pile,
                    &fixture.signer,
                    &fixture.path,
                    Some("waiting-reader"),
                    &options,
                    Duration::from_secs(180),
                    &mut Out::new(&mut emit),
                ),
            )
            .await
            .expect("pending persona selection must honor its timeout without fetching history")
            .unwrap();
        });
        assert!(started.elapsed() >= options.timeout.unwrap());
        assert!(text.contains("No fully readable attention view"));
        let snapshot = pile.snapshot().unwrap();
        assert!(!snapshot.contains_blob(blob.get_handle()).unwrap());
        assert!(snapshot.wants().unwrap().next().is_none());
        assert_eq!(
            snapshot
                .records()
                .unwrap()
                .map(Result::unwrap)
                .collect::<Vec<_>>(),
            vec![CollectionRecord::Commit(commit)],
        );
        pile.close().unwrap();
    }

    #[test]
    fn receipt_refresh_is_scoped_to_target_progress() {
        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, true)
                .await
                .unwrap();
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let receipt = orient_model::receipt_fragment([id(87)], clock::point_now().unwrap());
            let bytes = IntoBlob::<SimpleArchive>::to_blob(receipt.facts().clone());
            let commit = CollectionCommit::sign(
                &fixture.signer,
                sources.presentations.source.handle(),
                inlineencodings::Handle::<SimpleArchive>::to_hash(bytes.get_handle()),
                empty_metadata_handle(),
            );
            pile.insert(CollectionRecord::Commit(commit)).unwrap();
            let cold = pile.snapshot().unwrap();
            assert!(!cold.contains_blob(bytes.get_handle()).unwrap());
            // A new receipt commit is a new node of the receipts' parent, so
            // the receipt read moves, even before the commit's bytes arrive;
            // until they do, it is residual and presents nothing.
            assert!(!observation.is_current(&cold));
            assert!(!observation.facts.presentations.is_current(&cold));
            let observation = observe_snapshot(cold, &sources).unwrap();
            assert!(!observation.facts.presentations.contains(id(87)));
            assert!(observation.snapshot.wants().unwrap().next().is_none());

            let unrelated = pile
                .collection(
                    "unrelated-refresh-fixture",
                    crate::collection_names::private_policy(fixture.signer.verifying_key()),
                )
                .unwrap();
            pile.commit(
                unrelated,
                &fixture.signer,
                entity! { metadata::name: "elsewhere" },
            )
            .unwrap();
            assert!(
                observation.is_current(&pile.snapshot().unwrap()),
                "unrelated records do not invalidate the selected attention inputs"
            );

            // The bytes arriving moves the receipt read, and only it: the
            // receipt is read from them, and upkeep then attaches it.
            pile.put::<SimpleArchive, _>(bytes).unwrap();
            let arrived = pile.snapshot().unwrap();
            assert!(!observation.facts.presentations.is_current(&arrived));
            assert!(observe_snapshot(arrived, &sources)
                .unwrap()
                .facts
                .presentations
                .contains(id(87)));
            sources
                .presentations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let advanced = pile.snapshot().unwrap();
            assert!(!observation.is_current(&advanced));
            assert!(!observation.facts.presentations.is_current(&advanced));
            for fact in [
                &observation.facts.messages,
                &observation.facts.mail,
                &observation.facts.teams,
                &observation.facts.compass,
                &observation.facts.relations,
                &observation.facts.status,
                observation.facts.habits.as_ref().unwrap(),
            ] {
                assert!(fact.is_current(&advanced));
            }
            let refreshed = observe_snapshot(advanced, &sources).unwrap();
            assert!(refreshed.facts.presentations.contains(id(87)));
            assert!(observation.facts.presentations.is_empty());
            assert!(refreshed.snapshot.wants().unwrap().next().is_none());
            assert!(pile.health().started_at.is_none());
            pile.close().unwrap();
        });
    }

    #[test]
    fn foreground_acquires_cold_foundations_without_advancing_the_observation() {
        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false).await.unwrap();
            let cold_tag = id(122);
            let later_tag = id(123);
            let cold = entity! { metadata::tag: &cold_tag };
            let bytes = IntoBlob::<SimpleArchive>::to_blob(cold.facts().clone());
            let handle = bytes.get_handle();
            pile.insert(CollectionRecord::Commit(CollectionCommit::sign(
                &fixture.signer,
                sources.messages.source.handle(),
                inlineencodings::Handle::<SimpleArchive>::to_hash(handle),
                empty_metadata_handle(),
            ))).unwrap();
            let watermark = pile.snapshot().unwrap();
            assert!(!watermark.contains_blob(handle).unwrap());
            // Arrival and another foundation follow the frozen boundary. Exact
            // Leech lookup can supply the former; the latter must stay absent
            // from this observation even though it is now locally available.
            pile.put::<SimpleArchive, _>(bytes).unwrap();
            pile.commit(
                sources.messages.source, &fixture.signer,
                entity! { metadata::tag: &later_tag },
            ).unwrap();
            let observation = observe_snapshot_acquiring(watermark.clone(), &sources, false)
                .await.unwrap();
            assert!(exists!(pattern!(observation.facts.messages.view(), [
                { metadata::tag: &cold_tag }
            ])));
            assert!(!exists!(pattern!(observation.facts.messages.view(), [
                { metadata::tag: &later_tag }
            ])));
            assert!(observation.snapshot.changes_since(&watermark).is_empty());
            assert!(!observation.snapshot.contains_blob(handle).unwrap());
            assert!(!observation.is_current(&pile.snapshot().unwrap()));
            assert!(pile.health().started_at.is_none());
            pile.close().unwrap();
        });
    }

    #[test]
    fn a_closed_owner_cannot_complete_a_cold_foreground_as_empty() {
        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false).await.unwrap();
            let cold = entity! { metadata::tag: &id(124) };
            let bytes = IntoBlob::<SimpleArchive>::to_blob(cold.facts().clone());
            let handle = bytes.get_handle();
            pile.insert(CollectionRecord::Commit(CollectionCommit::sign(
                &fixture.signer,
                sources.messages.source.handle(),
                inlineencodings::Handle::<SimpleArchive>::to_hash(handle),
                empty_metadata_handle(),
            ))).unwrap();
            let watermark = pile.snapshot().unwrap();
            pile.close().unwrap();
            // No fallback owner remains. Missing selected support must not
            // masquerade as a successfully observed empty source.
            assert!(observe_snapshot_acquiring(watermark.clone(), &sources, false).await.is_err());
            assert!(!watermark.contains_blob(handle).unwrap());
        });
    }

    #[test]
    fn acquired_preparation_descriptor_is_used_by_the_same_retry() {
        runtime().unwrap().block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            // A preparation boundary can predate the descriptor's arrival.
            // Register the real sources afterward; no network mock is needed
            // because exact acquisition also handles newly resident bytes.
            let watermark = pile.snapshot().unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, true)
                .await
                .unwrap();
            // This older application time was still cooling. Once preparation
            // can select a fresh prefix, it must evaluate at a fresh time too.
            let earlier = clock::now().unwrap() - hifitime::Duration::from_seconds(7200.0);
            let (habit, habit_id) =
                habits::habit_fragment("preparation-clock", "every 1h", "due now", None, &[], &[])
                    .unwrap();
            let (done, _) =
                habits::completion_fragment(habit_id, clock::point(earlier).unwrap()).unwrap();
            let habit_source = sources.habits.as_ref().unwrap();
            pile.commit(habit_source.source, &fixture.signer, habit + done)
                .unwrap();
            habit_source
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let descriptor = Inline::<inlineencodings::Handle<blobencodings::UnknownBlob>>::new(
                sources.messages.rank9.handle().raw,
            );
            let resident = pile.snapshot().unwrap();
            assert!(!watermark.contains_blob(descriptor).unwrap());
            assert!(resident.contains_blob(descriptor).unwrap());
            assert!(sources.messages.observe(&watermark).is_err());
            assert_eq!(sources.observations.load(Ordering::Relaxed), 0);

            let mut preparation =
                PendingWaitFrame::awaiting_view(watermark.clone(), PendingWaitReason::Preparation);
            preparation.missing = Some(MissingBlob { handle: descriptor });
            let mut pending = Some(preparation);
            let persona = id(89);
            let attempt = load_wait_frame_before_health_deadline(
                &mut pile,
                &sources,
                watermark.clone(),
                &mut pending,
                &fixture.path,
                &fmt_id(persona),
                None,
                earlier,
            )
            .await
            .unwrap()
            .expect("no health deadline interrupts the preparation retry");
            let WaitFrameLoad::Ready(frame) = attempt else {
                panic!("successful descriptor acquisition must prepare from its resident prefix")
            };
            assert_eq!(frame.persona, persona);
            assert!(matches!(frame.news, News::Quiet));
            assert!(frame.habits.due.contains_key(&habit_id));
            let old_evaluation = observe_habits_in_observation(
                &frame.observation,
                &fixture.path,
                epoch_seconds(earlier),
                persona,
            )
            .unwrap();
            assert!(!old_evaluation.due.contains_key(&habit_id));
            assert_eq!(sources.observations.load(Ordering::Relaxed), 1);
            assert!(frame.watermark.contains_blob(descriptor).unwrap());
            assert!(frame
                .observation
                .snapshot
                .contains_blob(descriptor)
                .unwrap());
            assert!(!watermark.contains_blob(descriptor).unwrap());
            assert!(frame.watermark.changes_since(&resident).is_empty());
            let after = pile.snapshot().unwrap();
            assert!(after.changes_since(&resident).is_empty());
            assert!(after.wants().unwrap().next().is_none());
            assert!(pile.health().started_at.is_none());
            pile.close().unwrap();
        });
    }

    #[test]
    fn unchanged_pending_persona_reuses_the_scoped_observation() {
        runtime().unwrap().block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, true)
                .await
                .unwrap();
            let unrelated = pile
                .collection(
                    "unrelated-pending-fixture",
                    crate::collection_names::private_policy(fixture.signer.verifying_key()),
                )
                .unwrap();
            let mut pending = None;
            for iteration in 0..6 {
                if iteration == 1 {
                    pile.commit(
                        unrelated,
                        &fixture.signer,
                        entity! { metadata::name: "elsewhere" },
                    )
                    .unwrap();
                }
                let sampled = pile.snapshot().unwrap();
                let attempt = load_wait_frame_before_health_deadline(
                    &mut pile,
                    &sources,
                    sampled,
                    &mut pending,
                    &fixture.path,
                    "waiting-reader",
                    None,
                    Epoch::from_tai_seconds(42.0),
                )
                .await
                .unwrap()
                .unwrap();
                let WaitFrameLoad::Pending(frame) = attempt else {
                    panic!("an absent persona must remain pending")
                };
                assert_eq!(frame.reason, PendingWaitReason::PersonaSelection);
                assert!(frame.missing.is_none());
                assert!(
                    frame.observation.is_some(),
                    "retain the exact selected views"
                );
                pending = Some(frame);
            }
            assert_eq!(sources.observations.load(Ordering::Relaxed), 1);

            let persona = id(86);
            let (person, _, _) = relations::person_fragment(
                persona,
                relations::ProfileInput {
                    label: "waiting-reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, person)
                .unwrap();
            sources
                .relations
                .maintain(&mut pile, &fixture.signer)
                .await
                .unwrap();
            let sampled = pile.snapshot().unwrap();
            let attempt = load_wait_frame_before_health_deadline(
                &mut pile,
                &sources,
                sampled,
                &mut pending,
                &fixture.path,
                "waiting-reader",
                None,
                Epoch::from_tai_seconds(42.0),
            )
            .await
            .unwrap()
            .unwrap();
            let WaitFrameLoad::Ready(frame) = attempt else {
                panic!("newly maintained persona must complete the pending selection")
            };
            assert_eq!(frame.persona, persona);
            assert_eq!(sources.observations.load(Ordering::Relaxed), 2);
            assert!(pile.health().started_at.is_none());
            pile.close().unwrap();
        });
    }

    #[test]
    fn only_unavailable_realization_is_preparation_pending() {
        assert!(is_preparation_pending(
            &CollectionRealizationError::IncompleteCover {
                missing: Vec::new(),
                unsupported_members: Vec::new(),
            }
            .into()
        ));
        let member = triblespace::core::collection::CollectionData::new([73; 32]);
        let missing = anyhow::Error::new(CollectionRealizationError::MissingDependency { member })
            .context("maintain selected inputs");
        assert!(is_preparation_pending(&missing));
        assert_eq!(pending_blob(&missing).unwrap().handle.raw, member.raw);
        for error in [
            CollectionRealizationError::InvalidCover("bad support".to_owned()),
            CollectionRealizationError::Resolution("conflicting equations".to_owned()),
            CollectionRealizationError::Stalled { cover: Vec::new() },
            CollectionRealizationError::Unmappable {
                blocked: vec![(member, "capacity".to_owned())],
            },
            CollectionRealizationError::UnauthorizedProducer {
                collection: triblespace::core::inline::Inline::new([74; 32]),
            },
        ] {
            assert!(!is_preparation_pending(&error.into()));
        }
    }

    #[test]
    fn habit_targets_select_before_loading_payloads() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let snapshot = pile.snapshot().unwrap();
        let habit = id(11); // An opaque, not intrinsically derived entity id.
        let owner = id(12);
        let other = id(13);
        let absent = habits::TextHandle::new([14; 32]);
        let absent_script = habits::ScriptHandle::new([15; 32]);
        let fragment = entity! { ExclusiveId::force_ref(&habit) @
            metadata::tag: &KIND_HABIT_ID,
            habit_attrs::label: "other-persona-clock",
            habit_attrs::condition: absent,
            habit_attrs::nudge: absent,
            habit_attrs::script: absent_script,
            habit_attrs::persona: &owner,
        };
        let facts = archive(fragment.facts());
        for observer in [None, Some(other)] {
            let (rows, attention) = prepare_habits(&snapshot, &facts, observer).unwrap();
            assert!(rows.is_empty());
            assert!(attention.attention.is_empty());
            assert!(!render_passive_habits(&facts, observer).contains("other-persona-clock"));
        }
        assert!(prepare_habits(&snapshot, &facts, Some(owner)).is_err());
        assert!(render_passive_habits(&facts, Some(owner)).contains("other-persona-clock"));
        pile.close().unwrap();
    }

    #[test]
    fn habit_targets_are_exact_sets_and_omission_is_global() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let collection = open_configured(
            &mut pile,
            crate::schemas::habit::DEFAULT_SCOPE_ID,
            fixture.signer.verifying_key(),
        )
        .unwrap();
        let gpt = id(16);
        let cc = id(17);
        let mut fragment = Fragment::empty();
        let mut ids = Vec::new();
        for (label, targets) in [
            ("global", vec![]),
            ("gpt", vec![gpt]),
            ("cc", vec![cc]),
            ("both", vec![cc, gpt]),
        ] {
            let (definition, id) =
                habits::habit_fragment(label, "every 1h", "do it", None, &[], &targets).unwrap();
            fragment += definition;
            ids.push(id);
        }
        let facts = archive(fragment.facts());
        pile.commit(collection, &fixture.signer, fragment).unwrap();
        let snapshot = pile.snapshot().unwrap();
        for (observer, expected) in [
            (None, BTreeSet::from([ids[0]])),
            (Some(gpt), BTreeSet::from([ids[0], ids[1], ids[3]])),
            (Some(cc), BTreeSet::from([ids[0], ids[2], ids[3]])),
            (Some(id(18)), BTreeSet::from([ids[0]])),
        ] {
            let prepared = prepare_habits(&snapshot, &facts, observer).unwrap();
            let observed = observe_habits(prepared, &fixture.path, 100).unwrap();
            assert_eq!(
                observed.due.keys().copied().collect::<BTreeSet<_>>(),
                expected
            );
        }
        pile.close().unwrap();
    }

    #[test]
    fn wait_initial_and_timer_paths_do_not_execute_other_personas_predicates() {
        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, true)
                .await
                .unwrap();
            let gpt = id(19);
            let cc = id(20);
            for (persona, label) in [(gpt, "gpt"), (cc, "cc")] {
                let (person, _, _) = relations::person_fragment(
                    persona,
                    crate::relations::ProfileInput {
                        label: label.to_owned(),
                        ..Default::default()
                    },
                )
                .unwrap();
                pile.commit(sources.relations.source, &fixture.signer, person)
                    .unwrap();
            }
            let (clock, global) =
                habits::habit_fragment("global-clock", "every 1h", "global due", None, &[], &[])
                    .unwrap();
            let (done, _) = habits::completion_fragment(
                global,
                clock::point(Epoch::from_tai_seconds(99.0)).unwrap(),
            )
            .unwrap();
            let (other, cc_habit) = habits::habit_fragment(
                "cc-clock",
                "when printf x >> other-invocations",
                "cc due",
                None,
                &[],
                &[cc],
            )
            .unwrap();
            pile.commit(
                sources.habits.as_ref().unwrap().source,
                &fixture.signer,
                clock + done + other,
            )
            .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let watermark = pile.snapshot().unwrap();
            let WaitFrameLoad::Ready(frame) = load_wait_frame(
                &mut pile,
                &sources,
                watermark,
                &mut None,
                &fixture.path,
                "gpt",
                Epoch::from_tai_seconds(100.0),
            )
            .await
            .unwrap() else {
                panic!("complete local inputs must be ready")
            };
            assert_eq!(frame.persona, gpt);
            assert!(frame.habits.due.is_empty());
            let marker = fixture.dir.join("other-invocations");
            assert!(!marker.exists());
            let later = observe_habits_in_observation(&frame.observation, &fixture.path, 3700, gpt)
                .unwrap();
            assert_eq!(later.due.keys().copied().collect::<Vec<_>>(), vec![global]);
            assert!(!marker.exists(), "the timer must retain persona selection");
            let cc_observation =
                observe_habits_in_observation(&frame.observation, &fixture.path, 3700, cc).unwrap();
            assert!(cc_observation.due.contains_key(&cc_habit));
            assert_eq!(fs::read(marker).unwrap(), b"x");
            pile.close().unwrap();
        });
    }

    fn run_wait_for(
        pile: &mut FacultyStore,
        fixture: &TestPile,
        persona: &str,
        timeout: Duration,
        poll: Duration,
    ) -> String {
        let options = WaitOptions {
            timeout: Some(timeout),
            poll_interval: poll,
        };
        let mut text = String::new();
        let mut emit = |part| {
            let crate::out::Part::Text { text: part } = part else {
                bail!("expected text")
            };
            text.push_str(&part);
            Ok(())
        };
        runtime().unwrap().block_on(async {
            tokio::time::timeout(
                timeout + Duration::from_secs(5),
                cmd_wait(
                    pile,
                    &fixture.signer,
                    &fixture.path,
                    Some(persona),
                    &options,
                    Duration::from_secs(180),
                    &mut Out::new(&mut emit),
                ),
            )
            .await
            .expect("wait honors its timeout")
            .unwrap();
        });
        text
    }

    fn run_poll_for(pile: &mut FacultyStore, fixture: &TestPile, peek: bool) -> String {
        let mut text = String::new();
        let mut emit = |part| {
            let crate::out::Part::Text { text: part } = part else {
                bail!("expected text")
            };
            text.push_str(&part);
            Ok(())
        };
        runtime()
            .unwrap()
            .block_on(cmd_poll(
                pile,
                &fixture.signer,
                Some("eager-reader"),
                peek,
                Duration::from_secs(180),
                &mut Out::new(&mut emit),
            ))
            .unwrap();
        text
    }

    #[test]
    fn poll_carries_raw_inputs_and_rearmed_receipts_without_a_daemon() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources =
            test_block_on(OrientSources::open(&mut pile, &fixture.signer, false)).unwrap();
        let reader = id(91);
        let sender = id(92);
        for (person, label) in [(reader, "eager-reader"), (sender, "eager-sender")] {
            let (fragment, _, _) = relations::person_fragment(
                person,
                crate::relations::ProfileInput {
                    label: label.to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, fragment)
                .unwrap();
        }
        let (fragment, event) = message::message_fragment(
            sender,
            &message::Recipient::Person(reader),
            "visible on the very next poll",
            clock::point_now().unwrap(),
        );
        pile.commit(sources.messages.source, &fixture.signer, fragment)
            .unwrap();
        let before = pile.snapshot().unwrap();
        assert!(before
            .attached(sources.messages.rank9)
            .unwrap()
            .cover()
            .is_empty());

        let peek = run_poll_for(&mut pile, &fixture, true);
        assert!(
            peek.contains(&format!("News: new message [{}]", fmt_short_id(event))),
            "{peek}"
        );
        assert!(peek.contains("visible on the very next poll"), "{peek}");
        assert!(sources
            .presentations
            .observe(&pile.snapshot().unwrap())
            .unwrap()
            .is_empty());
        assert!(before
            .attached(sources.messages.rank9)
            .unwrap()
            .cover()
            .is_empty());

        let delivered = run_poll_for(&mut pile, &fixture, false);
        // Short, because that is what news now prints. It stays a usable
        // argument: resolve_id_prefix accepts a prefix, so the line can be
        // pasted straight into `message ack`.
        assert!(delivered.contains(&fmt_short_id(event)), "{delivered}");
        // No fixture carry between the consuming call and its fresh successor.
        let rearmed = run_poll_for(&mut pile, &fixture, true);
        assert!(rearmed.is_empty(), "a presented event repeated: {rearmed}");
        assert!(sources
            .presentations
            .observe(&pile.snapshot().unwrap())
            .unwrap()
            .contains(event));
        assert!(pile.snapshot().unwrap().wants().unwrap().next().is_none());
        pile.close().unwrap();
    }

    #[test]
    fn wait_timeout_cancels_pending_upkeep_and_allows_a_later_retry() {
        use std::cell::Cell;

        struct DroppedAttempt<'a>(&'a Cell<usize>);
        impl Drop for DroppedAttempt<'_> {
            fn drop(&mut self) {
                self.0.set(self.0.get() + 1);
            }
        }

        // A missing endorsed blob is not a reliable stall fixture: ordinary
        // maintenance can choose resident support or rebuild another image.
        // Inject a definitely-pending future at the actual upkeep boundary
        // instead, and count both entry and cancellation. This does not claim
        // that any particular Core selection must acquire that missing image.
        let started = Cell::new(0);
        let dropped = Cell::new(0);
        let completed = Cell::new(0);
        let mut upkeep_interrupted = false;
        runtime().unwrap().block_on(async {
            let timeout_at = tokio::time::Instant::now() + Duration::from_millis(50);
            let acquisition = async {
                started.set(started.get() + 1);
                let _attempt = DroppedAttempt(&dropped);
                std::future::pending::<()>().await;
                completed.set(completed.get() + 1);
                Ok(())
            };
            let result = tokio::time::timeout(
                Duration::from_secs(3),
                wait_upkeep_before_deadline(
                    acquisition,
                    &mut upkeep_interrupted,
                    None,
                    Some(timeout_at),
                ),
            )
            .await
            .expect("without the upkeep timeout arm this definitely-pending future never returns")
            .unwrap();
            assert!(!result, "a deadline must not claim upkeep completed");
            assert!(tokio::time::Instant::now() >= timeout_at);
            assert!(upkeep_interrupted);
            assert_eq!(started.get(), 1, "upkeep must actually be polled");
            assert_eq!(dropped.get(), 1, "the interrupted acquisition is cancelled");
            assert_eq!(completed.get(), 0, "no post-acquisition work executed");

            // A later attempt with enough remaining budget can take longer than
            // the initial short cap and succeed; interruption is not sticky.
            let acquisition = async {
                started.set(started.get() + 1);
                let _attempt = DroppedAttempt(&dropped);
                tokio::time::sleep(Duration::from_millis(75)).await;
                completed.set(completed.get() + 1);
                Ok(())
            };
            let result = tokio::time::timeout(
                Duration::from_secs(3),
                wait_upkeep_before_deadline(
                    acquisition,
                    &mut upkeep_interrupted,
                    None,
                    Some(tokio::time::Instant::now() + Duration::from_secs(1)),
                ),
            )
            .await
            .expect("a successful retry remains bounded")
            .unwrap();
            assert!(result);
            assert!(!upkeep_interrupted);
            assert_eq!(started.get(), 2);
            assert_eq!(dropped.get(), 2);
            assert_eq!(completed.get(), 1);
        });
    }

    #[test]
    fn an_owned_clock_falls_due_while_a_message_body_is_still_missing() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources =
            test_block_on(OrientSources::open(&mut pile, &fixture.signer, true)).unwrap();
        let cc = id(22);
        let sender = id(23);
        let (person, _, _) = relations::person_fragment(
            cc,
            crate::relations::ProfileInput {
                label: "cc".to_owned(),
                ..Default::default()
            },
        )
        .unwrap();
        pile.commit(sources.relations.source, &fixture.signer, person)
            .unwrap();
        // A message addressed to cc whose body has not arrived: directed news
        // stays pending for the whole wait.
        let mut remote = MemoryRepo::default();
        let body = remote
            .put::<blobencodings::UTF8String, _>("still on its way".to_owned())
            .unwrap();
        pile.commit(
            sources.messages.source,
            &fixture.signer,
            message::envelope_fragment(
                sender,
                cc,
                body,
                clock::point(Epoch::from_tai_seconds(42.0)).unwrap(),
                None,
                None,
            ),
        )
        .unwrap();
        // An owned clock that falls due about three seconds into the wait.
        let now = clock::now().unwrap();
        let done = clock::point(Epoch::from_tai_seconds(now.to_tai_seconds() - 1197.0)).unwrap();
        let (owned, owned_id) =
            habits::habit_fragment("cc-clock", "every 20m", "mark it", None, &[], &[cc]).unwrap();
        let (owned_done, _) = habits::completion_fragment(owned_id, done).unwrap();
        let habit_source = sources.habits.as_ref().unwrap().source;
        pile.commit(habit_source, &fixture.signer, owned + owned_done)
            .unwrap();
        test_block_on(maintain_sources(&mut pile, &fixture.signer, &sources)).unwrap();

        let started = Instant::now();
        let text = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(8),
            Duration::from_millis(200),
        );
        assert!(
            text.contains("News: habit became due: cc-clock"),
            "the clock must be reported while the body is still missing: {text}"
        );
        assert!(
            started.elapsed() >= Duration::from_millis(500),
            "fixture setup ate the due margin: the clock was already due at arm"
        );
        assert!(!text.contains("News: new message"), "{text}");
        assert!(
            stored_presentations(&mut pile, &fixture.signer, cc).is_empty(),
            "a habit-only report acknowledges no news"
        );

        // Its owner completes the clock, then the body arrives: a rearmed wait
        // reports the message and records it.
        let (fresh, _) = habits::completion_fragment(owned_id, clock::point(now).unwrap()).unwrap();
        pile.commit(habit_source, &fixture.signer, fresh).unwrap();
        pile.put::<blobencodings::UTF8String, _>("still on its way".to_owned())
            .unwrap();
        test_block_on(maintain_sources(&mut pile, &fixture.signer, &sources)).unwrap();
        // These are semantic delivery assertions, not 25 ms latency gates.
        // Frozen selection now yields while its blocking task runs, so the
        // command can enforce its timeout during otherwise resident work.
        let delivered = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(1),
            Duration::from_secs(1),
        );
        assert!(delivered.contains("News: new message"), "{delivered}");
        assert_eq!(
            stored_presentations(&mut pile, &fixture.signer, cc).len(),
            1
        );

        // Nothing new once the worker has carried that receipt: a further
        // rearm is quiet.
        test_block_on(maintain_sources(&mut pile, &fixture.signer, &sources)).unwrap();
        let quiet = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(1),
            Duration::from_secs(1),
        );
        assert!(!quiet.contains("News:"), "{quiet}");
        assert!(quiet.contains("No change detected"), "{quiet}");
        pile.close().unwrap();
    }

    #[test]
    fn a_persona_clock_already_due_at_arm_is_reported_until_completed() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources =
            test_block_on(OrientSources::open(&mut pile, &fixture.signer, true)).unwrap();
        let cc = id(21);
        let (person, _, _) = relations::person_fragment(
            cc,
            crate::relations::ProfileInput {
                label: "cc".to_owned(),
                ..Default::default()
            },
        )
        .unwrap();
        pile.commit(sources.relations.source, &fixture.signer, person)
            .unwrap();
        let now = clock::now().unwrap();
        let an_hour_ago =
            clock::point(Epoch::from_tai_seconds(now.to_tai_seconds() - 3600.0)).unwrap();
        let (owned, owned_id) =
            habits::habit_fragment("cc-clock", "every 20m", "mark it", None, &[], &[cc]).unwrap();
        let (owned_done, _) = habits::completion_fragment(owned_id, an_hour_ago).unwrap();
        let (shared, shared_id) =
            habits::habit_fragment("shared-clock", "every 20m", "anyone", None, &[], &[]).unwrap();
        let (shared_done, _) = habits::completion_fragment(shared_id, an_hour_ago).unwrap();
        let habit_source = sources.habits.as_ref().unwrap().source;
        pile.commit(
            habit_source,
            &fixture.signer,
            owned + owned_done + shared + shared_done,
        )
        .unwrap();
        // No fixture upkeep or daemon: the command must carry these raw
        // Relations and Habit writes before it can decide what is due.

        // Both fell due before the watcher armed, and BOTH are reported: no
        // watcher observed either transition, so leaving the shared one to
        // "whoever saw it" leaves it to nobody. Reporting is what writes the
        // receipt, which is what makes the rearm below quiet.
        // Leave room for the asynchronous frozen-selection boundary; this
        // fixture tests due/receipt semantics, not sub-25 ms preparation.
        let armed = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(1),
            Duration::from_secs(1),
        );
        assert!(
            armed.contains("News: habit became due: cc-clock"),
            "{armed}"
        );
        assert!(armed.contains("shared-clock"), "{armed}");

        // Completed now. The rearmed watcher is quiet about BOTH: the owned
        // clock because completion ended its occurrence, the shared one
        // because the first report receipted it. Neither needs a baseline
        // held in this process to stay quiet.
        let (fresh, _) = habits::completion_fragment(owned_id, clock::point(now).unwrap()).unwrap();
        pile.commit(habit_source, &fixture.signer, fresh).unwrap();
        let rearmed = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(1),
            Duration::from_secs(1),
        );
        assert!(!rearmed.contains("became due"), "{rearmed}");
        assert!(!rearmed.contains("shared-clock"), "{rearmed}");
        assert!(rearmed.contains("No change detected"), "{rearmed}");
        pile.close().unwrap();
    }

    #[test]
    fn an_owned_clock_already_due_is_reported_while_a_message_body_is_still_missing() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources =
            test_block_on(OrientSources::open(&mut pile, &fixture.signer, true)).unwrap();
        let cc = id(24);
        let sender = id(25);
        let (person, _, _) = relations::person_fragment(
            cc,
            crate::relations::ProfileInput {
                label: "cc".to_owned(),
                ..Default::default()
            },
        )
        .unwrap();
        pile.commit(sources.relations.source, &fixture.signer, person)
            .unwrap();
        let mut remote = MemoryRepo::default();
        let body = remote
            .put::<blobencodings::UTF8String, _>("still on its way".to_owned())
            .unwrap();
        pile.commit(
            sources.messages.source,
            &fixture.signer,
            message::envelope_fragment(
                sender,
                cc,
                body,
                clock::point(Epoch::from_tai_seconds(42.0)).unwrap(),
                None,
                None,
            ),
        )
        .unwrap();
        let now = clock::now().unwrap();
        let an_hour_ago =
            clock::point(Epoch::from_tai_seconds(now.to_tai_seconds() - 3600.0)).unwrap();
        let (owned, owned_id) =
            habits::habit_fragment("cc-clock", "every 20m", "mark it", None, &[], &[cc]).unwrap();
        let (owned_done, _) = habits::completion_fragment(owned_id, an_hour_ago).unwrap();
        pile.commit(
            sources.habits.as_ref().unwrap().source,
            &fixture.signer,
            owned + owned_done,
        )
        .unwrap();
        test_block_on(maintain_sources(&mut pile, &fixture.signer, &sources)).unwrap();

        // Already due when the watcher arms, body still missing: reported at
        // once, nothing acknowledged.
        let armed = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_millis(500),
            Duration::from_millis(100),
        );
        assert!(
            armed.contains("News: habit became due: cc-clock"),
            "{armed}"
        );
        assert!(!armed.contains("News: new message"), "{armed}");
        assert!(stored_presentations(&mut pile, &fixture.signer, cc).is_empty());
        pile.close().unwrap();
    }

    #[test]
    fn an_owned_clock_gets_its_turn_while_a_body_fetch_stalls() {
        use triblespace_net::peer::{PeerConfig, ReconcileDirection, ReconcileQos};
        let fixture = TestPile::new();
        // A leech that knows no peers: an exact acquisition of a handle no
        // provider has occupies its whole budget rather than failing fast.
        // The foreground store is a Leech now and deliberately refuses
        // construction from a wired peer, so the stall comes from having
        // nowhere to ask rather than from a dead wire.
        let mut pile: FacultyStore = FacultyStore::lazy(
            crate::storage::open_pile_strict(&fixture.path).unwrap(),
            fixture.signer.clone(),
            PeerConfig {
                peers: Vec::new(),
                qos: ReconcileQos {
                    direction: ReconcileDirection::ReadOnly,
                },
                provider_publication_budget: Some(0),
                bind: None,
            },
        );
        let sources =
            test_block_on(OrientSources::open(&mut pile, &fixture.signer, true)).unwrap();
        let cc = id(26);
        let sender_id = id(27);
        let (person, _, _) = relations::person_fragment(
            cc,
            crate::relations::ProfileInput {
                label: "cc".to_owned(),
                ..Default::default()
            },
        )
        .unwrap();
        pile.commit(sources.relations.source, &fixture.signer, person)
            .unwrap();
        let mut remote = MemoryRepo::default();
        let body = remote
            .put::<blobencodings::UTF8String, _>("stalled on the wire".to_owned())
            .unwrap();
        pile.commit(
            sources.messages.source,
            &fixture.signer,
            message::envelope_fragment(
                sender_id,
                cc,
                body,
                clock::point(Epoch::from_tai_seconds(42.0)).unwrap(),
                None,
                None,
            ),
        )
        .unwrap();
        let now = clock::now().unwrap();
        let done = clock::point(Epoch::from_tai_seconds(now.to_tai_seconds() - 1197.0)).unwrap();
        let (owned, owned_id) =
            habits::habit_fragment("cc-clock", "every 20m", "mark it", None, &[], &[cc]).unwrap();
        let (owned_done, _) = habits::completion_fragment(owned_id, done).unwrap();
        // A shared script intention counts its evaluations: once at the first
        // preparation and once per timer sweep, never per cancelled retry.
        let (probe, probe_id) = habits::habit_fragment(
            "evaluation-probe",
            "when printf x >> evaluation-probe-invocations",
            "probe",
            None,
            &[],
            &[],
        )
        .unwrap();
        pile.commit(
            sources.habits.as_ref().unwrap().source,
            &fixture.signer,
            owned + owned_done + probe,
        )
        .unwrap();

        // The probe is due the moment the watcher arms, and the clock is not
        // (it falls due about three seconds in). Receipt the probe's
        // occurrence first, so this test measures what it is named for: the
        // owned clock getting its turn while a body fetch stalls, rather than
        // a one-shot wait returning on the first thing that happens to be
        // reportable. An intention never completed has `since` 0.
        let seen_probe = orient_model::habit_receipt_fragment(
            [(
                probe_id,
                clock::point(Epoch::from_tai_seconds(0.0)).unwrap(),
            )],
            clock::point_now().unwrap(),
        );
        let receipts = ReceiptSource::register(&mut pile, &fixture.signer)
            .unwrap()
            .source;
        pile.commit(receipts, &fixture.signer, seen_probe).unwrap();
        test_block_on(maintain_sources(&mut pile, &fixture.signer, &sources)).unwrap();

        let started = Instant::now();
        let text = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(8),
            Duration::from_millis(200),
        );
        assert!(
            text.contains("News: habit became due: cc-clock"),
            "the clock must get its turn while the fetch stalls: {text}"
        );
        assert!(
            started.elapsed() >= Duration::from_millis(500),
            "fixture setup ate the due margin: the clock was already due at arm"
        );
        assert!(
            started.elapsed() < Duration::from_secs(8),
            "the clock must not wait for the stalled fetch budget"
        );
        // Receipted before the wait, so it stays quiet -- which is the whole
        // contract: being SHOWN silences an occurrence, not being unowned.
        assert!(!text.contains("evaluation-probe"), "{text}");
        let evaluations = fs::read(fixture.dir.join("evaluation-probe-invocations"))
            .unwrap()
            .len();
        assert!(
            evaluations <= 2,
            "condition scripts reran under cancelled body retries: {evaluations} evaluations"
        );
        // A habit-only report still acknowledges NO NEWS -- it must never
        // receipt a message body the reader did not see. That invariant is
        // preserved by shape, not by abstinence: a habit receipt carries
        // `habit`+`due_at` and no `event`, so it cannot be mistaken for one.
        assert!(stored_presentations(&mut pile, &fixture.signer, cc).is_empty());
        assert!(
            !stored_habit_presentations(&mut pile, &fixture.signer).is_empty(),
            "the report receipted the occurrences it displayed"
        );
        pile.close().unwrap();
    }

    /// Habit due occurrences this signer holds receipts for, as `(habit, due)`.
    fn stored_habit_presentations(pile: &mut FacultyStore, signer: &SigningKey) -> BTreeSet<Id> {
        let collection = ReceiptSource::register(pile, signer).unwrap().source;
        let snapshot = pile.snapshot().unwrap();
        let facts = snapshot
            .collection(collection)
            .unwrap()
            .view::<TribleSet>()
            .unwrap();
        find!(
            habit: Id,
            pattern!(&facts, [{ _?receipt @ presentation::habit: ?habit }])
        )
        .collect()
    }

    #[test]
    fn ready_first_sweep_refreshes_the_pending_habit_context_before_body_ready() {
        use triblespace::core::repo::async_store::AsyncBlobStoreAcquire;

        struct Supply<'a> {
            store: &'a mut FacultyStore,
            handle: Inline<inlineencodings::Handle<blobencodings::UnknownBlob>>,
            requested: usize,
        }

        impl SnapshotSource for Supply<'_> {
            type Snapshot = FacultySnapshot;
            type SnapshotError = <FacultyStore as SnapshotSource>::SnapshotError;

            fn snapshot(&mut self) -> std::result::Result<Self::Snapshot, Self::SnapshotError> {
                self.store.snapshot()
            }
        }

        impl AsyncBlobStoreAcquire for Supply<'_> {
            type AcquireError = io::Error;

            async fn acquire(
                &mut self,
                handle: Inline<inlineencodings::Handle<blobencodings::UnknownBlob>>,
            ) -> std::result::Result<Option<Bytes>, Self::AcquireError> {
                self.requested += 1;
                assert_eq!(handle, self.handle, "only the selected body is missing");
                let snapshot = self.store.snapshot().unwrap();
                if !snapshot.contains_blob(handle).unwrap() {
                    return Ok(None);
                }
                Ok(Some(
                    snapshot
                        .get::<Bytes, blobencodings::UnknownBlob>(handle)
                        .await
                        .unwrap(),
                ))
            }
        }

        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        test_block_on(async {
            let sources = OrientSources::open(&mut pile, &fixture.signer, true)
                .await
                .unwrap();
            let persona = id(28);
            let sender = id(29);
            let (person, _, _) = relations::person_fragment(
                persona,
                crate::relations::ProfileInput {
                    label: "ready-first".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, person)
                .unwrap();
            let marker = fixture.dir.join("ready-first-due");
            let evaluations = fixture.dir.join("ready-first-evaluations");
            fs::write(&marker, b"").unwrap();
            let (probe, probe_id) = habits::habit_fragment(
                "ready-first-probe",
                "when printf x >> ready-first-evaluations; test -e ready-first-due",
                "probe",
                None,
                &[],
                &[],
            )
            .unwrap();
            let habit_source = sources.habits.as_ref().unwrap().source;
            pile.commit(habit_source, &fixture.signer, probe).unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();

            // First reach Ready: the shared due intention is a quiet baseline.
            let snapshot = pile.snapshot().unwrap();
            let WaitFrameLoad::Ready(current) = load_wait_frame(
                &mut pile,
                &sources,
                snapshot,
                &mut None,
                &fixture.path,
                "ready-first",
                Epoch::from_tai_seconds(100.0),
            )
            .await
            .unwrap() else {
                panic!("the initial frame must be fully readable")
            };
            assert!(current.habits.due.contains_key(&probe_id));
            assert_eq!(current.habits.next_cooldown_at, None);
            // Due and unreceipted, so reported. Under the old rule this
            // stayed silent because the intention was not addressed to this
            // persona -- which is exactly how a shared intention that fell due
            // unobserved was lost.
            assert!(render_due_habits_unreceipted(&current.habits, &nothing_presented()).is_some());

            // A later selected frame has a missing body and a genuinely newer
            // Habit context. Its cooling row is absent from `current`, so a
            // sweep over that old context cannot correctly refresh this cache.
            let body_text = "ready after a Habit timer sweep";
            let mut remote = MemoryRepo::default();
            let body = remote
                .put::<blobencodings::UTF8String, _>(body_text.to_owned())
                .unwrap();
            let message = message::envelope_fragment(
                sender,
                persona,
                body,
                clock::point(Epoch::from_tai_seconds(110.0)).unwrap(),
                None,
                None,
            );
            let event = message.root().unwrap();
            pile.commit(sources.messages.source, &fixture.signer, message)
                .unwrap();
            let (cooling, cooling_id) =
                habits::habit_fragment("new-context", "every 20m", "later", None, &[], &[])
                    .unwrap();
            let (done, _) = habits::completion_fragment(
                cooling_id,
                clock::point(Epoch::from_tai_seconds(20.0)).unwrap(),
            )
            .unwrap();
            pile.commit(habit_source, &fixture.signer, cooling + done)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let selected = pile.snapshot().unwrap();
            let handle = Inline::new(body.raw);
            assert!(!selected.contains_blob(handle).unwrap());
            // Select exactly as load_wait_frame does, then inject a local-only
            // reader at its generic continuation seam. No network timing or
            // acquisition cancellation participates in this regression.
            let observation = observe_snapshot(selected.clone(), &sources).unwrap();
            let mut frame =
                PendingWaitFrame::awaiting_view(selected.clone(), PendingWaitReason::Payload);
            frame.observation = Some(observation);
            {
                let mut supply = Supply {
                    store: &mut pile,
                    handle,
                    requested: 0,
                };
                assert!(resume_wait_payloads(
                    &mut supply,
                    &mut frame,
                    &fixture.path,
                    "ready-first",
                    Epoch::from_tai_seconds(120.0),
                )
                .await
                .unwrap()
                .is_none());
                assert_eq!(supply.requested, 1);
            }
            assert_eq!(frame.missing.unwrap().handle, handle);
            assert_eq!(frame.persona, Some(persona));
            assert!(frame.habits.as_ref().unwrap().due.contains_key(&probe_id));
            assert_eq!(frame.habits.as_ref().unwrap().next_cooldown_at, Some(1220));
            let support = frame
                .observation
                .as_ref()
                .unwrap()
                .facts
                .messages
                .support()
                .clone();
            let mut pending = Some(frame);

            // The established wait's timer path observes Waiting and retains
            // that result in the *same newer* context before the body arrives.
            fs::remove_file(marker).unwrap();
            let seen = sweep_wait_habits(
                &current.observation,
                current.persona,
                &mut pending,
                &fixture.path,
                200,
            )
            .unwrap();
            assert!(!seen.due.contains_key(&probe_id));
            assert!(
                render_habit_transitions(&current.habits, &seen, &nothing_presented()).is_none()
            );
            assert_eq!(fs::read(&evaluations).unwrap(), b"xxx");
            assert!(stored_presentations(&mut pile, &fixture.signer, persona).is_empty());

            pile.put::<blobencodings::UTF8String, _>(body_text.to_owned())
                .unwrap();
            let ready = {
                let mut supply = Supply {
                    store: &mut pile,
                    handle,
                    requested: 0,
                };
                let ready = resume_wait_payloads(
                    &mut supply,
                    pending.as_mut().unwrap(),
                    &fixture.path,
                    "ready-first",
                    Epoch::from_tai_seconds(300.0),
                )
                .await
                .unwrap()
                .expect("only the selected body was missing");
                assert_eq!(supply.requested, 1);
                ready
            };
            assert_eq!(ready.persona, persona);
            assert!(
                render_habit_transitions(&seen, &ready.habits, &nothing_presented()).is_none(),
                "Ready must not resurrect the Due result cached before the timer sweep"
            );
            assert_eq!(ready.habits, seen);
            assert_eq!(
                ready.habits.next_cooldown_at,
                Some(1220),
                "the refreshed cache must retain the newer selected Habit context"
            );
            assert_eq!(
                fs::read(evaluations).unwrap(),
                b"xxx",
                "a body retry runs no scripts"
            );
            assert!(!wait_storage_changed(&ready.watermark, &selected));
            assert_eq!(ready.observation.facts.messages.support(), &support);
            let News::Report { events, .. } = ready.news else {
                panic!("the newly resident body must produce its selected news")
            };
            assert_eq!(events, vec![event]);
            assert!(stored_presentations(&mut pile, &fixture.signer, persona).is_empty());
        });
        pile.close().unwrap();
    }

    #[test]
    fn a_shared_intention_due_at_arm_is_presented_once_and_the_body_still_carries_the_frame_into_ready(
    ) {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources =
            test_block_on(OrientSources::open(&mut pile, &fixture.signer, true)).unwrap();
        let cc = id(22);
        let sender = id(23);
        let (person, _, _) = relations::person_fragment(
            cc,
            crate::relations::ProfileInput {
                label: "cc".to_owned(),
                ..Default::default()
            },
        )
        .unwrap();
        pile.commit(sources.relations.source, &fixture.signer, person)
            .unwrap();
        // A message whose body arrives only after the first timer sweep.
        let mut remote = MemoryRepo::default();
        let body = remote
            .put::<blobencodings::UTF8String, _>("arrives after the sweep".to_owned())
            .unwrap();
        pile.commit(
            sources.messages.source,
            &fixture.signer,
            message::envelope_fragment(
                sender,
                cc,
                body,
                clock::point(Epoch::from_tai_seconds(42.0)).unwrap(),
                None,
                None,
            ),
        )
        .unwrap();
        // A shared script intention that is due at arm and no longer due ten
        // seconds later. Due-ness is decided by receipt: the first watcher
        // presents it once, while the message body is still pending, and
        // receipts it; the rearmed watcher stays quiet about it and waits for
        // the body. Ten seconds, not two: the first read of an arm is budgeted
        // one poll interval and a loaded machine cuts it, so the window in
        // which the habits are first evaluated must hold many attempts.
        let marker = fixture.dir.join("due-marker");
        fs::write(&marker, b"").unwrap();
        let (probe, _) =
            habits::habit_fragment("probe", "when test -e due-marker", "probe", None, &[], &[])
                .unwrap();
        pile.commit(
            sources.habits.as_ref().unwrap().source,
            &fixture.signer,
            probe,
        )
        .unwrap();
        test_block_on(maintain_sources(&mut pile, &fixture.signer, &sources)).unwrap();

        let path = fixture.path.clone();
        let deliverer = std::thread::spawn(move || {
            std::thread::sleep(Duration::from_secs(10));
            fs::remove_file(&marker).unwrap();
            std::thread::sleep(Duration::from_secs(50));
            let mut second = open_store(&path).unwrap();
            second
                .put::<blobencodings::UTF8String, _>("arrives after the sweep".to_owned())
                .unwrap();
            second.close().unwrap();
        });
        let first = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(80),
            Duration::from_millis(200),
        );
        assert!(
            first.contains("News: habit became due: probe"),
            "a shared intention due at arm is presented while the body is pending: {first}"
        );
        assert!(
            !first.contains("News: new message"),
            "the body had not landed yet: {first}"
        );
        let second = run_wait_for(
            &mut pile,
            &fixture,
            "cc",
            Duration::from_secs(80),
            Duration::from_millis(200),
        );
        deliverer.join().unwrap();
        assert!(
            second.contains("News: new message"),
            "the body must land within the wait: {second}"
        );
        // The occurrence presented at the first arm was receipted there; the
        // rearmed watcher's own arm, sweep and Ready never present it again.
        assert!(
            !second.contains("became due"),
            "a presented occurrence was shown again by a rearmed watcher: {second}"
        );
        pile.close().unwrap();
    }

    fn stored_presentations(
        pile: &mut FacultyStore,
        signer: &SigningKey,
        _persona: Id,
    ) -> BTreeSet<Id> {
        // Inspect committed receipt facts, not the asynchronously maintained
        // projection. Routing aliases share the signing zooid's source.
        let collection = ReceiptSource::register(pile, signer).unwrap().source;
        let snapshot = pile.snapshot().unwrap();
        let facts = snapshot
            .collection(collection)
            .unwrap()
            .view::<TribleSet>()
            .unwrap();
        find!(event: Id, pattern!(&facts, [{ presentation::event: ?event }])).collect()
    }

    #[test]
    fn facts_do_not_cross_source_boundaries() {
        let person = entity! { metadata::tag: &KIND_PERSON_ID };
        let person_id = person.root().unwrap();
        let message_source = archive(person.facts());
        let relations_source = archive(&TribleSet::new());

        assert_eq!(person_anchors(&message_source), BTreeSet::from([person_id]));
        assert!(person_anchors(&relations_source).is_empty());
    }

    /// The status register's attachments may lag the facts' -- the facts
    /// attached a commit the register has not -- and a read still answers
    /// for every status the facts hold: the register's residual is built in
    /// memory. An observation taken earlier keeps its own answer.
    #[test]
    fn a_status_register_whose_attachments_lag_still_answers_for_the_facts() {
        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let goal = id(46);
            let unseen_goal = id(47);
            let initial = compass::status_fragment(
                goal,
                "todo",
                None,
                clock::point(Epoch::from_tai_seconds(1.0)).unwrap(),
            )
            .unwrap();
            let initial_id = initial.root().unwrap();
            pile.commit(sources.compass.source, &fixture.signer, initial)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let watermark = pile.snapshot().unwrap();
            let observation = observe_snapshot(watermark, &sources).unwrap();

            let next = compass::status_fragment(
                goal,
                "done",
                None,
                clock::point(Epoch::from_tai_seconds(2.0)).unwrap(),
            )
            .unwrap();
            let next_id = next.root().unwrap();
            let unseen = compass::status_fragment(
                unseen_goal,
                "doing",
                None,
                clock::point(Epoch::from_tai_seconds(3.0)).unwrap(),
            )
            .unwrap();
            let unseen_id = unseen.root().unwrap();
            pile.commit(sources.compass.source, &fixture.signer, next + unseen)
                .unwrap();
            drop(
                pile.maintain_attached(sources.compass.succinct, &fixture.signer)
                    .await
                    .unwrap(),
            );
            let snapshot = pile
                .maintain_attached(sources.compass.rank9, &fixture.signer)
                .await
                .unwrap();
            let lagging = observe_sources(snapshot, &sources).unwrap();
            // The facts attached the new commit and the status register has
            // not; its read builds that commit in memory and answers for it.
            let status = lagging.snapshot.attached(sources.compass_status).unwrap();
            assert_eq!(status.residual().len(), 1);
            drop(status);
            let query = lagging.query(&lagging.snapshot);
            assert_eq!(latest_goal_status(&query, goal).unwrap().0, next_id);
            assert_eq!(
                latest_goal_status(&query, unseen_goal).unwrap().0,
                unseen_id
            );

            let ready = pile
                .maintain_attached(sources.compass_status, &fixture.signer)
                .await
                .unwrap();
            let advanced = observe_sources(ready, &sources).unwrap();
            let query = advanced.query(&advanced.snapshot);
            assert_eq!(latest_goal_status(&query, goal).unwrap().0, next_id);
            assert_eq!(
                latest_goal_status(&query, unseen_goal).unwrap().0,
                unseen_id
            );
            let frozen = lagging.query(&lagging.snapshot);
            assert_eq!(latest_goal_status(&frozen, goal).unwrap().0, next_id);
            assert_eq!(
                latest_goal_status(&frozen, unseen_goal).unwrap().0,
                unseen_id
            );
            assert_eq!(
                latest_goal_status(&observation.query(&observation.snapshot), goal)
                    .unwrap()
                    .0,
                initial_id,
            );
            pile.close().unwrap();
        });
    }

    #[test]
    fn wait_selects_resident_targets_and_preserves_polling_watermark() {
        test_block_on(wait_selects_resident_targets_and_preserves_polling_watermark_async())
    }

    async fn wait_selects_resident_targets_and_preserves_polling_watermark_async() {
        use triblespace::core::repo::WantRead;

        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources = OrientSources::open(&mut pile, &fixture.signer, true)
            .await
            .unwrap();
        let persona = id(42);
        let profile = crate::relations::ProfileInput {
            label: "test-persona".to_owned(),
            ..Default::default()
        };
        let (person, _, _) = relations::person_fragment(persona, profile).unwrap();
        pile.commit(sources.relations.source, &fixture.signer, person)
            .unwrap();
        sources
            .relations
            .maintain(&mut pile, &fixture.signer)
            .await
            .unwrap();
        let watermark = pile.snapshot().unwrap();
        let expected_support = watermark
            .attached(sources.messages.rank9)
            .unwrap()
            .support()
            .clone();

        // This commit arrives after the wait watermark was frozen. Another
        // maintainer also realizes that newer support before this reader runs.
        // The selected view must still belong to the supplied snapshot. A new
        // polling sample will see the externally produced target progress.
        let message_collection = sources.messages.source;
        pile.commit(
            message_collection,
            &fixture.signer,
            entity! { metadata::tag: &KIND_MESSAGE_ID },
        )
        .unwrap();
        drop(
            pile.maintain_attached(sources.messages.succinct, &fixture.signer)
                .await
                .unwrap(),
        );
        drop(
            pile.maintain_attached(sources.messages.rank9, &fixture.signer)
                .await
                .unwrap(),
        );
        let attempt = load_wait_frame(
            &mut pile,
            &sources,
            watermark.clone(),
            &mut None,
            &fixture.path,
            "test-persona",
            Epoch::from_tai_seconds(42.0),
        )
        .await
        .unwrap();
        let WaitFrameLoad::Ready(frame) = attempt else {
            panic!("resident persona payload unexpectedly pending")
        };

        let resident_after = frame
            .observation
            .snapshot
            .attached(sources.messages.rank9)
            .unwrap();
        assert_eq!(
            frame.observation.facts.messages.support(),
            &expected_support,
            "later target progress cannot alter the selected immutable snapshot",
        );
        assert_eq!(
            frame.observation.facts.messages.support(),
            resident_after.support(),
            "the selected view must be the resident target at the observation snapshot",
        );
        assert_eq!(frame.observation.facts.messages.view().iter().count(), 0);
        assert!(frame.observation.snapshot.wants().unwrap().next().is_none());
        assert!(
            frame.watermark.changes_since(&watermark).is_empty()
                && watermark.changes_since(&frame.watermark).is_empty(),
            "the production wait frame must retain the exact input watermark",
        );
        assert!(
            frame
                .observation
                .snapshot
                .changes_since(&frame.watermark)
                .is_empty(),
            "passive attachment must use the exact polling snapshot",
        );
        let sampled = pile.snapshot().unwrap();
        assert!(wait_storage_changed(&sampled, &frame.watermark));
        let WaitFrameLoad::Ready(next) = load_wait_frame(
            &mut pile,
            &sources,
            sampled,
            &mut None,
            &fixture.path,
            "test-persona",
            Epoch::from_tai_seconds(42.0),
        )
        .await
        .unwrap() else {
            panic!("externally maintained target must be readable")
        };
        assert_eq!(next.observation.facts.messages.view().iter().count(), 1);
        assert_ne!(next.observation.facts.messages.support(), &expected_support);
        pile.close().unwrap();
    }

    #[test]
    fn tags_are_normalized_for_display() {
        assert_eq!(
            render_tags(&[
                "review".to_owned(),
                "#urgent".to_owned(),
                "review".to_owned(),
            ]),
            " #urgent #review"
        );
    }

    #[test]
    fn projected_receipt_ids_are_subtracted_from_attention() {
        let first = id(1);
        let second = id(2);
        let mut view = AttentionView::default();
        view.insert(AttentionEvent::Message(first));
        view.insert(AttentionEvent::Mail(second));

        let presented = presented([first]);
        assert_eq!(
            view.pending(&presented).ids().collect::<Vec<_>>(),
            vec![second]
        );
    }

    #[test]
    fn duration_wait_target_is_parsed_without_wall_clock_state() {
        let duration = parse_wait_target(Some(&WaitTarget::For {
            duration: "1500ms".to_owned(),
        }))
        .unwrap();
        assert_eq!(duration, Some(Duration::from_millis(1500)));
    }

    #[test]
    fn a_missing_selected_payload_is_pending() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let snapshot = pile.snapshot().unwrap();
        let handle = Inline::<inlineencodings::Handle<blobencodings::UTF8String>>::new([42; 32]);

        let error = read_utf8(&snapshot, handle, "selected test body").unwrap_err();
        assert!(is_payload_pending(&error));
        assert_eq!(
            error.downcast_ref::<MissingBlob>().unwrap().handle.raw,
            handle.raw,
        );
        assert!(!pile.snapshot().unwrap().contains_blob(handle).unwrap());
        pile.close().unwrap();
    }

    #[test]
    fn news_acquires_only_its_selected_body_before_presenting_frozen_events() {
        use triblespace::core::repo::async_store::AsyncBlobStoreAcquire;
        use triblespace::core::repo::WantRead;

        struct Supply {
            store: FacultyStore,
            source: Collection<blobencodings::SimpleArchive>,
            succinct: Collection<SuccinctArchiveBlob>,
            rank9: Collection<Rank9AcceleratedSuccinctArchiveBlob>,
            signer: SigningKey,
            handle: Inline<inlineencodings::Handle<blobencodings::UnknownBlob>>,
            bytes: Bytes,
            available: bool,
            requested: Vec<Inline<inlineencodings::Handle<blobencodings::UnknownBlob>>>,
        }

        impl SnapshotSource for Supply {
            type Snapshot = FacultySnapshot;
            type SnapshotError = <FacultyStore as SnapshotSource>::SnapshotError;

            fn snapshot(&mut self) -> std::result::Result<Self::Snapshot, Self::SnapshotError> {
                self.store.snapshot()
            }
        }

        impl AsyncBlobStoreAcquire for Supply {
            type AcquireError = io::Error;

            async fn acquire(
                &mut self,
                handle: Inline<inlineencodings::Handle<blobencodings::UnknownBlob>>,
            ) -> std::result::Result<Option<Bytes>, Self::AcquireError> {
                self.requested.push(handle);
                assert_eq!(handle, self.handle, "an unrelated body must stay cold");
                if !self.available {
                    return Ok(None);
                }
                let cached = self
                    .store
                    .put::<blobencodings::UnknownBlob, _>(self.bytes.clone())
                    .unwrap();
                assert_eq!(cached, handle);
                // Acquiring bytes races with an unrelated authoritative append
                // and its maintenance. The render must retain its chosen view
                // even though a newer target is now resident in the reader.
                self.store
                    .commit(
                        self.source,
                        &self.signer,
                        entity! { metadata::tag: &KIND_MESSAGE_ID },
                    )
                    .unwrap();
                drop(
                    self.store
                        .maintain_attached(self.succinct, &self.signer)
                        .await
                        .unwrap(),
                );
                drop(
                    self.store
                        .maintain_attached(self.rank9, &self.signer)
                        .await
                        .unwrap(),
                );
                Ok(Some(self.bytes.clone()))
            }
        }

        test_block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, true)
                .await
                .unwrap();
            let persona = id(43);
            let sender = id(44);
            let (person, _, _) = relations::person_fragment(
                persona,
                relations::ProfileInput {
                    label: "lazy-reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, person)
                .unwrap();

            let mut remote = MemoryRepo::default();
            let bytes: Bytes = Vec::from("newly synced message body").into();
            let body = remote
                .put::<blobencodings::UTF8String, _>("newly synced message body".to_owned())
                .unwrap();
            let unrelated = remote
                .put::<blobencodings::UTF8String, _>("not addressed to this reader".to_owned())
                .unwrap();
            let instant = Epoch::from_tai_seconds(42.0);
            let envelope = message::envelope_fragment(
                sender,
                persona,
                body,
                clock::point(instant).unwrap(),
                None,
                None,
            );
            let event = envelope.root().unwrap();
            pile.commit(sources.messages.source, &fixture.signer, envelope)
                .unwrap();
            pile.commit(
                sources.messages.source,
                &fixture.signer,
                message::envelope_fragment(
                    sender,
                    id(45),
                    unrelated,
                    clock::point(instant).unwrap(),
                    None,
                    None,
                ),
            )
            .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let watermark = pile.snapshot().unwrap();
            let observation = observe_snapshot(watermark.clone(), &sources).unwrap();
            let support = observation.facts.messages.support().clone();
            let handle = Inline::new(body.raw);
            let mut supply = Supply {
                store: pile,
                source: sources.messages.source,
                succinct: sources.messages.succinct,
                rank9: sources.messages.rank9,
                signer: fixture.signer.clone(),
                handle,
                bytes,
                available: false,
                requested: Vec::new(),
            };

            let frozen = observation.snapshot.clone();
            let mut pending =
                PendingWaitFrame::awaiting_view(watermark.clone(), PendingWaitReason::Payload);
            pending.observation = Some(observation);
            assert!(resume_wait_payloads(
                &mut supply,
                &mut pending,
                &fixture.path,
                "lazy-reader",
                instant,
            )
            .await
            .unwrap()
            .is_none());
            assert_eq!(pending.reason, PendingWaitReason::Payload);
            assert_eq!(pending.missing.unwrap().handle, handle);
            assert!(stored_presentations(&mut supply.store, &fixture.signer, persona).is_empty());

            let before_provider = supply.snapshot().unwrap();
            assert!(!wait_storage_changed(&before_provider, &pending.watermark));
            supply.available = true;
            assert!(!wait_storage_changed(
                &supply.snapshot().unwrap(),
                &before_provider
            ));
            let frame = resume_wait_payloads(
                &mut supply,
                &mut pending,
                &fixture.path,
                "lazy-reader",
                instant,
            )
            .await
            .unwrap()
            .expect("a provider-only change must complete the selected read");
            let reader = &frame.observation.snapshot;
            let news = frame.news;
            assert_eq!(frame.persona, persona);
            assert!(!wait_storage_changed(&frame.watermark, &watermark));
            assert_eq!(
                sources
                    .observations
                    .load(std::sync::atomic::Ordering::Relaxed),
                1
            );
            assert_eq!(supply.requested, [handle, handle]);
            assert!(!frozen.contains_blob(handle).unwrap());
            assert!(!reader.contains_blob(unrelated).unwrap());
            assert_eq!(frame.observation.facts.messages.support(), &support);
            assert_ne!(reader.attached(supply.rank9).unwrap().support(), &support);
            assert!(reader.wants().unwrap().next().is_none());
            let News::Report { text, events } = &news else {
                panic!("the acquired body must make the selected message readable")
            };
            assert!(text.contains("newly synced message body"));
            assert_eq!(events, &[event]);
            assert!(stored_presentations(&mut supply.store, &fixture.signer, persona).is_empty());

            let mut output = Vec::new();
            apply_news_to_writer(
                &mut supply.store,
                &fixture.signer,
                persona,
                false,
                &news,
                "",
                &mut output,
            )
            .unwrap();
            assert!(String::from_utf8(output)
                .unwrap()
                .contains("newly synced message body"));
            assert_eq!(
                stored_presentations(&mut supply.store, &fixture.signer, persona),
                BTreeSet::from([event]),
            );
            supply.store.close().unwrap();
        });
    }

    #[test]
    fn a_missing_persona_preserves_the_wait_watermark() {
        test_block_on(a_missing_persona_preserves_the_wait_watermark_async())
    }

    async fn a_missing_persona_preserves_the_wait_watermark_async() {
        let fixture = TestPile::new();
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let sources = OrientSources::open(&mut pile, &fixture.signer, true)
            .await
            .unwrap();
        let watermark = pile.snapshot().unwrap();

        let attempt = load_wait_frame(
            &mut pile,
            &sources,
            watermark.clone(),
            &mut None,
            &fixture.path,
            "not-yet-resident",
            Epoch::from_tai_seconds(42.0),
        )
        .await
        .unwrap();
        let WaitFrameLoad::Pending(pending) = attempt else {
            panic!("an absent persona unexpectedly produced a readable wait frame")
        };
        assert_eq!(pending.reason, PendingWaitReason::PersonaSelection);
        assert!(
            pending.watermark.changes_since(&watermark).is_empty()
                && watermark.changes_since(&pending.watermark).is_empty(),
            "a pending production frame must preserve its input watermark",
        );
        pile.close().unwrap();
    }

    #[test]
    fn report_is_written_before_its_flush_barrier() {
        #[derive(Default)]
        struct Writer {
            bytes: Vec<u8>,
            flushed: bool,
        }

        impl Write for Writer {
            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
                assert!(!self.flushed);
                self.bytes.extend_from_slice(bytes);
                Ok(bytes.len())
            }

            fn flush(&mut self) -> io::Result<()> {
                assert_eq!(self.bytes, b"complete");
                self.flushed = true;
                Ok(())
            }
        }

        let mut writer = Writer::default();
        write_report_to_writer(&mut writer, "complete", "test report").unwrap();
        assert!(writer.flushed);
    }

    #[test]
    fn failed_flush_is_retryable_and_cannot_present() {
        struct FailingFlush(Vec<u8>);

        impl Write for FailingFlush {
            fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
                self.0.extend_from_slice(bytes);
                Ok(bytes.len())
            }

            fn flush(&mut self) -> io::Result<()> {
                Err(io::Error::other("injected flush failure"))
            }
        }

        let fixture = TestPile::new();
        let persona = id(3);
        let event = id(4);
        let news = News::Report {
            text: "News: retry me\n".to_owned(),
            events: vec![event],
        };
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let error = apply_news_to_writer(
            &mut pile,
            &fixture.signer,
            persona,
            false,
            &news,
            "",
            &mut FailingFlush(Vec::new()),
        )
        .unwrap_err();
        assert!(format!("{error:#}").contains("injected flush failure"));
        assert!(stored_presentations(&mut pile, &fixture.signer, persona).is_empty());

        let mut output = Vec::new();
        apply_news_to_writer(
            &mut pile,
            &fixture.signer,
            persona,
            false,
            &news,
            "",
            &mut output,
        )
        .unwrap();
        assert_eq!(output, b"News: retry me\n");
        assert_eq!(
            stored_presentations(&mut pile, &fixture.signer, persona),
            BTreeSet::from([event])
        );
        pile.close().unwrap();
    }

    #[test]
    fn peek_reports_without_presenting() {
        let fixture = TestPile::new();
        let persona = id(5);
        let event = id(6);
        let news = News::Report {
            text: "News: peek\n".to_owned(),
            events: vec![event],
        };
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
        let mut output = Vec::new();
        apply_news_to_writer(
            &mut pile,
            &fixture.signer,
            persona,
            true,
            &news,
            "",
            &mut output,
        )
        .unwrap();

        assert_eq!(output, b"News: peek\n");
        assert!(stored_presentations(&mut pile, &fixture.signer, persona).is_empty());
        pile.close().unwrap();
    }

    #[test]
    fn baseline_is_exactly_the_current_attention_set() {
        let fixture = TestPile::new();
        let persona = id(7);
        let first = id(8);
        let second = id(9);
        let mut view = AttentionView::default();
        view.insert(AttentionEvent::Message(first));
        view.insert(AttentionEvent::Mail(second));
        let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();

        save_presentations(&mut pile, &fixture.signer, view.ids()).unwrap();
        let presented = stored_presentations(&mut pile, &fixture.signer, persona);
        assert_eq!(presented, BTreeSet::from([first, second]));
        let projected = super::tests::presented(presented);
        assert!(view.pending(&projected).is_empty());
        pile.close().unwrap();
    }

    #[test]
    fn becoming_relevant_late_does_not_retroactively_present_an_event() {
        let event = id(10);
        let initial = AttentionView::default();
        let presented = presented(initial.ids());

        let mut later = AttentionView::default();
        later.insert(AttentionEvent::Note {
            note: event,
            goal: id(11),
        });
        assert_eq!(
            later.pending(&presented).ids().collect::<Vec<_>>(),
            vec![event]
        );
    }

    /// A News line carries what the reader has to act on, not the identifier
    /// they would otherwise have to look up: a note names its goal, its author
    /// and one bounded line of its body, and a status move names both lanes.
    #[test]
    fn news_names_goals_authors_and_both_lanes_instead_of_bare_ids() {
        runtime().unwrap().block_on(async {
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let reader_id = id(90);
            let author_id = id(91);
            for (person, label) in [(reader_id, "cc"), (author_id, "astra")] {
                let (profile, _, _) = relations::person_fragment(
                    person,
                    relations::ProfileInput {
                        label: label.to_owned(),
                        ..Default::default()
                    },
                )
                .unwrap();
                pile.commit(sources.relations.source, &fixture.signer, profile)
                    .unwrap();
            }
            let moment =
                |seconds: f64| clock::point(Epoch::from_tai_seconds(seconds)).unwrap();
            let (goal, goal_id) = compass::goal_fragment(
                "Give the maintenance daemons their cores",
                vec!["cc".to_owned()],
                None,
                moment(1.0),
            )
            .unwrap();
            pile.commit(sources.compass.source, &fixture.signer, goal)
                .unwrap();
            for (status, when) in [("doing", moment(2.0)), ("done", moment(3.0))] {
                pile.commit(
                    sources.compass.source,
                    &fixture.signer,
                    compass::status_fragment(goal_id, status, Some(author_id), when).unwrap(),
                )
                .unwrap();
            }
            let (note, _) = compass::note_fragment(
                goal_id,
                "Slice B frozen, Core lock unchanged\nthe rest of the body stays in Compass",
                vec![],
                vec![],
                vec![],
                Some(author_id),
                moment(4.0),
            )
            .unwrap();
            pile.commit(sources.compass.source, &fixture.signer, note)
                .unwrap();
            let (envelope, _) = message::message_fragment(
                author_id,
                &message::Recipient::Person(reader_id),
                "the body itself still follows under New messages",
                moment(5.0),
            );
            pile.commit(sources.messages.source, &fixture.signer, envelope)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();

            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let news = read(&mut pile, &observation.snapshot, |reader| {
                let query = observation.query(reader);
                prepare_news_once(&query, reader_id)
            })
            .await
            .unwrap();
            let News::Report { text, .. } = news else {
                panic!("a goal tagged for the reader with a foreign note is news");
            };
            assert!(text.contains("News: new message ["), "{text}");
            assert!(text.contains("] from astra"), "{text}");
            let goal_hex = fmt_id(goal_id);
            let short = &goal_hex[..8];
            assert!(
                text.contains(&format!(
                    "News: goal [{short}] \"Give the maintenance daemons their cores\": doing -> done by astra"
                )),
                "{text}"
            );
            assert!(
                text.contains(&format!(
                    "News: note on [{short}] \"Give the maintenance daemons their cores\" by astra: Slice B frozen, Core lock unchanged…"
                )),
                "{text}"
            );
            // Bounded preview: the first line only, and the full id never has
            // to be printed once the title names the goal.
            assert!(!text.contains("the rest of the body stays in Compass"), "{text}");
            assert!(!text.contains(&goal_hex), "{text}");
            pile.close().unwrap();
        });
    }

    /// Discord is read like Teams: a message somebody else wrote is news with
    /// its author, channel, first line and attachments; the bot's own message
    /// and a system notice are not; and a presented message never repeats,
    /// not even once edited.
    #[test]
    fn discord_messages_from_others_are_news_once_and_the_bots_own_never() {
        crate::test_support::clear_ambient_environment();
        runtime().unwrap().block_on(async {
            use crate::discord::operations::observed_fragment;
            use serde_json::json;
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let reader_id = id(90);
            let (profile, _, _) = relations::person_fragment(
                reader_id,
                relations::ProfileInput {
                    label: "reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, profile)
                .unwrap();

            let channel = "100000000000000200";
            let bot = "100000000000000900";
            let message = |id: &str, author: &str, content: &str, edited: Option<&str>| {
                let (name, attachments) = if author == bot {
                    ("Bot", json!([]))
                } else {
                    let photo = json!({
                        "id": "100000000000000509", "url": "https://cdn.example/p.png",
                        "filename": "photo.png", "content_type": "image/png"
                    });
                    ("Ada", json!([photo]))
                };
                json!({
                    "id": id, "channel_id": channel, "guild_id": "100000000000000300",
                    "type": 0, "content": content,
                    "author": {"id": author, "username": author, "global_name": name},
                    "timestamp": "2026-09-26T08:00:00Z", "edited_timestamp": edited,
                    "attachments": attachments,
                })
            };
            let stored = |message: serde_json::Value| {
                observed_fragment(message, |_, _| Ok(b"image".to_vec())).unwrap()
            };
            let mut facts = crate::discord::bot_account_fragment(bot).unwrap();
            // Intake hears the channel from before every message here.
            facts += crate::discord::intake_fragment(channel, 100000000000000000).unwrap();
            facts += stored(message(
                "100000000000000501",
                "100000000000000400",
                "look at this\nsecond line",
                None,
            ));
            facts += stored(message("100000000000000502", bot, "my own reply", None));
            // Somebody pinning a message is a system notice, not writing.
            let mut pin = message("100000000000000503", "100000000000000400", "", None);
            pin["type"] = json!(6);
            pin["attachments"] = json!([]);
            facts += stored(pin);
            pile.commit(sources.discord.source, &fixture.signer, facts)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();

            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let news = read(&mut pile, &observation.snapshot, |reader| {
                let query = observation.query(reader);
                prepare_news_once(&query, reader_id)
            })
            .await
            .unwrap();
            let News::Report { text, events } = &news else {
                panic!("a Discord message from somebody else is news");
            };
            assert_eq!(events.len(), 1, "{text}");
            assert!(
                text.contains(&format!(
                    "News: Discord message from Ada in channel {channel}: look at this… (1 attachment)"
                )),
                "{text}"
            );
            // The attachment is named by the handle of its stored bytes.
            let photo: Blob<blobencodings::RawBytes> = b"image".to_vec().to_blob();
            let photo = crate::files::content_hash_hex(photo.get_handle());
            assert!(
                text.contains(&format!(
                    "New Discord messages:\n- Ada in channel {channel}: look at this\nsecond line\n  attachment photo.png: files get files:{photo} @-"
                )),
                "{text}"
            );
            assert!(!text.contains("my own reply"), "{text}");

            // Presented, it is not news again, and neither is its edit.
            let mut output = Vec::new();
            apply_news_to_writer(
                &mut pile,
                &fixture.signer,
                reader_id,
                false,
                &news,
                "",
                &mut output,
            )
            .unwrap();
            pile.commit(
                sources.discord.source,
                &fixture.signer,
                stored(message(
                    "100000000000000501",
                    "100000000000000400",
                    "look at this, edited",
                    Some("2026-09-26T08:05:00Z"),
                )),
            )
            .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let again = read(&mut pile, &observation.snapshot, |reader| {
                let query = observation.query(reader);
                prepare_news_once(&query, reader_id)
            })
            .await
            .unwrap();
            assert!(
                matches!(again, News::Quiet),
                "a presented Discord message repeated"
            );
            pile.close().unwrap();
        });
    }

    /// Two versions of one Discord message with the same newest time (as
    /// `discord read` keeps them both) are both in its one report: once
    /// presented, neither would be news again.
    #[test]
    fn discord_versions_with_the_same_time_are_all_presented() {
        crate::test_support::clear_ambient_environment();
        runtime().unwrap().block_on(async {
            use crate::discord::operations::observed_fragment;
            use serde_json::json;
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let reader_id = id(91);
            let (profile, _, _) = relations::person_fragment(
                reader_id,
                relations::ProfileInput {
                    label: "reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, profile)
                .unwrap();
            let channel = "100000000000000200";
            let version = |content: &str| {
                observed_fragment(
                    json!({
                        "id": "100000000000000501", "channel_id": channel, "type": 0,
                        "content": content,
                        "author": {"id": "100000000000000400", "username": "ada",
                                   "global_name": "Ada"},
                        "timestamp": "2026-09-26T08:00:00Z",
                        "edited_timestamp": "2026-09-26T08:05:00Z",
                        "attachments": [],
                    }),
                    |_, _| unreachable!("no attachments"),
                )
                .unwrap()
            };
            let mut facts = version("one wording");
            facts += version("another wording");
            facts += crate::discord::intake_fragment(channel, 100000000000000000).unwrap();
            pile.commit(sources.discord.source, &fixture.signer, facts)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let news = read(&mut pile, &observation.snapshot, |reader| {
                let query = observation.query(reader);
                prepare_news_once(&query, reader_id)
            })
            .await
            .unwrap();
            let News::Report { text, events } = &news else {
                panic!("a Discord message from somebody else is news");
            };
            assert_eq!(events.len(), 1, "{text}");
            assert!(
                text.contains(&format!(
                    "News: Discord message from Ada in channel {channel}: another wording \
                     (2 versions with the same time)"
                )),
                "{text}"
            );
            for wording in ["another wording", "one wording"] {
                assert!(
                    text.contains(&format!(
                        "- Ada in channel {channel} (one of 2 versions with the same time): \
                         {wording}"
                    )),
                    "{text}"
                );
            }
            pile.close().unwrap();
        });
    }

    /// The review's R4: versions with the same time are told apart as
    /// `discord read` tells them apart, by their attachment entities. Two
    /// large `report.bin` attachments with different Discord ids and sizes,
    /// neither stored, are two versions and two attachments, each printed
    /// with its size; two that differ in the Discord id alone are two
    /// versions and two attachments as well, and print alike, so once.
    #[test]
    fn discord_versions_are_told_apart_by_their_attachments() {
        crate::test_support::clear_ambient_environment();
        runtime().unwrap().block_on(async {
            use crate::discord::operations::observed_fragment;
            use serde_json::json;
            let fixture = TestPile::new();
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &fixture.signer, false)
                .await
                .unwrap();
            let reader_id = id(92);
            let (profile, _, _) = relations::person_fragment(
                reader_id,
                relations::ProfileInput {
                    label: "reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &fixture.signer, profile)
                .unwrap();
            let channel = "100000000000000200";
            let version = |message: &str, attachment: &str, size: u64| {
                observed_fragment(
                    json!({
                        "id": message, "channel_id": channel, "type": 0,
                        "content": "the report",
                        "author": {"id": "100000000000000400", "username": "ada",
                                   "global_name": "Ada"},
                        "timestamp": "2026-09-26T08:00:00Z",
                        "edited_timestamp": "2026-09-26T08:05:00Z",
                        "attachments": [{
                            "id": attachment, "url": "https://cdn.example/report.bin",
                            "filename": "report.bin",
                            "content_type": "application/octet-stream", "size": size
                        }],
                    }),
                    |_, _| unreachable!("too large to fetch"),
                )
                .unwrap()
            };
            let mut facts = version("100000000000000501", "100000000000000601", 30_000_000);
            facts += version("100000000000000501", "100000000000000602", 31_000_000);
            facts += version("100000000000000502", "100000000000000603", 30_000_000);
            facts += version("100000000000000502", "100000000000000604", 30_000_000);
            facts += crate::discord::intake_fragment(channel, 100000000000000000).unwrap();
            pile.commit(sources.discord.source, &fixture.signer, facts)
                .unwrap();
            maintain_sources(&mut pile, &fixture.signer, &sources)
                .await
                .unwrap();
            let observation = observe_current_sources(&mut pile, &sources).unwrap();
            let news = read(&mut pile, &observation.snapshot, |reader| {
                let query = observation.query(reader);
                prepare_news_once(&query, reader_id)
            })
            .await
            .unwrap();
            let News::Report { text, events } = &news else {
                panic!("Discord messages from somebody else are news");
            };
            assert_eq!(events.len(), 2, "{text}");
            let line = format!(
                "News: Discord message from Ada in channel {channel}: the report \
                 (2 attachments) (2 versions with the same time)\n"
            );
            assert_eq!(text.matches(&line).count(), 2, "{text}");
            let version = |size: u64| {
                format!(
                    "- Ada in channel {channel} (one of 2 versions with the same time): \
                     the report\n  attachment report.bin: {size} bytes, too large to store\n"
                )
            };
            assert_eq!(text.matches(&version(31_000_000)).count(), 1, "{text}");
            assert_eq!(
                text.matches(&version(30_000_000)).count(),
                2,
                "one per message: the second message's two versions print alike\n{text}"
            );
            pile.close().unwrap();
        });
    }

    /// Discord without a network: one channel's message objects, paged as
    /// Discord pages them (`after` gives the messages right after it, every
    /// page newest first) and fetched by id.
    #[derive(Clone, Default)]
    struct DiscordFake {
        messages: std::sync::Arc<std::sync::Mutex<Vec<serde_json::Value>>>,
    }

    fn discord_fake_id(message: &serde_json::Value) -> u64 {
        message["id"].as_str().unwrap().parse().unwrap()
    }

    impl crate::discord::Source for DiscordFake {
        fn page(
            &mut self,
            _channel_id: &str,
            request: crate::discord::PageRequest,
        ) -> Result<Vec<serde_json::Value>> {
            let mut page: Vec<serde_json::Value> = self
                .messages
                .lock()
                .unwrap()
                .iter()
                .filter(|message| {
                    let id = discord_fake_id(message);
                    request.after.is_none_or(|after| id > after)
                        && request.before.is_none_or(|before| id < before)
                })
                .cloned()
                .collect();
            page.sort_by_key(discord_fake_id);
            let limit = request.limit as usize;
            let mut page: Vec<serde_json::Value> = if request.after.is_some() {
                page.into_iter().take(limit).collect()
            } else {
                page.split_off(page.len().saturating_sub(limit))
            };
            page.reverse();
            Ok(page)
        }

        fn message(
            &mut self,
            _channel_id: &str,
            message_id: u64,
        ) -> Result<Option<serde_json::Value>> {
            Ok(self
                .messages
                .lock()
                .unwrap()
                .iter()
                .find(|message| discord_fake_id(message) == message_id)
                .cloned())
        }

        fn attachment(&mut self, url: &str, _limit: u64) -> Result<Vec<u8>> {
            bail!("no attachment {url}")
        }
    }

    /// Orient's news for `reader` from the pile as it stands, taken as a
    /// watcher takes it: reported, then presented.
    async fn take_news(path: &Path, signer: &SigningKey, reader: Id) -> News {
        let mut pile = open_store_as(path, signer.verifying_key()).unwrap();
        let sources = OrientSources::open(&mut pile, signer, false).await.unwrap();
        maintain_sources(&mut pile, signer, &sources).await.unwrap();
        let observation = observe_current_sources(&mut pile, &sources).unwrap();
        let news = read(&mut pile, &observation.snapshot, |view| {
            let query = observation.query(view);
            prepare_news_once(&query, reader)
        })
        .await
        .unwrap();
        apply_news_to_writer(
            &mut pile,
            signer,
            reader,
            false,
            &news,
            "",
            &mut Vec::<u8>::new(),
        )
        .unwrap();
        pile.close().unwrap();
        news
    }

    /// The live sequence on sky, 2026-09-25/26, with the channel's and the
    /// messages' own ids and times. A `discord read` pulled the channel before
    /// intake had ever run: its first pull, a bounded baseline of the bot
    /// joining, the bot's greeting and the reply to it, all printed to the
    /// window that read them. The next day `discord live` began intake of the
    /// channel (its floor the moment it started, the bot's account recorded,
    /// a backfill with nothing newer to store), and orient, reading Discord
    /// for the first time, presented the day-old reply as news. It is the
    /// channel's history, which `discord read` stored: never news. What is
    /// sent from where intake begins is news whoever stored it: one sent
    /// while intake was down that `discord read` brings in, although intake
    /// has stored nothing in the channel yet, and the gateway's next message.
    #[test]
    fn discord_history_read_before_intake_began_is_never_news() {
        use crate::discord::intake::{floor_at, Done, Intake, Work};
        use serde_json::json;
        crate::test_support::clear_ambient_environment();
        // The Discord faculty runs its own runtime, which must not end inside
        // this one's: only orient's reads run on it.
        let runtime = runtime().unwrap();
        let fixture = TestPile::new();
        let key = fixture.dir.join("discord.key");
        let signer = crate::storage::initialize_signer(&fixture.path, Some(&key)).unwrap();
        let discord = crate::discord::Discord::new(fixture.path.clone(), Some(key));
        let reader_id = id(93);
        runtime.block_on(async {
            let mut pile = open_store_as(&fixture.path, fixture.signer.verifying_key()).unwrap();
            let sources = OrientSources::open(&mut pile, &signer, false)
                .await
                .unwrap();
            let (profile, _, _) = relations::person_fragment(
                reader_id,
                relations::ProfileInput {
                    label: "reader".to_owned(),
                    ..Default::default()
                },
            )
            .unwrap();
            pile.commit(sources.relations.source, &signer, profile)
                .unwrap();
            pile.close().unwrap();
        });

        let channel = "1553078566332403894";
        let bot = "100000000000000900";
        let message = |id: &str, at: &str, author: &str, kind: u64, content: &str| {
            let name = if author == bot { "Bot" } else { "Ada" };
            json!({
                "id": id, "channel_id": channel, "type": kind, "content": content,
                "author": {"id": author, "username": name, "global_name": name},
                "timestamp": at, "edited_timestamp": null,
                "attachments": [], "referenced_message": null,
            })
        };
        let ada = "100000000000000400";
        let fake = DiscordFake::default();
        fake.messages.lock().unwrap().extend([
            // The bot joining the server.
            message(
                "1553078807873724499",
                "2026-09-25T16:20:42.606000+00:00",
                bot,
                7,
                "",
            ),
            message(
                "1553079260489711807",
                "2026-09-25T16:22:30.518000+00:00",
                bot,
                0,
                "hello, it is the bot",
            ),
            message(
                "1553079976620851252",
                "2026-09-25T16:25:21.257000+00:00",
                ada,
                0,
                "thank you, the disk is connected",
            ),
        ]);

        // 2026-09-25T16:25:38Z: `discord read`, the channel's first pull.
        let first = discord
            .pull_channel(channel, None, &mut fake.clone())
            .unwrap();
        assert_eq!(first.observations, 3);
        assert_eq!(
            first.coverage,
            Some(
                crate::discord::CoverageInterval::new(
                    1553078807873724498,
                    1553079976620851252,
                    true
                )
                .unwrap()
            ),
            "the bounded baseline the live commit covered"
        );

        // 2026-09-26T20:41:13.191Z: `discord live` begins intake.
        let floor = floor_at(UNIX_EPOCH + Duration::from_millis(1_790_455_273_191));
        assert_eq!(floor, 1553506755164504063, "the floor intake kept on sky");
        let mut intake = Intake::new(
            discord.clone(),
            Box::new(fake.clone()),
            vec![std::num::NonZeroU64::new(channel.parse().unwrap()).unwrap()],
            false,
            fixture.dir.join("intake"),
            floor,
        );
        assert_eq!(
            intake.handle(Work::Account(bot.to_owned())).unwrap(),
            Done::Recorded
        );
        assert_eq!(
            intake.handle(Work::Backfill).unwrap(),
            Done::Backfilled {
                channels: 1,
                failed: 0,
                more: false,
                unstored: 0
            }
        );
        let history = runtime.block_on(take_news(&fixture.path, &signer, reader_id));
        assert!(
            matches!(history, News::Quiet),
            "the channel's history is not news: {:?}",
            match &history {
                News::Report { text, .. } => text.as_str(),
                News::Quiet => "",
            }
        );

        // The process exits, as it did five times over; meanwhile a message is
        // sent, and `discord read` brings it in with the history it
        // reconciles. Intake stored none of it, and it is news.
        drop(intake);
        fake.messages.lock().unwrap().push(message(
            "1553511481344004096",
            "2026-09-26T21:00:00.000000+00:00",
            ada,
            0,
            "are you there?",
        ));
        let read = discord
            .pull_channel(channel, None, &mut fake.clone())
            .unwrap();
        assert_eq!(
            read.observations, 5,
            "the one read forward, and the recent page reconciled"
        );
        let News::Report { text, events } =
            runtime.block_on(take_news(&fixture.path, &signer, reader_id))
        else {
            panic!("a message sent while intake was down is news");
        };
        assert_eq!(events.len(), 1, "{text}");
        assert!(
            text.contains(&format!(
                "News: Discord message from Ada in channel {channel}: are you there?"
            )),
            "{text}"
        );
        assert!(!text.contains("the disk is connected"), "{text}");

        // Restarted, intake keeps the floor it began with, and the gateway's
        // next message is news.
        let mut intake = Intake::new(
            discord.clone(),
            Box::new(fake.clone()),
            vec![std::num::NonZeroU64::new(channel.parse().unwrap()).unwrap()],
            false,
            fixture.dir.join("intake"),
            floor_at(UNIX_EPOCH + Duration::from_millis(1_790_456_700_000)),
        );
        assert_eq!(
            intake.handle(Work::Account(bot.to_owned())).unwrap(),
            Done::Recorded
        );
        assert_eq!(
            intake.handle(Work::Backfill).unwrap(),
            Done::Backfilled {
                channels: 1,
                failed: 0,
                more: false,
                unstored: 0
            }
        );
        let heard = message(
            "1553512739635204096",
            "2026-09-26T21:05:00.000000+00:00",
            ada,
            0,
            "never mind",
        );
        fake.messages.lock().unwrap().push(heard.clone());
        let mut gateway = heard;
        gateway["guild_id"] = json!("1553078565199683675");
        assert_eq!(intake.handle(Work::Message(gateway)).unwrap(), Done::Stored);
        let News::Report { text, events } =
            runtime.block_on(take_news(&fixture.path, &signer, reader_id))
        else {
            panic!("a message sent after intake began is news");
        };
        assert_eq!(events.len(), 1, "{text}");
        assert!(
            text.contains(&format!(
                "News: Discord message from Ada in channel {channel}: never mind"
            )),
            "{text}"
        );
        assert!(!text.contains("the disk is connected"), "{text}");
    }
}