polyc-query 2026.8.3

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

use std::sync::Arc;

use polyc_eventlog::Event;

use super::*;
use crate::feed::test_commit;

fn turn_complete(turn: &str) -> Event {
    Event::new(format!("{}:{turn}", kinds::TURN_COMPLETE), Vec::new())
}

fn turn_start(turn: &str) -> Event {
    Event::new(format!("{}:{turn}", kinds::TURN_START), Vec::new())
}

fn user_msg(turn: &str) -> Event {
    Event::new(format!("{}:{turn}", kinds::USER_MSG), b"hello".to_vec())
}

/// Hand `marks` the feed chunk a subscription would deliver for `events`.
fn notify(marks: &CommitMarks, partition: &str, events: &[Event], positions: &[u64]) {
    marks.note_commit(partition, &[test_commit(partition, events, positions)]);
}

fn observer() -> (CommitMarks, Arc<DirtySet>) {
    let dirty = Arc::new(DirtySet::default());
    (CommitMarks::new(Arc::clone(&dirty)), dirty)
}

const UUID_A: &str = "11111111-1111-4111-8111-111111111111";
const UUID_B: &str = "22222222-2222-4222-8222-222222222222";

/// The boundary is the position AFTER a `turn_complete`: everything strictly
/// below it is committed.
#[test]
fn a_turn_complete_marks_the_partition_through_the_following_position() {
    let (observer, dirty) = observer();

    notify(
        &observer,
        "conv-a",
        &[turn_start(UUID_A), user_msg(UUID_A), turn_complete(UUID_A)],
        &[10, 11, 12],
    );

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::IndexThrough(13))
    );
}

/// The failure this guards: a turn's `turn_start` and inputs commit BEFORE the
/// harness is dialed, so an append carrying no `turn_complete` moves the
/// journal tail but not the committed tail. Marking on it would make an
/// interrupted turn searchable though the conversation never accepted it.
#[test]
fn an_append_without_a_turn_complete_marks_nothing() {
    let (observer, dirty) = observer();

    notify(
        &observer,
        "conv-a",
        &[turn_start(UUID_A), user_msg(UUID_A)],
        &[0, 1],
    );

    assert!(
        dirty.drain().is_empty(),
        "an in-flight turn must not advance the committed boundary"
    );
}

/// Within one batch, positions ascend with append order, so the highest
/// `turn_complete` is also the last — this cannot distinguish "take the max"
/// from "take the last". It pins the RESULT, and
/// `later_appends_advance_the_boundary_and_never_retreat` is what pins the max
/// across batches, where the two genuinely differ.
#[test]
fn the_last_turn_complete_in_one_batch_sets_the_boundary() {
    let (observer, dirty) = observer();

    notify(
        &observer,
        "conv-a",
        &[
            turn_complete(UUID_A),
            user_msg(UUID_B),
            turn_complete(UUID_B),
        ],
        &[4, 5, 6],
    );

    assert_eq!(dirty.drain().get("conv-a"), Some(&Pending::IndexThrough(7)));
}

#[test]
fn later_appends_advance_the_boundary_and_never_retreat() {
    let (observer, dirty) = observer();

    notify(&observer, "conv-a", &[turn_complete(UUID_A)], &[20]);
    notify(&observer, "conv-a", &[turn_complete(UUID_B)], &[5]);

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::IndexThrough(21)),
        "merging must keep the higher boundary, never the most recent"
    );
}

/// A rewrite — an excision, a repair's quarantine drop, or a migration
/// destination — renumbers positions, so the watermark a forward index would
/// resume from is meaningless.
#[test]
fn a_reported_rewrite_forces_a_rebuild() {
    let (observer, dirty) = observer();

    observer.note_partition_change("conv-a", PartitionChange::Rewritten);

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::Rebuild),
        "a rewrite must invalidate the watermark"
    );
}

/// A partition followed for the first time is marked for a full recompute.
///
/// A subscription registering against a partition starts from a snapshot taken
/// at the current head, so commits predating the registration never arrive on
/// the feed and `note_commit` never sees them. A forward index would start
/// above them; only a rebuild reaches them.
#[test]
fn a_bootstrap_marks_a_full_rebuild() {
    let dirty = std::sync::Arc::new(DirtySet::default());
    let marks = CommitMarks::new(std::sync::Arc::clone(&dirty));

    marks.note_bootstrap("conv-a");

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::Rebuild),
        "the prefix sits below any watermark a forward window would start from, so nothing short \
         of a rebuild covers it"
    );
}

/// A bootstrap mark is confined to conversation partitions, like every other
/// mark this type takes.
#[test]
fn a_bootstrap_on_a_non_conversation_partition_marks_nothing() {
    let dirty = std::sync::Arc::new(DirtySet::default());
    let marks = CommitMarks::new(std::sync::Arc::clone(&dirty));

    marks.note_bootstrap("persona-abc-mem");

    assert!(dirty.drain().is_empty());
}

/// Reporting the same change twice leaves the same pending state: an
/// invalidation is idempotent, so a caller that retried its command identity
/// never compounds the work.
#[test]
fn reporting_the_same_change_twice_is_the_same_pending_state() {
    let (observer, dirty) = observer();

    observer.note_partition_change("conv-a", PartitionChange::Destroyed);
    observer.note_partition_change("conv-a", PartitionChange::Destroyed);

    assert_eq!(dirty.drain().get("conv-a"), Some(&Pending::Destroy));
}

/// A Container whose own feed went dark says so, and the index refuses until a
/// full reconcile re-establishes coverage — the same honest answer an overflow
/// gets, for the same reason.
#[test]
fn coverage_doubt_degrades_the_set() {
    let (observer, dirty) = observer();
    assert!(!dirty.degraded());

    observer.note_coverage_doubt();

    assert!(
        dirty.degraded(),
        "a subscription that stopped means the marks are no longer the whole story"
    );
}

/// A destroyed partition and a migration source have no journal left, so their
/// records must go rather than sit unavailable forever — but they must go
/// DIFFERENTLY. A destroy records an authoritative tombstone that forbids every
/// later publish; a migration source must read as never indexed, because the
/// conversation is alive under its new id and resolves from the destination.
/// Collapsing the two either tombstones a live conversation or leaves a
/// destroyed one looking merely unindexed, so the worker cannot be handed one
/// mark for both.
#[test]
fn destructive_changes_are_told_apart() {
    for (change, expected) in [
        (PartitionChange::Destroyed, Pending::Destroy),
        (PartitionChange::MigratedAway, Pending::Remove),
    ] {
        let (observer, dirty) = observer();

        observer.note_partition_change("conv-a", change);

        assert_eq!(
            dirty.drain().get("conv-a"),
            Some(&expected),
            "{change:?} must reach the worker as {expected:?}"
        );
    }
}

/// The merge order is what stops a forward index resuming across a compaction.
#[test]
fn a_rebuild_outranks_a_forward_index_whichever_arrives_first() {
    for (first, second) in [
        (Pending::IndexThrough(9), Pending::Rebuild),
        (Pending::Rebuild, Pending::IndexThrough(9)),
    ] {
        let dirty = DirtySet::default();

        dirty.mark("conv-a", first);
        dirty.mark("conv-a", second);

        assert_eq!(
            dirty.drain().get("conv-a"),
            Some(&Pending::Rebuild),
            "a compaction must not be absorbed by a forward index"
        );
    }
}

#[test]
fn a_removal_outranks_everything_but_a_destroy() {
    for other in [Pending::IndexThrough(9), Pending::Rebuild] {
        let dirty = DirtySet::default();

        dirty.mark("conv-a", other);
        dirty.mark("conv-a", Pending::Remove);
        assert_eq!(dirty.drain().get("conv-a"), Some(&Pending::Remove));

        let dirty = DirtySet::default();
        dirty.mark("conv-a", Pending::Remove);
        dirty.mark("conv-a", other);
        assert_eq!(
            dirty.drain().get("conv-a"),
            Some(&Pending::Remove),
            "a partition whose journal moved cannot be indexed from where it no longer is"
        );
    }
}

/// The tombstone is terminal, so it has to survive a merge from either side: a
/// conversation id whose journal migrated away and was then re-created and
/// destroyed must end at the destroy, never back at a state that lets
/// something publish for it again.
#[test]
fn a_destroy_outranks_everything() {
    for other in [Pending::IndexThrough(9), Pending::Rebuild, Pending::Remove] {
        let dirty = DirtySet::default();

        dirty.mark("conv-a", other);
        dirty.mark("conv-a", Pending::Destroy);
        assert_eq!(dirty.drain().get("conv-a"), Some(&Pending::Destroy));

        let dirty = DirtySet::default();
        dirty.mark("conv-a", Pending::Destroy);
        dirty.mark("conv-a", other);
        assert_eq!(
            dirty.drain().get("conv-a"),
            Some(&Pending::Destroy),
            "a destroyed conversation cannot be rebuilt, indexed, or quietly erased"
        );
    }
}

#[test]
fn partitions_are_tracked_independently() {
    let (observer, dirty) = observer();

    notify(&observer, "conv-a", &[turn_complete(UUID_A)], &[0]);
    notify(&observer, "conv-b", &[turn_complete(UUID_B)], &[7]);

    let pending = dirty.drain();
    assert_eq!(pending.get("conv-a"), Some(&Pending::IndexThrough(1)));
    assert_eq!(pending.get("conv-b"), Some(&Pending::IndexThrough(8)));
}

#[test]
fn draining_leaves_the_set_empty() {
    let (observer, dirty) = observer();
    notify(&observer, "conv-a", &[turn_complete(UUID_A)], &[0]);

    assert_eq!(dirty.drain().len(), 1);
    assert!(dirty.drain().is_empty());
}

/// Overflow must not silently forget a partition: a forgotten one keeps
/// serving a stale watermark with nothing to disagree with it.
#[test]
fn overflow_degrades_the_index_rather_than_forgetting_a_partition() {
    let dirty = DirtySet::default();
    for i in 0..MAX_TRACKED_PARTITIONS {
        dirty.mark(&format!("conv-{i}"), Pending::IndexThrough(1));
    }
    assert!(!dirty.degraded(), "the set must hold its stated capacity");

    dirty.mark("conv-one-too-many", Pending::IndexThrough(1));

    assert!(dirty.degraded(), "overflow must be loud, not silent");
}

/// An already-tracked partition still merges once the set is full — otherwise
/// a busy fleet would degrade on partitions it is already watching.
#[test]
fn a_full_set_still_merges_a_partition_it_already_tracks() {
    let dirty = DirtySet::default();
    for i in 0..MAX_TRACKED_PARTITIONS {
        dirty.mark(&format!("conv-{i}"), Pending::IndexThrough(1));
    }

    dirty.mark("conv-0", Pending::IndexThrough(99));

    assert!(!dirty.degraded());
    assert_eq!(
        dirty.drain().get("conv-0"),
        Some(&Pending::IndexThrough(99))
    );
}

/// Draining proves the worker caught up with what it still knew about, never
/// with what overflow already discarded — so only a full reconcile may clear
/// the flag.
#[test]
fn draining_does_not_clear_the_degraded_flag() {
    let dirty = DirtySet::default();
    for i in 0..=MAX_TRACKED_PARTITIONS {
        dirty.mark(&format!("conv-{i}"), Pending::IndexThrough(1));
    }
    assert!(dirty.degraded());

    let _ = dirty.drain();
    assert!(dirty.degraded(), "a drain is not a reconcile");

    assert!(dirty.clear_degraded(dirty.degrade_count()));
    assert!(!dirty.degraded());
}

/// The race a sweep would otherwise lose. The observer runs on the event-log
/// write thread, so a mark can overflow AFTER the sweep visited that partition;
/// clearing on the flag alone would then declare the index whole over a mark
/// nobody accounted for.
#[test]
fn a_degrade_after_the_snapshot_blocks_the_clear() {
    let dirty = DirtySet::default();
    for i in 0..=MAX_TRACKED_PARTITIONS {
        dirty.mark(&format!("conv-{i}"), Pending::IndexThrough(1));
    }
    let observed = dirty.degrade_count();

    // A second overflow, standing in for one landing mid-sweep.
    dirty.mark("conv-later", Pending::IndexThrough(1));

    assert!(
        !dirty.clear_degraded(observed),
        "a sweep may only clear the degrade it actually swept for"
    );
    assert!(dirty.degraded());
    assert!(
        dirty.clear_degraded(dirty.degrade_count()),
        "a later sweep that saw the newer count may clear it"
    );
}

/// The counter counts events, not calls: it must not move on an ordinary mark,
/// or every sweep would look stale and the flag would never clear.
#[test]
fn only_a_degrade_moves_the_counter() {
    let dirty = DirtySet::default();
    let start = dirty.degrade_count();

    dirty.mark("conv-a", Pending::IndexThrough(1));
    dirty.mark("conv-a", Pending::Rebuild);
    let _ = dirty.drain();

    assert_eq!(dirty.degrade_count(), start);
}

/// A removal the store refused is carried by its re-queued mark, because the
/// sweep enumerates partitions that still EXIST and a destroyed or
/// migrated-away one is gone from that listing. Indexing work is not: the sweep
/// retries that itself.
#[test]
fn only_a_destroy_or_a_removal_counts_as_outstanding() {
    for pending in [Pending::Destroy, Pending::Remove] {
        let dirty = DirtySet::default();
        dirty.mark("conv-a", pending);
        assert!(
            dirty.has_pending_removal(),
            "{pending:?} is user text still on disk that the deployment was told to forget"
        );
    }

    for pending in [Pending::Rebuild, Pending::IndexThrough(4)] {
        let dirty = DirtySet::default();
        dirty.mark("conv-a", pending);
        assert!(
            !dirty.has_pending_removal(),
            "{pending:?} is work the sweep re-does for itself"
        );
    }
}

/// The host wraps observer callbacks in `catch_unwind`, so a panic here would
/// be swallowed and that partition's notification lost. A poisoned lock must
/// therefore recover and escalate rather than unwind.
#[test]
fn a_poisoned_lock_degrades_instead_of_panicking() {
    let dirty = Arc::new(DirtySet::default());

    let poisoner = Arc::clone(&dirty);
    let _ = std::thread::spawn(move || {
        let _guard = poisoner.inner.lock().expect("first lock");
        panic!("poison the lock");
    })
    .join();

    // Must not panic.
    dirty.mark("conv-a", Pending::IndexThrough(1));

    assert!(
        dirty.degraded(),
        "state that may have lost an update must refuse, not answer"
    );
}

/// A verified excision marker is an ORDINARY append: it fires `notify_append`
/// and never bumps `mutation_epoch`, so an invalidation scheme keyed on
/// mutations misses it entirely and keeps serving removed content — the worst
/// failure this feature can have.
///
/// It must force a REBUILD, not a forward index: the positions it names sit
/// below the current watermark, so a forward window would never contain them
/// and the postings carry-forward would restore the excised text verbatim.
#[test]
fn an_excision_append_forces_a_rebuild() {
    let (observer, dirty) = observer();

    notify(
        &observer,
        "conv-a",
        &[Event::new(
            kinds::TAINT_EXCISION.to_owned(),
            b"marker".to_vec(),
        )],
        &[7],
    );

    assert_eq!(dirty.drain().get("conv-a"), Some(&Pending::Rebuild));
}

/// An excision landing in the same batch as a completing turn must still force
/// a rebuild — the forward index it would otherwise get cannot apply it.
#[test]
fn an_excision_outranks_a_turn_completing_in_the_same_batch() {
    let (observer, dirty) = observer();

    notify(
        &observer,
        "conv-a",
        &[
            Event::new(kinds::TAINT_EXCISION.to_owned(), b"marker".to_vec()),
            turn_complete(UUID_A),
        ],
        &[7, 8],
    );

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::Rebuild),
        "a forward index cannot apply an excision naming earlier positions"
    );
}

/// `register_observer` is host-wide, so both callbacks see every partition in
/// the deployment. A partition this index never holds must not enter the set:
/// it would consume the tracking budget that overflow turns into a fleet-wide
/// refusal, and hand the worker something it cannot act on.
#[test]
fn a_non_conversation_partition_is_ignored_by_both_callbacks() {
    for partition in [
        "persona-abc-mem",
        "admin-audit",
        "skill-share-ledger",
        "query-audit",
        "conv-",
    ] {
        let (observer, dirty) = observer();

        notify(&observer, partition, &[turn_complete(UUID_A)], &[0]);
        observer.note_partition_change(partition, PartitionChange::Destroyed);

        assert!(
            dirty.drain().is_empty(),
            "{partition} is not a conversation and must never be tracked"
        );
    }
}

/// The filter must not exclude a real conversation, including the namespaced
/// form the chat edge produces once sanitized.
#[test]
fn conversation_partitions_are_still_tracked() {
    for partition in [
        "conv-01950000-0000-7000-8000-00000000aaaa",
        "conv-web_01950000-0000-7000-8000-00000000aaaa",
    ] {
        let (observer, dirty) = observer();

        notify(&observer, partition, &[turn_complete(UUID_A)], &[0]);

        assert_eq!(
            dirty.drain().get(partition),
            Some(&Pending::IndexThrough(1)),
            "{partition} is a conversation and must be tracked"
        );
    }
}

/// The production shape, end to end through the public callback rather than
/// through `DirtySet::mark`.
///
/// An excision marker is appended SOLO (`approval_grpc.rs` writes it as its own
/// one-event batch), so the same-batch test does not reflect how this actually
/// arrives. The merge ordering is tested directly elsewhere; this composes the
/// two real calls in the real order and asserts the excision still wins.
#[test]
fn a_solo_excision_after_a_completed_turn_still_forces_a_rebuild() {
    let (observer, dirty) = observer();

    notify(&observer, "conv-a", &[turn_complete(UUID_A)], &[10]);
    notify(
        &observer,
        "conv-a",
        &[Event::new(
            kinds::TAINT_EXCISION.to_owned(),
            b"marker".to_vec(),
        )],
        &[11],
    );

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::Rebuild),
        "a forward index cannot apply an excision naming earlier positions"
    );
}

/// The reverse arrival order, which the same-batch test never covers: a turn
/// completing AFTER an excision must not downgrade the pending rebuild.
#[test]
fn a_turn_completing_after_a_solo_excision_does_not_downgrade_it() {
    let (observer, dirty) = observer();

    notify(
        &observer,
        "conv-a",
        &[Event::new(
            kinds::TAINT_EXCISION.to_owned(),
            b"marker".to_vec(),
        )],
        &[10],
    );
    notify(&observer, "conv-a", &[turn_complete(UUID_A)], &[11]);

    assert_eq!(
        dirty.drain().get("conv-a"),
        Some(&Pending::Rebuild),
        "a later forward index must not absorb a pending rebuild"
    );
}