polyc-query 2026.8.3

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

use std::sync::Arc;

use super::store::{Coverage, CoverageState, ExcisionScan, SearchProjection, StoreError};
use super::terms::TermKey;
use super::*;

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

fn key_id() -> String {
    key().key_id()
}

const PARTITION: &str = "conv-web:11111111-2222-3333-4444-555555555555";

/// [`PARTITION`] as storage spells it.
///
/// Derived, never written out: the contract is that the projection holds a
/// conversation under its ENCODED name, so a literal here would pin one
/// spelling of the codec rather than the contract.
fn encoded_partition() -> String {
    polyc_eventlog_host::encode_partition(PARTITION)
        .expect("the fixture partition encodes")
        .to_string()
}

/// A projection rooted in a test-unique temp dir, removed on drop.
struct Fixture {
    projection: SearchProjection,
    dir: std::path::PathBuf,
}

impl Fixture {
    fn open(name: &str) -> Self {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-search-projection-{name}-{}",
            std::process::id()
        ));
        let _ = std::fs::remove_dir_all(&dir);
        Self {
            projection: SearchProjection::open(dir.clone()).expect("open"),
            dir,
        }
    }
}

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

fn coverage(indexed_through: u64) -> Coverage {
    Coverage {
        indexed_through,
        source_incarnation: vec![9u8; 32],
        available: true,
        excision_scanned_through: indexed_through,
    }
}

/// The coverage an indexed conversation carries, naming the state it got
/// instead when it is not indexed.
fn indexed(state: CoverageState) -> Coverage {
    match state {
        CoverageState::Indexed(coverage) => coverage,
        other => panic!("expected an indexed conversation, got {other:?}"),
    }
}

fn message(position: u64, turn: &str, text: &str) -> IndexedMessage {
    IndexedMessage {
        position,
        turn_id: turn.to_owned(),
        term_hashes: key().hash_text(text),
    }
}

/// The projection's real contract: what is appended must read back
/// identically, or a search answers from a file that no longer says what it
/// was written to say.
#[tokio::test]
async fn append_then_read_returns_the_rows_and_coverage_unchanged() {
    let fx = Fixture::open("round-trip");
    let expected = vec![
        message(3, "turn-a", "where did we decide the timeout"),
        message(9, "turn-b", "the deploy failed"),
    ];

    fx.projection
        .append(PARTITION, &expected, &coverage(42), &key_id())
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read"),
        Some(PostingsRecord { messages: expected })
    );
    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::Indexed(coverage(42))
    );
}

/// The reason segments exist: a turn appends only its delta, and the union
/// still reads as one conversation. A single rewritten file would cost O(N)
/// per turn and O(N squared) over a conversation's life.
#[tokio::test]
async fn appending_a_delta_leaves_earlier_segments_readable() {
    let fx = Fixture::open("delta");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("first");
    fx.projection
        .append(
            PARTITION,
            &[message(4, "turn-b", "beta")],
            &coverage(6),
            &key_id(),
        )
        .await
        .expect("second");

    let read = fx
        .projection
        .postings(PARTITION, &key_id())
        .await
        .expect("read")
        .expect("present");
    assert_eq!(
        read.messages.len(),
        2,
        "both segments must be visible: {:?}",
        read.messages
    );
    assert_eq!(read.messages[0].position, 1);
    assert_eq!(read.messages[1].position, 4);
    assert_eq!(
        indexed(
            fx.projection
                .coverage(PARTITION, &key_id())
                .await
                .expect("read"),
        )
        .indexed_through,
        6,
        "coverage is the newest segment's watermark"
    );
}

/// Recovery is at-least-once, so re-appending a range already indexed must
/// change nothing a reader sees.
#[tokio::test]
async fn re_appending_the_same_positions_is_idempotent() {
    let fx = Fixture::open("idempotent");
    let batch = vec![message(1, "turn-a", "alpha")];

    fx.projection
        .append(PARTITION, &batch, &coverage(3), &key_id())
        .await
        .expect("first");
    fx.projection
        .append(PARTITION, &batch, &coverage(3), &key_id())
        .await
        .expect("second");

    let read = fx
        .projection
        .postings(PARTITION, &key_id())
        .await
        .expect("read")
        .expect("present");
    assert_eq!(
        read.messages, batch,
        "a replayed range must not duplicate a position"
    );
}

/// Compaction folds segments without changing what they cover.
#[tokio::test]
async fn compaction_preserves_content_and_coverage() {
    let fx = Fixture::open("compact");
    for (position, watermark) in [(1u64, 3u64), (4, 6), (7, 9)] {
        fx.projection
            .append(
                PARTITION,
                &[message(position, "turn-a", "alpha")],
                &coverage(watermark),
                &key_id(),
            )
            .await
            .expect("append");
    }
    let before = fx
        .projection
        .postings(PARTITION, &key_id())
        .await
        .expect("read");

    fx.projection
        .compact(PARTITION, &key_id())
        .await
        .expect("compact");

    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read"),
        before,
        "compaction must not change what is readable"
    );
    assert_eq!(
        indexed(
            fx.projection
                .coverage(PARTITION, &key_id())
                .await
                .expect("read"),
        )
        .indexed_through,
        9
    );
    let files = std::fs::read_dir(
        fx.projection
            .root()
            .join(format!("conversation_id={}", encoded_partition())),
    )
    .expect("list")
    .count();
    assert_eq!(files, 1, "compaction must leave exactly one segment");
}

/// A write answers with what the directory now holds, so the caller deciding
/// whether to fold never has to keep a counter beside the store.
///
/// A counter has to be bounded, and past its bound it stops counting — which
/// silently disables the trigger it feeds for the rest of the process's life.
/// These numbers come off the listing the write already performs.
#[tokio::test]
async fn a_write_reports_the_directory_it_produced() {
    let fx = Fixture::open("write-stats");

    // A rebuild is the base: one segment, and nothing above it to merge.
    let rebuilt = fx
        .projection
        .rebuild(
            PARTITION,
            &[message(1, "turn-a", "alpha beta")],
            &coverage(2),
            &key_id(),
        )
        .await
        .expect("rebuild");
    assert_eq!(rebuilt.segments, 1);
    assert_eq!(rebuilt.rows, 2, "two distinct terms on one message");
    assert_eq!(
        rebuilt.unmerged_rows, 0,
        "a rebuild leaves only the base, so a fold has nothing to merge"
    );

    let appended = fx
        .projection
        .append(
            PARTITION,
            &[message(4, "turn-b", "gamma delta")],
            &coverage(5),
            &key_id(),
        )
        .await
        .expect("append");
    assert_eq!(appended.stats.segments, 2);
    assert_eq!(appended.stats.rows, 4);
    assert_eq!(
        appended.stats.unmerged_rows, 2,
        "only the delta sits outside the base"
    );

    // And a fold returns both quantities to the base — the property that makes
    // each compaction trigger an edge rather than a level.
    fx.projection
        .compact(PARTITION, &key_id())
        .await
        .expect("compact");
    let folded = fx
        .projection
        .segment_stats(PARTITION, &key_id())
        .await
        .expect("stats");
    assert_eq!(folded.segments, 1);
    assert_eq!(folded.rows, 4, "a fold merges rows, it never drops them");
    assert_eq!(
        folded.unmerged_rows, 0,
        "which is exactly why the trigger reads this and not the total"
    );
}

/// The decodability probe, footer-only: an older segment that will not open
/// must surface even though the newest one — the only footer `coverage` reads —
/// is perfectly healthy.
///
/// Decoding the rows to learn the same thing costs O(rows) on every committed
/// turn, which is the O(N squared) the segment layout exists to remove.
#[tokio::test]
async fn segment_stats_catches_an_older_segment_that_will_not_open() {
    let fx = Fixture::open("stats-probe");
    for (position, watermark) in [(1u64, 2u64), (4, 5)] {
        fx.projection
            .append(
                PARTITION,
                &[message(position, "turn-a", "alpha beta")],
                &coverage(watermark),
                &key_id(),
            )
            .await
            .expect("append");
    }

    let dir = fx
        .projection
        .root()
        .join(format!("conversation_id={}", encoded_partition()));
    let mut paths: Vec<_> = std::fs::read_dir(&dir)
        .expect("list")
        .filter_map(Result::ok)
        .map(|entry| entry.path())
        .collect();
    paths.sort();
    std::fs::write(&paths[0], b"not parquet").expect("break the older segment");

    assert!(
        matches!(
            fx.projection.coverage(PARTITION, &key_id()).await,
            Ok(CoverageState::Indexed(_))
        ),
        "the newest footer must still read, or this proves nothing"
    );
    assert!(
        fx.projection
            .segment_stats(PARTITION, &key_id())
            .await
            .is_err_and(|err| err.is_unreadable()),
        "a prefix nothing can open must not be advanced over"
    );
}

/// Nothing on disk is a state, not a failure — the caller asks `coverage`
/// which state it is.
#[tokio::test]
async fn segment_stats_reports_zeroes_for_a_conversation_with_no_segments() {
    let fx = Fixture::open("stats-empty");

    let stats = fx
        .projection
        .segment_stats(PARTITION, &key_id())
        .await
        .expect("stats");

    assert_eq!(stats.segments, 0);
    assert_eq!(stats.rows, 0);
    assert_eq!(stats.unmerged_rows, 0);
}

/// A rebuild discards everything: a rewrite or repair compacts journal
/// positions, so the old segments describe a prefix that no longer exists.
#[tokio::test]
async fn a_rebuild_discards_every_earlier_segment() {
    let fx = Fixture::open("rebuild");
    fx.projection
        .append(
            PARTITION,
            &[message(99, "gone", "removed")],
            &coverage(100),
            &key_id(),
        )
        .await
        .expect("append");

    let survivor = vec![message(1, "turn-a", "survivor")];
    fx.projection
        .rebuild(PARTITION, &survivor, &coverage(3), &key_id())
        .await
        .expect("rebuild");

    let read = fx
        .projection
        .postings(PARTITION, &key_id())
        .await
        .expect("read")
        .expect("present");
    assert_eq!(read.messages, survivor);
    assert!(
        !read
            .messages
            .iter()
            .any(|m| m.term_hashes.contains(&key().hash_term("removed"))),
        "nothing from the compacted-away prefix may survive"
    );
}

/// Excision removes messages, so a rebuild that SHRINKS is the invalidation
/// case this design is built around.
#[tokio::test]
async fn a_shrinking_rebuild_drops_the_excised_message() {
    let fx = Fixture::open("shrink");
    fx.projection
        .append(
            PARTITION,
            &[
                message(1, "turn-a", "kept"),
                message(2, "turn-a", "excised secret"),
            ],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");

    let shrunk = vec![message(1, "turn-a", "kept")];
    fx.projection
        .rebuild(PARTITION, &shrunk, &coverage(3), &key_id())
        .await
        .expect("rebuild");

    let read = fx
        .projection
        .postings(PARTITION, &key_id())
        .await
        .expect("read")
        .expect("present");
    assert_eq!(read.messages, shrunk);
    assert!(
        !read
            .messages
            .iter()
            .any(|m| m.term_hashes.contains(&key().hash_term("secret"))),
        "an excised term must not survive a rebuild"
    );
}

/// A conversation nothing has indexed reads as absent, never as empty
/// coverage: empty coverage would claim a conversation nothing looked at.
#[tokio::test]
async fn an_unindexed_conversation_reads_as_absent() {
    let fx = Fixture::open("absent");

    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::NeverIndexed
    );
    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read"),
        None
    );
}

/// Marking unavailable is a coverage statement, not a deletion: the newest
/// segment says unavailable while every row survives for a rebuild to reuse.
#[tokio::test]
async fn mark_unavailable_flips_coverage_and_keeps_the_rows() {
    let fx = Fixture::open("unavailable");
    let published = vec![message(1, "turn-a", "timeout")];
    fx.projection
        .append(PARTITION, &published, &coverage(11), &key_id())
        .await
        .expect("append");

    fx.projection
        .mark_unavailable(PARTITION, &key_id())
        .await
        .expect("mark");

    let read = indexed(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
    );
    assert!(!read.available, "the flag must be cleared");
    assert_eq!(read.indexed_through, 11, "the watermark survives");
    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read")
            .expect("present")
            .messages,
        published,
        "the rows survive"
    );
}

/// A conversation with no segments is already unsearchable, so marking it must
/// not fabricate one claiming a watermark nothing established.
#[tokio::test]
async fn mark_unavailable_on_an_unindexed_conversation_writes_nothing() {
    let fx = Fixture::open("mark-absent");

    fx.projection
        .mark_unavailable(PARTITION, &key_id())
        .await
        .expect("mark");

    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::NeverIndexed
    );
}

/// The migration source leaves no journal to rebuild from, so its segments go
/// rather than sitting unavailable forever — and it must leave NO trace, since
/// the conversation is alive under its new id and its coverage resolves from
/// the destination.
#[tokio::test]
async fn remove_deletes_every_segment() {
    let fx = Fixture::open("remove");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");

    fx.projection.remove(PARTITION).await.expect("remove");

    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::NeverIndexed
    );
}

/// The distinction the tombstone exists for: a destroyed conversation and one
/// nothing ever indexed must not read the same, because the first must never be
/// rebuilt and the second must be. Before this, destroy removed the directory
/// and both answered "absent".
#[tokio::test]
async fn destroyed_and_never_indexed_are_distinguishable() {
    let fx = Fixture::open("destroy-distinct");
    let never = "conv-web_99999999-9999-4999-8999-999999999999";
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");

    fx.projection
        .destroy(PARTITION, &key_id())
        .await
        .expect("destroy");

    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::Destroyed,
        "the destroyed state must be recorded, not merely absent"
    );
    assert_eq!(
        fx.projection
            .coverage(never, &key_id())
            .await
            .expect("read"),
        CoverageState::NeverIndexed
    );
    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read"),
        None,
        "the rows must be gone, not merely unavailable"
    );
}

/// Destroy is terminal. A publish that read its range before the destroy landed
/// must be refused rather than written: its segment would sort AFTER the
/// tombstone and resurrect a conversation the deployment was told to forget.
#[tokio::test]
async fn a_destroyed_conversation_refuses_every_republish() {
    let fx = Fixture::open("destroy-terminal");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");
    fx.projection
        .destroy(PARTITION, &key_id())
        .await
        .expect("destroy");

    let late = vec![message(4, "turn-b", "beta")];
    assert!(
        matches!(
            fx.projection
                .append(PARTITION, &late, &coverage(6), &key_id())
                .await,
            Err(StoreError::Destroyed)
        ),
        "an append after a destroy must be refused"
    );
    assert!(
        matches!(
            fx.projection
                .rebuild(PARTITION, &late, &coverage(6), &key_id())
                .await,
            Err(StoreError::Destroyed)
        ),
        "a rebuild after a destroy must be refused"
    );
    fx.projection
        .mark_unavailable(PARTITION, &key_id())
        .await
        .expect("marking is a no-op");

    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::Destroyed,
        "the conversation stays destroyed"
    );
    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read"),
        None
    );
}

/// A destroy of a conversation nothing indexed still records the state: destroy
/// is a statement about the conversation, not about what happened to be
/// indexed for it, and the worker must not go looking for a journal that is
/// gone.
#[tokio::test]
async fn destroying_an_unindexed_conversation_still_records_the_state() {
    let fx = Fixture::open("destroy-unindexed");

    fx.projection
        .destroy(PARTITION, &key_id())
        .await
        .expect("destroy");

    assert_eq!(
        fx.projection
            .coverage(PARTITION, &key_id())
            .await
            .expect("read"),
        CoverageState::Destroyed
    );
}

/// The tombstone holds no rows, so a term-key rotation cannot make it
/// unreadable. Reporting `Corrupt` for it would route a destroyed conversation
/// straight into the rebuild the tombstone exists to forbid.
#[tokio::test]
async fn a_tombstone_survives_a_key_rotation() {
    let fx = Fixture::open("destroy-rotated");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");
    fx.projection
        .destroy(PARTITION, &key_id())
        .await
        .expect("destroy");

    let rotated = TermKey::new([77u8; 32]).key_id();
    assert_eq!(
        fx.projection
            .coverage(PARTITION, &rotated)
            .await
            .expect("read"),
        CoverageState::Destroyed
    );
}

/// A message whose text yields no searchable term must still survive. The
/// tokenizer drops punctuation-only text and every non-ASCII term, so a
/// CJK-only message or a bare reaction lands here — and losing it would make a
/// conversation held in a non-Latin script round trip to nothing.
#[tokio::test]
async fn a_message_with_no_searchable_terms_survives_the_round_trip() {
    let fx = Fixture::open("no-terms");
    let published = vec![
        message(1, "turn-a", "\u{4f60}\u{597d}"),
        message(2, "turn-a", "timeout"),
    ];
    assert!(
        published[0].term_hashes.is_empty(),
        "fixture must actually produce a term-less message"
    );

    fx.projection
        .append(PARTITION, &published, &coverage(3), &key_id())
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .postings(PARTITION, &key_id())
            .await
            .expect("read")
            .expect("present")
            .messages,
        published,
        "a term-less message must not vanish"
    );
}

/// A file written under a different term key decodes perfectly and answers
/// every query with zero hits. That is the silent not-found the design exists
/// to prevent, so it must be refused as unreadable instead.
#[tokio::test]
async fn a_segment_written_under_another_key_is_refused() {
    let fx = Fixture::open("rotated-key");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "timeout")],
            &coverage(2),
            &key_id(),
        )
        .await
        .expect("append");

    let rotated = TermKey::new([99u8; 32]).key_id();

    assert!(
        matches!(
            fx.projection.coverage(PARTITION, &rotated).await,
            Err(StoreError::Corrupt)
        ),
        "a rotated key must be detected, not silently answered around"
    );
}

/// A corrupt footer must return `Corrupt` so the conversation is rebuilt.
/// Aborting the reading thread is not that — and an even BYTE length does not
/// make a string safe to slice.
#[test]
fn a_non_ascii_incarnation_is_refused_rather_than_panicking() {
    assert!(matches!(
        super::store::unhex_for_test("a\u{e9}b"),
        Err(StoreError::Corrupt)
    ));
    assert!(matches!(
        super::store::unhex_for_test("zz"),
        Err(StoreError::Corrupt)
    ));
    assert_eq!(
        super::store::unhex_for_test("00ff").expect("valid hex"),
        vec![0x00, 0xff]
    );
}

/// The premise of the whole layout, and the thing nothing tested: can
/// `DataFusion` actually read what this writes?
///
/// Note what a `ListingTable` needs and does NOT get for free —
/// `ListingOptions::default()` declares no partition columns, so
/// `conversation_id` is absent from the schema unless it is named. Pinning
/// that here means the search port inherits a working recipe rather than
/// rediscovering it.
#[tokio::test]
async fn datafusion_reads_the_projection_and_prunes_by_conversation() {
    use datafusion::datasource::listing::{
        ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,
    };
    use datafusion::prelude::SessionContext;

    let fx = Fixture::open("datafusion-read");
    fx.projection
        .append(
            "conv-a",
            &[message(1, "turn-a", "timeout deploy")],
            &coverage(2),
            &key_id(),
        )
        .await
        .expect("append a");
    fx.projection
        .append(
            "conv-b",
            &[message(5, "turn-b", "invoices billing")],
            &coverage(6),
            &key_id(),
        )
        .await
        .expect("append b");

    let ctx = SessionContext::new();
    let url =
        ListingTableUrl::parse(format!("file://{}/", fx.projection.root().display())).expect("url");
    let options = ListingOptions::new(Arc::new(
        datafusion::datasource::file_format::parquet::ParquetFormat::default(),
    ))
    .with_file_extension(".parquet")
    .with_table_partition_cols(vec![(
        "conversation_id".to_owned(),
        arrow::datatypes::DataType::Utf8,
    )]);
    let resolved = options
        .infer_schema(&ctx.state(), &url)
        .await
        .expect("schema");
    let config = ListingTableConfig::new(url)
        .with_listing_options(options)
        .with_schema(resolved);
    ctx.register_table(
        "search_terms",
        Arc::new(ListingTable::try_new(config).expect("table")),
    )
    .expect("register");

    let hash = key().hash_term("timeout");
    let rows = ctx
        .sql(&format!(
            "SELECT conversation_id, turn_id, position FROM search_terms \
             WHERE term_hash = {hash}"
        ))
        .await
        .expect("plan")
        .collect()
        .await
        .expect("execute");

    let total: usize = rows.iter().map(arrow::array::RecordBatch::num_rows).sum();
    assert_eq!(
        total, 1,
        "exactly the one conversation holding the term must match"
    );
}

/// A namespaced conversation is readable, and its two former twins stay apart.
///
/// Storage escapes a namespaced id to hold it as one directory, so the Hive
/// partition column carries the ENCODED name (#1748). Two things have to hold
/// at once, and only a real scan proves them together: the column value is
/// something DataFusion can actually read back and prune on, and `web:cafe`
/// and `web_cafe` land in separate directories rather than one.
///
/// Before the codec, both ids wrote to `conv-web_cafe` and this query could
/// not have told them apart at all.
#[tokio::test]
async fn datafusion_reads_a_namespaced_conversation_and_keeps_its_twin_apart() {
    use datafusion::datasource::listing::{
        ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,
    };
    use datafusion::prelude::SessionContext;

    let fx = Fixture::open("datafusion-namespaced");
    fx.projection
        .append(
            "conv-web:cafe",
            &[message(1, "turn-colon", "timeout deploy")],
            &coverage(2),
            &key_id(),
        )
        .await
        .expect("append the namespaced id");
    fx.projection
        .append(
            "conv-web_cafe",
            &[message(5, "turn-underscore", "timeout deploy")],
            &coverage(6),
            &key_id(),
        )
        .await
        .expect("append its former twin");

    // Two directories, because the two ids are two conversations.
    let dirs: Vec<String> = std::fs::read_dir(fx.projection.root())
        .expect("read the projection root")
        .flatten()
        .map(|entry| entry.file_name().to_string_lossy().into_owned())
        .filter(|name| name.starts_with("conversation_id="))
        .collect();
    assert_eq!(
        dirs.len(),
        2,
        "each conversation holds its own directory: {dirs:?}"
    );

    let ctx = SessionContext::new();
    let url =
        ListingTableUrl::parse(format!("file://{}/", fx.projection.root().display())).expect("url");
    let options = ListingOptions::new(Arc::new(
        datafusion::datasource::file_format::parquet::ParquetFormat::default(),
    ))
    .with_file_extension(".parquet")
    .with_table_partition_cols(vec![(
        "conversation_id".to_owned(),
        arrow::datatypes::DataType::Utf8,
    )]);
    let resolved = options
        .infer_schema(&ctx.state(), &url)
        .await
        .expect("schema");
    let config = ListingTableConfig::new(url)
        .with_listing_options(options)
        .with_schema(resolved);
    ctx.register_table(
        "search_terms",
        Arc::new(ListingTable::try_new(config).expect("table")),
    )
    .expect("register");

    // The column carries the ENCODED name, so that is what a predicate names.
    // A reader wanting the conversation id decodes it.
    let hash = key().hash_term("timeout");
    let rows_for = |partition: &str| {
        let encoded = polyc_eventlog_host::encode_partition(partition).expect("the name encodes");
        let sql = format!(
            "SELECT turn_id FROM search_terms \
             WHERE term_hash = {hash} AND conversation_id = '{encoded}'"
        );
        let ctx = &ctx;
        async move {
            ctx.sql(&sql)
                .await
                .expect("plan")
                .collect()
                .await
                .expect("execute")
                .iter()
                .map(arrow::array::RecordBatch::num_rows)
                .sum::<usize>()
        }
    };

    // Both conversations hold the same term. Each predicate reaches exactly its
    // own row, which is the whole point: one shared partition could not have
    // answered these two queries differently.
    assert_eq!(
        rows_for("conv-web:cafe").await,
        1,
        "the namespaced conversation holds its own row"
    );
    assert_eq!(
        rows_for("conv-web_cafe").await,
        1,
        "and its former twin holds a separate one"
    );

    let encoded = polyc_eventlog_host::encode_partition("conv-web:cafe").expect("the name encodes");
    assert_eq!(
        polyc_eventlog_host::decode_partition(encoded.as_str()).expect("the name decodes"),
        "conv-web:cafe",
        "and the id is recoverable from the column a reader scanned"
    );
}

/// The catalog boundary IS the authorization boundary, so the paths handed to
/// a session must be exactly the conversations it is authorized for.
#[tokio::test]
async fn authorized_paths_names_only_the_conversations_asked_for() {
    let fx = Fixture::open("authorized");
    for partition in ["conv-a", "conv-b", "conv-c"] {
        fx.projection
            .append(
                partition,
                &[message(1, "turn-a", "alpha")],
                &coverage(2),
                &key_id(),
            )
            .await
            .expect("append");
    }

    let paths = fx
        .projection
        .authorized_paths(vec!["conv-a".to_owned(), "conv-c".to_owned()])
        .await
        .expect("paths");

    assert_eq!(paths.len(), 2, "one segment each: {paths:?}");
    assert!(
        paths
            .iter()
            .all(|p| p.to_string_lossy().contains("conv-a")
                || p.to_string_lossy().contains("conv-c")),
        "an unauthorized conversation must never appear: {paths:?}"
    );
    assert!(
        !paths.iter().any(|p| p.to_string_lossy().contains("conv-b")),
        "conv-b was not authorized"
    );
}

/// Naming a conversation that was never indexed is not an error, and must not
/// invent a path.
#[tokio::test]
async fn authorized_paths_skips_conversations_with_no_segments() {
    let fx = Fixture::open("authorized-absent");

    assert!(
        fx.projection
            .authorized_paths(vec!["never-indexed".to_owned()])
            .await
            .expect("paths")
            .is_empty()
    );
}

#[test]
fn overlap_counts_each_distinct_query_term_once() {
    let key = key();
    let message = message(1, "turn-a", "timeout deploy");

    assert_eq!(message.overlap(&key.hash_text("timeout")), 1);
    assert_eq!(message.overlap(&key.hash_text("timeout deploy")), 2);
    assert_eq!(message.overlap(&key.hash_text("absent")), 0);
}

/// Defaults are starting points, not measured values — but they must at least
/// be usable: a zero-partition budget would refuse every search.
#[test]
fn the_default_bounds_permit_a_search_to_run() {
    let config = SearchIndexConfig::default();

    assert!(config.max_partitions_read > 0);
    assert!(config.max_term_document_frequency > 0.0);
    assert!(config.max_term_document_frequency <= 1.0);
}

/// The defect a whole class of data loss came from: the segment sequence used
/// to be a process-local counter written into a durable filename. A restart
/// rewound it to zero while the directory kept its old names, and `rename`
/// silently REPLACES a colliding destination — so a rebuild deleted the file
/// it had just written and left the conversation indistinguishable from never
/// indexed.
///
/// Reopening the projection is the restart: the sequence must come off disk.
#[tokio::test]
async fn the_segment_sequence_survives_a_restart() {
    let fx = Fixture::open("restart-sequence");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");

    // A fresh projection over the same directory — a restarted process.
    let restarted = SearchProjection::open(fx.dir.clone()).expect("reopen");
    restarted
        .append(
            PARTITION,
            &[message(4, "turn-b", "beta")],
            &coverage(6),
            &key_id(),
        )
        .await
        .expect("append after restart");

    let read = restarted
        .postings(PARTITION, &key_id())
        .await
        .expect("read")
        .expect("present");
    assert_eq!(
        read.messages.len(),
        2,
        "the pre-restart segment must survive: {:?}",
        read.messages
    );
}

/// The same defect, in the shape that destroyed data: a rebuild at the SAME
/// watermark after a restart would collide with the existing segment, replace
/// it, and then delete it.
#[tokio::test]
async fn a_rebuild_after_a_restart_does_not_erase_the_conversation() {
    let fx = Fixture::open("restart-rebuild");
    fx.projection
        .append(
            PARTITION,
            &[
                message(1, "turn-a", "kept"),
                message(2, "turn-a", "excised secret"),
            ],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");

    let restarted = SearchProjection::open(fx.dir.clone()).expect("reopen");
    let shrunk = vec![message(1, "turn-a", "kept")];
    restarted
        .rebuild(PARTITION, &shrunk, &coverage(3), &key_id())
        .await
        .expect("rebuild");

    assert_eq!(
        restarted
            .postings(PARTITION, &key_id())
            .await
            .expect("read")
            .expect("the conversation must still exist")
            .messages,
        shrunk
    );
}

/// A later segment must SUPERSEDE an earlier one at the same position, not
/// union with it. Re-indexing a position with fewer terms — a redaction, or
/// text edited until every term is non-ASCII — must actually lose the terms it
/// no longer has.
#[tokio::test]
async fn a_later_segment_replaces_an_earlier_positions_terms() {
    let fx = Fixture::open("supersede");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha secret")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("first");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(4),
            &key_id(),
        )
        .await
        .expect("second");

    let read = fx
        .projection
        .postings(PARTITION, &key_id())
        .await
        .expect("read")
        .expect("present");
    assert_eq!(read.messages.len(), 1);
    assert!(
        !read.messages[0]
            .term_hashes
            .contains(&key().hash_term("secret")),
        "a term removed by a later segment must not survive as a union"
    );
    assert!(
        read.messages[0]
            .term_hashes
            .contains(&key().hash_term("alpha"))
    );
}

/// An append must never restore availability. The design is explicit: an
/// append while unavailable stays unavailable and enqueues a rebuild. Without
/// this a routine turn landing after an excision marker made the conversation
/// searchable again, excised terms included.
#[tokio::test]
async fn an_append_after_mark_unavailable_does_not_restore_availability() {
    let fx = Fixture::open("stay-unavailable");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");
    fx.projection
        .mark_unavailable(PARTITION, &key_id())
        .await
        .expect("mark");

    // A later turn commits at a HIGHER watermark, so its segment sorts last.
    fx.projection
        .append(
            PARTITION,
            &[message(4, "turn-b", "beta")],
            &coverage(6),
            &key_id(),
        )
        .await
        .expect("append after mark");

    assert!(
        !indexed(
            fx.projection
                .coverage(PARTITION, &key_id())
                .await
                .expect("read")
        )
        .available,
        "only a rebuild may clear unavailability"
    );
}

/// A rebuild is what clears unavailability — otherwise the conversation could
/// never recover.
#[tokio::test]
async fn a_rebuild_clears_unavailability() {
    let fx = Fixture::open("rebuild-clears");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("append");
    fx.projection
        .mark_unavailable(PARTITION, &key_id())
        .await
        .expect("mark");

    fx.projection
        .rebuild(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("rebuild");

    assert!(
        indexed(
            fx.projection
                .coverage(PARTITION, &key_id())
                .await
                .expect("read")
        )
        .available
    );
}

/// A rebuild must clear segments written under a rotated key — that is the
/// state it exists to repair, and swallowing the listing error made it a
/// permanent no-op against exactly that case.
#[tokio::test]
async fn a_rebuild_repairs_a_conversation_written_under_a_rotated_key() {
    let fx = Fixture::open("rotate-repair");
    let old_key = TermKey::new([42u8; 32]);
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &old_key.key_id(),
        )
        .await
        .expect("append under the old key");

    assert!(
        matches!(
            fx.projection.coverage(PARTITION, &key_id()).await,
            Err(StoreError::Corrupt)
        ),
        "the rotated key must be detected first"
    );

    fx.projection
        .rebuild(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &coverage(3),
            &key_id(),
        )
        .await
        .expect("rebuild must be able to repair a rotated-key conversation");

    assert!(
        matches!(
            fx.projection.coverage(PARTITION, &key_id()).await,
            Ok(CoverageState::Indexed(_))
        ),
        "the conversation must be readable again"
    );
}

/// The journal side of verification, shaped exactly as the worker's will be: a
/// replay of the boundary position, hashed through the same
/// `project::incarnation_of` the write side used. One definition of the hash,
/// or the check compares two different things and always fails.
struct Journal(Vec<polyc_eventlog::Event>);

impl Journal {
    fn of(payloads: &[&str]) -> Self {
        Self(
            payloads
                .iter()
                .map(|payload| {
                    polyc_eventlog::Event::new("user_msg".to_owned(), payload.as_bytes().to_vec())
                })
                .collect(),
        )
    }

    fn positioned(&self) -> Vec<(u64, polyc_eventlog::Event)> {
        self.0
            .iter()
            .enumerate()
            .map(|(position, event)| (position as u64, event.clone()))
            .collect()
    }

    fn incarnation(&self, indexed_through: u64) -> Vec<u8> {
        super::project::incarnation_of(&self.positioned(), indexed_through)
    }
}

impl super::store::JournalState for Journal {
    async fn incarnation_at(&self, _partition: &str, indexed_through: u64) -> Option<Vec<u8>> {
        Some(self.incarnation(indexed_through))
    }

    async fn excision_since(&self, _partition: &str, _scanned_through: u64) -> ExcisionScan {
        ExcisionScan::Clear
    }
}

/// A journal that still holds an excision marker the stored footer never
/// accounted for.
struct ExcisedJournal(Journal);

impl super::store::JournalState for ExcisedJournal {
    async fn incarnation_at(&self, partition: &str, indexed_through: u64) -> Option<Vec<u8>> {
        self.0.incarnation_at(partition, indexed_through).await
    }

    async fn excision_since(&self, _partition: &str, _scanned_through: u64) -> ExcisionScan {
        ExcisionScan::Pending
    }
}

/// A journal that cannot answer the excision question at all.
struct UnknownExcisionJournal(Journal);

impl super::store::JournalState for UnknownExcisionJournal {
    async fn incarnation_at(&self, partition: &str, indexed_through: u64) -> Option<Vec<u8>> {
        self.0.incarnation_at(partition, indexed_through).await
    }

    async fn excision_since(&self, _partition: &str, _scanned_through: u64) -> ExcisionScan {
        ExcisionScan::Unknown
    }
}

/// A journal that cannot answer — the partition is gone, or the replay failed.
/// It also counts, so a test can assert it was never asked.
#[derive(Default)]
struct SilentJournal(std::sync::atomic::AtomicUsize);

impl SilentJournal {
    fn asked(&self) -> usize {
        self.0.load(std::sync::atomic::Ordering::Relaxed)
    }
}

impl super::store::JournalState for SilentJournal {
    async fn incarnation_at(&self, _partition: &str, _indexed_through: u64) -> Option<Vec<u8>> {
        self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        None
    }

    async fn excision_since(&self, _partition: &str, _scanned_through: u64) -> ExcisionScan {
        self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        ExcisionScan::Unknown
    }
}

fn covering(indexed_through: u64, source_incarnation: Vec<u8>) -> Coverage {
    Coverage {
        indexed_through,
        source_incarnation,
        available: true,
        excision_scanned_through: indexed_through,
    }
}

/// The gap the incarnation cannot close, and the reason a dropped excision
/// rebuild is recoverable at all.
///
/// A taint-excision marker is a pure APPEND: it never touches an event at or
/// below the watermark, so the boundary event hashes identically before and
/// after it and the identity check passes over a conversation whose segments
/// still hold the removed text. The only record that a rebuild was owed lived
/// in an in-memory dirty set that a restart discards — so this read had to
/// learn to ask the journal instead.
#[tokio::test]
async fn an_excision_the_index_never_applied_reads_as_stale() {
    let fx = Fixture::open("verify-excision-pending");
    let journal = Journal::of(&["one", "two", "three"]);
    let published = covering(3, journal.incarnation(3));
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &published,
            &key_id(),
        )
        .await
        .expect("append");

    // The identity side is unchanged and still agrees — that is the point.
    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &journal)
            .await
            .expect("verify"),
        CoverageState::Indexed(published)
    );
    assert_eq!(
        fx.projection
            .verified_coverage(
                PARTITION,
                &key_id(),
                &ExcisedJournal(Journal::of(&["one", "two", "three"]))
            )
            .await
            .expect("verify"),
        CoverageState::Stale,
        "a marker past the stored frontier means the segments may still hold removed text"
    );
}

/// "Cannot prove there is no excision" and "there is an excision" call for the
/// same refusal. A scan that ran out of budget, or a journal that would not
/// answer, must not be read as a clean bill of health.
#[tokio::test]
async fn an_unprovable_excision_scan_reads_as_stale() {
    let fx = Fixture::open("verify-excision-unknown");
    let journal = Journal::of(&["one", "two", "three"]);
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &covering(3, journal.incarnation(3)),
            &key_id(),
        )
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .verified_coverage(
                PARTITION,
                &key_id(),
                &UnknownExcisionJournal(Journal::of(&["one", "two", "three"]))
            )
            .await
            .expect("verify"),
        CoverageState::Stale
    );
}

/// The excision frontier is a property of the whole directory, so a later
/// append may not retract a scan an earlier one already performed. If it
/// could, an ordinary short window would reopen a question a rebuild had
/// already settled — and every read would then refuse a conversation that is
/// in fact clean.
#[tokio::test]
async fn the_excision_frontier_never_rewinds() {
    let fx = Fixture::open("frontier-monotonic");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &Coverage {
                indexed_through: 3,
                source_incarnation: vec![9u8; 32],
                available: true,
                excision_scanned_through: 99,
            },
            &key_id(),
        )
        .await
        .expect("append");
    fx.projection
        .append(
            PARTITION,
            &[message(4, "turn-b", "beta")],
            &coverage(6),
            &key_id(),
        )
        .await
        .expect("append");

    assert_eq!(
        indexed(
            fx.projection
                .coverage(PARTITION, &key_id())
                .await
                .expect("read")
        )
        .excision_scanned_through,
        99,
        "the newest segment must carry the highest frontier the directory has reached"
    );
}

/// The ordinary case: the journal still holds what was indexed, so the stored
/// coverage stands.
#[tokio::test]
async fn a_matching_incarnation_verifies() {
    let fx = Fixture::open("verify-match");
    let journal = Journal::of(&["one", "two", "three"]);
    let published = covering(3, journal.incarnation(3));
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &published,
            &key_id(),
        )
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &journal)
            .await
            .expect("verify"),
        CoverageState::Indexed(published)
    );
}

/// The hole this closes. A rewrite commits durably and the process dies before
/// its notification: the segment still decodes, still reports a watermark, and
/// still says available, while describing a journal that no longer exists.
/// Nothing computed the incarnation and nothing compared it, so nothing caught
/// this.
#[tokio::test]
async fn a_rewritten_journal_makes_the_conversation_uncovered() {
    let fx = Fixture::open("verify-rewrite");
    let indexed = Journal::of(&["one", "two", "three"]);
    let published = covering(3, indexed.incarnation(3));
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &published,
            &key_id(),
        )
        .await
        .expect("append");

    // A repair replayed, dropped an event, and re-appended the survivors, so
    // the event AT the watermark is a different one.
    let rewritten = Journal::of(&["one", "three", "four"]);

    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &rewritten)
            .await
            .expect("verify"),
        CoverageState::Stale,
        "an index describing a journal that no longer exists must not be searched"
    );
}

/// A journal that cannot answer at all — destroyed partition, failed replay —
/// is a mismatch, not a pass. Failing closed is the whole point of verifying.
#[tokio::test]
async fn a_journal_that_cannot_answer_makes_the_conversation_uncovered() {
    let fx = Fixture::open("verify-silent");
    let journal = Journal::of(&["one", "two", "three"]);
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &covering(3, journal.incarnation(3)),
            &key_id(),
        )
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &SilentJournal::default())
            .await
            .expect("verify"),
        CoverageState::Stale
    );
}

/// An empty incarnation beside a NON-ZERO watermark is already known invalid —
/// the projection writes empty only for "nothing indexed yet". It has to be
/// caught before the comparison, because the journal side also returns empty
/// for a position it cannot read: comparing the two would let an unreadable
/// journal MATCH an uninitialized footer and hand back coverage for a prefix
/// nobody has seen.
#[tokio::test]
async fn an_empty_incarnation_beside_a_watermark_fails() {
    let fx = Fixture::open("verify-empty");
    fx.projection
        .append(
            PARTITION,
            &[message(1, "turn-a", "alpha")],
            &covering(3, Vec::new()),
            &key_id(),
        )
        .await
        .expect("append");

    // The journal that cannot reach position 2 hashes to empty as well, which
    // is exactly the pairing that must NOT be read as agreement.
    let short = Journal::of(&["one"]);
    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &short)
            .await
            .expect("verify"),
        CoverageState::Stale
    );
    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &SilentJournal::default())
            .await
            .expect("verify"),
        CoverageState::Stale
    );
}

/// A zero watermark with an empty incarnation is the one honest empty pairing:
/// there is no event before position zero to name. It must verify, or a
/// conversation whose indexable prefix is empty — every turn still in flight —
/// would read as corrupt rather than as covering nothing.
#[tokio::test]
async fn a_zero_watermark_with_an_empty_incarnation_verifies() {
    let fx = Fixture::open("verify-zero");
    let published = covering(0, Vec::new());
    fx.projection
        .append(PARTITION, &[], &published, &key_id())
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &SilentJournal::default())
            .await
            .expect("verify"),
        CoverageState::Indexed(published),
        "an empty prefix names no event, so there is nothing to disagree with"
    );
}

/// A non-empty incarnation beside a zero watermark is a pairing no publish
/// produces, so one of the two is corrupt. Refuse rather than reason about
/// which.
#[tokio::test]
async fn an_incarnation_beside_a_zero_watermark_fails() {
    let fx = Fixture::open("verify-zero-hash");
    fx.projection
        .append(PARTITION, &[], &covering(0, vec![7u8; 32]), &key_id())
        .await
        .expect("append");

    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &SilentJournal::default())
            .await
            .expect("verify"),
        CoverageState::Stale
    );
}

/// Neither a never-indexed nor a destroyed conversation claims a prefix, so
/// neither has one to check — and asking the journal for a destroyed partition
/// is a replay that can only fail. Verification must pass those through
/// untouched, which is also what keeps the per-search cost at one journal read
/// per conversation that actually has coverage.
#[tokio::test]
async fn verification_never_asks_the_journal_about_a_conversation_with_no_prefix() {
    let fx = Fixture::open("verify-passthrough");
    let journal = SilentJournal::default();

    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &journal)
            .await
            .expect("verify"),
        CoverageState::NeverIndexed
    );

    fx.projection
        .destroy(PARTITION, &key_id())
        .await
        .expect("destroy");
    assert_eq!(
        fx.projection
            .verified_coverage(PARTITION, &key_id(), &journal)
            .await
            .expect("verify"),
        CoverageState::Destroyed
    );
    assert_eq!(journal.asked(), 0, "neither state has a prefix to verify");
}

/// The Container-facing wiring: one [`SearchIndex`] hands out an observer and
/// owns a worker over the SAME dirty set, so registering the one and
/// supervising the other cannot be done to different sets.
///
/// This is the end-to-end proof of the invariant the whole change exists for —
/// an observer registered without a running worker fills a set nothing drains,
/// and the read path then refuses forever.
mod wiring {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

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

    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::store::{CoverageState, SearchProjection};
    use super::super::terms::TermKey;
    use super::super::{SearchIndex, TermKeyOrigin};

    const PARTITION: &str = "conv-web_99999999-8888-7777-6666-555555555555";
    const TURN: &str = "01950000-0000-7000-8000-0000000000ff";
    const TERM_KEY: [u8; 32] = [23u8; 32];

    /// One fully committed turn carrying a single searchable text block.
    fn committed_turn(text: &str) -> Vec<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()
        };
        vec![
            Event::new(format!("{}:{TURN}", kinds::TURN_START), Vec::new()),
            Event::new(
                format!("{}:{TURN}", kinds::USER_MSG),
                message.encode_to_vec(),
            ),
            Event::new(format!("{}:{TURN}", kinds::TURN_COMPLETE), Vec::new()),
        ]
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn a_driven_marks_handle_and_a_supervised_worker_index_a_committed_turn() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-search-wiring-{}-{:?}",
            std::process::id(),
            std::thread::current().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(9).relabel_for_test(),
            )
            .expect("spawn eventlog host"),
        );

        let root = dir.join("search-index");
        let index = SearchIndex::open(
            root.clone(),
            TERM_KEY,
            TermKeyOrigin::Stored,
            crate::journal::over_host(Arc::clone(&eventlog)),
        )
        .expect("open index");
        // Exactly the Container's ordering: take the marks handle, then
        // supervise.
        let marks = index.marks();
        let worker = tokio::spawn(index.run(shutdown.clone()));

        let events = committed_turn("timeout decision");
        let positions = eventlog
            .append_batch(PARTITION.to_owned(), events.clone())
            .await
            .expect("append");
        let boundary = positions.last().copied().expect("positions") + 1;
        marks.note_commit(
            PARTITION,
            &[crate::feed::test_commit(PARTITION, &events, &positions)],
        );

        // A second, read-only handle over the same directory — the worker owns
        // the writing one.
        let reader = SearchProjection::open(root).expect("open reader");
        let key_id = TermKey::new(TERM_KEY).key_id();
        let mut indexed = None;
        for _ in 0..60 {
            if let CoverageState::Indexed(coverage) = reader
                .coverage(PARTITION, &key_id)
                .await
                .expect("read coverage")
            {
                indexed = Some(coverage);
                break;
            }
            tokio::time::sleep(Duration::from_millis(250)).await;
        }
        let coverage = indexed.expect("the supervised worker never published a segment");
        assert!(
            coverage.available,
            "a committed turn must leave the conversation searchable"
        );
        assert_eq!(
            coverage.indexed_through, boundary,
            "coverage must reach the committed turn boundary the feed marked"
        );

        shutdown.cancel();
        worker.await.expect("the worker must return on shutdown");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A minted term key orphans every segment already on disk: they decode,
    /// they report a watermark, and every query against them returns zero hits
    /// for text they demonstrably hold. Detection alone never repaired that —
    /// nothing sweeps on a key change, so a conversation that never commits
    /// another turn keeps its orphaned segments and its participants' search
    /// refuses forever. Opening degraded is what schedules the sweep.
    #[test]
    fn a_minted_term_key_opens_the_index_degraded() {
        let dir = std::env::temp_dir().join(format!(
            "polychrome-search-minted-key-{}-{}",
            std::process::id(),
            uuid::Uuid::now_v7().as_simple()
        ));
        let shutdown = CancellationToken::new();
        let eventlog = Arc::new(
            EventLogHost::spawn(
                dir.join("journal"),
                shutdown.clone(),
                ApprovalSigner::from_seed(11).relabel_for_test(),
            )
            .expect("spawn eventlog host"),
        );

        let stored = SearchIndex::open(
            dir.join("stored"),
            TERM_KEY,
            TermKeyOrigin::Stored,
            crate::journal::over_host(Arc::clone(&eventlog)),
        )
        .expect("open");
        assert!(
            !stored.dirty.degraded(),
            "a key read back from custody describes the segments already on disk, so degrading \
             on it would rebuild the whole deployment on every restart"
        );

        let minted = SearchIndex::open(
            dir.join("minted"),
            TERM_KEY,
            TermKeyOrigin::Minted,
            crate::journal::over_host(Arc::clone(&eventlog)),
        )
        .expect("open");
        assert!(
            minted.dirty.degraded(),
            "a minted key must open degraded, or the orphaned segments refuse forever with \
             nothing scheduled to replace them"
        );

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

/// The write layout the read side is being built against, pinned against the
/// file itself rather than against the builder call that requested it.
///
/// Every pruning claim this module makes depends on three physical facts: a
/// Bloom filter on `term_hash`, rows actually sorted by it, and the declared
/// `SortingColumn` naming that column's LEAF index. A `WriterProperties`
/// builder call proves none of them -- a filter can be silently dropped, and
/// `column_idx` is an ordinal into the row group's leaf columns, so a schema
/// gaining a field ahead of `term_hash` would leave the declaration pointing
/// at the wrong column while still compiling.
#[tokio::test]
async fn the_segment_carries_the_bloom_filter_and_sort_the_read_side_needs() {
    let fx = Fixture::open("write-layout");
    fx.projection
        .append(
            PARTITION,
            &[
                message(3, "turn-a", "where did we decide the timeout"),
                message(9, "turn-b", "the deploy failed again today"),
            ],
            &coverage(42),
            &key_id(),
        )
        .await
        .expect("append");

    let segment = std::fs::read_dir(
        fx.dir
            .join(format!("conversation_id={}", encoded_partition())),
    )
    .expect("the partition directory")
    .filter_map(Result::ok)
    .map(|entry| entry.path())
    .find(|path| path.extension().is_some_and(|ext| ext == "parquet"))
    .expect("a published segment");

    let reader = parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(
        std::fs::File::open(&segment).expect("open the segment"),
    )
    .expect("parse the footer");
    let metadata = reader.metadata();

    let term_hash_leaf = metadata
        .file_metadata()
        .schema_descr()
        .columns()
        .iter()
        .position(|column| column.name() == "term_hash")
        .expect("a term_hash leaf column");

    for index in 0..metadata.num_row_groups() {
        let group = metadata.row_group(index);
        assert_eq!(
            group.sorting_columns().map(Vec::as_slice),
            Some(
                [parquet::file::metadata::SortingColumn {
                    column_idx: i32::try_from(term_hash_leaf).expect("a small leaf index"),
                    descending: false,
                    nulls_first: false,
                }]
                .as_slice()
            ),
            "the declared sort must name term_hash's leaf ordinal",
        );
        for column in 0..group.num_columns() {
            let chunk = group.column(column);
            let is_term_hash = chunk.column_path().string() == "term_hash";
            assert_eq!(
                chunk.bloom_filter_offset().is_some(),
                is_term_hash,
                "a Bloom filter belongs on term_hash and nowhere else, found on {}",
                chunk.column_path(),
            );
        }
    }

    // The declaration above is only true if the rows honour it.
    let hashes: Vec<u32> = reader
        .build()
        .expect("build the reader")
        .map(|batch| batch.expect("decode a batch"))
        .flat_map(|batch| {
            batch
                .column(term_hash_leaf)
                .as_any()
                .downcast_ref::<arrow::array::UInt32Array>()
                .expect("term_hash is UInt32")
                .values()
                .to_vec()
        })
        .collect();
    assert!(
        hashes.windows(2).all(|pair| pair[0] <= pair[1]),
        "rows must be written in term_hash order: {hashes:?}",
    );
}