polyc-eventlog-host 2026.8.3

//! Post-append and post-mutation observers on [`EventLogHost`](crate::EventLogHost).
//!
//! Two later features share one primitive: a per-partition decode cache needs
//! to know when a partition's on-disk content changed underneath it
//! (`destroy`/`rewrite`/`migrate`), and a maintained dashboard projection needs
//! to see every append that lands on a conversation partition, regardless of
//! which control-plane code path produced it. Rather than growing two
//! independent hooks, this host exposes one: register any number of
//! [`EventLogObserver`]s and each is called after every durable write this host
//! performs.
//!
//! # Ordering
//!
//! Every partition is pinned, for the process's life, to its own lazily
//! spawned worker task (`#1711`) — one shard's dedicated Commonware-runtime
//! thread multiplexes many such workers as independent, genuinely concurrent
//! tasks, rather than running every partition's commands through one shared
//! sequential loop. Each worker's OWN loop, though, is single-threaded: it
//! processes one command for its partition at a time, start to finish,
//! before touching the next. A notification fires synchronously, inline,
//! from inside that same processing step, strictly after the write it
//! reports already durably committed. This gives per-partition commit
//! ordering: two writes to the SAME partition are notified in the order they
//! committed. There is **no ordering guarantee across two DIFFERENT
//! partitions** — whether they land on the same shard or different ones,
//! their notifications may interleave in either order relative to each
//! other, even if a caller happened to issue the underlying calls in a
//! particular order.
//!
//! # Failure containment
//!
//! An observer must never fail, block, or panic the write path. This host
//! enforces that by policy, not by convention: every call into
//! [`EventLogObserver::on_append`] or [`EventLogObserver::on_mutation`] is
//! wrapped in [`std::panic::catch_unwind`]. A panicking observer is logged
//! (`tracing::error!`) and skipped — the write that triggered it already
//! durably succeeded before the observer ran, and every later write to the
//! same or a different partition proceeds normally; a broken observer can
//! never poison the host. The other half of the contract is the caller's:
//! both methods run synchronously, inline, on the shard's dedicated thread, so
//! an observer **must be cheap** — no I/O, no blocking, no lock the write path
//! could itself need. An observer that needs to do real work should hand it
//! off to its own background task or bounded queue rather than doing it here.
//!
//! `repair_partition`'s quarantine rewrite fires [`MutationKind::Repaired`]
//! exactly like the other three mutations — see that variant's docs for why
//! a repair, which drops quarantined (corrupted) events, changes durable
//! partition content just as a rewrite does.
//!
//! Callbacks run OUTSIDE the observer-registration lock: both
//! [`WriteObservers::notify_append`] and
//! [`WriteObservers::notify_mutation_bumping_epoch`] clone the registered
//! observer list out from under its read lock before invoking anything, so
//! an observer that calls
//! [`EventLogHost::register_observer`](crate::EventLogHost::register_observer)
//! back on the host from inside its own callback never reenters that lock.

use std::collections::HashMap;
use std::panic::AssertUnwindSafe;
use std::sync::{Arc, Mutex, RwLock};

use polyc_eventlog::Event;

/// Every observer registered on an [`EventLogHost`](crate::EventLogHost),
/// shared (the same [`Arc`]) with every shard's dedicated thread, so a
/// registration is visible to whichever shard next durably commits —
/// including shards that were already running when it was registered.
type ObserverList = Arc<RwLock<Vec<Arc<dyn EventLogObserver>>>>;

/// Per-partition mutation epoch, shared the same way as [`ObserverList`]. See
/// [`EventLogHost::mutation_epoch`](crate::EventLogHost::mutation_epoch).
type EpochMap = Arc<Mutex<HashMap<String, u64>>>;

/// The shared, cross-shard write-observation state — registered observers
/// plus per-partition mutation epochs — bundled into one handle so every
/// write path threads a single parameter instead of two. Cheap to clone: both
/// fields are `Arc`s, so every shard's dedicated thread (and every helper
/// function it calls into) shares the SAME underlying registry and epoch map
/// through its own clone of this handle.
#[derive(Clone)]
pub(crate) struct WriteObservers {
    /// Registered observers, shared across every shard.
    observers: ObserverList,
    /// Per-partition mutation epochs, shared across every shard.
    epochs: EpochMap,
}

impl WriteObservers {
    /// A fresh handle with no observers registered and every partition's
    /// epoch implicitly `0`.
    pub(crate) fn new() -> Self {
        Self {
            observers: Arc::new(RwLock::new(Vec::new())),
            epochs: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    /// Register `observer`. See
    /// [`EventLogHost::register_observer`](crate::EventLogHost::register_observer).
    pub(crate) fn register(&self, observer: Arc<dyn EventLogObserver>) {
        self.observers.write().expect("poison").push(observer);
    }

    /// `partition`'s current mutation epoch (`0` if never mutated). See
    /// [`EventLogHost::mutation_epoch`](crate::EventLogHost::mutation_epoch).
    pub(crate) fn epoch(&self, partition: &str) -> u64 {
        self.epochs
            .lock()
            .expect("poison")
            .get(partition)
            .copied()
            .unwrap_or(0)
    }

    /// Invoke every registered observer's [`EventLogObserver::on_append`],
    /// containing any panic so the write that already durably succeeded is
    /// never affected by a broken observer. A no-op when `events` is empty
    /// (nothing was actually appended) or no observer is registered.
    ///
    /// The registration lock is held only long enough to clone the observer
    /// list out; every callback below runs with it already released, so an
    /// observer that registers another observer from inside its own callback
    /// cannot reenter it.
    pub(crate) fn notify_append(&self, partition: &str, events: &[Event], positions: &[u64]) {
        if events.is_empty() {
            return;
        }
        let observers = {
            let guard = self.observers.read().expect("poison");
            if guard.is_empty() {
                return;
            }
            guard.clone()
        };
        let notification = AppendNotification {
            partition,
            events,
            positions,
        };
        for observer in &observers {
            if let Err(payload) =
                std::panic::catch_unwind(AssertUnwindSafe(|| observer.on_append(&notification)))
            {
                log_observer_panic("on_append", partition, &*payload);
            }
        }
    }

    /// Bump `partition`'s mutation epoch and invoke every registered
    /// observer's [`EventLogObserver::on_mutation`] with the new value,
    /// containing any panic the same way [`Self::notify_append`] does.
    /// Returns the new epoch. The single place every mutation path
    /// (destroy/rewrite/migrate/repair) advances the epoch AND notifies, so a
    /// call site can never bump without notifying or vice versa. Never
    /// called from the append path: appends do not change which decoded
    /// state a cache keyed by this epoch should serve.
    pub(crate) fn notify_mutation_bumping_epoch(
        &self,
        partition: &str,
        kind: &MutationKind,
    ) -> u64 {
        let mut guard = self.epochs.lock().expect("poison");
        let epoch = guard.entry(partition.to_owned()).or_insert(0);
        *epoch += 1;
        let epoch = *epoch;
        drop(guard);
        let observers = {
            let guard = self.observers.read().expect("poison");
            if guard.is_empty() {
                return epoch;
            }
            guard.clone()
        };
        let notification = MutationNotification {
            partition,
            kind,
            epoch,
        };
        for observer in &observers {
            if let Err(payload) = std::panic::catch_unwind(AssertUnwindSafe(|| {
                observer.on_mutation(&notification);
            })) {
                log_observer_panic("on_mutation", partition, &*payload);
            }
        }
        epoch
    }
}

/// Observer over an [`EventLogHost`](crate::EventLogHost)'s durable writes.
///
/// Notified strictly after an append or a partition mutation lands on disk —
/// never before, and never for a write that failed. See the module docs for
/// the ordering guarantee this host provides and the failure-containment
/// policy it enforces around every call.
///
/// Both methods default to doing nothing, so an implementor overrides only
/// the notification it cares about.
pub trait EventLogObserver: Send + Sync {
    /// Called after `notification`'s events are durably appended (via
    /// [`EventLogHost::append_batch`](crate::EventLogHost::append_batch)) to
    /// their partition.
    fn on_append(&self, _notification: &AppendNotification<'_>) {}

    /// Called after a partition is destroyed, rewritten, or migrated — see
    /// [`MutationKind`] for which host calls this covers.
    fn on_mutation(&self, _notification: &MutationNotification<'_>) {}
}

/// One durable [`EventLogHost::append_batch`](crate::EventLogHost::append_batch)
/// call's events, with the journal positions they were assigned.
#[derive(Debug, Clone, Copy)]
pub struct AppendNotification<'a> {
    /// LOGICAL name of the partition the events were appended to.
    ///
    /// The name the caller addressed, never the spelling storage holds it
    /// under. A projector keys its own durable state on this, so it must be
    /// the id it was given — see [`PartitionRef`](crate::PartitionRef).
    pub partition: &'a str,
    /// Appended events, in append order.
    pub events: &'a [Event],
    /// Durable journal positions assigned to `events`: same length and order,
    /// so `positions[i]` is `events[i]`'s position.
    pub positions: &'a [u64],
}

/// What kind of durable partition mutation a [`MutationNotification`] reports.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MutationKind {
    /// The partition was destroyed wholesale
    /// ([`EventLogHost::destroy_partition`](crate::EventLogHost::destroy_partition),
    /// the #216 erasure primitive): a subsequent open of the same name starts
    /// empty.
    Destroyed,
    /// The partition was rewritten in place
    /// ([`EventLogHost::rewrite_partition`](crate::EventLogHost::rewrite_partition),
    /// #216/#860): `dropped` events were removed. A replaced event is not
    /// counted here — it survives under its existing kind with a new payload,
    /// so it is neither a drop nor an untouched keep, but the partition's
    /// content still changed and this still fires.
    Rewritten {
        /// Number of events the rewrite dropped.
        dropped: usize,
    },
    /// This partition was a
    /// [`EventLogHost::migrate_partition`](crate::EventLogHost::migrate_partition)
    /// SOURCE: its events (if any) were copied to `to` and this partition was
    /// then destroyed.
    MigrationSource {
        /// Destination partition the events were copied to.
        to: String,
    },
    /// This partition was a
    /// [`EventLogHost::migrate_partition`](crate::EventLogHost::migrate_partition)
    /// DESTINATION: `appended` new events (already deduped against what this
    /// partition already held) were copied in from `from`.
    MigrationDestination {
        /// Source partition the events were copied from.
        from: String,
        /// Number of new events appended, after dedup.
        appended: usize,
    },
    /// The partition was recovered from a corrupted item
    /// ([`EventLogHost::repair_partition`](crate::EventLogHost::repair_partition),
    /// `#799`'s quarantine primitive): `quarantined` events were unreadable
    /// and dropped, so the partition's durable content shrank exactly like a
    /// [`Self::Rewritten`] drop does — a cache keyed on the mutation epoch
    /// must invalidate on this exactly as it would on a rewrite. A repair
    /// that finds nothing corrupted is a no-op and does not fire this (or
    /// bump the epoch) at all.
    Repaired {
        /// Number of events the repair quarantined (dropped as unreadable).
        quarantined: usize,
    },
}

/// A durable partition mutation: a destroy, a rewrite, or one side of a
/// migrate (see [`MutationKind`]).
#[derive(Debug, Clone, Copy)]
pub struct MutationNotification<'a> {
    /// LOGICAL name of the partition that was mutated, as
    /// [`AppendNotification::partition`] reports it.
    pub partition: &'a str,
    /// What happened.
    pub kind: &'a MutationKind,
    /// This partition's mutation epoch AFTER this mutation — see
    /// [`EventLogHost::mutation_epoch`](crate::EventLogHost::mutation_epoch).
    pub epoch: u64,
}

/// Log a contained observer panic. `payload` is [`std::panic::catch_unwind`]'s
/// `Err` payload — a `Box<dyn Any + Send>` that is a `&str` or `String` for
/// every panic produced by `panic!`/`assert!`/`.expect()`, and unnamed
/// otherwise.
fn log_observer_panic(hook: &str, partition: &str, payload: &(dyn std::any::Any + Send)) {
    let message = payload
        .downcast_ref::<&str>()
        .map(|s| (*s).to_owned())
        .or_else(|| payload.downcast_ref::<String>().cloned())
        .unwrap_or_else(|| "non-string panic payload".to_owned());
    tracing::error!(
        hook,
        partition = %partition,
        message,
        "event-log observer panicked; the write it observed already durably succeeded and is unaffected"
    );
}

#[cfg(test)]
pub(crate) mod tests {
    use super::{AppendNotification, EventLogObserver, MutationKind, MutationNotification};
    use crate::{EventLogHost, PartitionMigration, RewriteDecision};
    use polyc_eventlog::Event;
    use std::sync::{Arc, Mutex};
    use tokio_util::sync::CancellationToken;

    /// One recorded [`AppendNotification`]: partition, events, positions.
    type RecordedAppend = (String, Vec<Event>, Vec<u64>);
    /// One recorded [`MutationNotification`]: partition, kind, epoch.
    type RecordedMutation = (String, MutationKind, u64);

    /// Observer double: records every append/mutation notification it sees, so
    /// a test can assert exactly what fired without reaching into host-private
    /// state.
    #[derive(Default)]
    pub(crate) struct RecordingObserver {
        pub(crate) appends: Mutex<Vec<RecordedAppend>>,
        pub(crate) mutations: Mutex<Vec<RecordedMutation>>,
    }

    impl EventLogObserver for RecordingObserver {
        fn on_append(&self, notification: &AppendNotification<'_>) {
            self.appends.lock().expect("poison").push((
                notification.partition.to_owned(),
                notification.events.to_vec(),
                notification.positions.to_vec(),
            ));
        }

        fn on_mutation(&self, notification: &MutationNotification<'_>) {
            self.mutations.lock().expect("poison").push((
                notification.partition.to_owned(),
                notification.kind.clone(),
                notification.epoch,
            ));
        }
    }

    /// Observer double that panics on every append — proves the host's
    /// failure-containment policy: a broken observer never blocks or takes
    /// down the write path, and never stops any OTHER observer from running.
    struct PanickingObserver;

    impl EventLogObserver for PanickingObserver {
        fn on_append(&self, _notification: &AppendNotification<'_>) {
            panic!("deliberate test panic: proving the host contains an observer panic");
        }
    }

    /// A committed batch counts its signed-root outcome (`#1386` risk 9).
    ///
    /// The metric's own unit tests prove it counts when called. They do not
    /// prove `append_batch_one` calls it, and a counter nothing increments
    /// reads exactly like a healthy zero. This drives the real host and
    /// asserts the count moved.
    #[tokio::test]
    async fn a_committed_batch_counts_its_attestation() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-attestation-count-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");

        let before = crate::metrics::attestation_count(crate::metrics::AttestationOutcome::Signed);

        host.append_batch(
            "conv-attestation-count".to_owned(),
            vec![Event::new("output_msg", b"one".to_vec())],
        )
        .await
        .expect("append");

        assert_eq!(
            crate::metrics::attestation_count(crate::metrics::AttestationOutcome::Signed),
            before + 1,
            "a committed batch must count its signed-root outcome"
        );

        shutdown.cancel();
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// An observer sees every `append_batch` call — single-event and
    /// multi-event alike — with the exact same positions the caller itself
    /// got back, in the same order as the events.
    #[tokio::test]
    async fn append_observer_sees_correct_positions_for_single_and_batch_appends() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-append-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");

        let recorder = Arc::new(RecordingObserver::default());
        host.register_observer(recorder.clone());

        let first_positions = host
            .append_batch(
                "conv-observer-append".to_owned(),
                vec![Event::new("output_msg", b"one".to_vec())],
            )
            .await
            .expect("single append");
        let batch_positions = host
            .append_batch(
                "conv-observer-append".to_owned(),
                vec![
                    Event::new("output_msg", b"two".to_vec()),
                    Event::new("output_msg", b"three".to_vec()),
                ],
            )
            .await
            .expect("batch append");

        let seen = recorder.appends.lock().expect("poison");
        assert_eq!(seen.len(), 2, "one notification per append_batch call");

        let (partition, events, positions) = &seen[0];
        assert_eq!(partition, "conv-observer-append");
        assert_eq!(
            events.len(),
            1,
            "the single-event call carries exactly one event"
        );
        assert_eq!(positions, &first_positions);

        let (partition, events, positions) = &seen[1];
        assert_eq!(partition, "conv-observer-append");
        assert_eq!(positions, &batch_positions);
        assert_eq!(
            events.iter().map(|e| e.payload.clone()).collect::<Vec<_>>(),
            vec![b"two".to_vec(), b"three".to_vec()],
            "batch events arrive in append order"
        );
        drop(seen);

        drop(host);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// `destroy_partition` and `rewrite_partition` each bump the partition's
    /// mutation epoch by one and fire exactly one `on_mutation` notification;
    /// `append_batch` bumps neither the epoch nor fires a mutation
    /// notification, no matter how many events land.
    #[tokio::test]
    async fn destroy_and_rewrite_bump_the_epoch_and_notify_append_does_not() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-mutation-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");

        let recorder = Arc::new(RecordingObserver::default());
        host.register_observer(recorder.clone());

        assert_eq!(
            host.mutation_epoch("conv-observer-mutation"),
            0,
            "a partition this process never mutated starts at epoch 0"
        );

        host.append_batch(
            "conv-observer-mutation".to_owned(),
            vec![Event::new("memory_added", b"fact".to_vec())],
        )
        .await
        .expect("seed");
        assert_eq!(
            host.mutation_epoch("conv-observer-mutation"),
            0,
            "append_batch must never bump the mutation epoch"
        );
        assert!(
            recorder.mutations.lock().expect("poison").is_empty(),
            "append_batch must never fire a mutation notification"
        );

        // Only the content event is offered to `decide`: the host drops
        // every signed-root marker itself (#799), because each one covers
        // content the rewrite is changing.
        host.rewrite_partition_for_test(
            "conv-observer-mutation".to_owned(),
            Box::new(|event| {
                if event.kind == "memory_added" {
                    RewriteDecision::Drop
                } else {
                    RewriteDecision::Keep
                }
            }),
        )
        .await
        .expect("rewrite");
        assert_eq!(
            host.mutation_epoch("conv-observer-mutation"),
            1,
            "rewrite is the first mutation"
        );

        host.destroy_partition("conv-observer-mutation".to_owned())
            .await
            .expect("destroy");
        assert_eq!(
            host.mutation_epoch("conv-observer-mutation"),
            2,
            "destroy is the second mutation"
        );

        let mutations = recorder.mutations.lock().expect("poison");
        assert_eq!(
            mutations.len(),
            2,
            "one notification for the rewrite, one for the destroy"
        );
        assert_eq!(mutations[0].0, "conv-observer-mutation");
        assert_eq!(mutations[0].1, MutationKind::Rewritten { dropped: 1 });
        assert_eq!(mutations[0].2, 1);
        assert_eq!(mutations[1].0, "conv-observer-mutation");
        assert_eq!(mutations[1].1, MutationKind::Destroyed);
        assert_eq!(mutations[1].2, 2);
        drop(mutations);

        drop(host);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Every observer registered on the host fires for the same durable
    /// append — registering a second observer never crowds out the first.
    #[tokio::test]
    async fn multiple_observers_all_fire_for_the_same_append() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-multi-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");

        let first = Arc::new(RecordingObserver::default());
        let second = Arc::new(RecordingObserver::default());
        host.register_observer(first.clone());
        host.register_observer(second.clone());

        host.append_batch(
            "conv-observer-multi".to_owned(),
            vec![Event::new("output_msg", b"hi".to_vec())],
        )
        .await
        .expect("append");

        assert_eq!(
            first.appends.lock().expect("poison").len(),
            1,
            "the first registered observer sees the append"
        );
        assert_eq!(
            second.appends.lock().expect("poison").len(),
            1,
            "the second registered observer sees the SAME append"
        );

        drop(host);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A panicking observer never blocks or fails the write path, and never
    /// stops a well-behaved observer registered alongside it from seeing
    /// every subsequent append — the host's failure-containment policy in
    /// force across repeated writes, not just a single one.
    #[tokio::test]
    async fn a_panicking_observer_never_blocks_or_poisons_the_write_path() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-panic-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");

        let recorder = Arc::new(RecordingObserver::default());
        host.register_observer(Arc::new(PanickingObserver));
        host.register_observer(recorder.clone());

        for i in 0..3u32 {
            host.append_batch(
                "conv-observer-panic".to_owned(),
                vec![Event::new(format!("k{i}"), b"x".to_vec())],
            )
            .await
            .expect("append must succeed even though an observer panics on every call");
        }

        assert_eq!(
            recorder.appends.lock().expect("poison").len(),
            3,
            "the well-behaved observer keeps seeing every append despite the other one \
             panicking every time"
        );

        drop(host);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// `migrate_partition` fires one notification per side: the source is
    /// notified as a `MigrationSource` (and its own epoch bumps), and the
    /// destination is notified as a `MigrationDestination` (and its own,
    /// independent epoch bumps) — not just one notification for the call as a
    /// whole.
    #[tokio::test]
    async fn migrate_partition_notifies_both_the_source_and_the_destination() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-migrate-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");

        let recorder = Arc::new(RecordingObserver::default());
        host.register_observer(recorder.clone());

        host.append_batch(
            "persona-observer-migrate-source".to_owned(),
            vec![Event::new("memory_added", b"fact-a".to_vec())],
        )
        .await
        .expect("seed source");

        let migration = host
            .migrate_partition(
                "persona-observer-migrate-source".to_owned(),
                "persona-observer-migrate-dest".to_owned(),
            )
            .await
            .expect("migrate");
        // 1, not 2 (`#2653`). `append_batch` also lands its own MMR
        // signed-root marker (#799), and that marker does NOT move: a root
        // commits to a leaf sequence at its own positions, so it cannot
        // describe the same event at a different position in the destination.
        // See `a_migrated_partition_verifies_at_its_destination`.
        assert_eq!(migration, PartitionMigration::Migrated(1));

        assert_eq!(
            host.mutation_epoch("persona-observer-migrate-source"),
            1,
            "the source's destroy bumps its own epoch"
        );
        assert_eq!(
            host.mutation_epoch("persona-observer-migrate-dest"),
            1,
            "the destination's append-in bumps its own, independent epoch"
        );

        let mutations = recorder.mutations.lock().expect("poison");
        assert_eq!(
            mutations.len(),
            2,
            "one notification per side of the migration"
        );

        let source_notification = mutations
            .iter()
            .find(|(partition, ..)| partition == "persona-observer-migrate-source")
            .expect("the source is notified");
        assert_eq!(
            source_notification.1,
            MutationKind::MigrationSource {
                to: "persona-observer-migrate-dest".to_owned(),
            }
        );

        let destination_notification = mutations
            .iter()
            .find(|(partition, ..)| partition == "persona-observer-migrate-dest")
            .expect("the destination is notified");
        // 1, not 2 (`#2653`). This field counts what came FROM the source. The
        // destination also writes a freshly minted signed root, but that was
        // not copied in from anywhere, and counting it here contradicted this
        // variant's own doc and disagreed with the `Migrated(n)` the same
        // migration returns.
        assert_eq!(
            destination_notification.1,
            MutationKind::MigrationDestination {
                from: "persona-observer-migrate-source".to_owned(),
                appended: 1,
            }
        );
        drop(mutations);

        drop(host);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// `repair_partition` (`#799`'s quarantine primitive) is a first-class
    /// mutation exactly like destroy/rewrite/migrate: when it actually
    /// quarantines something, it bumps the partition's mutation epoch and
    /// fires a `MutationKind::Repaired` notification — a cache keyed on the
    /// epoch must invalidate on a repair exactly as it would on a rewrite,
    /// since a repair also durably drops content (the corrupted events).
    #[tokio::test]
    async fn repair_bumps_the_epoch_and_notifies_when_it_quarantines_something() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-repair-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let partition = "conv-observer-repair".to_owned();

        {
            let shutdown = CancellationToken::new();
            let host = EventLogHost::spawn(
                dir.clone(),
                shutdown.clone(),
                polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
            )
            .expect("spawn");
            host.append_batch(
                partition.clone(),
                vec![
                    Event::new("user_msg", b"what is 2+2?".to_vec()),
                    Event::new("output_msg", b"the reply is four".to_vec()),
                ],
            )
            .await
            .expect("append 2-event turn");
            drop(host);
        }

        // Flip a byte inside the second event's payload — a content tamper
        // `repair_partition` must quarantine (mirrors
        // `active_section_payload_content_tamper_is_a_hard_verify_failure_not_a_false_pass`,
        // which explains why the WHOLE active section is invalidated, not
        // just the touched event).
        let data_file = dir
            .join(format!("{partition}_data"))
            .join("0000000000000000");
        let mut bytes = std::fs::read(&data_file).expect("read section 0");
        let payload_at = bytes
            .windows(b"the reply is four".len())
            .position(|w| w == b"the reply is four")
            .expect("the second event's payload bytes are present on disk");
        bytes[payload_at] ^= 0xFF;
        std::fs::write(&data_file, &bytes).expect("write corrupted section 0");

        let shutdown2 = CancellationToken::new();
        let host2 = EventLogHost::spawn(
            dir.clone(),
            shutdown2,
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("reopen");
        let recorder = Arc::new(RecordingObserver::default());
        host2.register_observer(recorder.clone());

        assert_eq!(host2.mutation_epoch(&partition), 0);

        let quarantined = host2
            .repair_partition(partition.clone())
            .await
            .expect("repair completes");
        assert!(
            !quarantined.is_empty(),
            "the corrupted section must quarantine at least one item for this test to be \
             meaningful"
        );

        assert_eq!(
            host2.mutation_epoch(&partition),
            1,
            "a repair that quarantines something bumps the epoch exactly like a rewrite"
        );

        let mutations = recorder.mutations.lock().expect("poison");
        assert_eq!(mutations.len(), 1, "one notification for the repair");
        assert_eq!(mutations[0].0, partition);
        assert_eq!(
            mutations[0].1,
            MutationKind::Repaired {
                quarantined: quarantined.len(),
            }
        );
        assert_eq!(mutations[0].2, 1);
        drop(mutations);

        drop(host2);
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A notification names the partition its caller asked about (#1748).
    ///
    /// A projector keys its own durable state on this string. Storage escapes a
    /// namespaced id to hold it as a directory, and a notification that carried
    /// that spelling would make every projector key on a name no caller ever
    /// used — and, before the codec, on a name shared with another
    /// conversation.
    #[tokio::test]
    async fn a_notification_names_the_logical_partition_not_its_storage_spelling() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-eventlog-observer-logical-{}-{}",
            std::process::id(),
            uuid::Uuid::now_v7().as_simple()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let host = EventLogHost::spawn(
            dir.clone(),
            shutdown.clone(),
            polyc_crypto::signing_role::JournalAttestationSigner::from_seed(1),
        )
        .expect("spawn host");
        let recorder = Arc::new(RecordingObserver::default());
        host.register_observer(recorder.clone());

        let partition = "conv-app:persona-1:standup";
        host.append_batch(partition.to_owned(), vec![Event::new("k", b"v".to_vec())])
            .await
            .expect("append");
        // Every mutation that notifies, not just the cheap one. Each of these
        // is a separate call site, and a site that reached for the storage
        // spelling would be invisible to a test that only destroys.
        host.rewrite_partition_for_test(partition.to_owned(), Box::new(|_| RewriteDecision::Keep))
            .await
            .expect("rewrite");
        host.migrate_partition(partition.to_owned(), "conv-app:persona-1:moved".to_owned())
            .await
            .expect("migrate");

        let appends = recorder.appends.lock().expect("poison");
        assert_eq!(appends.len(), 1, "one append notification");
        assert_eq!(
            appends[0].0, partition,
            "the append notification names the logical partition"
        );
        drop(appends);

        let mutations = recorder.mutations.lock().expect("poison");
        assert!(
            mutations.len() >= 2,
            "the rewrite and the migration both notify: {mutations:?}"
        );
        for (name, kind, _) in mutations.iter() {
            assert!(
                name.contains(':'),
                "every notification names the logical partition, and this one does not:                  {name} for {kind:?}"
            );
        }
        assert!(
            mutations.iter().any(|(name, kind, _)| name == partition
                && matches!(kind, MutationKind::Rewritten { .. })),
            "the rewrite names its own partition logically: {mutations:?}"
        );
        assert!(
            mutations.iter().any(|(name, kind, _)| name == partition
                && matches!(kind, MutationKind::MigrationSource { .. })),
            "and so does the migration source: {mutations:?}"
        );
        drop(mutations);

        // The epoch is reachable under the same name a notification reported,
        // which is what a projector holding that name needs.
        assert!(
            host.mutation_epoch(partition) >= 2,
            "the epoch answers under the LOGICAL name, and every mutation bumped it: {}",
            host.mutation_epoch(partition)
        );

        drop(host);
        let _ = std::fs::remove_dir_all(&dir);
    }
}