polyc-query 2026.9.0

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

use polyc_eventlog::Event;
use polyc_proto::kinds;

use super::*;

fn turn_a() -> String {
    uuid::Uuid::parse_str("01950000-0000-7000-8000-00000000aaaa")
        .expect("uuid")
        .simple()
        .to_string()
}

fn turn_b() -> String {
    uuid::Uuid::parse_str("01950000-0000-7000-8000-00000000bbbb")
        .expect("uuid")
        .simple()
        .to_string()
}

/// The turn id as a projected fact reports it. Kind tags carry the simple
/// (unhyphenated) form, but `CommittedMessageFact` normalizes to the canonical
/// hyphenated one, so the two are not interchangeable in an assertion.
fn turn_a_id() -> String {
    "01950000-0000-7000-8000-00000000aaaa".to_owned()
}

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

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(base: &str, 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!("{base}:{turn}"), message.encode_to_vec())
}

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

/// The term set a conversation's rows amount to.
///
/// There is no separate membership record — the Bloom filter over
/// these rows is it. The tests that pinned "membership is exactly the union of
/// the postings" now compute that union here, so the property is still
/// asserted rather than quietly dropped.
fn union_terms(built: &Projected) -> Vec<u32> {
    let mut terms: Vec<u32> = built
        .messages
        .iter()
        .flat_map(|m| m.term_hashes.iter().copied())
        .collect();
    terms.sort_unstable();
    terms.dedup();
    terms
}

fn project(events: &[(u64, Event)], end: u64) -> Projected {
    super::project(
        &key(),
        "conv-a",
        PartitionIncarnation::from_bytes([7; 32]),
        events,
        end,
    )
}

#[test]
fn a_committed_turn_yields_one_indexed_message_at_its_journal_position() {
    let events = committed_turn(&turn_a(), "the deploy timeout", 0);

    let built = project(&events, 3);

    assert_eq!(built.messages.len(), 1);
    let message = &built.messages[0];
    assert_eq!(message.position, 1, "the message's own journal position");
    assert_eq!(message.turn_id, turn_a_id());
    assert_eq!(message.term_hashes, key().hash_text("the deploy timeout"));
}

/// A paused turn's WITHHELD narration is never indexed (`#2702`).
///
/// A turn that pauses for a human records a marker retracting the narration it
/// already streamed, and the marker is applied at READ time — the durable
/// event carries no flag of its own. So this projection has to apply the
/// marker itself. Skipping messages that arrive already flagged is not enough:
/// nothing flags them.
///
/// The index stores no text, but a term hash is what makes a turn findable, so
/// indexing these would leave a withheld turn findable by the words it was
/// withheld for.
#[test]
fn a_withheld_turns_narration_is_not_indexed() {
    use buffa::Message as _;
    use polyc_proto::proto::polychrome::agent::v1::{Content, Message, TextContent, content};

    // Model-role narration: the only shape the marker retracts.
    let narration = |turn: &str, text: &str| {
        let message = Message {
            role: "model".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::OUTPUT_MSG),
            message.encode_to_vec(),
        )
    };

    let turn = turn_a();
    let events = vec![
        (0, marker(kinds::TURN_START, &turn)),
        (1, text_msg(kinds::USER_MSG, &turn, "run the migration")),
        (2, narration(&turn, "zarquon")),
        // The completion batch's retraction, one event ahead of the complete
        // marker, exactly as the dispatch writes it.
        (3, marker(kinds::TURN_TEXT_WITHHELD, &turn)),
        (4, marker(kinds::TURN_COMPLETE, &turn)),
    ];

    let built = project(&events, 5);

    let terms = union_terms(&built);
    for hash in key().hash_text("zarquon") {
        assert!(
            !terms.contains(&hash),
            "a withheld narration's terms reached the index, so the turn stays \
             findable by the words it was withheld for"
        );
    }
    // The person's own words are untouched — the marker retracts a status
    // claim the agent made, not what the human said.
    for hash in key().hash_text("run the migration") {
        assert!(
            terms.contains(&hash),
            "the human's own message must stay searchable, or this proves nothing"
        );
    }
}

/// The failure this exists to prevent: a turn's `turn_start` and inputs commit
/// before the harness is dialed, so an interrupted turn must never become
/// searchable.
#[test]
fn an_uncommitted_turn_is_not_indexed() {
    let events = vec![
        (0, marker(kinds::TURN_START, &turn_a())),
        (1, text_msg(kinds::USER_MSG, &turn_a(), "never committed")),
    ];

    let built = project(&events, 2);

    assert!(
        built.messages.is_empty(),
        "a turn with no turn_complete must contribute nothing"
    );
    assert!(union_terms(&built).is_empty());
}

#[test]
fn only_the_committed_turn_survives_when_another_is_still_open() {
    let mut events = committed_turn(&turn_a(), "committed text", 0);
    events.push((3, marker(kinds::TURN_START, &turn_b())));
    events.push((4, text_msg(kinds::USER_MSG, &turn_b(), "in flight text")));

    let built = project(&events, 5);

    assert_eq!(built.messages.len(), 1);
    assert_eq!(built.messages[0].turn_id, turn_a_id());
    assert_eq!(union_terms(&built), key().hash_text("committed text"));
}

/// The membership set is derived from the postings rather than accumulated
/// beside them, so the two cannot disagree about what a conversation contains.
#[test]
fn the_membership_set_is_exactly_the_union_of_the_postings() {
    let mut events = committed_turn(&turn_a(), "alpha beta", 0);
    events.extend(committed_turn(&turn_b(), "beta gamma", 3));

    let built = project(&events, 6);

    let mut expected: Vec<u32> = built
        .messages
        .iter()
        .flat_map(|m| m.term_hashes.iter().copied())
        .collect();
    expected.sort_unstable();
    expected.dedup();

    assert_eq!(union_terms(&built), expected);
    assert_eq!(
        union_terms(&built),
        key().hash_text("alpha beta gamma"),
        "a term shared by two turns is stored once"
    );
}

/// A forward pass publishes the DELTA and nothing else. The earlier messages
/// are not missing — they are in the segments this one appends after, and a
/// read unions them. Folding them back in here would rewrite the whole
/// conversation once per turn.
#[test]
fn a_forward_pass_publishes_only_the_replayed_range() {
    let events = committed_turn(&turn_b(), "beta", 3);

    let built = project(&events, 6);

    assert_eq!(
        built.messages.len(),
        1,
        "only the replayed range may be published: {:?}",
        built.messages
    );
    assert_eq!(built.messages[0].position, 4);
    assert_eq!(union_terms(&built), key().hash_text("beta"));
}

/// Recovery is at-least-once, so a range can be replayed and folded twice. What
/// makes that harmless is determinism: the same events yield the same rows at
/// the same positions, so the second publish is redundant rather than
/// conflicting. (One entry per position across segments is the store's job; see
/// `a_later_segment_replaces_an_earlier_positions_terms`.)
#[test]
fn folding_the_same_range_twice_yields_the_same_rows() {
    let events = committed_turn(&turn_a(), "alpha", 3);

    let first = project(&events, 6);
    let second = project(&events, 6);

    assert_eq!(
        first, second,
        "the fold must be a pure function of its range"
    );
    assert_eq!(first.messages.len(), 1, "and it must not duplicate entries");
}

#[test]
fn projection_carries_the_exact_source_even_at_watermark_zero() {
    let source = PartitionIncarnation::from_bytes([7; 32]);
    assert_eq!(project(&[], 0).coverage.source_incarnation, source);
    assert_eq!(
        project(&committed_turn(&turn_a(), "alpha", 0), 99)
            .coverage
            .source_incarnation,
        source
    );
}

/// A published record always claims availability; only the store's
/// `mark_unavailable` path clears it.
#[test]
fn a_projected_record_is_available_and_carries_its_watermark() {
    let built = project(&committed_turn(&turn_a(), "alpha", 0), 3);

    assert!(built.coverage.available);
    assert_eq!(built.coverage.indexed_through, 3);
}

/// The forward-window trap the review caught: `committed_message_facts` needs
/// BOTH markers in the slice it is given, so a watermark that advanced past an
/// open turn's `turn_start` would make that turn read as uncommitted forever —
/// its text permanently invisible behind a watermark claiming to cover it.
#[test]
fn the_watermark_stops_short_of_an_open_turns_start() {
    // Turn A opens at 0 and stays open. Turn B runs to completion at 2..5.
    let mut events = vec![
        (0, marker(kinds::TURN_START, &turn_a())),
        (1, text_msg(kinds::USER_MSG, &turn_a(), "still in flight")),
    ];
    events.extend(committed_turn(&turn_b(), "committed text", 2));

    let built = project(&events, 5);

    assert_eq!(
        built.coverage.indexed_through, 0,
        "the watermark must not pass turn A's start at position 0"
    );
}

/// The interleaved-turn scenario end to end: with the barrier in place, a turn
/// that completes after a later one still gets indexed.
#[test]
fn a_turn_completing_out_of_order_is_still_indexed() {
    let mut first = vec![
        (0, marker(kinds::TURN_START, &turn_a())),
        (1, text_msg(kinds::USER_MSG, &turn_a(), "alpha from turn a")),
    ];
    first.extend(committed_turn(&turn_b(), "beta from turn b", 2));

    let pass_one = project(&first, 5);
    let watermark = pass_one.coverage.indexed_through;
    assert_eq!(watermark, 0, "turn A is still open, so nothing may advance");

    // Turn A now completes. The next window starts at the barrier, which held
    // at zero, so it re-reads turn A's start along with its completion.
    let mut second = first;
    second.push((
        5,
        text_msg(kinds::OUTPUT_MSG, &turn_a(), "omega from turn a"),
    ));
    second.push((6, marker(kinds::TURN_COMPLETE, &turn_a())));

    let pass_two = project(&second, 7);

    assert_eq!(pass_two.coverage.indexed_through, 7);
    let turns: Vec<&str> = pass_two
        .messages
        .iter()
        .map(|m| m.turn_id.as_str())
        .collect();
    assert!(
        turns.contains(&turn_a_id().as_str()),
        "turn A committed and must be searchable: {turns:?}"
    );
    assert!(
        union_terms(&pass_two).contains(&key().hash_term("omega")),
        "turn A's committed text must be searchable once its turn closed"
    );
}

/// A range with no open turn keeps the caller's boundary.
#[test]
fn a_fully_committed_range_keeps_the_requested_boundary() {
    let events = committed_turn(&turn_a(), "alpha", 0);

    assert_eq!(project(&events, 3).coverage.indexed_through, 3);
}

#[test]
fn excision_stripping_preserves_the_exact_source_lineage() {
    use polyc_crypto::approval::{ApprovalSigner, EXCISION_SCOPE_SOURCE_ONLY, excision_payload};

    const CONVERSATION: &str = "web:01950000-0000-7000-8000-0000000000cc";
    const PARTITION: &str = "conv-web:01950000-0000-7000-8000-0000000000cc";

    // A real excision must be in range, or `strip_excised` never runs and the
    // assertion holds with the bug present — an earlier version of this test
    // had no marker at all and proved nothing.
    //
    // The message is excised, not the `turn_complete` marker: excising the
    // marker would make the turn read as open, hold the watermark at zero, and
    // test something else entirely.
    let mut events = committed_turn(&turn_a(), "alpha", 0);
    let (payload, _, _) = excision_payload(
        CONVERSATION,
        EXCISION_SCOPE_SOURCE_ONLY,
        &[1],
        "persona-1",
        "test excision",
        &ApprovalSigner::from_seed(1),
    );
    events.push((3, Event::new(kinds::TAINT_EXCISION.to_owned(), payload)));
    let source = PartitionIncarnation::from_bytes([8; 32]);
    let built = super::project(&key(), PARTITION, source, &events, 2);

    assert_eq!(built.coverage.source_incarnation, source);
}

/// The highest-value test in this module: an excised message's terms must be
/// absent from BOTH records.
///
/// The design calls serving excised content "the worst failure this feature can
/// have", and an architecture review found this path was broken in two
/// independent ways — the marker was matched against the wrong identifier, and
/// the append that carries it marked nothing at all. Neither had a test.
///
/// Note what the partition argument is: `project_records` matches on the
/// SANITIZED partition name, so a marker bound to conversation `web:<id>` must
/// be matched under `conv-web_<id>`. Passing the raw conversation id here would
/// make this test pass while production silently matched nothing.
mod excision {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

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

    use super::*;

    const CONVERSATION: &str = "web:01950000-0000-7000-8000-0000000000cc";
    const PARTITION: &str = "conv-web:01950000-0000-7000-8000-0000000000cc";

    fn marker(scope: &str, positions: &[u64]) -> Event {
        let (payload, _, _) = excision_payload(
            CONVERSATION,
            scope,
            positions,
            "persona-1",
            "test excision",
            &ApprovalSigner::from_seed(1),
        );
        Event::new(kinds::TAINT_EXCISION.to_owned(), payload)
    }

    /// One committed turn holding a secret, plus a verified marker naming the
    /// message's position.
    fn log_with_excision(scope: &str) -> Vec<(u64, Event)> {
        let mut events = committed_turn(&turn_a(), "the passphrase is hunter2", 0);
        events.push((3, marker(scope, &[1])));
        events
    }

    #[test]
    fn an_excised_message_leaves_no_terms_in_either_record() {
        let built = super::super::project(
            &key(),
            PARTITION,
            PartitionIncarnation::from_bytes([7; 32]),
            &log_with_excision(EXCISION_SCOPE_SOURCE_ONLY),
            4,
        );

        assert!(
            !union_terms(&built).contains(&key().hash_term("passphrase")),
            "an excised term must not survive in the membership record"
        );
        assert!(
            built
                .messages
                .iter()
                .all(|m| !m.term_hashes.contains(&key().hash_term("hunter2"))),
            "an excised term must not survive in the postings record"
        );
    }

    /// The mutation-test guard that proves the case above is not vacuous.
    ///
    /// Projecting under the bare conversation id, where the partition name
    /// belongs, admits no marker: the matcher adds the `conv-` prefix, so the
    /// two cannot be equal. The excised term therefore survives. If the case
    /// above ever passed for the wrong reason, this one would fail with it.
    #[test]
    fn matching_on_the_raw_conversation_id_would_leave_the_secret_indexed() {
        let built = super::super::project(
            &key(),
            CONVERSATION,
            PartitionIncarnation::from_bytes([7; 32]),
            &log_with_excision(EXCISION_SCOPE_SOURCE_ONLY),
            4,
        );

        assert!(
            union_terms(&built).contains(&key().hash_term("passphrase")),
            "this asserts the BUG, so the test above is proven non-vacuous: matching on the raw \
             conversation id admits no marker and the excised term survives"
        );
    }

    /// A marker bound to a different conversation must never strip anything —
    /// fail-closed in the other direction.
    #[test]
    fn a_marker_for_another_conversation_strips_nothing() {
        let built = super::super::project(
            &key(),
            "conv-web_01950000-0000-7000-8000-0000000000ff",
            PartitionIncarnation::from_bytes([7; 32]),
            &log_with_excision(EXCISION_SCOPE_SOURCE_ONLY),
            4,
        );

        assert!(
            union_terms(&built).contains(&key().hash_term("passphrase")),
            "a marker bound elsewhere must not strip this conversation"
        );
    }

    #[test]
    fn a_cascade_excision_also_leaves_no_terms() {
        let built = super::super::project(
            &key(),
            PARTITION,
            PartitionIncarnation::from_bytes([7; 32]),
            &log_with_excision(EXCISION_SCOPE_CASCADE),
            4,
        );

        assert!(!union_terms(&built).contains(&key().hash_term("passphrase")));
    }
}