polyc-query 2026.9.0

#![allow(clippy::pedantic, clippy::nursery, missing_docs)]

use std::{
    path::PathBuf,
    sync::atomic::{AtomicUsize, Ordering},
};

use polyc_crypto::approval::ApprovalSigner;
use polyc_eventlog::Event;
use polyc_eventlog_host::EventLogHost;
use polyc_proto::kinds;
use tokio_util::sync::CancellationToken;

use super::super::marks::CommitMarks;
use super::super::store::Coverage;
use super::*;

const CONVERSATION: &str = "conv-web:01950000-0000-7000-8000-0000000000aa";
const OTHER_CONVERSATION: &str = "conv-web:01950000-0000-7000-8000-0000000000bb";

fn key() -> TermKey {
    TermKey::new([11u8; 32])
}

fn turn(nth: u8) -> String {
    format!("01950000-0000-7000-8000-0000000000{nth:02x}")
}

fn marker(base: &str, turn: &str) -> Event {
    Event::new(format!("{base}:{turn}"), Vec::new())
}

/// A `user_msg` carrying one text content block — the shape
/// `committed_message_facts` projects.
fn text_msg(turn: &str, text: &str) -> Event {
    use buffa::Message as _;
    use polyc_proto::proto::polychrome::agent::v1::{Content, Message, TextContent, content};

    let message = Message {
        role: "user".to_owned(),
        content: buffa::MessageField::some(Content {
            r#type: Some(content::Type::Text(Box::new(TextContent {
                text: text.to_owned(),
                ..Default::default()
            }))),
            ..Default::default()
        }),
        ..Default::default()
    };
    Event::new(
        format!("{}:{turn}", kinds::USER_MSG),
        message.encode_to_vec(),
    )
}

/// One fully committed turn: start, one message, complete.
fn committed_turn(turn: &str, text: &str) -> Vec<Event> {
    vec![
        marker(kinds::TURN_START, turn),
        text_msg(turn, text),
        marker(kinds::TURN_COMPLETE, turn),
    ]
}

/// A real event-log host plus a projection rooted beside it, both removed on
/// drop — the same spawn/cleanup shape `dashboard`'s own fixture uses.
struct Fixture {
    eventlog: Arc<EventLogHost>,
    projection: SearchProjection,
    dirty: Arc<DirtySet>,
    shutdown: CancellationToken,
    dir: PathBuf,
}

struct ControlledSourceJournal {
    inner: Arc<EventLogHost>,
    before: polyc_state::revision::PartitionIncarnation,
    after: polyc_state::revision::PartitionIncarnation,
    source_reads: AtomicUsize,
    source_error: Option<&'static str>,
    stopping: bool,
}

#[async_trait::async_trait]
impl PartitionJournal for ControlledSourceJournal {
    async fn partition_incarnation(
        &self,
        _partition: String,
    ) -> Result<Option<polyc_state::revision::PartitionIncarnation>, JournalError> {
        if let Some(reason) = self.source_error {
            return Err(JournalError::Unreachable(reason.to_owned()));
        }
        Ok(Some(
            if self.source_reads.fetch_add(1, Ordering::SeqCst) == 0 {
                self.before
            } else {
                self.after
            },
        ))
    }

    async fn list_partitions(&self) -> Result<Vec<String>, JournalError> {
        PartitionJournal::list_partitions(self.inner.as_ref()).await
    }

    async fn partition_event_count(&self, partition: String) -> Result<u64, JournalError> {
        PartitionJournal::partition_event_count(self.inner.as_ref(), partition).await
    }

    async fn replay_with_positions(
        &self,
        partition: String,
    ) -> Result<Vec<(u64, Event)>, JournalError> {
        PartitionJournal::replay_with_positions(self.inner.as_ref(), partition).await
    }

    async fn replay_with_positions_bounded(
        &self,
        partition: String,
        max_bytes: u64,
    ) -> Result<polyc_eventlog::BoundedReplay, JournalError> {
        PartitionJournal::replay_with_positions_bounded(self.inner.as_ref(), partition, max_bytes)
            .await
    }

    async fn replay_from_with_positions_bounded(
        &self,
        partition: String,
        start: u64,
        max_bytes: u64,
    ) -> Result<polyc_eventlog::BoundedReplay, JournalError> {
        PartitionJournal::replay_from_with_positions_bounded(
            self.inner.as_ref(),
            partition,
            start,
            max_bytes,
        )
        .await
    }

    async fn replay_range_with_positions_bounded(
        &self,
        partition: String,
        start: u64,
        end: u64,
        max_bytes: u64,
    ) -> Result<polyc_eventlog::BoundedReplay, JournalError> {
        PartitionJournal::replay_range_with_positions_bounded(
            self.inner.as_ref(),
            partition,
            start,
            end,
            max_bytes,
        )
        .await
    }

    async fn append_batch(
        &self,
        partition: String,
        events: Vec<Event>,
    ) -> Result<(), JournalError> {
        PartitionJournal::append_batch(self.inner.as_ref(), partition, events).await
    }

    fn is_stopping(&self) -> bool {
        self.stopping || PartitionJournal::is_stopping(self.inner.as_ref())
    }
}

impl Fixture {
    fn build(name: &str) -> Self {
        let dir =
            std::env::temp_dir().join(format!("polyc-search-worker-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        let shutdown = CancellationToken::new();
        let eventlog = Arc::new(
            EventLogHost::spawn(
                dir.join("journal"),
                shutdown.clone(),
                ApprovalSigner::from_seed(7).relabel_for_test(),
            )
            .expect("spawn eventlog host"),
        );
        let projection = SearchProjection::open(dir.join("projection")).expect("open projection");
        Self {
            eventlog,
            projection,
            dirty: Arc::new(DirtySet::default()),
            shutdown,
            dir,
        }
    }

    fn worker(&self) -> SearchIndexWorker {
        SearchIndexWorker::new(
            self.projection.clone(),
            crate::journal::over_host(Arc::clone(&self.eventlog)),
            Arc::clone(&self.dirty),
            key(),
        )
    }

    /// Append one committed turn and mark it dirty exactly as the commit feed
    /// would, so the worker sees the boundary a real deployment gives it.
    ///
    /// Returns the turn's message position and the exclusive boundary its
    /// `turn_complete` establishes. Both are read back from the append rather
    /// than assumed: the host appends its own signed MMR root after every
    /// batch, so a turn does not occupy the three consecutive positions its
    /// events suggest.
    async fn commit_turn(&self, partition: &str, nth: u8, text: &str) -> (u64, u64) {
        self.append(partition, committed_turn(&turn(nth), text))
            .await
    }

    /// Append a batch, mark it exactly as the commit feed would, and hand
    /// back every position the host assigned — for a test that needs a
    /// MARKER's position rather than a message's.
    async fn append_positions(&self, partition: &str, events: &[Event]) -> Vec<u64> {
        self.ensure_incarnation(partition).await;
        let marks = CommitMarks::new(Arc::clone(&self.dirty));
        let positions = self
            .eventlog
            .append_batch(partition.to_owned(), events.to_vec())
            .await
            .expect("append");
        marks.note_commit(
            partition,
            &[crate::feed::test_commit(partition, events, &positions)],
        );
        positions
    }

    /// The position of the first event in `events` whose kind base is `base`,
    /// read off the host's own answer rather than inferred from the batch's
    /// shape — the host appends its own signed MMR root after every batch.
    fn position_of(events: &[Event], positions: &[u64], base: &str) -> u64 {
        events
            .iter()
            .zip(positions.iter().copied())
            .find(|(event, _)| polyc_proto::kinds::parse(&event.kind).0 == base)
            .map_or(0, |(_, position)| position)
    }

    /// Append a batch and mark it exactly as the commit feed would.
    async fn append(&self, partition: &str, events: Vec<Event>) -> (u64, u64) {
        let positions = self.append_positions(partition, &events).await;
        let message = events
            .iter()
            .zip(positions.iter().copied())
            .find(|(event, _)| polyc_proto::kinds::parse(&event.kind).0 == kinds::USER_MSG)
            .map(|(_, position)| position)
            .unwrap_or_default();
        let boundary = events
            .iter()
            .zip(positions.iter().copied())
            .find(|(event, _)| polyc_proto::kinds::parse(&event.kind).0 == kinds::TURN_COMPLETE)
            .map_or(0, |(_, position)| position + 1);
        (message, boundary)
    }

    /// Append a batch and tell the index NOTHING — the case where the feed
    /// stopped delivering, or a chunk was lost, and only the sweep is left.
    async fn append_unmarked(&self, partition: &str, events: Vec<Event>) -> Vec<u64> {
        self.ensure_incarnation(partition).await;
        self.eventlog
            .append_batch(partition.to_owned(), events)
            .await
            .expect("append")
    }

    async fn ensure_incarnation(&self, partition: &str) {
        if self
            .eventlog
            .partition_event_count(partition.to_owned())
            .await
            .expect("read fixture partition length")
            != 0
        {
            return;
        }
        let incarnation = polyc_state::revision::PartitionIncarnation::from_bytes(
            polyc_crypto::random_secret_bytes(),
        );
        self.eventlog
            .append_batch(
                partition.to_owned(),
                vec![Event::new(
                    polyc_state::journal::INCARNATION_MARKER_KIND,
                    polyc_state::journal::incarnation_marker_payload(incarnation),
                )],
            )
            .await
            .expect("seed fixture source incarnation");
    }

    fn segments(&self, partition: &str) -> Vec<PathBuf> {
        let dir = self.projection.root().join(format!(
            "conversation_id={}",
            polyc_eventlog_host::encode_partition(partition).expect("the partition encodes")
        ));
        let Ok(entries) = std::fs::read_dir(dir) else {
            return Vec::new();
        };
        let mut paths: Vec<PathBuf> = entries
            .filter_map(Result::ok)
            .map(|entry| entry.path())
            .filter(|path| path.extension().and_then(std::ffi::OsStr::to_str) == Some("parquet"))
            .collect();
        paths.sort();
        paths
    }

    async fn coverage(&self, partition: &str) -> CoverageState {
        self.projection
            .coverage(partition, &key().key_id())
            .await
            .expect("coverage")
    }

    async fn indexed_positions(&self, partition: &str) -> Vec<u64> {
        self.projection
            .postings(partition, &key().key_id())
            .await
            .expect("postings")
            .map(|record| {
                record
                    .messages
                    .iter()
                    .map(|message| message.position)
                    .collect()
            })
            .unwrap_or_default()
    }
}

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

/// A directory shape for the pure trigger tests. `rows` is irrelevant to
/// [`should_compact`] by design, so it tracks `unmerged_rows` here rather than
/// pretending to be a second input.
const fn stats(segments: u32, unmerged_rows: u64) -> SegmentStats {
    SegmentStats {
        segments,
        rows: unmerged_rows,
        unmerged_rows,
    }
}

fn indexed(state: CoverageState) -> Coverage {
    match state {
        CoverageState::Indexed(coverage) => coverage,
        other => panic!("expected an indexed conversation, got {other:?}"),
    }
}

/// The steady state: a committed turn becomes searchable coverage, behind a
/// watermark naming the event it actually read.
#[tokio::test]
async fn a_committed_turn_is_published_with_verifiable_coverage() {
    let fx = Fixture::build("publish");
    let (_, boundary) = fx
        .commit_turn(CONVERSATION, 1, "where did we decide the timeout")
        .await;

    let outcomes = fx.worker().drain_once(&CancellationToken::new()).await;

    assert_eq!(
        outcomes,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary
            }
        )]
    );

    let coverage = indexed(fx.coverage(CONVERSATION).await);
    assert!(coverage.available);
    assert_eq!(coverage.indexed_through, boundary);

    // The incarnation must survive the read path's own re-derivation, which is
    // the check a search actually performs.
    let journal = LiveJournal::new(crate::journal::over_host(Arc::clone(&fx.eventlog)));
    assert_eq!(
        fx.projection
            .verified_coverage(CONVERSATION, &key().key_id(), &journal)
            .await
            .expect("verified"),
        CoverageState::Indexed(coverage),
        "the incarnation this worker published must re-derive from the live journal"
    );
}

/// A forward index appends the DELTA. Publishing the whole conversation each
/// turn is the O(N²) the segment layout exists to remove, and the earlier
/// segment is what keeps the prefix readable.
#[tokio::test]
async fn a_forward_index_appends_only_the_new_window() {
    let fx = Fixture::build("forward");
    let mut worker = fx.worker();

    let (first, _) = fx.commit_turn(CONVERSATION, 1, "alpha").await;
    worker.drain_once(&CancellationToken::new()).await;
    let (second, boundary) = fx.commit_turn(CONVERSATION, 2, "beta").await;
    let outcomes = worker.drain_once(&CancellationToken::new()).await;

    assert_eq!(
        outcomes,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary
            }
        )]
    );
    assert_eq!(
        fx.segments(CONVERSATION).len(),
        2,
        "the second pass must append a segment, not rewrite the conversation"
    );
    assert_eq!(
        fx.indexed_positions(CONVERSATION).await,
        vec![first, second],
        "both turns' messages must still be readable across the two segments"
    );
}

/// The rule an architecture review put in: a decodable newest footer beside
/// rows nothing can decode must NOT be appended to. Advancing the watermark
/// over a prefix no reader can open leaves the conversation refusing until
/// something rebuilds it, so the worker rebuilds it here.
#[tokio::test]
async fn undecodable_rows_fall_back_to_a_rebuild() {
    let fx = Fixture::build("decode-fallback");
    let mut worker = fx.worker();

    let (first, _) = fx.commit_turn(CONVERSATION, 1, "alpha").await;
    worker.drain_once(&CancellationToken::new()).await;
    let (second, _) = fx.commit_turn(CONVERSATION, 2, "beta").await;
    worker.drain_once(&CancellationToken::new()).await;
    let segments = fx.segments(CONVERSATION);
    assert_eq!(segments.len(), 2, "fixture must have two segments to break");

    // Corrupt the OLDER segment. Its rows hold the prefix, and the newest
    // segment — the only one `coverage` reads — stays perfectly readable, so
    // nothing but the probe can catch this.
    std::fs::write(&segments[0], b"not parquet").expect("corrupt the older segment");

    let (third, boundary) = fx.commit_turn(CONVERSATION, 3, "gamma").await;
    let outcomes = worker.drain_once(&CancellationToken::new()).await;

    assert_eq!(
        outcomes,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary
            }
        )]
    );
    assert_eq!(
        fx.segments(CONVERSATION).len(),
        1,
        "a rebuild replaces every segment, including the unreadable one"
    );
    assert_eq!(
        fx.indexed_positions(CONVERSATION).await,
        vec![first, second, third],
        "the rebuild must recover every turn from the journal, not just the new window"
    );
}

/// A replay that stopped short read only part of a turn, so a segment built
/// from it would sit behind a clean-looking watermark. Publish nothing.
#[tokio::test]
async fn a_replay_over_budget_publishes_nothing() {
    let fx = Fixture::build("budget");
    // One byte: the first message payload trips it, and the `turn_complete`
    // after it proves the replay stopped short of the boundary.
    let mut worker = fx.worker().with_replay_budget(1);

    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    fx.commit_turn(CONVERSATION, 2, "beta").await;

    let outcomes = worker.drain_once(&CancellationToken::new()).await;

    assert_eq!(
        outcomes,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Unavailable {
                reason: UnavailableReason::ReplayBudgetExceeded
            }
        )]
    );
    assert_eq!(
        fx.coverage(CONVERSATION).await,
        CoverageState::NeverIndexed,
        "a partial read must publish nothing at all — not even an unavailable segment claiming a \
         watermark"
    );
    assert!(
        fx.dirty.drain().is_empty(),
        "a budget failure fails identically on retry, so re-enqueuing it would spin the worker \
         on one conversation forever"
    );
}

/// The other half of the same rule: a cause that could resolve on its own MUST
/// be re-enqueued, or a quiet conversation stays refused until it happens to
/// receive another turn — which may be never.
#[tokio::test]
async fn a_transient_failure_schedules_its_own_repair() {
    let fx = Fixture::build("transient");
    let mut worker = fx.worker();

    let outcome = worker
        .fail_unavailable(CONVERSATION, UnavailableReason::StoreFailed)
        .await;

    assert_eq!(
        outcome,
        Outcome::Unavailable {
            reason: UnavailableReason::StoreFailed
        }
    );
    assert_eq!(
        fx.dirty.drain().get(CONVERSATION),
        Some(&Pending::Rebuild),
        "a transient failure must queue the rebuild rather than wait for the next turn"
    );
}

#[test]
fn only_a_self_resolving_cause_is_retried() {
    assert!(UnavailableReason::ReplayFailed.is_transient());
    assert!(UnavailableReason::StoreFailed.is_transient());
    assert!(
        !UnavailableReason::ReplayBudgetExceeded.is_transient(),
        "the identical replay reads the identical bytes and trips the identical budget"
    );
    assert!(
        !UnavailableReason::RecordTooLarge.is_transient(),
        "the row count is a property of the conversation, not of the attempt"
    );
    assert!(
        !UnavailableReason::SourceEmpty.is_transient(),
        "a journal that is gone does not come back on the next pass"
    );
}

/// A rebuild against a journal with no events must never publish
/// `{ indexed_through: 0, available: true }` — that claims complete coverage of
/// a conversation nothing has read.
#[tokio::test]
async fn a_rebuild_with_no_journal_publishes_no_coverage() {
    let fx = Fixture::build("empty-source");
    let mut worker = fx.worker();

    let outcome = worker.index_partition(CONVERSATION, None).await;

    assert_eq!(
        outcome,
        Outcome::Unavailable {
            reason: UnavailableReason::SourceEmpty
        }
    );
    assert_eq!(fx.coverage(CONVERSATION).await, CoverageState::NeverIndexed);
    assert!(
        fx.dirty.drain().is_empty(),
        "re-enqueuing a partition that no longer exists spins the worker forever"
    );
}

/// Destroy is terminal in both directions: the rows go, the tombstone stays,
/// and no later pass may publish over it.
#[tokio::test]
async fn a_destroyed_conversation_is_tombstoned_and_never_indexed_again() {
    let fx = Fixture::build("destroy");
    let mut worker = fx.worker();

    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    worker.drain_once(&CancellationToken::new()).await;

    CommitMarks::new(Arc::clone(&fx.dirty))
        .note_partition_change(CONVERSATION, crate::feed::PartitionChange::Destroyed);
    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(CONVERSATION.to_owned(), Outcome::Destroyed)]
    );
    assert_eq!(fx.coverage(CONVERSATION).await, CoverageState::Destroyed);

    // A late mark for the same conversation must not resurrect it.
    assert_eq!(
        worker.index_partition(CONVERSATION, Some(3)).await,
        Outcome::Destroyed
    );
    assert_eq!(fx.coverage(CONVERSATION).await, CoverageState::Destroyed);
}

#[tokio::test]
async fn a_recreation_queued_after_destroy_publishes_only_the_new_source() {
    let fx = Fixture::build("destroy-then-recreate");
    let mut worker = fx.worker();
    fx.commit_turn(CONVERSATION, 1, "old lineage text").await;
    worker.drain_once(&CancellationToken::new()).await;

    fx.eventlog
        .destroy_partition(CONVERSATION.to_owned())
        .await
        .expect("destroy the old physical source");
    let replacement = committed_turn(&turn(2), "replacement text");
    fx.append_unmarked(CONVERSATION, replacement).await;
    let boundary = fx
        .eventlog
        .partition_event_count(CONVERSATION.to_owned())
        .await
        .expect("read the replacement source length");
    let marks = CommitMarks::new(Arc::clone(&fx.dirty));
    marks.note_partition_change(CONVERSATION, crate::feed::PartitionChange::Destroyed);
    marks.note_source_replacement(CONVERSATION);

    assert_eq!(
        fx.dirty.pending_for(CONVERSATION),
        Some(Pending::Replace),
        "the later physical source must supersede queued destruction"
    );
    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary,
            },
        )]
    );
    let postings = fx
        .projection
        .postings(CONVERSATION, &key().key_id())
        .await
        .expect("replacement postings are readable")
        .expect("the replacement source is published");
    assert_eq!(postings.messages.len(), 1);
    assert_eq!(
        postings.messages[0].term_hashes,
        key().hash_text("replacement text"),
        "only the replacement source's terms may survive"
    );
}

/// The migration source takes the opposite treatment: it must read as though
/// it were never indexed, because the conversation is alive under its new id
/// and its coverage resolves from the destination. A tombstone here would
/// report a live conversation as destroyed.
#[tokio::test]
async fn a_migrated_conversation_leaves_no_tombstone_behind() {
    let fx = Fixture::build("migrate-source");
    let mut worker = fx.worker();

    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    worker.drain_once(&CancellationToken::new()).await;

    CommitMarks::new(Arc::clone(&fx.dirty))
        .note_partition_change(CONVERSATION, crate::feed::PartitionChange::MigratedAway);

    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(CONVERSATION.to_owned(), Outcome::Removed)]
    );
    assert_eq!(
        fx.coverage(CONVERSATION).await,
        CoverageState::NeverIndexed,
        "a migrated-away conversation must leave no trace, never a tombstone"
    );
}

#[tokio::test]
async fn a_same_name_source_replacement_removes_old_rows_and_publishes_the_rebuild() {
    let fx = Fixture::build("source-replacement");
    let mut worker = fx.worker();
    fx.commit_turn(CONVERSATION, 1, "old lineage text").await;
    worker.drain_once(&CancellationToken::new()).await;
    let old = fx
        .projection
        .postings(CONVERSATION, &key().key_id())
        .await
        .expect("old postings are readable")
        .expect("the old lineage is indexed");

    fx.eventlog
        .destroy_partition(CONVERSATION.to_owned())
        .await
        .expect("replace the physical source");
    fx.append_unmarked(CONVERSATION, committed_turn(&turn(2), "replacement text"))
        .await;
    let boundary = fx
        .eventlog
        .partition_event_count(CONVERSATION.to_owned())
        .await
        .expect("replacement source length");
    let marks = CommitMarks::new(Arc::clone(&fx.dirty));
    marks.note_partition_change(CONVERSATION, crate::feed::PartitionChange::Rewritten);
    marks.note_source_replacement(CONVERSATION);
    marks.note_bootstrap(CONVERSATION);
    assert_eq!(
        fx.dirty.pending_for(CONVERSATION),
        Some(Pending::Replace),
        "concurrent rewrite/bootstrap marks cannot turn replacement into removal"
    );

    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary,
            },
        )]
    );
    let replacement = fx
        .projection
        .postings(CONVERSATION, &key().key_id())
        .await
        .expect("replacement postings are readable")
        .expect("the replacement lineage is indexed");
    assert_ne!(replacement, old, "stale source rows do not survive");
    assert_eq!(replacement.messages.len(), 1);
    assert_eq!(
        indexed(fx.coverage(CONVERSATION).await).indexed_through,
        boundary
    );
}

#[tokio::test]
async fn a_migration_after_queued_replacement_leaves_no_projection_trace() {
    let fx = Fixture::build("replacement-then-migration");
    let mut worker = fx.worker();
    fx.commit_turn(CONVERSATION, 1, "old lineage text").await;
    worker.drain_once(&CancellationToken::new()).await;

    fx.eventlog
        .destroy_partition(CONVERSATION.to_owned())
        .await
        .expect("migration removes the local source");
    let marks = CommitMarks::new(Arc::clone(&fx.dirty));
    marks.note_source_replacement(CONVERSATION);
    marks.note_partition_change(CONVERSATION, crate::feed::PartitionChange::MigratedAway);

    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(CONVERSATION.to_owned(), Outcome::Removed)]
    );
    assert_eq!(fx.coverage(CONVERSATION).await, CoverageState::NeverIndexed);
    assert!(
        fx.segments(CONVERSATION).is_empty(),
        "a migrated-away source leaves neither stale rows nor unavailable coverage"
    );
}

#[tokio::test]
async fn a_failed_source_removal_revokes_same_boundary_stale_reads_across_restart() {
    let fx = Fixture::build("source-replacement-remove-failure");
    let mut worker = fx.worker();
    let (_, old_boundary) = fx.commit_turn(CONVERSATION, 1, "secret alpha").await;
    worker.drain_once(&CancellationToken::new()).await;
    let old_source =
        PartitionJournal::partition_incarnation(fx.eventlog.as_ref(), CONVERSATION.to_owned())
            .await
            .expect("read old source")
            .expect("old source exists");

    fx.eventlog
        .destroy_partition(CONVERSATION.to_owned())
        .await
        .expect("replace the physical source");
    let replacement_events = committed_turn(&turn(1), "public alpha");
    let replacement_positions = fx
        .append_unmarked(CONVERSATION, replacement_events.clone())
        .await;
    let replacement_boundary = Fixture::position_of(
        &replacement_events,
        &replacement_positions,
        kinds::TURN_COMPLETE,
    ) + 1;
    let replacement_source =
        PartitionJournal::partition_incarnation(fx.eventlog.as_ref(), CONVERSATION.to_owned())
            .await
            .expect("read replacement source")
            .expect("replacement source exists");
    assert_eq!(old_boundary, replacement_boundary);
    assert_ne!(old_source, replacement_source);

    CommitMarks::new(Arc::clone(&fx.dirty)).note_source_replacement(CONVERSATION);
    fx.projection.fail_next_remove();
    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Unavailable {
                reason: UnavailableReason::StoreFailed,
            },
        )]
    );
    let coverage = indexed(fx.coverage(CONVERSATION).await);
    assert!(!coverage.available, "stale read authority is revoked first");
    assert_eq!(
        fx.projection
            .verified_coverage(
                CONVERSATION,
                &key().key_id(),
                &LiveJournal::new(crate::journal::over_host(Arc::clone(&fx.eventlog))),
            )
            .await
            .expect("verify stale coverage"),
        CoverageState::Stale,
        "the exact source mismatch refuses independently of cleanup"
    );

    let reopened = SearchProjection::open(fx.projection.root().to_path_buf())
        .expect("reopen projection after failed removal");
    assert_eq!(
        reopened
            .verified_coverage(
                CONVERSATION,
                &key().key_id(),
                &LiveJournal::new(crate::journal::over_host(Arc::clone(&fx.eventlog))),
            )
            .await
            .expect("verify after restart"),
        CoverageState::Stale,
        "restart cannot turn old same-boundary rows back into authority"
    );
    assert_eq!(
        fx.dirty.pending_for(CONVERSATION),
        Some(Pending::Replace),
        "failed cleanup remains retryable"
    );
}

#[tokio::test]
async fn a_source_rotation_during_replay_cannot_publish_mixed_lineage_coverage() {
    let fx = Fixture::build("source-rotates-during-replay");
    fx.append_unmarked(CONVERSATION, committed_turn(&turn(1), "alpha"))
        .await;
    let before =
        PartitionJournal::partition_incarnation(fx.eventlog.as_ref(), CONVERSATION.to_owned())
            .await
            .expect("read fixture source")
            .expect("source exists");
    let journal: Arc<dyn PartitionJournal> = Arc::new(ControlledSourceJournal {
        inner: Arc::clone(&fx.eventlog),
        before,
        after: polyc_state::revision::PartitionIncarnation::from_bytes([99; 32]),
        source_reads: AtomicUsize::new(0),
        source_error: None,
        stopping: false,
    });
    let mut worker =
        SearchIndexWorker::new(fx.projection.clone(), journal, Arc::clone(&fx.dirty), key());
    CommitMarks::new(Arc::clone(&fx.dirty)).note_bootstrap(CONVERSATION);

    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Unavailable {
                reason: UnavailableReason::ReplayFailed,
            },
        )]
    );
    assert_eq!(fx.coverage(CONVERSATION).await, CoverageState::NeverIndexed);
    assert_eq!(
        fx.dirty.pending_for(CONVERSATION),
        Some(Pending::Replace),
        "the exact replacement cleanup is retried instead of publishing a mixed-source fold"
    );
}

#[tokio::test]
async fn a_first_registration_replacement_mark_publishes_instead_of_removing() {
    let fx = Fixture::build("first-registration-replacement");
    fx.append_unmarked(CONVERSATION, committed_turn(&turn(3), "first registration"))
        .await;
    let boundary = fx
        .eventlog
        .partition_event_count(CONVERSATION.to_owned())
        .await
        .expect("first source length");
    let marks = CommitMarks::new(Arc::clone(&fx.dirty));
    marks.note_source_replacement(CONVERSATION);
    marks.note_bootstrap(CONVERSATION);

    assert_eq!(
        fx.worker().drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary,
            },
        )]
    );
    assert!(
        matches!(fx.coverage(CONVERSATION).await, CoverageState::Indexed(_)),
        "an empty old lineage still leads to a replacement rebuild"
    );
}

/// Both triggers, independently. A count alone misses the conversation with
/// three enormous segments; a row ceiling alone misses the one with forty tiny
/// ones.
#[test]
fn either_compaction_trigger_fires_on_its_own() {
    assert!(!should_compact(
        stats(COMPACT_SEGMENT_THRESHOLD - 1, 0),
        COMPACT_ROW_CEILING
    ));
    assert!(should_compact(
        stats(COMPACT_SEGMENT_THRESHOLD, 0),
        COMPACT_ROW_CEILING
    ));
    assert!(
        should_compact(stats(1, COMPACT_ROW_CEILING), COMPACT_ROW_CEILING),
        "one enormous segment must fold on rows alone"
    );
    assert!(!should_compact(
        stats(1, COMPACT_ROW_CEILING - 1),
        COMPACT_ROW_CEILING
    ));
}

/// The row trigger measures UNMERGED rows, so a conversation whose total is
/// permanently past the ceiling is not permanently due for a fold.
///
/// Compaction merges rows; it never drops them. A trigger on the total would
/// therefore fire on every append forever once crossed — three O(rows) passes
/// per committed turn, which is exactly the quadratic the segment layout
/// exists to remove.
#[test]
fn the_row_trigger_reads_a_quantity_a_fold_resets() {
    let folded = SegmentStats {
        segments: 1,
        rows: COMPACT_ROW_CEILING * 10,
        unmerged_rows: 0,
    };
    assert!(
        !should_compact(folded, COMPACT_ROW_CEILING),
        "a conversation an order of magnitude past the ceiling must be DONE once it is folded"
    );
}

/// The count trigger, end to end: a run of appends folds back to one segment
/// rather than growing a directory every reader has to open in full.
#[tokio::test]
async fn a_run_of_appends_compacts_back_to_one_segment() {
    let fx = Fixture::build("compact");
    let mut worker = fx.worker();

    // The first pass rebuilds to one segment and each later pass appends one,
    // so the directory reaches the threshold on pass COMPACT_SEGMENT_THRESHOLD.
    for nth in 0..u8::try_from(COMPACT_SEGMENT_THRESHOLD).unwrap() {
        fx.commit_turn(CONVERSATION, nth, "alpha beta").await;
        worker.drain_once(&CancellationToken::new()).await;
    }

    assert_eq!(
        fx.segments(CONVERSATION).len(),
        1,
        "crossing the segment threshold must fold the conversation back to one file"
    );
    let coverage = indexed(fx.coverage(CONVERSATION).await);
    assert!(
        coverage.available,
        "compaction changes storage, not coverage"
    );
    assert_eq!(
        fx.indexed_positions(CONVERSATION).await.len(),
        usize::try_from(COMPACT_SEGMENT_THRESHOLD).unwrap(),
        "every turn must survive the fold"
    );
}

/// The trigger reads the directory, never a counter the worker remembers.
///
/// A per-conversation counter map has to be bounded, and past its bound it
/// answers zero for every conversation thereafter — silently disabling the
/// count trigger for the rest of the process's life. A worker that has never
/// seen this conversation before must still fold it on time, which is what a
/// fresh worker per pass proves.
#[tokio::test]
async fn a_worker_with_no_memory_of_the_conversation_still_folds_it() {
    let fx = Fixture::build("compact-fresh-worker");

    for nth in 0..u8::try_from(COMPACT_SEGMENT_THRESHOLD).unwrap() {
        fx.commit_turn(CONVERSATION, nth, "alpha beta").await;
        // A NEW worker every pass: nothing carries over but the directory.
        fx.worker().drain_once(&CancellationToken::new()).await;
    }

    assert_eq!(
        fx.segments(CONVERSATION).len(),
        1,
        "the fold must come from the segment listing, not from what this process remembers"
    );
    assert_eq!(
        fx.indexed_positions(CONVERSATION).await.len(),
        usize::try_from(COMPACT_SEGMENT_THRESHOLD).unwrap(),
        "every turn must survive the fold"
    );
}

/// The row trigger, end to end, and the reason it must be an edge.
///
/// Each turn here writes two rows ("alpha beta" is two distinct terms on one
/// message), and the ceiling is three. So the directory must grow to two
/// segments, fold on the pass that takes the unmerged count to four, and then
/// grow again — the sawtooth a working trigger produces.
///
/// Against the conversation TOTAL the same run folds on every single pass after
/// the first, because a fold never lowers the total. The `2`s in this
/// expectation are what that bug erases.
#[tokio::test]
async fn the_row_trigger_folds_on_an_edge_rather_than_on_every_append() {
    let fx = Fixture::build("compact-rows");
    let mut worker = fx.worker().with_row_ceiling(3);

    let mut segments_after_each_pass = Vec::new();
    for nth in 0..5u8 {
        fx.commit_turn(CONVERSATION, nth, "alpha beta").await;
        worker.drain_once(&CancellationToken::new()).await;
        segments_after_each_pass.push(fx.segments(CONVERSATION).len());
    }

    assert_eq!(
        segments_after_each_pass,
        vec![1, 2, 1, 2, 1],
        "the row ceiling must fold on the pass that crosses it and then let the directory grow \
         again, never fold on every append forever"
    );
    assert_eq!(
        fx.indexed_positions(CONVERSATION).await.len(),
        5,
        "every turn must survive the folds"
    );
}

/// Overflow is not a one-way door: the reconcile is the only thing that may
/// clear the flag, and it does so only after re-establishing coverage for
/// every conversation in the deployment.
#[tokio::test]
async fn the_reconcile_clears_degraded_after_covering_every_conversation() {
    let fx = Fixture::build("reconcile");
    let mut worker = fx.worker();

    let (_, first_boundary) = fx.commit_turn(CONVERSATION, 1, "alpha").await;
    let (_, other_boundary) = fx.commit_turn(OTHER_CONVERSATION, 2, "beta").await;
    // Overflow the dirty set exactly as a burst of live conversations would,
    // which is what sets the flag in the first place.
    for nth in 0..=super::super::marks::MAX_TRACKED_PARTITIONS {
        fx.dirty
            .mark(&format!("conv-flood-{nth}"), Pending::Rebuild);
    }
    assert!(fx.dirty.degraded(), "the fixture must actually be degraded");

    let summary = worker
        .reconcile(&CancellationToken::new())
        .await
        .expect("reconcile");

    assert!(summary.complete);
    assert_eq!(summary.visited, 2, "both conversations, and nothing else");
    assert_eq!(summary.refused, 0);
    assert!(
        !fx.dirty.degraded(),
        "a completed sweep is what re-opens the door overflow closed"
    );
    // A sweep has no marked boundary to work from, so it reads to the
    // journal's CEILING and lets the open-turn barrier decide what within it is
    // committed. With no turn in flight that lands at or past the boundary a
    // mark would have reported, never short of it.
    assert!(
        indexed(fx.coverage(CONVERSATION).await).indexed_through >= first_boundary,
        "a sweep must cover at least what the queue would have"
    );
    assert!(
        indexed(fx.coverage(OTHER_CONVERSATION).await).indexed_through >= other_boundary,
        "the conversation whose mark overflow discarded must be covered by the sweep, not by \
         the queue"
    );
}

/// The overflow the sweep never saw.
///
/// A mark lands on whatever task drains the feed, so it can be dropped
/// AFTER the sweep visited that partition — and this worker's own
/// `fail_unavailable` re-marks every transiently failed conversation during the
/// sweep, so a fleet-wide store failure produces exactly this. A pass that
/// clears on the flag alone declares the index whole over a mark nobody
/// accounted for.
#[tokio::test]
async fn a_degrade_landing_mid_sweep_blocks_the_clear() {
    let fx = Fixture::build("reconcile-late-overflow");
    let mut worker = fx.worker();

    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    for nth in 0..=super::super::marks::MAX_TRACKED_PARTITIONS {
        fx.dirty
            .mark(&format!("conv-flood-{nth}"), Pending::Rebuild);
    }
    assert!(fx.dirty.degraded());

    let token = CancellationToken::new();
    let mut sweep = std::pin::pin!(worker.reconcile(&token));
    // One poll: enough for the pass to snapshot the degrade counter and reach
    // its first await, which is the partition enumeration.
    let first = std::future::poll_fn(|cx| {
        std::task::Poll::Ready(std::future::Future::poll(sweep.as_mut(), cx))
    })
    .await;
    assert!(
        first.is_pending(),
        "the sweep must still be running for this to be a mid-sweep overflow"
    );

    // The set is already full, so this is a dropped mark — the same event
    // overflow produces on the write path.
    fx.dirty.mark("conv-dropped-mid-sweep", Pending::Rebuild);

    let summary = sweep.await.expect("reconcile");
    assert!(
        summary.complete,
        "the sweep did reach every conversation; the clear is what must be refused"
    );
    assert!(
        fx.dirty.degraded(),
        "a mark dropped after the sweep passed that partition is one the sweep cannot account for"
    );
}

/// A destroy the store refused must keep the index from reporting itself whole.
///
/// The sweep cannot repair it: `reconcile` enumerates partitions that still
/// EXIST, and a destroyed one is gone from that listing. So the outstanding
/// work lives in the dirty set as its re-queued mark, and the clear consults
/// it — a deployment told to forget user text must not report the index whole
/// while those rows are still on disk.
#[tokio::test]
async fn an_outstanding_removal_blocks_the_clear() {
    let fx = Fixture::build("reconcile-pending-removal");
    let mut worker = fx.worker();

    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    // What a failed destroy leaves behind, marked before the flood so the
    // overflow cannot be what drops it.
    fx.dirty.mark(OTHER_CONVERSATION, Pending::Destroy);
    for nth in 0..=super::super::marks::MAX_TRACKED_PARTITIONS {
        fx.dirty
            .mark(&format!("conv-flood-{nth}"), Pending::Rebuild);
    }
    assert!(fx.dirty.degraded());

    let summary = worker
        .reconcile(&CancellationToken::new())
        .await
        .expect("reconcile");
    assert!(summary.complete);
    assert!(
        fx.dirty.degraded(),
        "rows the deployment was told to forget are still on disk, so the index is not whole"
    );

    // Once the removal is off the queue the identical sweep may clear.
    let _ = fx.dirty.drain();
    assert!(fx.dirty.degraded(), "a drain is not a reconcile");
    worker
        .reconcile(&CancellationToken::new())
        .await
        .expect("reconcile");
    assert!(
        !fx.dirty.degraded(),
        "with no removal outstanding the same sweep re-opens the door"
    );
}

/// A sweep that did not finish proves nothing about the conversations it never
/// looked at.
#[tokio::test]
async fn an_interrupted_reconcile_leaves_the_index_degraded() {
    let fx = Fixture::build("reconcile-cancel");
    let mut worker = fx.worker();

    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    for nth in 0..=super::super::marks::MAX_TRACKED_PARTITIONS {
        fx.dirty
            .mark(&format!("conv-flood-{nth}"), Pending::Rebuild);
    }

    let cancelled = CancellationToken::new();
    cancelled.cancel();
    let summary = worker.reconcile(&cancelled).await.expect("reconcile");

    assert!(!summary.complete);
    assert_eq!(summary.visited, 0);
    assert!(
        fx.dirty.degraded(),
        "a cancelled sweep must not clear the flag it never earned"
    );
}

/// Cancellation loses no work: whatever the drain took but did not apply goes
/// back on the queue.
#[tokio::test]
async fn a_cancelled_drain_puts_its_work_back() {
    let fx = Fixture::build("drain-cancel");
    let mut worker = fx.worker();

    let (_, boundary) = fx.commit_turn(CONVERSATION, 1, "alpha").await;
    let cancelled = CancellationToken::new();
    cancelled.cancel();

    assert!(worker.drain_once(&cancelled).await.is_empty());
    assert_eq!(
        fx.dirty.drain().get(CONVERSATION),
        Some(&Pending::IndexThrough(boundary)),
        "a mark the drain took but never applied must survive the cancellation"
    );
}

/// The barrier case a naive port gets wrong: an open turn holds the watermark
/// at its start. A source-only prefix may publish before that position. The pass
/// must still report the barrier. It must not call the conversation caught up
/// while completed turns remain unindexable behind the barrier.
#[tokio::test]
async fn an_open_turn_holds_the_watermark_without_refusing() {
    let fx = Fixture::build("open-turn");
    let mut worker = fx.worker();

    // Turn one opens and stays open; turn two runs to completion behind it.
    let open = vec![
        marker(kinds::TURN_START, &turn(1)),
        text_msg(&turn(1), "still in flight"),
    ];
    let positions = fx.append_positions(CONVERSATION, &open).await;
    let open_start = Fixture::position_of(&open, &positions, kinds::TURN_START);
    let open_message = Fixture::position_of(&open, &positions, kinds::USER_MSG);
    let (committed_message, committed_boundary) =
        fx.commit_turn(CONVERSATION, 2, "committed text").await;

    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::BarrierHeld {
                open_turn_at: open_start,
                journal_lag: committed_boundary.saturating_sub(open_start),
            }
        )],
        "the barrier sits at the open turn's start, so the pass must SAY that rather than \
         answering the same `Published` or `AlreadyCurrent` a caught-up conversation does"
    );
    let prefix = indexed(fx.coverage(CONVERSATION).await);
    assert_eq!(prefix.indexed_through, open_start);
    assert!(
        fx.indexed_positions(CONVERSATION).await.is_empty(),
        "publishing exact-source coverage before the barrier must not publish either open or \
         out-of-order turn text"
    );

    // The open turn completes: both turns become indexable at once.
    let (_, boundary) = fx
        .append(CONVERSATION, vec![marker(kinds::TURN_COMPLETE, &turn(1))])
        .await;

    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary
            }
        )]
    );
    assert_eq!(
        fx.indexed_positions(CONVERSATION).await,
        vec![open_message, committed_message],
        "the turn that completed out of order must be indexed, not lost behind the barrier"
    );
}

/// A shutdown that lands mid-conversation must publish NOTHING.
///
/// The Container supervises this worker on the same `CancellationToken` that
/// drives the event-log host, so a graceful stop can close the log while a
/// pass is holding a conversation. Treating that as an ordinary replay failure
/// would write `available: false` for a perfectly healthy conversation — and
/// under all-or-nothing coverage that takes participation-wide search down for
/// everyone in it until something rebuilds, with the re-enqueue that would
/// have scheduled the rebuild dying with the process.
#[tokio::test]
async fn a_shut_down_event_log_defers_rather_than_marking_a_conversation_unsearchable() {
    let fx = Fixture::build("shutdown-defer");
    let (_message, boundary) = fx.commit_turn(CONVERSATION, 1, "timeout decision").await;
    let mut worker = fx.worker();
    assert!(matches!(
        worker.index_partition(CONVERSATION, Some(boundary)).await,
        Outcome::Published { .. }
    ));
    let published = fx.segments(CONVERSATION);
    assert_eq!(published.len(), 1, "one segment published before shutdown");

    fx.shutdown.cancel();
    // The host answers `Closed` only after its own post-cancel drain has run.
    for _ in 0..200 {
        if fx
            .eventlog
            .partition_event_count(CONVERSATION.to_owned())
            .await
            .is_err()
        {
            break;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }

    // The rebuild path, which reaches the host through `partition_event_count`.
    assert_eq!(
        tokio::time::timeout(
            Duration::from_secs(10),
            worker.index_partition(CONVERSATION, None),
        )
        .await
        .expect("index_partition hung for 10s after the event log shut down"),
        Outcome::Deferred
    );
    // The forward path, which reaches it through the bounded range replay.
    assert_eq!(
        tokio::time::timeout(
            Duration::from_secs(10),
            worker.index_partition(CONVERSATION, Some(boundary + 5)),
        )
        .await
        .expect("index_partition hung for 10s after the event log shut down"),
        Outcome::Deferred
    );

    assert_eq!(
        fx.segments(CONVERSATION),
        published,
        "a shutdown must not publish a refusal over a healthy conversation"
    );
    match fx.coverage(CONVERSATION).await {
        CoverageState::Indexed(coverage) => assert!(
            coverage.available,
            "the conversation must still be searchable after the restart"
        ),
        other => panic!("coverage must survive a shutdown untouched: {other:?}"),
    }
}

/// A reconcile interrupted by the same shutdown must not clear `degraded`.
///
/// The flag means "coverage is unknown somewhere". A pass that stopped because
/// the log went away examined nothing after that point, so clearing would
/// declare the index whole on the strength of conversations it never looked
/// at.
#[tokio::test]
async fn a_reconcile_stopped_by_a_shut_down_event_log_leaves_the_index_degraded() {
    let fx = Fixture::build("shutdown-reconcile");
    fx.commit_turn(CONVERSATION, 1, "timeout decision").await;
    fx.dirty.mark(OTHER_CONVERSATION, Pending::Rebuild);
    // Overflow's own signal, set the way the observer sets it.
    for index in 0..=super::super::marks::MAX_TRACKED_PARTITIONS {
        fx.dirty
            .mark(&format!("conv-overflow-{index}"), Pending::Rebuild);
    }
    assert!(fx.dirty.degraded(), "the dirty set must report overflow");

    let mut worker = fx.worker();
    fx.shutdown.cancel();
    for _ in 0..200 {
        if fx.eventlog.list_partitions().await.is_err() {
            break;
        }
        tokio::time::sleep(Duration::from_millis(25)).await;
    }

    // `list_partitions` may itself fail once the host is closed, which is
    // already a refusal to clear; when it still answers from the snapshot the
    // per-conversation reads are what stop the pass. Bounded well above the
    // 5s poll loop above: a hang here must fail fast and name itself, not
    // take down the whole test partition at the 180s suite timeout (#2620).
    let reconciled = tokio::time::timeout(
        Duration::from_secs(10),
        worker.reconcile(&CancellationToken::new()),
    )
    .await
    .expect("reconcile hung for 10s after the event log shut down");
    if let Ok(summary) = reconciled {
        assert!(
            !summary.complete,
            "a pass the log cut short cannot be called complete"
        );
    }
    assert!(
        fx.dirty.degraded(),
        "only a pass that reached every conversation may clear the flag"
    );
}

/// Excision fixtures. The conversation id must sanitize to [`CONVERSATION`],
/// or the marker is verified and then correctly ignored as belonging to
/// another conversation — which is the failure `verified_excisions_matching`
/// exists to prevent, not a state worth asserting here.
mod excision {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use polyc_crypto::approval::{ApprovalSigner, EXCISION_SCOPE_SOURCE_ONLY, excision_payload};

    use super::*;

    /// The unsanitized conversation id behind [`CONVERSATION`].
    const CONVERSATION_ID: &str = "web:01950000-0000-7000-8000-0000000000aa";

    /// A verified excision marker naming `positions`, as `approval_grpc`
    /// appends it: a plain event, solo, at the journal tail.
    fn excision_marker(positions: &[u64]) -> Event {
        let (payload, _, _) = excision_payload(
            CONVERSATION_ID,
            EXCISION_SCOPE_SOURCE_ONLY,
            positions,
            "persona-1",
            "test excision",
            &ApprovalSigner::from_seed(1),
        );
        Event::new(kinds::TAINT_EXCISION.to_owned(), payload)
    }

    /// Whether the conversation's stored postings still carry `term`.
    async fn holds_term(fx: &Fixture, term: &str) -> bool {
        fx.projection
            .postings(CONVERSATION, &key().key_id())
            .await
            .expect("postings")
            .is_some_and(|record| {
                record
                    .messages
                    .iter()
                    .any(|message| message.term_hashes.contains(&key().hash_term(term)))
            })
    }

    /// A restart after the post-rewrite taint marker but before the index
    /// rebuild discards the in-memory work mark. This test isolates the marker
    /// defense. The marker itself is a pure append and preserves the source. The
    /// durable scan frontier must still refuse it.
    ///
    /// Before the excision frontier, `verified_coverage` returned `Indexed`
    /// here — so a reconcile took the forward catch-up branch over a window
    /// that cannot contain the excised positions, `strip_excised` did nothing,
    /// the earlier segments kept the removed terms, and the sweep then cleared
    /// `degraded` over a conversation still serving excised text.
    #[tokio::test]
    async fn an_excision_a_restart_forgot_is_still_refused() {
        let fx = Fixture::build("excision-restart");
        let (message, _) = fx.commit_turn(CONVERSATION, 1, "hunter2 passphrase").await;
        fx.worker().drain_once(&CancellationToken::new()).await;
        assert!(holds_term(&fx, "hunter2").await, "the term must be indexed");

        // The excision lands, and the process dies before the drain: the mark
        // is gone and only the journal remembers.
        fx.append(CONVERSATION, vec![excision_marker(&[message])])
            .await;
        fx.dirty.drain();

        let journal = LiveJournal::new(crate::journal::over_host(Arc::clone(&fx.eventlog)));
        assert_eq!(
            fx.projection
                .verified_coverage(CONVERSATION, &key().key_id(), &journal)
                .await
                .expect("verified"),
            CoverageState::Stale,
            "an excision the index never applied must read as uncovered, whatever the \
             incarnation says"
        );
    }

    /// And the reconcile REPAIRS it rather than re-blessing it: the `Stale`
    /// above lands in the rebuild arm, which replays from zero and strips.
    #[tokio::test]
    async fn a_reconcile_rebuilds_the_excision_a_restart_forgot() {
        let fx = Fixture::build("excision-reconcile");
        let (secret, _) = fx.commit_turn(CONVERSATION, 1, "hunter2 passphrase").await;
        fx.commit_turn(CONVERSATION, 2, "routine follow up").await;
        let mut worker = fx.worker();
        worker.drain_once(&CancellationToken::new()).await;

        // The marker names the whole message, so the excision removes that
        // message's terms and nothing else.
        fx.append(CONVERSATION, vec![excision_marker(&[secret])])
            .await;
        fx.dirty.drain();

        worker
            .reconcile(&CancellationToken::new())
            .await
            .expect("reconcile");

        assert!(
            !holds_term(&fx, "hunter2").await,
            "the reconcile must strip the excised message, not catch the conversation up over it"
        );
        assert!(
            holds_term(&fx, "routine").await,
            "a rebuild must keep every message the excision did not name"
        );
    }

    /// The rule that keeps the stored frontier honest. A forward pass CAN see
    /// the marker — it sits above the watermark — but it cannot apply it,
    /// because the positions it names live in segments an append never
    /// rewrites. Publishing anyway would advance the frontier past a marker
    /// that was only half applied, which is worse than missing it: the read
    /// path would then certify the conversation as clean.
    #[tokio::test]
    async fn a_forward_pass_that_sees_an_excision_rebuilds_instead() {
        let fx = Fixture::build("excision-forward");
        let (message, _) = fx.commit_turn(CONVERSATION, 1, "hunter2 passphrase").await;
        let mut worker = fx.worker();
        worker.drain_once(&CancellationToken::new()).await;

        // The marker lands, then a routine turn commits. The observer's own
        // `Rebuild` mark is thrown away, leaving exactly what a restart leaves:
        // a forward boundary and nothing else.
        fx.append(CONVERSATION, vec![excision_marker(&[message])])
            .await;
        let (_, boundary) = fx.commit_turn(CONVERSATION, 2, "routine follow up").await;
        fx.dirty.drain();

        // A rebuild's watermark is the journal's own tail rather than the
        // boundary the mark named, because it replays the whole partition.
        assert!(
            matches!(
                worker.index_partition(CONVERSATION, Some(boundary)).await,
                Outcome::Published { indexed_through } if indexed_through >= boundary
            ),
            "the escalated rebuild must publish"
        );
        assert!(
            !holds_term(&fx, "hunter2").await,
            "a forward pass must escalate to a rebuild rather than append over an excision"
        );
        assert_eq!(
            fx.segments(CONVERSATION).len(),
            1,
            "a rebuild leaves exactly one segment; an append would have left more"
        );
        assert!(
            holds_term(&fx, "routine").await,
            "the escalated rebuild must still index the turn that triggered it"
        );
    }
}

/// Invalidation row 6, the half that was quoted and never executed. A
/// conversation refused for any reason has no recovery path of its own: only a
/// rebuild clears the flag, and nothing revisits a conversation that has
/// stopped being marked. So the next committed turn's append becomes that
/// rebuild.
#[tokio::test]
async fn an_append_onto_a_refused_conversation_becomes_its_rebuild() {
    let fx = Fixture::build("recover-unavailable");
    let mut worker = fx.worker();
    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    worker.drain_once(&CancellationToken::new()).await;

    fx.projection
        .mark_unavailable(CONVERSATION, &key().key_id())
        .await
        .expect("mark unavailable");
    assert!(!indexed(fx.coverage(CONVERSATION).await).available);

    let (_, boundary) = fx.commit_turn(CONVERSATION, 2, "beta").await;
    let outcomes = worker.drain_once(&CancellationToken::new()).await;
    assert!(
        matches!(
            outcomes.as_slice(),
            [(partition, Outcome::Published { indexed_through })]
                if partition == CONVERSATION && *indexed_through >= boundary
        ),
        "the pass must publish: {outcomes:?}"
    );

    assert!(
        indexed(fx.coverage(CONVERSATION).await).available,
        "the append must have become a rebuild; a plain append can never clear the flag, so \
         the conversation would have stayed refused forever"
    );
    let positions = fx.indexed_positions(CONVERSATION).await;
    assert_eq!(
        positions.len(),
        2,
        "the rebuild must carry both turns, not only the one that triggered it: {positions:?}"
    );
}

/// And that recovery is BOUNDED. A conversation whose rebuild fails the same
/// way every time must not buy a full replay per committed turn forever —
/// which is the livelock the design refuses elsewhere, reached by the recovery
/// path instead of the retry path.
#[tokio::test]
async fn a_doomed_recovery_rebuild_is_attempted_once() {
    let fx = Fixture::build("recover-bounded");
    // One large turn, so a rebuild from zero must read it and a forward window
    // above it need not.
    fx.commit_turn(CONVERSATION, 1, &"alpha ".repeat(2_000))
        .await;
    fx.worker().drain_once(&CancellationToken::new()).await;
    fx.projection
        .mark_unavailable(CONVERSATION, &key().key_id())
        .await
        .expect("mark unavailable");

    // A budget that a small forward window fits inside and a rebuild does not.
    let mut worker = fx.worker().with_replay_budget(4_096);

    fx.commit_turn(CONVERSATION, 2, "beta").await;
    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Unavailable {
                reason: UnavailableReason::ReplayBudgetExceeded
            }
        )],
        "the first append escalates to a rebuild, which trips the budget"
    );

    let (_, boundary) = fx.commit_turn(CONVERSATION, 3, "gamma").await;
    assert_eq!(
        worker.drain_once(&CancellationToken::new()).await,
        vec![(
            CONVERSATION.to_owned(),
            Outcome::Published {
                indexed_through: boundary
            }
        )],
        "the second append must NOT escalate again: the doomed rebuild is remembered, so this \
         pass appends and leaves the conversation refused"
    );
    assert!(
        !indexed(fx.coverage(CONVERSATION).await).available,
        "the suppressed recovery must not quietly restore availability"
    );
}

/// A sweep is the deployment-wide retry, so it is also what gives a suppressed
/// recovery another attempt.
#[tokio::test]
async fn a_reconcile_lifts_the_recovery_suppression() {
    let fx = Fixture::build("recover-sweep");
    let mut worker = fx.worker();
    worker.recovery_blocked.insert(CONVERSATION.to_owned());

    worker
        .reconcile(&CancellationToken::new())
        .await
        .expect("reconcile");

    assert!(
        worker.recovery_blocked.is_empty(),
        "a sweep must clear the suppression, or one doomed rebuild suppresses recovery for the \
         life of the process"
    );
}

/// `AppendError::Closed` has two producers and they demand opposite answers. A
/// SHUTDOWN says nothing about this conversation, so the pass defers. A
/// PANICKED partition worker answers the identical error on a host that is
/// still serving, and reading that as a shutdown froze that one conversation
/// behind an `available: true` watermark with nothing queued to move it — and
/// aborted any reconcile that reached it, so `degraded` never cleared and the
/// whole fleet refused forever over one bad worker.
#[tokio::test]
async fn a_closed_partition_on_a_live_host_refuses_rather_than_defers() {
    let fx = Fixture::build("closed-live");
    fx.commit_turn(CONVERSATION, 1, "alpha").await;
    let mut worker = fx.worker();
    worker.drain_once(&CancellationToken::new()).await;

    assert_eq!(
        worker
            .replay_failed(
                CONVERSATION,
                &JournalError::Unreachable("test".to_owned()),
                "test"
            )
            .await,
        Outcome::Unavailable {
            reason: UnavailableReason::ReplayFailed
        },
        "a journal that is unreachable while this process serves is about this conversation"
    );
    assert!(
        !indexed(fx.coverage(CONVERSATION).await).available,
        "the refusal must be recorded, not merely logged"
    );
    assert_eq!(
        fx.dirty.drain().get(CONVERSATION),
        Some(&Pending::Rebuild),
        "and the repair must be scheduled: the cause is transient"
    );

    // The same error while this process really is going away still defers.
    fx.shutdown.cancel();
    assert_eq!(
        worker
            .replay_failed(
                CONVERSATION,
                &JournalError::Unreachable("test".to_owned()),
                "test"
            )
            .await,
        Outcome::Deferred
    );
}

/// A deferred pass read nothing and published nothing, so the work is still
/// owed. Dropping the mark froze the conversation with no re-enqueue and no
/// unavailability record — behind a watermark `verified_coverage` accepts.
#[tokio::test]
async fn a_deferred_pass_puts_its_mark_back() {
    let fx = Fixture::build("deferred-remark");
    let (_, boundary) = fx.commit_turn(CONVERSATION, 1, "alpha").await;
    let source = polyc_state::revision::PartitionIncarnation::from_bytes([1; 32]);
    let journal: Arc<dyn PartitionJournal> = Arc::new(ControlledSourceJournal {
        inner: Arc::clone(&fx.eventlog),
        before: source,
        after: source,
        source_reads: AtomicUsize::new(0),
        source_error: Some("the test journal is stopping"),
        stopping: true,
    });
    let mut worker =
        SearchIndexWorker::new(fx.projection.clone(), journal, Arc::clone(&fx.dirty), key());

    let outcomes = tokio::time::timeout(
        Duration::from_secs(10),
        worker.drain_once(&CancellationToken::new()),
    )
    .await
    .expect("drain_once hung for 10s after the event log shut down");
    assert_eq!(
        outcomes,
        vec![(CONVERSATION.to_owned(), Outcome::Deferred)],
        "a shutdown failure must be deferred rather than made into a partition refusal"
    );
    assert_eq!(
        fx.dirty.drain().get(CONVERSATION),
        Some(&Pending::IndexThrough(boundary)),
        "a deferred pass must leave its work owed"
    );
}

/// Every non-shutdown read failure must not map to one retryable reason: a
/// tamper-evidence failure, an undecodable blob, and an over-cap payload are
/// all deterministic, so mapping them to "retryable" would re-queue them into
/// a two-second loop replaying up to 32 MiB of one partition forever.
///
/// Driven through the whole chain a real failure takes — the storage error, the
/// classification into a [`JournalError`], and the reason that earns — so the
/// property is checked where it is decided rather than restated on a class name.
#[test]
fn a_deterministic_replay_failure_is_never_retried() {
    for error in [
        polyc_eventlog_host::AppendError::Verify(
            polyc_eventlog_host::VerifyError::TruncatedReplay {
                expected: 9,
                actual: 3,
            },
        ),
        polyc_eventlog_host::AppendError::PayloadTooLarge {
            kind: "user_msg".to_owned(),
            len: usize::MAX,
        },
    ] {
        let reason = replay_reason(&crate::journal::classify_host_error(&error));
        assert!(
            !reason.is_transient(),
            "{error:?} is a property of the conversation's own bytes, so the identical replay \
             fails identically"
        );
    }

    assert!(
        replay_reason(&crate::journal::classify_host_error(
            &polyc_eventlog_host::AppendError::Listing(
                polyc_eventlog_host::ListPartitionsError::Storage(std::io::Error::other(
                    "transient"
                ))
            )
        ))
        .is_transient(),
        "a storage-directory read may well succeed next time"
    );

    // The other half of the same error type is NOT transient: a directory that
    // does not describe one fleet reads the same way on every attempt.
    assert!(
        !replay_reason(&crate::journal::classify_host_error(
            &polyc_eventlog_host::AppendError::Listing(
                polyc_eventlog_host::ListPartitionsError::Corrupt {
                    entry: "conv-x_data".to_owned()
                }
            )
        ))
        .is_transient(),
        "corruption is a property of what the volume holds, so a retry reports it again"
    );
}

/// The other half of the open-turn split: a conversation that really is caught
/// up must still answer `AlreadyCurrent`, or the barrier signal means nothing.
#[tokio::test]
async fn a_caught_up_conversation_is_not_reported_as_barrier_held() {
    let fx = Fixture::build("caught-up");
    let (_, boundary) = fx.commit_turn(CONVERSATION, 1, "alpha").await;
    let mut worker = fx.worker();
    worker.drain_once(&CancellationToken::new()).await;

    assert_eq!(
        worker.index_partition(CONVERSATION, Some(boundary)).await,
        Outcome::AlreadyCurrent
    );
}

/// The correctness fallback under the durable feed (#1565, chunk B6): a
/// committed turn lands, NOTHING marks it — the subscription died, the chunk
/// was lost, nobody was left to hear — and the coverage sweep still makes the
/// conversation searchable.
///
/// This is the property that lets the feed be a latency mechanism. Without a
/// sweep that runs whether or not anything reported a change, a lost mark is a
/// conversation that is never indexed again.
#[tokio::test]
async fn a_sweep_indexes_a_conversation_no_mark_ever_reached() {
    let fx = Fixture::build("sweep-unmarked");
    let mut worker = fx.worker();

    let positions = fx
        .append_unmarked(CONVERSATION, committed_turn(&turn(1), "alpha beta"))
        .await;
    let boundary = positions.last().copied().expect("positions") + 1;
    assert!(
        fx.dirty.drain().is_empty(),
        "this test is only meaningful with an empty dirty set"
    );
    assert!(
        !fx.dirty.degraded(),
        "nothing reported a problem — the sweep is the only thing that can notice"
    );

    let summary = worker
        .reconcile(&CancellationToken::new())
        .await
        .expect("reconcile");

    assert!(summary.complete);
    assert_eq!(summary.visited, 1);
    assert!(
        indexed(fx.coverage(CONVERSATION).await).indexed_through >= boundary,
        "the sweep must cover a turn no mark ever reported"
    );
}

/// The sweep's schedule: a fresh worker does not sweep at startup (the design
/// record dropped that barrier), and it sweeps once the coverage interval has
/// passed even with nothing degraded and nothing pending.
#[tokio::test]
async fn the_coverage_sweep_comes_due_on_its_own_without_a_degrade() {
    let fx = Fixture::build("sweep-schedule");
    let mut worker = fx.worker();

    assert!(
        !worker.sweep_is_due(),
        "a fresh worker must not sweep the whole deployment at startup"
    );

    worker.last_sweep = std::time::Instant::now() - COVERAGE_SWEEP_INTERVAL;
    assert!(
        worker.sweep_is_due(),
        "the coverage sweep is unconditional: it does not wait for a degrade"
    );
}